X-Git-Url: http://mj.ucw.cz/gitweb/?a=blobdiff_plain;f=t%2Fmoe%2Fpipeline.py;h=bffca80ffc408e4a900a9c1835b3fe504bdfc3d9;hb=a249fb797c925052e3ad692d887eb2004d66c455;hp=30c4c587e4417bb71c0d0114b1a1a7c65951729f;hpb=8da1e0880f38cdd74fa1b73339860a46f3fa7227;p=moe.git diff --git a/t/moe/pipeline.py b/t/moe/pipeline.py index 30c4c58..bffca80 100644 --- a/t/moe/pipeline.py +++ b/t/moe/pipeline.py @@ -1,57 +1,61 @@ #!/usr/bin/env python -import sys import bisect -import imp -import moe.log +import moe -class MoePipeError(Exception): +class MoePipeError(moe.MoeError): """Failure of the MoePipeline.""" -class MoePipeline: +class MoeAbortPipeline(Exception): + + def __init__(self, skip_to=999): + self.skip_to = skip_to + +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, name, fun): - triple = (pri,name,fun) + "Insert callable `fun` to time `pri`, `name` is only informative." + assert(isinstance(pri, int)) + assert(callable(fun)) + triple = (pri, name, 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): - file.write(">>> Pipeline %s\n" % self.name) - for pri,name,fun in self.pipe: - file.write("%03d %s\n" % (pri,name)) + def dump(self, prefix=""): + """ + Debugging dump of the pipe. + Returns a list of lines. + """ + l=["%s >>> Pipeline %s\n" % (prefix, self.name)] + for pri, name, fun in self.pipe: + l.append("%s% 3d %s, %s\n" % (prefix, pri, name, fun)) - def run(self, *args): + def run(self, *args, **kwargs): self.index = 0 + min_pri = -1 while self.index < len(self.pipe): (pri,name,fun) = self.pipe[self.index] - moe.log.verbose(">> Running %s:%s\n" % (self.name,name)) - fun(*args) + if pri >= min_pri: + self.e.log.debug("Pipeline %r:%d running: %s\n" % (self.name, pri, name)) + try: + fun(*args, **kwargs) + except MoeAbortPipeline, err: + min_pri = self.skip_to + else: + self.e.log.debug("Pipeline %r:d skipping: %s\n" % (self.name, pri, name)) self.index += 1 self.index = -1 + self.e.log.debug("Pipeline %r finished\n" % (self.name)) - def add_hook(self, name): - modname = "moe.hooks." + name - moe.log.verbose(">> Loading hook %s\n" % name) - if not sys.modules.has_key(modname): - ## FIXME: Configuration variable for the hook directory? - try: - fp, path, desc = imp.find_module(name, ["moe/hooks"]) - except ImportError: - raise MoePipeError, "Cannot find hook module " + modname - try: - imp.load_module(modname, fp, path, desc) - finally: - fp.close() - sys.modules[modname].init(self) - - def configure(self, names): - for name in names.split(): - self.add_hook(name)