How to Split a Python List - Update materials

Author: bzaczynski

python-split-list/README.md

@@ -1,3 +1,3 @@
-# How to Split a Python List
+# How to Split a Python List Into Chunks
 
-This folder holds the code for the Real Python How to Split a Python List tutorial.
+This folder holds sample code that supplements the Real Python tutorial on [How to Split a Python List Into Chunks](https://realpython.com/how-to-split-a-python-list-into-chunks/).

python-split-list/parallelization.py python-split-list/parallel_demo.pypython-split-list/parallelization.py renamed to python-split-list/parallel_demo.py

@@ -1,29 +1,29 @@
 """
-Plot the Mandelbrot set using parallel workers.
+Synthesize an image in chunks using parallel workers.
 
 Usage:
-$ python parallelization.py
+$ python parallel_demo.py
 """
 
 import functools
 import multiprocessing
-import os
 import time
 from dataclasses import dataclass
 from math import log
-from typing import Callable, Iterable
+from os import cpu_count
+from typing import Callable, Iterable, Iterator
 
 import numpy as np
 from PIL import Image
 
-from splitting.multidimensional import Bounds, split_multi
+from spatial_splitting import Bounds, split_multi
 
-IMAGE_WIDTH, IMAGE_HEIGHT = 1280, 720
+IMAGE_WIDTH, IMAGE_HEIGHT = 1920, 1080
 CENTER = -0.7435 + 0.1314j
-SCALE = 0.0000025
+SCALE = 0.0000015
 MAX_ITERATIONS = 256
 ESCAPE_RADIUS = 1000
-NUM_CHUNKS = os.cpu_count() or 4
+NUM_CHUNKS = cpu_count() or 4
 
 
 class Chunk:
@@ -35,12 +35,10 @@ def __init__(self, bounds: Bounds) -> None:
         self.width = bounds.size[1]
         self.pixels = np.zeros((self.height, self.width), dtype=np.uint8)
 
-    def __getitem__(self, coordinates) -> int:
-        """Get the value of a pixel at the given absolute coordinates."""
+    def __getitem__(self, coordinates: tuple[int, int]) -> int:
         return self.pixels[self.bounds.offset(*coordinates)]
 
-    def __setitem__(self, coordinates, value: int) -> None:
-        """Set the value of a pixel at the given absolute coordinates."""
+    def __setitem__(self, coordinates: tuple[int, int], value: int) -> None:
         self.pixels[self.bounds.offset(*coordinates)] = value
 
 
@@ -49,14 +47,14 @@ class MandelbrotSet:
     max_iterations: int
     escape_radius: float = 2.0
 
-    def __contains__(self, c: complex) -> bool:
+    def __contains__(self, c):
         return self.stability(c) == 1
 
-    def stability(self, c: complex, smooth=False, clamp=True) -> float:
+    def stability(self, c, smooth=False, clamp=True):
         value = self.escape_count(c, smooth) / self.max_iterations
         return max(0.0, min(value, 1.0)) if clamp else value
 
-    def escape_count(self, c: complex, smooth=False) -> int | float:
+    def escape_count(self, c, smooth=False):
         z = 0 + 0j
         for iteration in range(self.max_iterations):
             z = z**2 + c
@@ -67,6 +65,32 @@ def escape_count(self, c: complex, smooth=False) -> int | float:
         return self.max_iterations
 
 
+def transform(y: int, x: int) -> complex:
+    """Transform the given pixel coordinates to the complex plane."""
+    im = SCALE * (IMAGE_HEIGHT / 2 - y)
+    re = SCALE * (x - IMAGE_WIDTH / 2)
+    return complex(re, im) + CENTER
+
+
+def generate_chunk(bounds: Bounds) -> Chunk:
+    """Generate a chunk of pixels for the given bounds."""
+    chunk = Chunk(bounds)
+    mandelbrot_set = MandelbrotSet(MAX_ITERATIONS, ESCAPE_RADIUS)
+    for y, x in bounds:
+        c = transform(y, x)
+        instability = 1 - mandelbrot_set.stability(c, smooth=True)
+        chunk[y, x] = int(instability * 255)
+    return chunk
+
+
+def combine(chunks: Iterable[Chunk]) -> Image.Image:
+    """Combine the chunks into a single image."""
+    pixels = np.zeros((IMAGE_HEIGHT, IMAGE_WIDTH), dtype=np.uint8)
+    for chunk in chunks:
+        pixels[chunk.bounds.slices()] = chunk.pixels
+    return Image.fromarray(pixels, mode="L")
+
+
 def timed(function: Callable) -> Callable:
     @functools.wraps(function)
     def wrapper(*args, **kwargs):
@@ -79,53 +103,23 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
-def main() -> None:
-    for worker in (process_chunks_parallel, process_chunks_sequential):
-        image = compute(worker)
-        image.show()
-
+def process_sequentially(bounds_iter: Iterator[Bounds]) -> Iterator[Chunk]:
+    return map(generate_chunk, bounds_iter)
 
-@timed
-def process_chunks_sequential(chunked_bounds: Iterable[Bounds]) -> list[Chunk]:
-    return list(map(generate_chunk, chunked_bounds))
 
-
-@timed
-def process_chunks_parallel(chunked_bounds: Iterable[Bounds]) -> list[Chunk]:
+def process_in_parallel(bounds_iter: Iterator[Bounds]) -> list[Chunk]:
     with multiprocessing.Pool() as pool:
-        return pool.map(generate_chunk, chunked_bounds)
-
-
-def generate_chunk(bounds: Bounds) -> Chunk:
-    """Generate a chunk of pixels for the given bounds."""
-    chunk = Chunk(bounds)
-    mandelbrot_set = MandelbrotSet(MAX_ITERATIONS, ESCAPE_RADIUS)
-    for y, x in bounds:
-        c = transform(y, x)
-        instability = 1 - mandelbrot_set.stability(c, smooth=True)
-        chunk[y, x] = int(instability * 255)
-    return chunk
-
-
-def transform(y: int, x: int) -> complex:
-    """Transform the given pixel coordinates to the complex plane."""
-    im = SCALE * (IMAGE_HEIGHT / 2 - y)
-    re = SCALE * (x - IMAGE_WIDTH / 2)
-    return complex(re, im) + CENTER
+        return pool.map(generate_chunk, bounds_iter)
 
 
+@timed
 def compute(worker: Callable) -> Image.Image:
-    """Render the image using the given worker function."""
-    chunks = worker(split_multi(NUM_CHUNKS, IMAGE_HEIGHT, IMAGE_WIDTH))
-    return combine(chunks)
+    return combine(worker(split_multi(NUM_CHUNKS, IMAGE_HEIGHT, IMAGE_WIDTH)))
 
 
-def combine(chunks: list[Chunk]) -> Image.Image:
-    """Combine the chunks into a single image."""
-    pixels = np.zeros((IMAGE_HEIGHT, IMAGE_WIDTH), dtype=np.uint8)
-    for chunk in chunks:
-        pixels[chunk.bounds.slices()] = chunk.pixels
-    return Image.fromarray(pixels, mode="L")
+def main() -> None:
+    for worker in (process_sequentially, process_in_parallel):
+        compute(worker).show()
 
 
 if __name__ == "__main__":

python-split-list/requirements.txt

@@ -1,3 +1,3 @@
-Pillow==9.3.0
+Pillow==9.4.0
 more-itertools==9.0.0
-numpy==1.23.5
+numpy==1.24.1

…split-list/splitting/multidimensional.py python-split-list/spatial_splitting.pypython-split-list/splitting/multidimensional.py renamed to python-split-list/spatial_splitting.py python-split-list/splitting/multidimensional.py renamed to python-split-list/spatial_splitting.py

@@ -0,0 +1,120 @@
+"""
+Split into fixed-size chunks or a fixed number of chunks.
+"""
+
+import sys
+from itertools import cycle, islice, zip_longest
+from typing import Any, Iterable, Iterator, Sequence
+
+import numpy as np
+
+if sys.version_info >= (3, 12):
+    from itertools import batched
+else:
+    try:
+        from more_itertools import batched
+    except ImportError:
+
+        def batched(
+            iterable: Iterable[Any], chunk_size: int
+        ) -> Iterator[tuple[Any, ...]]:
+            iterator = iter(iterable)
+            while chunk := tuple(islice(iterator, chunk_size)):
+                yield chunk
+
+
+def batched_with_padding(
+    iterable: Iterable[Any], batch_size: int, fill_value: Any = None
+) -> Iterator[tuple[Any, ...]]:
+    for batch in batched(iterable, batch_size):
+        yield batch + (fill_value,) * (batch_size - len(batch))
+
+
+def batched_functional(
+    iterable: Iterable[Any], chunk_size: int
+) -> Iterator[tuple[Any, ...]]:
+    iterator = iter(iterable)
+    return iter(lambda: tuple(islice(iterator, chunk_size)), tuple())
+
+
+def split_with_padding(
+    iterable: Iterable[Any], chunk_size: int, fill_value: Any = None
+) -> Iterator[tuple[Any, ...]]:
+    return zip_longest(*[iter(iterable)] * chunk_size, fillvalue=fill_value)
+
+
+def split_into_pairs(
+    iterable: Iterable[Any], fill_value: Any = None
+) -> Iterator[tuple[Any, Any]]:
+    iterator = iter(iterable)
+    return zip_longest(iterator, iterator, fillvalue=fill_value)
+
+
+def split_drop_last(
+    iterable: Iterable[Any], chunk_size: int
+) -> Iterator[tuple[Any, ...]]:
+    return zip(*[iter(iterable)] * chunk_size)
+
+
+def split_sequence(
+    sequence: Sequence[Any], chunk_size: int
+) -> Iterator[Sequence[Any]]:
+    for i in range(0, len(sequence), chunk_size):
+        yield sequence[i : i + chunk_size]
+
+
+def split_str(
+    sequence: str, chunk_size: int, fill_value: str = " "
+) -> Iterator[str]:
+    for chunk in split_sequence(sequence, chunk_size):
+        padding = "".join([fill_value] * (chunk_size - len(chunk)))
+        yield chunk + padding
+
+
+def split_list(
+    sequence: list[Any], chunk_size: int, fill_value: Any = None
+) -> Iterator[list[Any]]:
+    for chunk in split_sequence(sequence, chunk_size):
+        padding = [fill_value] * (chunk_size - len(chunk))
+        yield chunk + padding
+
+
+def split_into_slices(
+    sequence: Sequence[Any], slice_size: int
+) -> Iterator[slice]:
+    for i in range(0, len(sequence), slice_size):
+        yield slice(i, i + slice_size)
+
+
+def split_strides(
+    sequence: Sequence[Any], num_chunks: int
+) -> Iterator[Sequence[Any]]:
+    for i in range(num_chunks):
+        yield sequence[i::num_chunks]
+
+
+def split_round_robin(
+    iterable: Iterable[Any], num_chunks: int
+) -> list[list[Any]]:
+    chunks: list[list[Any]] = [list() for _ in range(num_chunks)]
+    index = cycle(range(num_chunks))
+    for value in iterable:
+        chunks[next(index)].append(value)
+    return chunks
+
+
+def split_n(
+    sequence: Sequence[Any], num_chunks: int
+) -> Iterator[Sequence[Any]]:
+    chunk_size, remaining = divmod(len(sequence), num_chunks)
+    for i in range(num_chunks):
+        begin = i * chunk_size + min(i, remaining)
+        end = (i + 1) * chunk_size + min(i + 1, remaining)
+        yield sequence[begin:end]
+
+
+def split_with_numpy(
+    numbers: Sequence[float | int], chunk_size: int
+) -> list[np.ndarray]:
+    indices = np.arange(chunk_size, len(numbers), chunk_size)
+    return np.array_split(numbers, indices)