Add source code examples

Author: gjbex

source-code/README.md

@@ -0,0 +1,21 @@
+# Source code
+
+This source code illustrates concepts in the training and
+was used to develop it.
+
+
+## What is it?
+
+1. `context-manager`: illustrates how to write your own context managers.
+1. `coroutines`: illustrates how to write coroutines in Python.
+1. `design-patterns`: illustrates some common design patterns in Python.
+1. `functional-programming`: illustrates some concepts of functional
+   programming in Python.
+1. `object-orientation`: illustrates some concepts of object-oriented
+   programming in Python.
+1. `operators-functools`: illustrates some applications of the `operator`
+   and `functools` modules in Python's standard library.
+1. `typing`: illustrates how to use type annotation in Python, and
+   demonstrates `mypy`.
+1. `unit-testing`: illustrates writing unit tests with `unittest` and
+   `pytest`.

source-code/context-manager/README.md

@@ -0,0 +1,19 @@
+Context manager
+===============
+
+A context manager can be used to initialize and finalize resources.  Once
+defined, they can be used with Python's `with` construct.
+
+What is it?
+-----------
+`context.py`: illustrates two ways to define a context manager,
+* as a class, providing hooks on instantiation, entering, and exiting
+  the context.
+* as function decorator, where code to be executed for initialization,
+  entering, and exiting sandwiches a `yield` statement that returns the
+  value.
+
+Note
+----
+The exception is raised on purpose to illustrate exception handling by
+a context manager.

source-code/context-manager/context.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python
+
+from contextlib import contextmanager
+import sys
+
+
+class ContextTest(object):
+
+    def __init__(self, context_nr):
+        self._context_nr = context_nr
+        print('created with {0}'.format(self._context_nr))
+
+    def __enter__(self):
+        print('entering {0}'.format(self._context_nr))
+        return self
+
+    def __exit__(self, exception_type, exception_value, backtrace):
+        print('exiting {0}'.format(self._context_nr))
+        if exception_type:
+            print('exception in context {0}:'.format(self._context_nr))
+            print('\t', exception_type, exception_value, backtrace)
+            return
+        else:
+            print('no exception in context {0}'.format(self._context_nr))
+
+
+@contextmanager
+def label(name):
+    print('entering label({0})'.format(name))
+    yield name
+    print('exiting label({0})'. format(name))
+
+
+def main():
+    with ContextTest(1) as c1, ContextTest(2) as c2:
+        print('in context {0}'.format(c1._context_nr))
+        print('in context {0}'.format(c2._context_nr))
+    with label('foo') as foo, label('bar') as bar:
+        print(foo, bar)
+    with ContextTest(1) as c1, ContextTest(2) as c2:
+        print('in context {0}'.format(c1._context_nr))
+        raise Exception()
+        print('in context {0}'.format(c2._context_nr))
+    return 0
+
+if __name__ == '__main__':
+    status = main()
+    sys.exit(status)

source-code/coroutines/README.md

@@ -0,0 +1,18 @@
+Coroutines
+==========
+
+Coroutines provide a powerful programming paradigm for functions that
+preserve state between invocations.
+
+What is it?
+-----------
+1. `coroutine.py`: implements a coroutine `my_printer` that can can be
+    "created" by invocating the function, "started" using the `next()`
+    method, "invoked" using the `send()` method, and terminated using
+    the `close()` method.
+1. `longest_run.py`: stateful streaming parser implemented using a
+    coroutine.  It keeps track of the longest run of characters in
+    a string that is presented one batch at the time.
+1. `stats.py`: coroutine implementation for collecting statistics (mean,
+    standard deviation).
+1. `data.txt`: data file to compute statistical measures from.

source-code/coroutines/coroutines.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python
+
+
+def my_printer():
+    try:
+        counter = 0
+        while True:
+            buffer = (yield)
+            counter += 1
+            print('{0:03d}: {1}'.format(counter, buffer))
+    except GeneratorExit:
+        pass
+
+
+def main():
+    coroutine = my_printer()
+    next(coroutine)
+    for i in range(26):
+        str = chr(ord('A') + i)
+        coroutine.send(str)
+    coroutine.close()
+
+if __name__ == '__main__':
+    main()

source-code/coroutines/data.txt

@@ -0,0 +1,6 @@
+alpha, beta, gamma
+1.0, 17.3, -3.3
+2.0, 14.5, -2.9
+3.0, 15.7, -4.9
+4.0, 14.3, -3.4
+

source-code/coroutines/longest_run.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python
+
+from collections import defaultdict
+
+
+def longest_run():
+    runs = defaultdict(lambda: 0)
+    buffer = None
+    while True:
+        data = yield runs
+        if data:
+            if not buffer:
+                prev = data.pop(0)
+                buffer = [prev]
+                run_length = 1
+            else:
+                prev = buffer[-1]
+                if prev != data[0]:
+                    run_length = len(buffer)
+                    if run_length > runs[prev]:
+                        runs[prev] = run_length
+                    prev = data.pop(0)
+                    buffer = [prev]
+                    run_length = 1
+                else:
+                    buffer.append(prev)
+            while data:
+                current = data.pop(0)
+                if current == prev:
+                    run_length += 1
+                    buffer.append(current)
+                else:
+                    if run_length > runs[prev]:
+                        runs[prev] = run_length
+                    prev = current
+                    buffer = [prev]
+                    run_length = 1
+    return runs
+
+
+def print_run_lengths(runs, prefix=''):
+    for char, run_length in runs.items():
+        print(f'{prefix}{char}: {run_length}')
+
+if __name__ == '__main__':
+    data_stream = [
+        'aabbbba',
+        'aaaabb',
+        'bbbbbaa',
+    ]
+    longest_runner = longest_run()
+    next(longest_runner)
+    for block_nr, data in enumerate(data_stream):
+        print(data)
+        runs = longest_runner.send(list(data))
+        print(f'  result for block {block_nr + 1}')
+        print_run_lengths(runs, '    ')
+    longest_runner.close()

source-code/coroutines/stats.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python
+
+from collections import namedtuple
+from math import sqrt
+import sys
+
+Result = namedtuple('Result', ['mean', 'stddev'])
+
+
+def stats():
+    sum = 0.0
+    sum2 = 0.0
+    count = 0
+    while True:
+        number = yield
+        if number is None:
+            break
+        sum += number
+        sum2 += number**2
+        count += 1
+    return Result(sum/count, sqrt((sum2 - sum**2/count)/(count - 1)))
+
+
+def process_line(line, mode=float):
+    line = line.strip()
+    if not line:
+        return None
+    items = [item.strip() for item in line.split(',')]
+    if mode == float:
+        return [float(item) for item in items]
+    else:
+        return items
+
+if __name__ == '__main__':
+    line = sys.stdin.readline()
+    names = process_line(line, mode='text')
+    averagers = [stats() for name in names]
+    for averager in averagers:
+        next(averager)
+    for line in sys.stdin:
+        data = process_line(line)
+        if not data:
+            continue
+        for number, averager in zip(data, averagers):
+            averager.send(number)
+    for name, averager in zip(names, averagers):
+        try:
+            averager.send(None)
+        except StopIteration as exception:
+            result = exception.value
+            print('{0}: {1:.3f}, {2:.3f}'.format(name,
+                                                 result.mean,
+                                                 result.stddev))

source-code/design-patters/README.md

@@ -0,0 +1,8 @@
+# Design patterns
+
+Code to illustrate some design patterns in Python.
+
+## What is it?
+
+1. `decorator_design_pattern.ipynb`: notebook that illustrate how decorators
+   can be used to change the behaviour of objects.