Publisher / Subscriber

SUMMARY

Run detection and disk I/O in separate processes using DetectionLoggerPublisher and DetectionLoggerSubscriber. The publisher sends frames over a Zenoh topic; the subscriber receives them and writes to disk — so file I/O never blocks the capture loop.

When to Use This

The inline DetectionLogger works when detection and disk writes happen in the same process. The pub/sub pattern is better when:

• Detection runs at high frequency and disk I/O would introduce latency.
• Detection and storage run on different machines on the same Zenoh network.
• You want to replay or redirect captured frames to multiple loggers.

Import

from telekinesis.dataengine import DetectionLoggerPublisher, DetectionLoggerSubscriber

DetectionLoggerPublisher

Sends frames and annotations over a Zenoh topic.

publisher = DetectionLoggerPublisher(
    topic,                  # Zenoh topic string
    compression="lz4",      # "lz4" | "zstd" | None
    color_order="rgb",      # "rgb" | "bgr"
)

Parameters

Parameter	Type	Description
topic	str	Zenoh topic to publish on. Must match the subscriber's topic.
compression	str | None	Payload compression. "lz4" (default) or "zstd". None disables compression.
color_order	str	Channel order of input numpy arrays.

publish()

publisher.publish(
    image,              # np.ndarray | Image datatype
    annotations=None,   # list[dict] | ObjectDetectionAnnotations | None
    split=None,         # "train" | "val" | "test" | None
    file_name=None,     # optional filename stem
)

Usage

with DetectionLoggerPublisher(topic="detection_logger/frames") as pub:
    for image, annotations in detections:
        pub.publish(image, annotations)

DetectionLoggerSubscriber

Receives frames over Zenoh and writes them to a DetectionLogger.

subscriber = DetectionLoggerSubscriber(
    topic,              # Zenoh topic to subscribe to
    logger,             # a DetectionLogger instance
    queue_depth=32,     # max frames buffered in memory
)

Parameters

Parameter	Type	Description
topic	str	Zenoh topic to subscribe to.
logger	DetectionLogger	The logger that writes frames to disk.
queue_depth	int	Internal queue size. Frames are dropped with a warning if the queue is full.

Properties

Property	Type	Description
written	int	Number of frames successfully written to disk.
dropped	int	Number of frames dropped due to a full queue.

Usage

from telekinesis.dataengine import DetectionLogger, DetectionLoggerSubscriber

logger = DetectionLogger.create("rfdetr", "results/dataset", categories)

with DetectionLoggerSubscriber(topic="detection_logger/frames", logger=logger) as sub:
    input("Recording... press Enter to stop.")

print(f"Written: {sub.written}, Dropped: {sub.dropped}")

Full Example

Start the subscriber first, then run the publisher in a separate process or terminal.

Process A — Subscriber (start first)

from telekinesis.dataengine import DetectionLogger, DetectionLoggerSubscriber

# ------------------------------------------------
# 1. Create the logger and start the subscriber
# ------------------------------------------------
categories = [{"id": 1, "name": "box", "supercategory": "object"}]
logger = DetectionLogger.create("rfdetr", "results/subscriber_dataset", categories)

with DetectionLoggerSubscriber(topic="detection_logger/frames", logger=logger) as sub:
    input("Subscriber running — press Enter to stop.")

# ------------------------------------------------
# 2. Report results after stopping
# ------------------------------------------------
print(f"Written: {sub.written} frames, Dropped: {sub.dropped}")

Process B — Publisher

import numpy as np
from telekinesis.dataengine import DetectionLoggerPublisher

# ------------------------------------------------
# 1. Publish frames over Zenoh
# ------------------------------------------------
with DetectionLoggerPublisher(topic="detection_logger/frames") as pub:
    for i in range(50):
        image = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
        annotations = [
            {
                "id": i,
                "category_id": 1,
                "bbox": [50, 40, 200, 150],
                "area": 200 * 150,
                "iscrowd": 0,
            }
        ]
        pub.publish(image, annotations, file_name=f"frame_{i:04d}")