Image Processing Pipeline

PracticalmediumPremium

45:00

Desktop only

Coding Challenges are optimized for Desktop

Come back on a wider screen - the live editor needs room to breathe. You can still read the problem brief above.

Problem

A media service ingests user images and applies named processing pipelines to them - thumbnails, blurs, grayscale previews. Each pipeline is an ordered list of operations; the service must apply many pipelines to many images, fast and without one bad input taking down the batch. You build that pipeline runner on top of Pillow, scaling it up one constraint at a time: a single image, then a full cross-product, then parallel execution, then fault isolation, then discovery from disk.

Pillow is available in the sandbox (from PIL import Image, ImageFilter). Operations are dicts, already parsed from JSON, for example {"op": "resize", "width": 200, "height": 100}, {"op": "rotate", "degrees": 90}, {"op": "crop", "left": 10, "top": 10, "right": 90, "bottom": 90}, {"op": "blur", "radius": 2}, {"op": "grayscale"}.

For the parallel levels, build the pool with from _compat import get_executor rather than constructing a ThreadPoolExecutor directly - it uses real threads where the runtime supports them and runs serially in this browser sandbox, which has no OS threads. The observable results are identical either way.

Level 1 - Single image, single pipeline

def apply_pipeline(image_path: str, pipeline: list[dict], output_path: str) -> None

Open the image, apply each operation in order, and save the result to output_path. The output dimensions and mode must reflect the operations (a resize changes size; grayscale yields mode "L").

Example:

apply_pipeline("red.png", [{"op": "resize", "width": 50, "height": 30}], "out.png")
# out.png is 50x30

Level 2 - All combinations, serial

def process_all(image_paths, pipelines, output_dir) -> dict[tuple[str, str], str]

Apply every pipeline to every image (M images by N pipelines). pipelines maps a name to its op list. Write each result as <image_stem>_<pipeline_name>.png under output_dir, and return {(image_path, pipeline_name): output_path} for every combination.

Example:

process_all(["red.png"], {"thumb": [...], "gray": [...]}, "out/")
# -> {("red.png","thumb"): "out/red_thumb.png", ("red.png","gray"): "out/red_gray.png"}

Level 3 - Parallel execution

def process_all_parallel(image_paths, pipelines, output_dir, max_workers=4) -> dict[tuple[str, str], str]

Same inputs and same results as process_all, but process the combinations through a thread pool. Each combination is an independent unit of work.

Example:

process_all_parallel(paths, pipelines, "out/", max_workers=4)
# same keys and output files as process_all

Level 4 - Partial failure handling

def process_all_safe(image_paths, pipelines, output_dir, max_workers=4)
    -> tuple[dict[tuple[str, str], str], dict[tuple[str, str], Exception]]

A bad operation (unknown op, out-of-bounds crop) must not abort the batch. Return (results, errors): successful combinations map to their output path; failed ones map to the exception raised. Every combination ends up in exactly one of the two dicts.

Example:

process_all_safe(["red.png"], {"ok": [grayscale], "bad": [unknown_op]}, "out/")
# -> ({("red.png","ok"): "..."} , {("red.png","bad"): ValueError(...)})

Level 5 - Discover inputs from disk

def process_from_dirs(images_dir, pipelines_dir, output_dir, max_workers=4)
    -> tuple[dict[tuple[str, str], str], dict[tuple[str, str], Exception]]

Read images from images_dir and pipeline JSON files from pipelines_dir (one pipeline per .json file, keyed by filename stem), then run every combination through the fault-tolerant path. A single corrupt pipeline file must not block the rest.

Example:

images_dir has img_00.png, img_01.png; pipelines_dir has gray.json, resize.json
process_from_dirs(images_dir, pipelines_dir, "out/")
# -> 4 combinations across (results, errors)

Loading editor...

The live editor for this challenge unlocks with Premium.

Browse free problems

0/0 passed