Publish Video with BabyROS

SUMMARY

Publish Video with BabyROS streams live video using BabyROS as the communication layer. Two scripts share responsibility: a camera server that owns the hardware connection, and a client that controls streaming remotely via BabyROS topics. Optionally, an advanced client wraps capture, streaming, and parameter access behind a small helper class so the rest of your application talks to the camera entirely over BabyROS.

Available on: IDS, Webcam.

The Skill

The camera class registers BabyROS topics under a base path derived from its name:

medulla/v1/camera/<vendor>/<ClassName>/<name>

Clients control streaming by publishing to the start_video_stream and stop_video_stream topics, and receive frames by subscribing to video. Advanced clients can also call connect, capture, and parameter topics for full remote control.

Example

IDS

Webcam

The camera server (camera_loop.py) runs in Terminal 1 — it connects to the hardware and keeps BabyROS nodes alive. The client (publish_video_example.py) runs in Terminal 2 — it publishes a start signal, subscribes to the video topic to receive and log frames to Rerun, then publishes a stop signal after 10 seconds.

The Code

Camera Server (Terminal 1)

IDS

"""
Camera loop for the IDS camera.
"""
import time

from loguru import logger

from babyros import node
from telekinesis.medulla.cameras import ids


def main():
    camera = None
    try:
        camera = ids.IDS(name="my_ids_camera", serial_number="4108909352")
        camera.connect()
        logger.info("Camera Server is running... Press Ctrl+C to stop.")
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        logger.info("Shutting down server...")
    finally:
        if camera is not None:
            camera.disconnect()
            node.SessionManager.delete()
        logger.info("Completed cleanup.")


if __name__ == "__main__":
    main()

Webcam

"""
Camera loop for the Webcam.
"""
import time

from loguru import logger

from babyros import node
from telekinesis.medulla.cameras import webcam


def main():
    camera = None
    try:
        camera = webcam.Webcam(name="my_webcam_camera", camera_id=0)
        camera.connect()
        logger.info("Camera Server is running... Press Ctrl+C to stop.")
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        logger.info("Shutting down server...")
    finally:
        if camera is not None:
            camera.disconnect()
            node.SessionManager.delete()
        logger.info("Completed cleanup.")


if __name__ == "__main__":
    main()

BabyROS Client (Terminal 2)

IDS

"""
Stream live video from an IDS camera over BabyROS.
"""
import time

from loguru import logger
import rerun as rr
import numpy as np

from babyros import node


def log_video(data: np.ndarray) -> None:
    rr.log(
        "Continuous_Image_Capture",
        rr.EncodedImage(contents=data, media_type="image/jpeg"),
    )


def main():
    rr.init("IDS_Example", spawn=True)

    name = "my_ids_camera"
    base_topic = f"medulla/v1/camera/ids/IDS/{name}"

    start_video_stream_publisher = node.Publisher(topic=f"{base_topic}/start_video_stream")
    stop_video_stream_publisher = node.Publisher(topic=f"{base_topic}/stop_video_stream")
    video_subscriber = node.Subscriber(topic=f"{base_topic}/video", callback=log_video)

    try:
        start_video_stream_publisher.publish(data={})
        time.sleep(10)
        stop_video_stream_publisher.publish(data={})
    except KeyboardInterrupt:
        logger.info("Shutting down.")
    finally:
        start_video_stream_publisher.delete()
        stop_video_stream_publisher.delete()
        video_subscriber.delete()
        node.SessionManager.delete()
        logger.info("Completed cleanup.")


if __name__ == "__main__":
    main()

Webcam

"""
Stream live video from a webcam over BabyROS.
"""
import time

from loguru import logger
import rerun as rr
import numpy as np

from babyros import node


def log_video(image: np.ndarray) -> None:
    rr.log("Continuous_Image_Capture", rr.Image(image))


