Olympict

Olympict runs image and video processing pipelines in parallel using Python multiprocessing. Build composable, typed chains — resize, crop, draw annotations, run AI models on every frame — without managing threads or processes yourself.

Built on olympipe >=2.2. Requires Python >=3.10 (tested on 3.10, 3.12, 3.13 and 3.14).

pip install olympict
pip install "olympict[hf]"   # Hugging Face / PyTorch
uv add olympict

Development:

uv sync --group dev
uv run pytest
uv run ruff check .
uv run ruff format .
uv run mypy .

CLI

Common operations are available directly from the terminal — processed in parallel automatically.

Standalone utilities

# Extract frames from video files
olympict extract-frames ./videos ./frames --discard 4

Chained syntax (single invocation)

Commands can be chained in a single olympict call. The pipeline is built lazily and only executed by the terminal command (save or to-video). Transform commands have no positional args — only options.

# Crop → resize → save
olympict source ./photos crop --top 50 --bottom 50 resize -w 640 -h 480 save ./output

# Rescale then convert format
olympict source ./photos rescale --x 0.5 --y 0.5 convert --format png save ./output

# Video → resize → reassemble
olympict source --type video ./videos resize -w 640 -h 480 to-video ./out.mp4 --fps 30

# Detect → draw → save (requires [hf])
olympict source ./photos detect --model facebook/detr-resnet-50 save ./annotated

# Subsample (keep 1 frame every 2)
olympict source ./frames subsample --keep 1 --discard 1 save ./subsampled

Unix pipe syntax

Same commands, connected by |. Useful for composing with other shell tools.

olympict source ./photos \
  | olympict crop --top 50 --bottom 50 \
  | olympict resize -w 640 -h 480 \
  | olympict save ./output

Note: source options must come before its INPUT argument in both modes: olympict source --limit 10 ./photos resize -w 640 save ./output

Run olympict --help or olympict COMMAND --help for full options.

Key concepts

• Pipelines are lazy — processing only starts when you call wait_for_completion() or wait_for_results()
• Each .step() returns a new pipeline — chain freely
• count=N on any step runs it across N parallel processes
• class_task instantiates a class once per worker — ideal for loading AI models

OPipeline

OPipeline wraps an olympipe Pipeline and returns OPipeline (or the same subclass, such as ImagePipeline, when the packet type does not change). ImagePipeline, VideoPipeline and BatchPipeline keep their existing methods.

from olympict import OPipeline

total = OPipeline(data=range(5)).wait_and_reduce(0, lambda item, acc: acc + item)

values = (
    OPipeline(data=["a", "b", "c"])
    .task(str.upper)
    .filter(lambda item: item != "B")
    .limit(2)
    .wait_for_results()
)

task accepts the same on_error callback as olympipe: return a replacement packet, or None to drop the failing one.

def fail_on_odd(value: int) -> int:
    if value % 2:
        raise ValueError(value)
    return value


def drop(_item: int, _exc: Exception) -> None:
    return None


OPipeline(data=range(6)).task(fail_on_odd, on_error=drop).wait_for_results()

def split_even_odd(value: int) -> tuple[int | None, int | None]:
    if value % 2 == 0:
        return (value, None)
    return (None, value)


evens, odds = OPipeline(data=range(6)).split(split_even_odd)
merged = evens.gather(odds)  # also accepts a raw olympipe Pipeline

(
    OPipeline(data=[1, 2, 3])
    .cached_task(lambda value: value * 2, cache_dir="/tmp/olympict-cache")
    .uncache()
    .checkpoint([], lambda item, acc: acc + [item], lambda acc: reversed(acc))
    .timeout(30)
    .debug()
    .start()
    .wait_for_completion()
)

timeout, debug, checkpoint and limit keep the concrete class:

from olympict import ImagePipeline

pipe = ImagePipeline.load_folder("./images").timeout(30).debug()
assert isinstance(pipe, ImagePipeline)

ImagePipeline

Resize and save to a new folder

from olympict import ImagePipeline

(
    ImagePipeline.load_folder("./images")
    .resize((640, 480))
    .save_to_folder("./resized")
    .wait_for_completion()
)

Resize preserving aspect ratio (letterbox padding)

from olympict import ImagePipeline

blue = (255, 0, 0)  # BGR — matches OpenCV convention

(
    ImagePipeline.load_folder("./images")
    .resize((640, 480), pad_color=blue)
    .save()  # overwrite originals
    .wait_for_completion()
)

Rescale by ratio instead of absolute size

from olympict import ImagePipeline

(
    ImagePipeline.load_folder("./images")
    .rescale((0.5, 0.5))  # halve width and height
    .save_to_folder("./half")
    .wait_for_completion()
)

Custom pixel operation

from olympict import ImagePipeline, Img


def swap_channels(img: Img) -> Img:
    img[:, :, 1] = (img[:, :, 0] + img[:, :, 2]) / 2
    return img


