コードの何が問題だったのかを理解するよりも、これを書き直す方が簡単でした。
import os.path
import glob
import re
import itertools
from collections import namedtuple, deque
from operator import attrgetter
R_PREFIX_VALUE = re.compile(r'^(?P<prefix>[A-Z]+)(?P<suffix>\d+)\s+(?P<value>\d+)\s*$')
getvalue = attrgetter('value')
def interleave(seq, val):
return itertools.chain.from_iterable(itertools.izip(seq, itertools.repeat(val)))
class Fileline(namedtuple('Fileline', 'filename prefix suffix value')):
@classmethod
def _fromstr(cls, s, filename=None, rematch=R_PREFIX_VALUE.match):
m = rematch(s)
if not m:
raise ValueError('No valid line found in %r' % s)
d = m.groupdict()
d['value'] = int(d['value'])
d['filename'] = filename
return cls(**d)
def _asstr(self):
return '{}{} {}'.format(self.prefix, self.suffix, self.value)
def max_value_with_prefix(lineseq, prefix, getvalue=getvalue):
withprefix = (line for line in lineseq if line.prefix==prefix)
return max_value(withprefix)
def filter_lt_line(lineseq, maxline):
for line in lineseq:
if line.prefix != maxline.prefix or line.value >= maxline.value:
yield line
def extreme_value(fn, lineseq, getvalue=getvalue):
try:
return fn((l for l in lineseq if l is not None), key=getvalue)
except ValueError:
return None
def max_value(lineseq):
return extreme_value(max, lineseq)
def min_value(lineseq):
return extreme_value(min, lineseq)
def read_lines(fn, maker=Fileline._fromstr):
with open(fn, 'rb') as f:
return deque(maker(l, fn) for l in f)
def write_file(fn, lineseq):
lines = (l._asstr() for l in lineseq)
newlines = interleave(lines, '\n')
with open(fn, 'wb') as f:
f.writelines(newlines)
def write_output_file(fn, lineseq):
lines = ("{} {}".format(l.filename, l.value) for l in lineseq)
newlines = interleave(lines, "\n")
with open(fn, 'wb') as f:
f.writelines(newlines)
def filter_max_returning_min(fn, prefix):
lineseq = read_lines(fn)
maxvalue = max_value_with_prefix(lineseq, prefix)
filteredlineseq = deque(filter_lt_line(lineseq, maxvalue))
write_file(fn, filteredlineseq)
minline = min_value(filteredlineseq)
return minline
def main(fileglob, prefix, outputfile):
minlines = []
for fn in glob.iglob(fileglob):
minlines.append(filter_max_returning_min(fn, prefix))
write_output_file(outputfile, minlines)
エントリポイントはmain()、のように呼ばれるですmain('txtdir', 'ENSG', 'output.txt')。ファイルごとにファイルfilter_max_returning_min()を開いて書き換え、最小値を返します。アクセスしたすべてのファイルのすべての行の辞書またはリストを保持する必要はありません。
（ところで、ファイルを破壊的に上書きするのは悪い考えのようです！他の場所にコピーすることを検討しましたか？）
個別の関心事を個別の機能に分離すると、さまざまな実行動作のためにそれらを再構成することが非常に簡単になります。たとえば、次の2つの小さな関数を追加して、すべてのファイルでこのタスクを並行して実行するのは簡単です。
def _worker(args):
return filter_max_returning_min(*args)
def multi_main(fileglob, prefix, outputfile, processes):
from multiprocessing import Pool
pool = Pool(processes=processes)
workerargs = ((fn, prefix) for fn in glob.iglob(fileglob))
minlines = pool.imap_unordered(_worker, workerargs, processes)
write_file(outputfile, minlines)
これで、構成可能な数のワーカーを起動できます。各ワーカーは1つのファイルで動作し、完了時に最小値を収集します。非常に大きなファイルまたは多数のファイルがあり、IOバウンドではない場合、これはより高速になる可能性があります。
楽しみのために、これをCLIユーティリティに簡単に変換することもできます。
def _argparse():
import argparse
def positive_int(s):
v = int(s)
if v < 1:
raise argparse.ArgumentTypeError('{:r} must be a positive integer'.format(s))
return v
parser = argparse.ArgumentParser(
formatter_class=argparse.RawDescriptionHelpFormatter,
description="""Filter text files and write min value.
Performs these operations on the text files in supplied `filedir`:
1. In each file, identify lines starting with the matching `maxprefix`
which do *not* contain the maximum value for that prefix in that file.
2. DESTRUCTIVELY REWRITE each file with lines found in step 1 removed!
3. Write the minimum value (for all lines in all files) to `outputfile`.
""")
parser.add_argument('filedir',
help="Directory containg the text files to process. WILL REWRITE FILES!")
parser.add_argument('maxprefix', nargs="?", default="ENSG",
help="Line prefix which should have values less than max value removed in each file")
parser.add_argument('outputfile', nargs="?", default="output.txt",
help="File in which to write min value found. WILL REWRITE FILES!")
parser.add_argument('-p', '--parallel', metavar="N", nargs="?", type=positive_int, const=10,
help="Process files in parallel, with N workers. Default is to process a file at a time.")
return parser.parse_args()
if __name__ == '__main__':
args = _argparse()
fileglob = os.path.join(args.filedir, '*.txt')
prefix = args.maxprefix
outputfile = args.outputfile
if args.parallel:
multi_main(fileglob, prefix, outputfile, args.parallel)
else:
main(fileglob, prefix, outputfile)
これで、コマンドラインから呼び出すことができます。
$ python ENSG.py txtdir ENSCAFG --parallel=4