def main():
    rr.init("Webcam_Example", spawn=True)

    name = "my_webcam_camera"
    base_topic = f"medulla/v1/camera/webcam/{name}"

    start_video_stream_publisher = node.Publisher(topic=f"{base_topic}/start_video_stream")
    stop_video_stream_publisher = node.Publisher(topic=f"{base_topic}/stop_video_stream")
    video_subscriber = node.Subscriber(topic=f"{base_topic}/video", callback=log_video)

    try:
        start_video_stream_publisher.publish(data={})
        time.sleep(10)
        stop_video_stream_publisher.publish(data={})
    except KeyboardInterrupt:
        logger.info("Shutting down.")
    finally:
        start_video_stream_publisher.delete()
        stop_video_stream_publisher.delete()
        video_subscriber.delete()
        node.SessionManager.delete()
        logger.info("Completed cleanup.")


if __name__ == "__main__":
    main()

Running the Example

Start the server first, then the client. Both must run from a terminal.

Terminal 1:

python camera_loop.py

Wait for Camera Server is running... Press Ctrl+C to stop., then:

Terminal 2:

python publish_video_example.py

Frames stream under Continuous_Image_Capture for 10 seconds. Press Ctrl+C in Terminal 1 to shut down the server.

---

Advanced: Remote Camera Control with a Client Helper

The plain client only starts and stops streaming. To call connect, capture, get_parameter, and set_parameter remotely as well, wrap the BabyROS nodes in a small client-side helper class. The example below targets the IDS server but the same pattern applies to any camera that registers connect, capture, and parameter topics.

The camera server (camera_loop.py) must be running first — see the Camera Server section above.

"""
Capture a single image and stream live video over BabyROS via a client helper.
"""
from typing import Any
import time

from loguru import logger
import rerun as rr
import numpy as np

from babyros import node


class IDS:
    """Helper class for communicating with the IDS camera over BabyROS."""

    def __init__(self, name: str, serial_number: str) -> None:
        self._name = name
        self._serial_number = serial_number
        self._base_topic = f"medulla/v1/camera/ids/IDS/{self._name}"

        self._connection_client = node.Client(topic=f"{self._base_topic}/connect")
        self._capture_client = node.Client(topic=f"{self._base_topic}/capture")
        self._parameter_client = node.Client(topic=f"{self._base_topic}/parameter")
        self._start_video_stream_publisher = node.Publisher(topic=f"{self._base_topic}/start_video_stream")
        self._stop_video_stream_publisher = node.Publisher(topic=f"{self._base_topic}/stop_video_stream")

    def connect(self) -> dict:
        return self._connection_client.request(data={"connect": True})

    def disconnect(self) -> dict:
        return self._connection_client.request(data={"connect": False})

    def capture(self) -> np.ndarray:
        image = self._capture_client.request()
        if isinstance(image, list) and len(image) > 0:
            image = image[0]
        if image is None:
            raise RuntimeError("No image received from camera server.")
        return np.array(image)

    def start_video_stream(self) -> None:
        self._start_video_stream_publisher.publish(data={})

    def stop_video_stream(self) -> None:
        self._stop_video_stream_publisher.publish(data={})

    def get_parameter(self, name: str) -> Any:
        response = self._parameter_client.request(data={"mode": "get", "name": name})
        return response[0]["value"]

    def set_parameter(self, name: str, value: Any) -> Any:
        response = self._parameter_client.request(data={"mode": "set", "name": name, "value": value})
        return response[0]["value"]

    def delete(self) -> None:
        self._connection_client.delete()
        self._capture_client.delete()
        self._parameter_client.delete()
        self._start_video_stream_publisher.delete()
        self._stop_video_stream_publisher.delete()


def log_image(data: np.ndarray) -> None:
    rr.log("Single_Image_Capture", rr.Image(data))


def log_video(data: np.ndarray) -> None:
    rr.log("Continuous_Image_Capture", rr.EncodedImage(contents=data, media_type="image/jpeg"))


