X-Git-Url: http://mj.ucw.cz/gitweb/?a=blobdiff_plain;f=t%2Fmoe%2Fpipeline.py;h=1b2d98a581fbf9e24a94120e273ec27a5d520900;hb=9f4eaf4234e6a7074fca72a3a9229d75035daac8;hp=a064c619f35da7c0baa7120c1580c4f6db61795d;hpb=68ef17dd956952f2a559bf9250628aa366884521;p=moe.git diff --git a/t/moe/pipeline.py b/t/moe/pipeline.py index a064c61..1b2d98a 100644 --- a/t/moe/pipeline.py +++ b/t/moe/pipeline.py @@ -1,10 +1,7 @@ #!/usr/bin/env python -import sys import bisect -import imp import moe -import moe.log class MoePipeError(moe.MoeError): """Failure of the MoePipeline.""" @@ -14,25 +11,46 @@ class MoeAbortPipeline(Exception): def __init__(self, skip_to=999): self.skip_to = skip_to -class MoePipeline: +class Pipeline: """Moe pipeline.""" - def __init__(self, name): + def __init__(self, e, name, skip_to = 70): + # e is Eval + self.e = e self.pipe = [] self.index = -1 self.name = name + self.skip_to = skip_to + + def insert(self, pri, fun, desc='', name=None): + """Insert callable `fun` to time `pri`, + `desc` is optional description, + `name` is the "module.function" name, guessed by default. + """ + assert(isinstance(pri, int)) + assert(callable(fun)) + if name is None: + name = fun.__module__ + '.' + fun.__name__ + if desc: + desc += ' ' + desc += '[' + name + ']' - def insert(self, pri, name, fun): - triple = (pri,name,fun) + triple = (pri, desc, fun) pos = bisect.bisect(self.pipe, triple) if pos <= self.index: - raise MoePipeError, "Pipeline insert cannot alter the past" + raise MoePipeError, "Pipeline %r at time %d: Insert cannot alter the past (time %d)" \ + % (self.name, self.index, pri) self.pipe.insert(pos, triple) - def dump(self, file=sys.stdout, prefix=""): - file.write(">>> Pipeline %s\n" % self.name) - for pri,name,fun in self.pipe: - file.write("%s%03d %s\n" % (prefix,pri,name)) + def dump(self, prefix=""): + """ + Debugging dump of the pipe. + Returns a list of lines. + """ + l=["%s >>> Pipeline %s" % (prefix, self.name)] + for pri, name, fun in self.pipe: + l.append("%s% 3d %s" % (prefix, pri, name)) + return l def run(self, *args, **kwargs): self.index = 0 @@ -40,17 +58,14 @@ class MoePipeline: while self.index < len(self.pipe): (pri,name,fun) = self.pipe[self.index] if pri >= min_pri: - moe.log.default.verbose(">> Running %s:%s\n" % (self.name,name)) + self.e.log.debug("Pipeline %r:%d running: %s" % (self.name, pri, name)) try: fun(*args, **kwargs) except MoeAbortPipeline, err: - min_pri = err.skip_to + min_pri = self.skip_to else: - moe.log.default.verbose(">> Skipping %s:%s\n" % (self.name,name)) + self.e.log.debug("Pipeline %r:d skipping: %s" % (self.name, pri, name)) self.index += 1 self.index = -1 - moe.log.default.verbose(">> Pipeline %s finished\n" % self.name) + self.e.log.debug("Pipeline %r finished" % (self.name)) - def configure(self, names): - for name in names.split(): - self.add_hook(name)