1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
|
#!/usr/bin/env python
# Written by Michael Denkowski
#
# This file is part of moses. Its use is licensed under the GNU Lesser General
# Public License version 2.1 or, at your option, any later version.
"""Parallelize decoding with simulated post-editing via moses XML input.
(XML entities need to be escaped in tokenization). Memory mapped
dynamic phrase tables (Ulrich Germann,
www.statmt.org/moses/?n=Moses.AdvancedFeatures#ntoc40) and language models
(Kenneth Heafield,
http://www.statmt.org/moses/?n=FactoredTraining.BuildingLanguageModel#ntoc19)
facilitate memory efficient multi process decoding. Input is divided into
batches, each of which is decoded sequentially. Each batch pre-loads the
data from previous batches.
To use in tuning, run mert-moses.pl with --sim-pe=SYMAL where SYMAL is the
alignment from input to references. Specify the number of jobs with
--decoder-flags="-threads N".
"""
import gzip
import itertools
import math
import os
import shutil
import subprocess
import sys
import tempfile
import threading
HELP = '''Moses with simulated post-editing
Usage:
{} moses-cmd -config moses.ini -input-file text.src -ref text.tgt \
-symal text.src-tgt.symal [options] [decoder flags]
Options:
-threads N: number of decoders to run in parallel \
(default read from moses.ini, 1 if not present)
-n-best-list nbest.out N [distinct]: location and size of N-best list
-show-weights: for mert-moses.pl, just call moses and exit
-tmp: location of temp directory (default /tmp)
Other options (decoder flags) are passed through to moses-cmd\n'''
class ProgramFailure(Exception):
"""Known kind of failure, with a known presentation to the user.
Error message will be printed, and the program will return an error,
but no traceback will be shown to the user.
"""
class Progress:
"""Provides progress bar."""
def __init__(self):
self.i = 0
self.lock = threading.Lock()
def inc(self):
self.lock.acquire()
self.i += 1
if self.i % 100 == 0:
sys.stderr.write('.')
if self.i % 1000 == 0:
sys.stderr.write(' [{}]\n'.format(self.i))
sys.stderr.flush()
self.lock.release()
def done(self):
self.lock.acquire()
if self.i % 1000 != 0:
sys.stderr.write('\n')
self.lock.release()
def atomic_io(cmd, in_file, out_file, err_file, prog=None):
"""Run with atomic (synchronous) I/O."""
with open(in_file, 'r') as inp, open(out_file, 'w') as out, open(err_file, 'w') as err:
p = subprocess.Popen(
cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=err)
while True:
line = inp.readline()
if not line:
break
p.stdin.write(line)
out.write(p.stdout.readline())
out.flush()
if prog:
prog.inc()
p.stdin.close()
p.wait()
def gzopen(f):
"""Open plain or gzipped text."""
return gzip.open(f, 'rb') if f.endswith('.gz') else open(f, 'r')
def wc(f):
"""Word count."""
i = 0
for line in gzopen(f):
i += 1
return i
def write_gzfile(lines, f):
"""Write lines to gzipped file."""
out = gzip.open(f, 'wb')
for line in lines:
out.write('{}\n'.format(line))
out.close()
def main(argv):
# Defaults
moses_ini = None
moses_ini_lines = None
text_src = None
text_tgt = None
text_symal = None
text_len = None
threads_found = False
threads = 1
n_best_out = None
n_best_size = None
n_best_distinct = False
hg_ext = None
hg_dir = None
tmp_dir = '/tmp'
xml_found = False
xml_input = 'exclusive'
show_weights = False
mmsapt_dynamic = []
mmsapt_static = []
mmsapt_l1 = None
mmsapt_l2 = None
# Decoder command
cmd = argv[1:]
# Parse special options and remove from cmd
i = 1
while i < len(cmd):
if cmd[i] in ('-f', '-config'):
moses_ini = cmd[i + 1]
cmd = cmd[:i] + cmd[i + 2:]
elif cmd[i] in ('-i', '-input-file'):
text_src = cmd[i + 1]
cmd = cmd[:i] + cmd[i + 2:]
elif cmd[i] == '-ref':
text_tgt = cmd[i + 1]
cmd = cmd[:i] + cmd[i + 2:]
elif cmd[i] == '-symal':
text_symal = cmd[i + 1]
cmd = cmd[:i] + cmd[i + 2:]
elif cmd[i] in ('-th', '-threads'):
threads_found = True
threads = int(cmd[i + 1])
cmd = cmd[:i] + cmd[i + 2:]
elif cmd[i] == '-n-best-list':
n_best_out = cmd[i + 1]
n_best_size = cmd[i + 2]
# Optional "distinct"
if i + 3 < len(cmd) and cmd[i + 3] == 'distinct':
n_best_distinct = True
cmd = cmd[:i] + cmd[i + 4:]
else:
cmd = cmd[:i] + cmd[i + 3:]
elif cmd[i] == '-output-search-graph-hypergraph':
# cmd[i + 1] == true
hg_ext = cmd[i + 2]
if i + 3 < len(cmd) and cmd[i + 3][0] != '-':
hg_dir = cmd[i + 3]
cmd = cmd[:i] + cmd[i + 4:]
else:
hg_dir = 'hypergraph'
cmd = cmd[:i] + cmd[i + 3:]
elif cmd[i] == '-tmp':
tmp_dir = cmd[i + 1]
cmd = cmd[:i] + cmd[i + 2:]
# Handled specially to make sure XML input is turned on somewhere
elif cmd[i] in ('-xi', '-xml-input'):
xml_found = True
xml_input = cmd[i + 1]
cmd = cmd[:i] + cmd[i + 2:]
# Handled specially for mert-moses.pl
elif cmd[i] == '-show-weights':
show_weights = True
# Do not remove from cmd
i += 1
else:
i += 1
# Read moses.ini
if moses_ini:
moses_ini_lines = [line.strip() for line in open(moses_ini, 'r')]
i = 0
while i < len(moses_ini_lines):
# PhraseDictionaryBitextSampling name=TranslationModel0
# output-factor=0 num-features=7 path=corpus. L1=src L2=tgt
# pfwd=g pbwd=g smooth=0 sample=1000 workers=1
if moses_ini_lines[i].startswith('PhraseDictionaryBitextSampling'):
for (k, v) in (pair.split('=') for pair in moses_ini_lines[i].split()[1:]):
if k == 'name':
# Dynamic means update this model
if v.startswith('Dynamic'):
mmsapt_dynamic.append(v)
moses_ini_lines[i] += '{mmsapt_extra}'
else:
mmsapt_static.append(v)
elif k == 'L1':
if mmsapt_l1 and v != mmsapt_l1:
raise ProgramFailure(
'Error: All PhraseDictionaryBitextSampling '
'entries should have same L1: '
'{} != {}\n'.format(v, mmsapt_l1))
mmsapt_l1 = v
elif k == 'L2':
if mmsapt_l2 and v != mmsapt_l2:
raise ProgramFailure(
'Error: All PhraseDictionaryBitextSampling '
'entries should have same L2: '
'{} != {}\n'.format(v, mmsapt_l2))
mmsapt_l2 = v
# [threads]
# 8
elif moses_ini_lines[i] == '[threads]':
# Prefer command line over moses.ini
if not threads_found:
threads = int(moses_ini_lines[i + 1])
i += 1
# [xml-input]
# exclusive
elif moses_ini_lines[i] == '[xml-input]':
# Prefer command line over moses.ini
if not xml_found:
xml_found = True
xml_input = moses_ini_lines[i + 1]
i += 1
i += 1
# If mert-moses.pl passes -show-weights, just call moses
if show_weights:
# re-append original moses.ini
cmd.append('-config')
cmd.append(moses_ini)
sys.stdout.write(subprocess.check_output(cmd))
sys.stdout.flush()
sys.exit(0)
# Input length
if text_src:
text_len = wc(text_src)
# Check inputs
if not (len(cmd) > 0 and all((moses_ini, text_src, text_tgt, text_symal))):
sys.stderr.write(HELP.format(argv[0]))
sys.exit(2)
if not (os.path.isfile(cmd[0]) and os.access(cmd[0], os.X_OK)):
raise ProgramFailure(
'Error: moses-cmd "{}" is not executable\n'.format(cmd[0]))
if not mmsapt_dynamic:
raise ProgramFailure((
'Error: no PhraseDictionaryBitextSampling entries named '
'"Dynamic..." found in {}. See '
'http://www.statmt.org/moses/?n=Moses.AdvancedFeatures#ntoc40\n'
).format(moses_ini))
if wc(text_tgt) != text_len or wc(text_symal) != text_len:
raise ProgramFailure(
'Error: length mismatch between "{}", "{}", and "{}"\n'.format(
text_src, text_tgt, text_symal))
# Setup
work_dir = tempfile.mkdtemp(prefix='moses.', dir=os.path.abspath(tmp_dir))
threads = min(threads, text_len)
batch_size = int(math.ceil(float(text_len) / threads))
# Report settings
sys.stderr.write(
'Moses flags: {}\n'.format(
' '.join('\'{}\''.format(s) if ' ' in s else s for s in cmd[1:])))
for (i, n) in enumerate(mmsapt_dynamic):
sys.stderr.write(
'Dynamic mmsapt {}: {} {} {}\n'.format(
i, n, mmsapt_l1, mmsapt_l2))
for (i, n) in enumerate(mmsapt_static):
sys.stderr.write(
'Static mmsapt {}: {} {} {}\n'.format(i, n, mmsapt_l1, mmsapt_l2))
sys.stderr.write('XML mode: {}\n'.format(xml_input))
sys.stderr.write(
'Inputs: {} {} {} ({})\n'.format(
text_src, text_tgt, text_symal, text_len))
sys.stderr.write('Jobs: {}\n'.format(threads))
sys.stderr.write('Batch size: {}\n'.format(batch_size))
if n_best_out:
sys.stderr.write(
'N-best list: {} ({}{})\n'.format(
n_best_out, n_best_size,
', distinct' if n_best_distinct else ''))
if hg_dir:
sys.stderr.write('Hypergraph dir: {} ({})\n'.format(hg_dir, hg_ext))
sys.stderr.write('Temp dir: {}\n'.format(work_dir))
# Accumulate seen lines
src_lines = []
tgt_lines = []
symal_lines = []
# Current XML source file
xml_out = None
# Split into batches. Each batch after 0 gets extra files with data from
# previous batches.
# Data from previous lines in the current batch is added using XML input.
job = -1
lc = -1
lines = itertools.izip(
gzopen(text_src), gzopen(text_tgt), gzopen(text_symal))
for (src, tgt, symal) in lines:
(src, tgt, symal) = (src.strip(), tgt.strip(), symal.strip())
lc += 1
if lc % batch_size == 0:
job += 1
xml_file = os.path.join(work_dir, 'input.{}.xml'.format(job))
extra_src_file = os.path.join(
work_dir, 'extra.{}.{}.txt.gz'.format(job, mmsapt_l1))
extra_tgt_file = os.path.join(
work_dir, 'extra.{}.{}.txt.gz'.format(job, mmsapt_l2))
extra_symal_file = os.path.join(
work_dir, 'extra.{}.{}-{}.symal.gz'.format(
job, mmsapt_l1, mmsapt_l2))
if job > 0:
xml_out.close()
write_gzfile(src_lines, extra_src_file)
write_gzfile(tgt_lines, extra_tgt_file)
write_gzfile(symal_lines, extra_symal_file)
xml_out = open(xml_file, 'w')
ini_file = os.path.join(work_dir, 'moses.{}.ini'.format(job))
with open(ini_file, 'w') as moses_ini_out:
if job == 0:
extra = ''
else:
extra = ' extra={}'.format(
os.path.join(work_dir, 'extra.{}.'.format(job)))
moses_ini_out.write(
'{}\n'.format(
'\n'.join(moses_ini_lines).format(mmsapt_extra=extra)))
src_lines.append(src)
tgt_lines.append(tgt)
symal_lines.append(symal)
# Lines after first start with update tag including previous
# translation.
# Translation of last line of each batch is included in extra for
# next batch.
xml_tags = []
if lc % batch_size != 0:
tag_template = (
'<update '
'name="{}" source="{}" target="{}" alignment="{}" /> ')
for n in mmsapt_dynamic:
# Note: space after tag.
xml_tags.append(
tag_template.format(
n, src_lines[-2], tgt_lines[-2], symal_lines[-2]))
xml_out.write('{}{}\n'.format(''.join(xml_tags), src))
xml_out.close()
# Run decoders in parallel
workers = []
prog = Progress()
for i in range(threads):
work_cmd = cmd[:]
work_cmd.append('-config')
work_cmd.append(os.path.join(work_dir, 'moses.{}.ini'.format(i)))
# Workers use 1 CPU each
work_cmd.append('-threads')
work_cmd.append('1')
if not xml_found:
work_cmd.append('-xml-input')
work_cmd.append(xml_input)
if n_best_out:
work_cmd.append('-n-best-list')
work_cmd.append(os.path.join(work_dir, 'nbest.{}'.format(i)))
work_cmd.append(str(n_best_size))
if n_best_distinct:
work_cmd.append('distinct')
if hg_dir:
work_cmd.append('-output-search-graph-hypergraph')
work_cmd.append('true')
work_cmd.append(hg_ext)
work_cmd.append(os.path.join(work_dir, 'hg.{}'.format(i)))
in_file = os.path.join(work_dir, 'input.{}.xml'.format(i))
out_file = os.path.join(work_dir, 'out.{}'.format(i))
err_file = os.path.join(work_dir, 'err.{}'.format(i))
t = threading.Thread(
target=atomic_io,
args=(work_cmd, in_file, out_file, err_file, prog))
workers.append(t)
t.start()
# Wait for all to finish
for t in workers:
t.join()
prog.done()
# Gather N-best lists
if n_best_out:
with open(n_best_out, 'w') as out:
for i in range(threads):
path = os.path.join(work_dir, 'nbest.{}'.format(i))
for line in open(path, 'r'):
entry = line.partition(' ')
out.write(
'{} {}'.format(
int(entry[0]) + (i * batch_size), entry[2]))
# Gather hypergraphs
if hg_dir:
if not os.path.exists(hg_dir):
os.mkdir(hg_dir)
shutil.copy(
os.path.join(work_dir, 'hg.0', 'weights'),
os.path.join(hg_dir, 'weights'))
for i in range(threads):
for j in range(batch_size):
shutil.copy(
os.path.join(
work_dir, 'hg.{}'.format(i),
'{}.{}'.format(j, hg_ext)),
os.path.join(
hg_dir, '{}.{}'.format((i * batch_size) + j, hg_ext)))
# Gather stdout
for i in range(threads):
for line in open(os.path.join(work_dir, 'out.{}'.format(i)), 'r'):
sys.stdout.write(line)
# Cleanup
shutil.rmtree(work_dir)
if __name__ == '__main__':
try:
main(sys.argv)
except ProgramFailure as error:
sys.stderr.write("%s\n" % error)
sys.exit(1)
|