DetectionLogger

DetectionLogger writes image frames and their bounding-box annotations to a YOLO or RF-DETR/COCO dataset on disk. Frames are split automatically across train, val, and test according to configurable ratios.

Import

from telekinesis.dataengine import DetectionLogger

Creating a Logger

Use the DetectionLogger.create() factory to get a format-specific logger.

logger = DetectionLogger.create(
    dataset_format,     # "yolo" | "rfdetr"
    dataset_dir,        # output directory
    categories=None,    # see Categories below
    *,
    task="detect",      # "detect" | "segment"
    mode="create",      # "create" | "overwrite" | "append"
    split_ratios=None,  # see Split Ratios below
    color_order="rgb",  # "rgb" | "bgr"
    image_ext=".jpg",
)

Parameters

Parameter	Type	Description
dataset_format	str	"yolo" or "rfdetr".
dataset_dir	str | Path	Root directory for the dataset. Created if it does not exist.
categories	list[dict] | None	Category definitions (see below). None enables dynamic registration.
task	str	"detect" for bounding boxes, "segment" for instance segmentation masks.
mode	str	How to handle an existing dataset at dataset_dir (see Modes).
split_ratios	dict | None	Per-split fractions, e.g. {"train": 0.8, "val": 0.1, "test": 0.1}. Default 80/10/10.
color_order	str	Channel order of input numpy arrays. Converted to RGB before writing.
image_ext	str	Output image extension. Default .jpg.

Categories

Pass a list of COCO-style category dicts:

categories = [
    {"id": 1, "name": "box",    "supercategory": "object"},
    {"id": 2, "name": "carton", "supercategory": "object"},
]

Passing categories=None enables dynamic registration: unknown category ids are registered automatically on first use. Useful when the full class list is not known upfront.

Modes

Mode	Behavior
"create"	Fails if dataset_dir already contains dataset files. Safe default.
"overwrite"	Deletes existing dataset files in dataset_dir and starts fresh.
"append"	Resumes an existing dataset, continuing image and annotation counters.

Split Ratios

Splits are assigned using a largest-deficit algorithm so that the running per-split proportions track the configured ratios at every step — no batching required.

# Custom split — e.g. 90% train, 5% val, 5% test
logger = DetectionLogger.create(
    "yolo", "results/dataset", categories,
    split_ratios={"train": 0.9, "val": 0.05, "test": 0.05}
)

To assign a split explicitly when logging, pass split= to log().

Logging Frames

logger.log(
    image,              # np.ndarray | Image datatype
    annotations=None,   # list[dict] | ObjectDetectionAnnotations | None
    split=None,         # "train" | "val" | "test" | None (auto-assigned)
    file_name=None,     # optional filename stem override
)

Annotation Format

Annotations are COCO-style dicts, one per object:

annotations = [
    {
        "id": 1,
        "image_id": 0,          # filled automatically if omitted
        "category_id": 1,
        "bbox": [x, y, w, h],   # absolute pixel coordinates
        "area": w * h,
        "iscrowd": 0,
    },
    ...
]

The logger also accepts ObjectDetectionAnnotations datatypes directly, which is the native output format of Telekinesis detection Skills.

Closing

Always call close() when done. It flushes the final dataset manifest (data.yaml for YOLO, _annotations.coco.json for RF-DETR).

logger.close()

The logger is also a context manager:

with DetectionLogger.create("yolo", "results/dataset", categories) as logger:
    for image, annotations in detections:
        logger.log(image, annotations)
# close() called automatically

Example

import numpy as np
from telekinesis.dataengine import DetectionLogger

# ------------------------------------------------
# 1. Define categories and create logger
# ------------------------------------------------
categories = [
    {"id": 1, "name": "box",    "supercategory": "object"},
    {"id": 2, "name": "carton", "supercategory": "object"},
]

logger = DetectionLogger.create(
    "yolo",
    "results/warehouse_dataset",
    categories,
    mode="create",
    split_ratios={"train": 0.8, "val": 0.1, "test": 0.1},
)

# ------------------------------------------------
# 2. Simulate detections and log frames
# ------------------------------------------------
for i in range(20):
    image = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
    annotations = [
        {
            "id": i * 2,
            "category_id": 1,
            "bbox": [100, 80, 200, 150],
            "area": 200 * 150,
            "iscrowd": 0,
        }
    ]
    logger.log(image, annotations)

# ------------------------------------------------
# 3. Close — writes data.yaml
# ------------------------------------------------
logger.close()

Output Layout

YOLO

results/warehouse_dataset/
├── images/
│   ├── train/   frame_0000.jpg ...
│   ├── val/     frame_0016.jpg ...
│   └── test/    frame_0018.jpg ...
├── labels/
│   ├── train/   frame_0000.txt ...
│   ├── val/     frame_0016.txt ...
│   └── test/    frame_0018.txt ...
└── data.yaml

RF-DETR / COCO

results/warehouse_dataset/
├── train/
│   ├── frame_0000.jpg ...
│   └── _annotations.coco.json
├── valid/
│   ├── frame_0016.jpg ...
│   └── _annotations.coco.json
└── test/
    ├── frame_0018.jpg ...
    └── _annotations.coco.json

Publisher / Subscriber

Decouple disk writes from detection so I/O never stalls the capture loop.

Read more →