(
    ImagePipeline.load_folder("./images")
    .task_img(swap_channels)
    .save()
    .wait_for_completion()
)

Visual preview at any step

from olympict import ImagePipeline, Img


def edge_blur(img: Img) -> Img:
    import cv2

    return cv2.GaussianBlur(img, (21, 21), 0)


(
    ImagePipeline.load_folder("./images")
    .debug_window("before")
    .task_img(edge_blur)
    .debug_window("after")
    .save()
    .wait_for_completion()
)

Crop with optional padding

from olympict import ImagePipeline

(
    ImagePipeline.load_folder("./images")
    .crop(left=50, top=50, right=50, bottom=50)
    .save_to_folder("./cropped")
    .wait_for_completion()
)

Keep 1 frame every N (temporal subsampling)

Useful when processing video sequences where consecutive frames are too similar.

from olympict import ImagePipeline

(
    ImagePipeline.load_folder("./frames")
    .keep_each_frame_in(keep_n=1, discard_n=4)  # keep 1, skip 4
    .save_to_folder("./sampled")
    .wait_for_completion()
)

Collect results instead of saving

from olympict import ImagePipeline
from olympict.files.o_image import OlympImage

results: list[OlympImage] = (
    ImagePipeline.load_folder("./images").resize((224, 224)).wait_for_results()
)

print(f"Processed {len(results)} images")

Draw bounding boxes

draw_bboxes expects a function returning a list of ((x1, y1, x2, y2, label, score), color) tuples.

from olympict import ImagePipeline
from olympict.files.o_image import OlympImage
from olympict import BBoxAbsolute, Color


def my_bboxes(o: OlympImage) -> list[tuple[BBoxAbsolute, Color]]:
    boxes = o.metadata.get("boxes", [])
    return [(b, (0, 255, 0)) for b in boxes]


(
    ImagePipeline.load_folder("./images")
    .draw_bboxes(my_bboxes)
    .save_to_folder("./annotated")
    .wait_for_completion()
)

Draw relative bounding boxes (HuggingFace format)

After .detect(), predictions are stored in metadata["pred_bboxes"] as relative coordinates.

from olympict import ImagePipeline

(
    ImagePipeline.load_folder("./images")
    .draw_relative_bboxes()  # reads metadata["pred_bboxes"] automatically
    .save_to_folder("./annotated")
    .wait_for_completion()
)

Stateful processing with class_task

class_task loads the class once per worker — perfect for models, database connections, or any expensive initialization.

from olympict import ImagePipeline, Img
from olympict.files.o_image import OlympImage


class MyModel:
    def __init__(self, model_path: str):
        import torch

        self.model = torch.jit.load(model_path)

    def run(self, o: OlympImage) -> OlympImage:
        score = self.model(o.img)
        o.metadata["score"] = score.item()
        return o


(
    ImagePipeline.load_folder("./images")
    .class_task(MyModel, MyModel.run, class_args=["./model.pt"], count=2)
    .save_to_folder("./scored")
    .wait_for_completion()
)

Generate images from code

import numpy as np
from olympict import ImagePipeline
from olympict.files.o_image import OlympImage


def generator():
    for i in range(96):
        img = np.zeros((256, 256, 3), np.uint8)
        img[i % 256, :, :] = 255
        o = OlympImage()
        o.path = f"/tmp/{i}.png"
        o.img = img
        yield o


(
    ImagePipeline(generator())
    .task_img(lambda x: x[::-1, :, :])  # flip vertically
    .save_to_folder("/tmp/frames")
    .wait_for_completion()
)

BatchPipeline

Group images into numpy batches for vectorized GPU processing.

GPU batch inference with a custom model

from olympict import ImagePipeline
from olympict.files.o_batch import OlympBatch
import numpy as np


class MyBatchModel:
    def __init__(self, model_path: str):
        import torch

        self.model = torch.jit.load(model_path)

    def run_batch(self, batch: OlympBatch) -> OlympBatch:
        import torch

        tensor = torch.from_numpy(batch.data).float() / 255.0
        scores = self.model(tensor).detach().numpy()
        for i, score in enumerate(scores):
            batch.metadata[i]["score"] = float(score)
        return batch


(
    ImagePipeline.load_folder("./images")
    .resize((224, 224))
    .batch(32)
    .class_task(MyBatchModel, MyBatchModel.run_batch, class_args=["./model.pt"])
    .to_images()
    .save_to_folder("./out")
    .wait_for_completion()
)

VideoPipeline

Process every frame of a video

from olympict import VideoPipeline, ImagePipeline, Img


def sharpen(img: Img) -> Img:
    import cv2

    kernel = np.array([[0, -1, 0], [-1, 5, -1], [0, -1, 0]])
    return cv2.filter2D(img, -1, kernel)


