# A generator function def fib_generator(): """A generator function for Fibonacci numbers. >>> fg = fib_generator() >>> [(i, n) for i, n in zip(range(1, 9), fg)] [(1, 0), (2, 1), (3, 1), (4, 2), (5, 3), (6, 5), (7, 8), (8, 13)] """ yield 0 prev, current = 0, 1 while True: yield current prev, current = current, prev + current # An iterable Rlist class Rlist(object): """A recursive list consisting of a first element and the rest. >>> s = Rlist(1, Rlist(2, Rlist(3))) >>> len(s) 3 >>> s[0] 1 >>> s[1] 2 >>> s[2] 3 >>> for e in s: print(e) 1 2 3 """ class EmptyList(object): def __repr__(self): return 'Rlist.empty' def __len__(self): return 0 empty = EmptyList() def __init__(self, first, rest=empty): self.first = first self.rest = rest def __repr__(self): f = repr(self.first) if self.rest is Rlist.empty: return 'Rlist({0})'.format(f) else: return 'Rlist({0}, {1})'.format(f, repr(self.rest)) def __len__(self): return 1 + len(self.rest) def __getitem__(self, k): if k == 0: return self.first elif self.rest is Rlist.empty: raise IndexError('index out of range') return self.rest[k - 1] def __iter__(self): current = self while current is not Rlist.empty: yield current.first current = current.rest # A lazily evaluated Rlist class Stream(Rlist): """A lazily computed recursive list. >>> s = Stream(1, lambda: Stream(2+3, lambda: Stream(9))) >>> s.first 1 >>> s.rest.first 5 >>> s.rest Stream(5, <...>) >>> s.rest.rest.first 9 >>> s = Stream(1, lambda: Stream(1+2, lambda: Stream(9))) >>> s[2] 9 >>> s[0] 1 >>> s[1] 3 >>> for e in s: print(e) 1 3 9 """ def __init__(self, first, compute_rest=lambda: Stream.empty): if not callable(compute_rest): raise TypeError('compute_rest must be callable') self.first = first self._compute_rest = compute_rest self._rest = None @property def rest(self): """Return the rest of the stream, computing it if necessary.""" if self._compute_rest is not None: self._rest = self._compute_rest() self._compute_rest = None return self._rest def __len__(self): raise NotImplementedError('length not supported on Streams') def __repr__(self): return 'Stream({0}, <...>)'.format(repr(self.first)) # An infinite stream of integers. def integer_stream(first=1): """Return a stream of consecutive integers, starting with first. >>> s = integer_stream(3) >>> s Stream(3, <...>) >>> m = map_stream(lambda x: x*x, s) >>> first_k_as_list(m, 5) [9, 16, 25, 36, 49] """ def compute_rest(): return integer_stream(first+1) return Stream(first, compute_rest) # Map and filter on streams def map_stream(fn, s): """Map a function fn over the elements of a stream s.""" if s is Stream.empty: return s def compute_rest(): return map_stream(fn, s.rest) return Stream(fn(s.first), compute_rest) def filter_stream(fn, s): """Filter stream s with predicate function fn.""" if s is Stream.empty: return s def compute_rest(): return filter_stream(fn, s.rest) if fn(s.first): return Stream(s.first, compute_rest) else: return compute_rest() # More stream examples def first_k_as_list(s, k): """Return the first k elements of stream s as a list.""" first_k = [] for item in s: if len(first_k) == k: break first_k.append(item) return first_k def fib_stream(a=0, b=1): """A stream of Fibonacci numbers. >>> first_k_as_list(fib_stream(), 8) [0, 1, 1, 2, 3, 5, 8, 13] """ return Stream(a, lambda: fib_stream(b, a+b)) ones = Stream(1, lambda: ones) def add_streams(s1, s2): """Return the sum of two streams as a stream. >>> ints = Stream(1, lambda: add_streams(ints, ones)) >>> first_k_as_list(ints, 6) [1, 2, 3, 4, 5, 6] >>> fibs = Stream(0, lambda: Stream(1, lambda: add_streams(fibs, fibs.rest))) >>> first_k_as_list(fibs, 8) [0, 1, 1, 2, 3, 5, 8, 13] """ def compute_rest(): return add_streams(s1.rest, s2.rest) return Stream(s1.first + s2.first, compute_rest) def primes(pos_stream): """Return a stream of primes, given a stream of consecutive integers. >>> ints = Stream(2, lambda: add_streams(ints, ones)) >>> first_k_as_list(primes(ints), 8) [2, 3, 5, 7, 11, 13, 17, 19] """ def not_divisible(x): return x % pos_stream.first != 0 def compute_rest(): return primes(filter_stream(not_divisible, pos_stream.rest)) return Stream(pos_stream.first, compute_rest) # An infinite sequence of bitstrings from itertools import product def bitstrings(): """Generate bitstrings in order of increasing size. >>> bs = bitstrings() >>> [next(bs) for _ in range(0, 10)] ['', '0', '1', '00', '01', '10', '11', '000', '001', '010'] """ size = 0 while True: tuples = product(('0', '1'), repeat=size) for elem in tuples: yield ''.join(elem) size += 1 # An infinite sequence of distinct functions from random import Random def functions(n): """Generate a stream of random, distinct 1-argument boolean functions.""" def compute_rest(): return functions(n+1) def f(x): return Random(x*n+n+x).choice((True, False)) return Stream(f, compute_rest) def func_to_str(fn, seq, n=None): """Return a string representation of the result of mapping fn across the elements in seq. If n is provided, then the nth element is bracketed.""" values = list(map(fn, seq)) for i in range(len(values)): fmt = '[{0}]' if i == n else ' {0} ' values[i] = fmt.format(str(values[i])[0]) values.append(' . . .') return ''.join(values) def print_func_table(s, k): """Print a tabular representation of the first k functions in s.""" fs = first_k_as_list(s, k) n = 0 for f in fs: print(func_to_str(f, range(k), n)) n += 1 for _ in range(3): print(func_to_str(lambda _: '.', range(k))) print() def func_not_in_stream(s): """Given an infinite stream s, return a function that is not in s.""" return lambda n: not s[n](n) def diagonalize(num_fn=10): """Print out functions from an infinite stream, and their diagonalization.""" funcs = functions(0) print_func_table(funcs, num_fn) print(func_to_str(func_not_in_stream(funcs), range(num_fn)))