Add a basic mandelbrot implementation.

Author: ericsnowcurrently

mandelbrot-script.py

@@ -0,0 +1,4 @@
+assert __name__ == '__main__'
+from mandelbrot.__main__ import parse_args, main
+kwargs = parse_args()
+main(**kwargs)

mandelbrot/__init__.py

@@ -0,0 +1,76 @@
+# Inspired by: https://realpython.com/mandelbrot-set-python/
+
+from math import log
+import os.path
+
+
+IMPORT_PATH_ENTRY = os.path.dirname(os.path.dirname(__file__))
+
+BOUNDS = (-2, 0.5, -1.5, 1.5)
+
+
+def xy_to_complex(x, y):
+    return x + y * 1j
+
+
+def iter_xy_to_complex(points):
+    for x, y in points:
+        yield x + y * 1j
+
+
+def apply_to_gridarray(mbs, grid, concurrent=None):
+    if concurrent is None:
+        values = grid.values
+        for i, p in enumerate(grid.iter_points()):
+            values[i] = mbs.stability_xy(*p)
+    else:
+        concurrent(mbs, grid)
+
+
+class MandelbrotSet:
+
+    def __init__(self, maxiterations, *, escape_radius=2.0, clamp=True):
+        self.maxiterations = maxiterations
+        self.escape_radius = escape_radius
+        self.clamp = clamp
+
+    def __contains__(self, c):
+        return self.stability(c, clamp=True) == 1
+
+    def escape_count(self, c, smooth=False):
+        z = 0
+        for iteration in range(self.maxiterations):
+            z = z ** 2 + c
+            if abs(z) > self.escape_radius:
+                if smooth:
+                    return iteration + 1 - log(log(abs(z))) / log(2)
+                return iteration
+        return self.maxiterations
+
+    def stability(self, c, smooth=None, clamp=None):
+        if clamp is None:
+            clamp = smooth if smooth is not None else self.clamp
+        if smooth is None:
+            smooth = False
+        value = self.escape_count(c, smooth) / self.maxiterations
+        return max(0.0, min(value, 1.0)) if clamp else value
+
+    def stability_xy(self, x, y, smooth=None, clamp=None):
+        c = x + y * 1j
+        return self.stability(c, smooth, clamp)
+
+    def map_xy(self, points, smooth=None, clamp=None):
+        for p in points:
+            x, y = p
+            c = x + y * 1j
+            yield p, self.stability_xy(*p, smooth, clamp)
+
+
+#############################
+# aliases
+
+from ._xypoints import (
+    GridSpec,
+    GridArray, BasicGridArray,
+    MultiprocessingGridArray, BasicMultiprocessingGridArray,
+)

mandelbrot/__main__.py

@@ -0,0 +1,133 @@
+import argparse
+import array
+import sys
+
+from . import BOUNDS, MandelbrotSet, apply_to_gridarray
+from ._xypoints import GridSpec, BasicGridArray, BasicMultiprocessingGridArray
+from ._concurrency import CONCURRENCY
+
+
+def _get_grid(spec, concurrentname):
+    if concurrentname == 'multiprocessing-not-shared':
+        return BasicGridArray.from_spec(spec)
+    if 'multiprocessing' in concurrentname:
+        return BasicMultiprocessingGridArray.from_spec(spec)
+    else:
+        return BasicGridArray.from_spec(spec)
+
+
+def prep(maxiterations, density, concurrent=None):
+    mbs = MandelbrotSet(maxiterations)
+    gridspec = GridSpec(*BOUNDS, density)
+    grid = _get_grid(gridspec, getattr(concurrent, '_concurrent', ''))
+    return mbs, grid
+
+
+#######################################
+# output
+
+def output_text(values, width):
+    hit = ' *'
+    hit = '\u26aa'
+    #hit = '\U0001f7e1'
+    miss = '  '
+
+    def iter_lines():
+        for rowi in range(width):
+            line = ''
+            for i in range(rowi*width, rowi*width+width):
+                if values[i] == 1:
+                    line += hit
+                else:
+                    line += miss
+            yield line
+
+    print('*'*(width*2+4))
+    print('*', ' '*(width*2), '*')
+    for line in iter_lines():
+        print('*', line, '*')
+    print('*', ' '*(width*2), '*')
+    print('*'*(width*2+4))
+
+
+def output_image(values, width):
+    from PIL import Image
+
+    height = width
+    BLACK_AND_WHITE = "1"
+
+    image = Image.new(mode=BLACK_AND_WHITE, size=(width, height))
+    image.putdata(values)
+    image.show()
+
+
+OUTPUT = {
+    'stdout': output_text,
+    'image': output_image,
+}
+
+
+#######################################
+# the script
+
+'''
+ (with GIL)
+ size: 100/100
+
+ threads-not-shared                86s
+ threads-shared-direct            105s
+ threads-shared-executor           93s
+
+ interpreters-shared-pipe          ---
+ interpreters-not-shared-pipe      28s
+ interpreters-shared-channel       19s
+ interpreters-not-shared-channel   19s
+
+ multiprocessing-not-shared        19s
+ multiprocessing-shared-direct     24s
+ multiprocessing-shared-executor   ---
+
+ async                             76s
+'''
+
+def parse_args(argv=sys.argv[1:], prog=sys.argv[0]):
+    parser = argparse.ArgumentParser(prog=prog)
+
+    parser.add_argument('--maxiterations', type=int)
+    parser.add_argument('--density', type=int)
+    parser.add_argument('--output', choices=list(OUTPUT))
+    parser.add_argument('--concurrent', choices=list(CONCURRENCY))
+
+    args = parser.parse_args(argv)
+    ns = vars(args)
+
+    if args.concurrent is not None:
+        args.concurrent = CONCURRENCY[args.concurrent]
+
+    for key, value in list(ns.items()):
+        if value is None:
+            del ns[key]
+
+    return ns
+
+
+def main(maxiterations=100, density=None, output='stdout', concurrent=None):
+    if not density:
+        if output == 'stdout':
+            density = 80
+        else:
+            density = 512
+
+    try:
+        output = OUTPUT[output]
+    except KeyError:
+        raise ValueError(f'unsupported output {output!r}')
+
+    mbs, grid = prep(maxiterations, density, concurrent)
+    apply_to_gridarray(mbs, grid, concurrent)
+    output(grid.values, grid.density)
+
+
+if __name__ == '__main__':
+    kwargs = parse_args()
+    main(**kwargs)