(
    VideoPipeline.load_folder("./videos")
    .to_sequence()
    .task_img(sharpen)
    .save_to_folder("./frames")
    .wait_for_completion()
)

Extract frames and reassemble into a new video

from olympict import VideoPipeline
from olympict.files.o_image import OlympImage


def annotate(o: OlympImage) -> OlympImage:
    import cv2

    cv2.putText(
        o.img, "processed", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2
    )
    return o


(
    VideoPipeline.load_folder("./videos")
    .to_sequence()
    .task(annotate)
    .to_video(lambda o: o.path.replace("./videos", "./out"), fps=30)
    .wait_for_completion()
)

Hugging Face models — requires [hf]

pip install olympict[hf]

Important: always set the multiprocessing start method to "spawn" when using PyTorch models.

Image classification

from olympict import ImagePipeline
from olympict.files.o_image import OlympImage


def print_label(o: OlympImage) -> OlympImage:
    print(o.path, "→", o.metadata["label"])
    return o


if __name__ == "__main__":
    from torch.multiprocessing import set_start_method

    set_start_method("spawn")

    (
        ImagePipeline.load_folder("./images")
        .classify("google/mobilenet_v2_1.0_224")
        .task(print_label)
        .wait_for_completion()
    )

Object detection + draw predictions

from olympict import ImagePipeline

if __name__ == "__main__":
    from torch.multiprocessing import set_start_method

    set_start_method("spawn")

    (
        ImagePipeline.load_folder("./images")
        .detect("facebook/detr-resnet-50")  # stores results in metadata["pred_bboxes"]
        .draw_relative_bboxes()  # draws them on the image
        .save_to_folder("./annotated")
        .wait_for_completion()
    )

Batch object detection (GPU-efficient)

from olympict import ImagePipeline

if __name__ == "__main__":
    from torch.multiprocessing import set_start_method

    set_start_method("spawn")

    (
        ImagePipeline.load_folder("./images")
        .resize((800, 800))
        .batch(16)
        .detect(
            "facebook/detr-resnet-50"
        )  # batch inference, results in metadata[i]["pred_bboxes"]
        .to_images()
        .draw_relative_bboxes()
        .save_to_folder("./annotated")
        .wait_for_completion()
    )

HTTP server pipeline

OPipeline.server turns any pipeline into an HTTP endpoint. Requests flow through the pipeline as packets — the route handler converts the JSON body into whatever type the pipeline expects.

from socket import socket
from typing import Any, Dict, Tuple
from olympict import OPipeline, send_json_response


def parse(data: Dict[str, Any]) -> Dict[str, Any]:
    return data  # pass the JSON body as-is


(
    OPipeline.server([("POST", "/echo", parse)], port=8080)
    .task(lambda packet: (packet[0], {**packet[1], "echo": True}))
    .task(lambda packet: send_json_response(packet[0], packet[1]))
    .wait_for_completion()
)

The socket must be propagated through the pipeline and used at the end to send the response with send_json_response.

AI model HTTP server (via olympipe)

olympict is built on olympipe which exposes a Pipeline.server constructor. This turns a model into an HTTP endpoint — each request is processed through the pipeline, with class_task loading the model once per worker.

from socket import socket
from typing import Any, Dict, Tuple

import torch
import transformers
from olympipe import Pipeline
from olympipe.helpers.server import send_json_response


class LLM:
    def __init__(self, model_name: str):
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model = transformers.AutoModelForCausalLM.from_pretrained(model_name).to(
            self.device
        )
        self.tokenizer = transformers.AutoTokenizer.from_pretrained(model_name)

    def generate(
        self, packet: Tuple[socket, Dict[str, Any]]
    ) -> Tuple[socket, Dict[str, Any]]:
        connection, data = packet
        inputs = self.tokenizer.encode(data["prompt"], return_tensors="pt").to(
            self.device
        )
        with torch.no_grad():
            outputs = self.model.generate(
                inputs, max_new_tokens=data.get("max_new_tokens", 100)
            )
        responses = [
            self.tokenizer.decode(o, skip_special_tokens=True) for o in outputs
        ]
        return connection, {"responses": responses}


def respond(packet: Tuple[socket, Dict[str, Any]]):
    connection, body = packet
    send_json_response(connection, body)


if __name__ == "__main__":
    from torch.multiprocessing import set_start_method

    set_start_method("spawn")

    (
        Pipeline.server(
            [("POST", "/generate", lambda x: x)], port=8000, host="localhost"
        )
        .class_task(LLM, LLM.generate, ["mistralai/Mistral-7B-Instruct-v0.2"])
        .task(respond)
        .wait_for_completion()
    )

curl -X POST http://localhost:8000/generate \
     -H "Content-Type: application/json" \
     -d '{"prompt": "Explain multiprocessing in one sentence.", "max_new_tokens": 80}'

The server handles concurrency via the pipeline — you can add .class_task(..., count=2) to run two model workers in parallel.