+++ /dev/null
-#!/usr/bin/env python\r
-\r
-"""\r
-Uses for generators\r
-* Pull pipelining (iterators)\r
-* Push pipelining (coroutines)\r
-* State machines (coroutines)\r
-* "Cooperative multitasking" (coroutines)\r
-* Algorithm -> Object transform for cohesiveness (for example context managers) (coroutines)\r
-\r
-Design considerations\r
-* When should a stage pass on exceptions or have it thrown within it?\r
-* When should a stage pass on GeneratorExits?\r
-* Is there a way to either turn a push generator into a iterator or to use\r
- comprehensions syntax for push generators (I doubt it)\r
-* When should the stage try and send data in both directions\r
-* Since pull generators (generators), push generators (coroutines), subroutines, and coroutines are all coroutines, maybe we should rename the push generators to not confuse them, like signals/slots? and then refer to two-way generators as coroutines\r
-** If so, make s* and co* implementation of functions\r
-"""\r
-\r
-import threading\r
-import Queue\r
-import pickle\r
-import functools\r
-import itertools\r
-import xml.sax\r
-import xml.parsers.expat\r
-\r
-\r
-def autostart(func):\r
- """\r
- >>> @autostart\r
- ... def grep_sink(pattern):\r
- ... print "Looking for %s" % pattern\r
- ... while True:\r
- ... line = yield\r
- ... if pattern in line:\r
- ... print line,\r
- >>> g = grep_sink("python")\r
- Looking for python\r
- >>> g.send("Yeah but no but yeah but no")\r
- >>> g.send("A series of tubes")\r
- >>> g.send("python generators rock!")\r
- python generators rock!\r
- >>> g.close()\r
- """\r
-\r
- @functools.wraps(func)\r
- def start(*args, **kwargs):\r
- cr = func(*args, **kwargs)\r
- cr.next()\r
- return cr\r
-\r
- return start\r
-\r
-\r
-@autostart\r
-def printer_sink(format = "%s"):\r
- """\r
- >>> pr = printer_sink("%r")\r
- >>> pr.send("Hello")\r
- 'Hello'\r
- >>> pr.send("5")\r
- '5'\r
- >>> pr.send(5)\r
- 5\r
- >>> p = printer_sink()\r
- >>> p.send("Hello")\r
- Hello\r
- >>> p.send("World")\r
- World\r
- >>> # p.throw(RuntimeError, "Goodbye")\r
- >>> # p.send("Meh")\r
- >>> # p.close()\r
- """\r
- while True:\r
- item = yield\r
- print format % (item, )\r
-\r
-\r
-@autostart\r
-def null_sink():\r
- """\r
- Good for uses like with cochain to pick up any slack\r
- """\r
- while True:\r
- item = yield\r
-\r
-\r
-def itr_source(itr, target):\r
- """\r
- >>> itr_source(xrange(2), printer_sink())\r
- 0\r
- 1\r
- """\r
- for item in itr:\r
- target.send(item)\r
-\r
-\r
-@autostart\r
-def cofilter(predicate, target):\r
- """\r
- >>> p = printer_sink()\r
- >>> cf = cofilter(None, p)\r
- >>> cf.send("")\r
- >>> cf.send("Hello")\r
- Hello\r
- >>> cf.send([])\r
- >>> cf.send([1, 2])\r
- [1, 2]\r
- >>> cf.send(False)\r
- >>> cf.send(True)\r
- True\r
- >>> cf.send(0)\r
- >>> cf.send(1)\r
- 1\r
- >>> # cf.throw(RuntimeError, "Goodbye")\r
- >>> # cf.send(False)\r
- >>> # cf.send(True)\r
- >>> # cf.close()\r
- """\r
- if predicate is None:\r
- predicate = bool\r
-\r
- while True:\r
- try:\r
- item = yield\r
- if predicate(item):\r
- target.send(item)\r
- except StandardError, e:\r
- target.throw(e.__class__, e.message)\r
-\r
-\r
-@autostart\r
-def comap(function, target):\r
- """\r
- >>> p = printer_sink()\r
- >>> cm = comap(lambda x: x+1, p)\r
- >>> cm.send(0)\r
- 1\r
- >>> cm.send(1.0)\r
- 2.0\r
- >>> cm.send(-2)\r
- -1\r
- >>> # cm.throw(RuntimeError, "Goodbye")\r
- >>> # cm.send(0)\r
- >>> # cm.send(1.0)\r
- >>> # cm.close()\r
- """\r
- while True:\r
- try:\r
- item = yield\r
- mappedItem = function(item)\r
- target.send(mappedItem)\r
- except StandardError, e:\r
- target.throw(e.__class__, e.message)\r
-\r
-\r
-def func_sink(function):\r
- return comap(function, null_sink())\r
-\r
-\r
-def expand_positional(function):\r
-\r
- @functools.wraps(function)\r
- def expander(item):\r
- return function(*item)\r
-\r
- return expander\r
-\r
-\r
-@autostart\r
-def append_sink(l):\r
- """\r
- >>> l = []\r
- >>> apps = append_sink(l)\r
- >>> apps.send(1)\r
- >>> apps.send(2)\r
- >>> apps.send(3)\r
- >>> print l\r
- [1, 2, 3]\r
- """\r
- while True:\r
- item = yield\r
- l.append(item)\r
-\r
-\r
-@autostart\r
-def last_n_sink(l, n = 1):\r
- """\r
- >>> l = []\r
- >>> lns = last_n_sink(l)\r
- >>> lns.send(1)\r
- >>> lns.send(2)\r
- >>> lns.send(3)\r
- >>> print l\r
- [3]\r
- """\r
- del l[:]\r
- while True:\r
- item = yield\r
- extraCount = len(l) - n + 1\r
- if 0 < extraCount:\r
- del l[0:extraCount]\r
- l.append(item)\r
-\r
-\r
-@autostart\r
-def coreduce(target, function, initializer = None):\r
- """\r
- >>> reduceResult = []\r
- >>> lns = last_n_sink(reduceResult)\r
- >>> cr = coreduce(lns, lambda x, y: x + y, 0)\r
- >>> cr.send(1)\r
- >>> cr.send(2)\r
- >>> cr.send(3)\r
- >>> print reduceResult\r
- [6]\r
- >>> cr = coreduce(lns, lambda x, y: x + y)\r
- >>> cr.send(1)\r
- >>> cr.send(2)\r
- >>> cr.send(3)\r
- >>> print reduceResult\r
- [6]\r
- """\r
- isFirst = True\r
- cumulativeRef = initializer\r
- while True:\r
- item = yield\r
- if isFirst and initializer is None:\r
- cumulativeRef = item\r
- else:\r
- cumulativeRef = function(cumulativeRef, item)\r
- target.send(cumulativeRef)\r
- isFirst = False\r
-\r
-\r
-@autostart\r
-def cotee(targets):\r
- """\r
- Takes a sequence of coroutines and sends the received items to all of them\r
-\r
- >>> ct = cotee((printer_sink("1 %s"), printer_sink("2 %s")))\r
- >>> ct.send("Hello")\r
- 1 Hello\r
- 2 Hello\r
- >>> ct.send("World")\r
- 1 World\r
- 2 World\r
- >>> # ct.throw(RuntimeError, "Goodbye")\r
- >>> # ct.send("Meh")\r
- >>> # ct.close()\r
- """\r
- while True:\r
- try:\r
- item = yield\r
- for target in targets:\r
- target.send(item)\r
- except StandardError, e:\r
- for target in targets:\r
- target.throw(e.__class__, e.message)\r
-\r
-\r
-class CoTee(object):\r
- """\r
- >>> ct = CoTee()\r
- >>> ct.register_sink(printer_sink("1 %s"))\r
- >>> ct.register_sink(printer_sink("2 %s"))\r
- >>> ct.stage.send("Hello")\r
- 1 Hello\r
- 2 Hello\r
- >>> ct.stage.send("World")\r
- 1 World\r
- 2 World\r
- >>> ct.register_sink(printer_sink("3 %s"))\r
- >>> ct.stage.send("Foo")\r
- 1 Foo\r
- 2 Foo\r
- 3 Foo\r
- >>> # ct.stage.throw(RuntimeError, "Goodbye")\r
- >>> # ct.stage.send("Meh")\r
- >>> # ct.stage.close()\r
- """\r
-\r
- def __init__(self):\r
- self.stage = self._stage()\r
- self._targets = []\r
-\r
- def register_sink(self, sink):\r
- self._targets.append(sink)\r
-\r
- def unregister_sink(self, sink):\r
- self._targets.remove(sink)\r
-\r
- def restart(self):\r
- self.stage = self._stage()\r
-\r
- @autostart\r
- def _stage(self):\r
- while True:\r
- try:\r
- item = yield\r
- for target in self._targets:\r
- target.send(item)\r
- except StandardError, e:\r
- for target in self._targets:\r
- target.throw(e.__class__, e.message)\r
-\r
-\r
-def _flush_queue(queue):\r
- while not queue.empty():\r
- yield queue.get()\r
-\r
-\r
-@autostart\r
-def cocount(target, start = 0):\r
- """\r
- >>> cc = cocount(printer_sink("%s"))\r
- >>> cc.send("a")\r
- 0\r
- >>> cc.send(None)\r
- 1\r
- >>> cc.send([])\r
- 2\r
- >>> cc.send(0)\r
- 3\r
- """\r
- for i in itertools.count(start):\r
- item = yield\r
- target.send(i)\r
-\r
-\r
-@autostart\r
-def coenumerate(target, start = 0):\r
- """\r
- >>> ce = coenumerate(printer_sink("%r"))\r
- >>> ce.send("a")\r
- (0, 'a')\r
- >>> ce.send(None)\r
- (1, None)\r
- >>> ce.send([])\r
- (2, [])\r
- >>> ce.send(0)\r
- (3, 0)\r
- """\r
- for i in itertools.count(start):\r
- item = yield\r
- decoratedItem = i, item\r
- target.send(decoratedItem)\r
-\r
-\r
-@autostart\r
-def corepeat(target, elem):\r
- """\r
- >>> cr = corepeat(printer_sink("%s"), "Hello World")\r
- >>> cr.send("a")\r
- Hello World\r
- >>> cr.send(None)\r
- Hello World\r
- >>> cr.send([])\r
- Hello World\r
- >>> cr.send(0)\r
- Hello World\r
- """\r
- while True:\r
- item = yield\r
- target.send(elem)\r
-\r
-\r
-@autostart\r
-def cointercept(target, elems):\r
- """\r
- >>> cr = cointercept(printer_sink("%s"), [1, 2, 3, 4])\r
- >>> cr.send("a")\r
- 1\r
- >>> cr.send(None)\r
- 2\r
- >>> cr.send([])\r
- 3\r
- >>> cr.send(0)\r
- 4\r
- >>> cr.send("Bye")\r
- Traceback (most recent call last):\r
- File "/usr/lib/python2.5/doctest.py", line 1228, in __run\r
- compileflags, 1) in test.globs\r
- File "<doctest __main__.cointercept[5]>", line 1, in <module>\r
- cr.send("Bye")\r
- StopIteration\r
- """\r
- item = yield\r
- for elem in elems:\r
- target.send(elem)\r
- item = yield\r
-\r
-\r
-@autostart\r
-def codropwhile(target, pred):\r
- """\r
- >>> cdw = codropwhile(printer_sink("%s"), lambda x: x)\r
- >>> cdw.send([0, 1, 2])\r
- >>> cdw.send(1)\r
- >>> cdw.send(True)\r
- >>> cdw.send(False)\r
- >>> cdw.send([0, 1, 2])\r
- [0, 1, 2]\r
- >>> cdw.send(1)\r
- 1\r
- >>> cdw.send(True)\r
- True\r
- """\r
- while True:\r
- item = yield\r
- if not pred(item):\r
- break\r
-\r
- while True:\r
- item = yield\r
- target.send(item)\r
-\r
-\r
-@autostart\r
-def cotakewhile(target, pred):\r
- """\r
- >>> ctw = cotakewhile(printer_sink("%s"), lambda x: x)\r
- >>> ctw.send([0, 1, 2])\r
- [0, 1, 2]\r
- >>> ctw.send(1)\r
- 1\r
- >>> ctw.send(True)\r
- True\r
- >>> ctw.send(False)\r
- >>> ctw.send([0, 1, 2])\r
- >>> ctw.send(1)\r
- >>> ctw.send(True)\r
- """\r
- while True:\r
- item = yield\r
- if not pred(item):\r
- break\r
- target.send(item)\r
-\r
- while True:\r
- item = yield\r
-\r
-\r
-@autostart\r
-def coslice(target, lower, upper):\r
- """\r
- >>> cs = coslice(printer_sink("%r"), 3, 5)\r
- >>> cs.send("0")\r
- >>> cs.send("1")\r
- >>> cs.send("2")\r
- >>> cs.send("3")\r
- '3'\r
- >>> cs.send("4")\r
- '4'\r
- >>> cs.send("5")\r
- >>> cs.send("6")\r
- """\r
- for i in xrange(lower):\r
- item = yield\r
- for i in xrange(upper - lower):\r
- item = yield\r
- target.send(item)\r
- while True:\r
- item = yield\r
-\r
-\r
-@autostart\r
-def cochain(targets):\r
- """\r
- >>> cr = cointercept(printer_sink("good %s"), [1, 2, 3, 4])\r
- >>> cc = cochain([cr, printer_sink("end %s")])\r
- >>> cc.send("a")\r
- good 1\r
- >>> cc.send(None)\r
- good 2\r
- >>> cc.send([])\r
- good 3\r
- >>> cc.send(0)\r
- good 4\r
- >>> cc.send("Bye")\r
- end Bye\r
- """\r
- behind = []\r
- for target in targets:\r
- try:\r
- while behind:\r
- item = behind.pop()\r
- target.send(item)\r
- while True:\r
- item = yield\r
- target.send(item)\r
- except StopIteration:\r
- behind.append(item)\r
-\r
-\r
-@autostart\r
-def queue_sink(queue):\r
- """\r
- >>> q = Queue.Queue()\r
- >>> qs = queue_sink(q)\r
- >>> qs.send("Hello")\r
- >>> qs.send("World")\r
- >>> qs.throw(RuntimeError, "Goodbye")\r
- >>> qs.send("Meh")\r
- >>> qs.close()\r
- >>> print [i for i in _flush_queue(q)]\r
- [(None, 'Hello'), (None, 'World'), (<type 'exceptions.RuntimeError'>, 'Goodbye'), (None, 'Meh'), (<type 'exceptions.GeneratorExit'>, None)]\r
- """\r
- while True:\r
- try:\r
- item = yield\r
- queue.put((None, item))\r
- except StandardError, e:\r
- queue.put((e.__class__, e.message))\r
- except GeneratorExit:\r
- queue.put((GeneratorExit, None))\r
- raise\r
-\r
-\r
-def decode_item(item, target):\r
- if item[0] is None:\r
- target.send(item[1])\r
- return False\r
- elif item[0] is GeneratorExit:\r
- target.close()\r
- return True\r
- else:\r
- target.throw(item[0], item[1])\r
- return False\r
-\r
-\r
-def queue_source(queue, target):\r
- """\r
- >>> q = Queue.Queue()\r
- >>> for i in [\r
- ... (None, 'Hello'),\r
- ... (None, 'World'),\r
- ... (GeneratorExit, None),\r
- ... ]:\r
- ... q.put(i)\r
- >>> qs = queue_source(q, printer_sink())\r
- Hello\r
- World\r
- """\r
- isDone = False\r
- while not isDone:\r
- item = queue.get()\r
- isDone = decode_item(item, target)\r
-\r
-\r
-def threaded_stage(target, thread_factory = threading.Thread):\r
- messages = Queue.Queue()\r
-\r
- run_source = functools.partial(queue_source, messages, target)\r
- thread_factory(target=run_source).start()\r
-\r
- # Sink running in current thread\r
- return functools.partial(queue_sink, messages)\r
-\r
-\r
-@autostart\r
-def pickle_sink(f):\r
- while True:\r
- try:\r
- item = yield\r
- pickle.dump((None, item), f)\r
- except StandardError, e:\r
- pickle.dump((e.__class__, e.message), f)\r
- except GeneratorExit:\r
- pickle.dump((GeneratorExit, ), f)\r
- raise\r
- except StopIteration:\r
- f.close()\r
- return\r
-\r
-\r
-def pickle_source(f, target):\r
- try:\r
- isDone = False\r
- while not isDone:\r
- item = pickle.load(f)\r
- isDone = decode_item(item, target)\r
- except EOFError:\r
- target.close()\r
-\r
-\r
-class EventHandler(object, xml.sax.ContentHandler):\r
-\r
- START = "start"\r
- TEXT = "text"\r
- END = "end"\r
-\r
- def __init__(self, target):\r
- object.__init__(self)\r
- xml.sax.ContentHandler.__init__(self)\r
- self._target = target\r
-\r
- def startElement(self, name, attrs):\r
- self._target.send((self.START, (name, attrs._attrs)))\r
-\r
- def characters(self, text):\r
- self._target.send((self.TEXT, text))\r
-\r
- def endElement(self, name):\r
- self._target.send((self.END, name))\r
-\r
-\r
-def expat_parse(f, target):\r
- parser = xml.parsers.expat.ParserCreate()\r
- parser.buffer_size = 65536\r
- parser.buffer_text = True\r
- parser.returns_unicode = False\r
- parser.StartElementHandler = lambda name, attrs: target.send(('start', (name, attrs)))\r
- parser.EndElementHandler = lambda name: target.send(('end', name))\r
- parser.CharacterDataHandler = lambda data: target.send(('text', data))\r
- parser.ParseFile(f)\r
-\r
-\r
-if __name__ == "__main__":\r
- import doctest\r
- doctest.testmod()\r
projects / gonvert / blobdiff