def main():
    rr.init("IDS_Example", spawn=True)

    camera = None
    video_subscriber = None
    try:
        name = "my_ids_camera"
        camera = IDS(name=name, serial_number="4108909352")
        base_topic = f"medulla/v1/camera/ids/IDS/{name}"

        video_subscriber = node.Subscriber(topic=f"{base_topic}/video", callback=log_video)

        logger.debug(f"AcquisitionFrameRate: {camera.get_parameter('AcquisitionFrameRate')}")
        logger.debug(f"ExposureTime: {camera.get_parameter('ExposureTime')}")
        logger.debug(f"DeviceLinkThroughputLimit: {camera.get_parameter('DeviceLinkThroughputLimit')}")

        camera.set_parameter('ExposureTime', 35000.0)
        image = camera.capture()
        log_image(image)

        camera.set_parameter('AcquisitionFrameRate', 12.0)
        camera.set_parameter('ExposureTime', 60000.0)
        camera.start_video_stream()
        time.sleep(5)
        camera.stop_video_stream()
    except KeyboardInterrupt:
        logger.info("Shutting down.")
    finally:
        if camera is not None:
            camera.delete()
        if video_subscriber is not None:
            video_subscriber.delete()
        node.SessionManager.delete()
        logger.info("Cleanup complete.")


if __name__ == "__main__":
    main()

The constructor builds the base topic path from name — this must match exactly the name given to the server-side camera instance in camera_loop.py. It then creates a node.Client for each request-response interaction (/connect, /capture, /parameter) and a node.Publisher for the fire-and-forget start_video_stream / stop_video_stream triggers.

get_parameter and set_parameter send a structured request to the /parameter topic, carrying a mode field ("get" or "set") along with the parameter name and, for writes, the value. The response carries the confirmed value back. capture calls the /capture server, unwraps the first element of the response list (the server always returns a list), and converts it to a NumPy array. If the server returns nothing, a RuntimeError is raised immediately rather than silently returning an invalid frame.

get_parameter, set_parameter, and capture can be freely interleaved with streaming — all mediated by the server over BabyROS. The video_subscriber is created before acquisition starts to ensure no frames are missed in the window between start_video_stream and the first frame arriving.

Run

Start camera_loop.py in Terminal 1 first, then:

python advanced_example.py

A Rerun viewer opens showing the captured frame under Single_Image_Capture, followed by 5 seconds of video under Continuous_Image_Capture.

Parameter Configuration

Parameter	Type	Default	Description
name	str	required	Camera name given to the server-side camera instance. The client must reuse it unchanged to derive the same base topic (medulla/v1/camera/webcam/<name> or medulla/v1/camera/ids/IDS/<name>).
start_video_stream / stop_video_stream payload	dict	{}	No fields are read from the payload; publishing any message, including an empty dict, triggers the action on the server.
parameter request mode	str	required	"get" or "set" — selects whether the request reads or writes the named device parameter.
parameter request name	str	required	The device parameter name, e.g. "ExposureTime", "AcquisitionFrameRate".
parameter request value	Any	required for "set"	The value to write. Ignored for "get" requests.

Returns

Type	Description
np.ndarray	Frame delivered to the video subscriber's callback — JPEG-encoded bytes for IDS, a raw RGB array for Webcam.
dict	connect() / disconnect() on the client helper return the server's {"status": bool} response.
np.ndarray	capture() on the client helper returns the captured RGB frame.
Any	get_parameter() / set_parameter() on the client helper return the confirmed parameter value.
None	start_video_stream() / stop_video_stream() — fire-and-forget triggers; nothing is returned.

Raises

Exception	Condition
RuntimeError	capture() on the client helper — the server returned no image for the capture request.
RuntimeError	Webcam server only — start_video_stream() could not read a frame mid-stream.
RuntimeError	IDS server parameter requests — the camera is not connected. Caught and logged server-side; not propagated back to the client.
KeyError / ValueError	IDS server parameter requests — unknown parameter name, or an unsupported parameter type, respectively.

Where to Use the Skill

• Distributed pipelines — keep the camera process pinned to the machine with USB access while running vision workloads on a different machine.
• Multi-consumer streaming — multiple clients can subscribe to the same video topic without duplicating hardware connections.
• Remote tuning — adjust exposure or frame rate from a control script without restarting the camera server.