Coroutines & Generators.ipynb

{
 "metadata": {
  "name": ""
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "Example of a logging tool that uses coroutines and decorators"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import logging\n",
      "import sys\n",
      "import time"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 1
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def gen_timestamp():\n",
      "    ''' A timestamp generator '''\n",
      "    ts_format = \"[%Y-%m-%d %H:%M:%S]\"\n",
      "    while True:\n",
      "        yield time.strftime(ts_format, time.localtime())"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 2
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def coroutine(func):\n",
      "    ''' Coroutine wrapper to 'hide' the initial next() call '''\n",
      "    def start(*arg, **kwarg):\n",
      "        cr = func(*arg, **kwarg)\n",
      "        next(cr)\n",
      "        return cr\n",
      "    return start"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 3
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "@coroutine\n",
      "def logger(filename=None):\n",
      "    ''' Does some setup then returns the correct logger func '''\n",
      "    opts = {\n",
      "        'msgfmt'   : \"{T} {L}: {M}\\n\",\n",
      "        'levels'   : [\"info\", \"debug\", \"warn\"],\n",
      "        'ts'       : gen_timestamp(),\n",
      "        'filename' : filename,\n",
      "    }\n",
      "    if filename:\n",
      "        return logger_file(opts)\n",
      "    return logger_stdout(opts)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 4
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def logsender(func):\n",
      "    ''' Decorator to 'hide' the tuple-ization required for send calls\n",
      "        * Doesn't actually work.\n",
      "    '''\n",
      "    def inner(*arg, **kwarg):\n",
      "        lf = func(*arg, **kwarg)\n",
      "        old_send = lf.send\n",
      "        lf.tsend = lambda: old_send(tuple(*arg))\n",
      "        return lf\n",
      "    return inner "
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 5
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def logger_stdout(opts):\n",
      "    ''' Logger to stdout '''\n",
      "    while True:\n",
      "        message, level = (yield)\n",
      "        if level in opts['levels']:\n",
      "            sys.stdout.write(opts['msgfmt'].format(T=next(opts['ts']), L=level, M=message))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 6
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def logger_file(opts):\n",
      "    ''' Filesystem logger with context manager '''\n",
      "    with open(opts['filename'], \"a\") as fd:\n",
      "        fd.seek(0, 2)\n",
      "        while True:\n",
      "            message, level = (yield)\n",
      "            if level in opts['levels']:\n",
      "                fd.writelines(opts['msgfmt'].format(T=next(opts['ts']), L=level, M=message))\n",
      "                fd.flush()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 7
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def logger_foo(logger=None):\n",
      "    log_targets = {\n",
      "        'info': logger.info,\n",
      "        'debug': logger.debug,\n",
      "        'error': logger.error,\n",
      "    }\n",
      "    while True:\n",
      "        message, level = (yield)\n",
      "        if level in log_targets:\n",
      "            log_targets[level](message)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 8
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "class FooThing(object):\n",
      "    def __init__(self, logger=None):\n",
      "        self.logger = logger\n",
      "\n",
      "    def log(self, message, level='info'):\n",
      "        if self.logger:\n",
      "            log_targets = {'info': self.logger.info,\n",
      "                           'debug': self.logger.debug,\n",
      "                           'error': self.logger.error}\n",
      "            if level in log_targets:\n",
      "                log_targets[level](message)\n",
      "        else:\n",
      "            print(message)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 9
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def set_logger(opts):\n",
      "    log_file = '/tmp/FooThing_%Y-%m-%d_%H_%M_%S.log'\n",
      "    #if opts.log:\n",
      "    #    file_, ext = os.path.splitext(opts.log)\n",
      "    #    file_ += '_%Y-%m-%d_%H_%M_%S'\n",
      "    #    log_file = file_ + ext\n",
      "    log_file = time.strftime(log_file, time.localtime())\n",
      "    logger = logging.getLogger('isi_km_stress_log')\n",
      "    logger.setLevel(logging.DEBUG)\n",
      "    loggerfh = logging.FileHandler(log_file)\n",
      "    loggerfh.setLevel(logging.INFO)\n",
      "    loggerch = logging.StreamHandler()\n",
      "    loggerch.setLevel(logging.ERROR)\n",
      "    loggerfh.setFormatter(logging.Formatter(\n",
      "        '%(asctime)s - %(levelname)s - %(message)s'))\n",
      "    logger.addHandler(loggerfh)\n",
      "    logger.addHandler(loggerch)\n",
      "    logger.info('Starting isi_km_d Signal Stress Test Run')\n",
      "    logger.info('Run Date: %s' % (time.strftime('%Y-%m-%d %H:%M:%S')))\n",
      "    #logger.info('Build Version: %s' % (misc.version()))\n",
      "    logger.info('=' * 80)\n",
      "    logger.info('')\n",
      "    return logger"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 10
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "foo test"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "foo = logger_foo(logger=set_logger(None))\n",
      "next(foo)\n",
      "foo.send(('new message', 'info'))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "INFO:isi_km_stress_log:Starting isi_km_d Signal Stress Test Run\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "INFO:isi_km_stress_log:Run Date: 2013-10-09 12:00:58\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "INFO:isi_km_stress_log:================================================================================\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "INFO:isi_km_stress_log:\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "INFO:isi_km_stress_log:new message\n"
       ]
      }
     ],
     "prompt_number": 11
    },
    {
     "cell_type": "heading",
     "level": 3,
     "metadata": {},
     "source": [
      "stdout test"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "slog = logger()\n",
      "slog.send((\"a message\", \"info\"))\n",
      "slog.send((\"b message\", \"foo\"))\n",
      "slog.send((complex(2), \"warn\"))\n",
      "slog.close()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "[2013-10-09 12:01:04] info: a message\n",
        "[2013-10-09 12:01:04] warn: (2+0j)\n"
       ]
      }
     ],
     "prompt_number": 12
    },
    {
     "cell_type": "heading",
     "level": 3,
     "metadata": {},
     "source": [
      "file system test"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "flog = logger(\"/tmp/flogger.log\")\n",
      "flog.send((\"line 1\", \"info\"))\n",
      "flog.send((\"no line\", \"bar\"))\n",
      "flog.send((\"line 2\", \"debug\"))\n",
      "flog.close()\n",
      "try:\n",
      "    flog.send((\"too late\", \"warn\"))\n",
      "except StopIteration:\n",
      "    print \"Nyah nyah nyah\""
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "Nyah nyah nyah\n"
       ]
      }
     ],
     "prompt_number": 13
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [],
     "language": "python",
     "metadata": {},
     "outputs": []
    }
   ],
   "metadata": {}
  }
 ]
}