Metadata-Version: 2.5
Name: olympict
Version: 2.0.0
Summary: A powerful parallel pipelining tool for image processing
Project-URL: Homepage, https://gitlab.com/gabraken/olympict
Project-URL: Repository, https://gitlab.com/gabraken/olympict
Project-URL: Documentation, https://gitlab.com/gabraken/olympict/-/blob/master/README.md
Author-email: Gabriel Kasser <gabriel.kasser@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: image,multiprocessing,parallel,pipeline
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Python: >=3.10
Requires-Dist: click>=8.0.0
Requires-Dist: natsort>=8.3.1
Requires-Dist: numpy>=1.26
Requires-Dist: olympipe<3,>=2.2
Requires-Dist: opencv-python>=4.5
Provides-Extra: hf
Requires-Dist: huggingface-hub>=0.26; extra == 'hf'
Requires-Dist: pillow>=10.4; extra == 'hf'
Requires-Dist: timm>=1.0; extra == 'hf'
Requires-Dist: torch>=2.4; extra == 'hf'
Requires-Dist: transformers>=4.46; extra == 'hf'
Description-Content-Type: text/markdown

# Olympict

![coverage](https://gitlab.com/gabraken/olympict/badges/master/coverage.svg?job=coverage)![status](https://gitlab.com/gabraken/olympict/badges/master/pipeline.svg)

![Olympict](https://gitlab.com/gabraken/olympict/-/raw/master/Olympict.png)

**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](https://pypi.org/project/olympipe/) `>=2.2`. Requires Python `>=3.10` (tested on 3.10, 3.12, 3.13 and 3.14).

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

Development:

```bash
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

```bash
# 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.

```bash
# 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.

```bash
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](https://pypi.org/project/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.

```python
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.

```python
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()
```

```python
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:

```python
from olympict import ImagePipeline

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

---

## ImagePipeline

### Resize and save to a new folder

```python
from olympict import ImagePipeline

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

### Resize preserving aspect ratio (letterbox padding)

```python
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

```python
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

```python
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

```python
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

```python
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.

```python
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

```python
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.

```python
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.

```python
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.

```python
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

```python
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

```python
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

```python
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

```python
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]`

```bash
pip install olympict[hf]
```

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

### Image classification

```python
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

```python
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)

```python
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.

```python
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](https://pypi.org/project/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.

```python
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()
    )
```

```bash
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.
