Telekinesis Agentic Skill Library

Appearance

Stream Live Color Video ​

SUMMARY

Stream Live Color Video acquires continuous color frames in a loop, returning each frame as a NumPy array ready to log, encode, or process. The camera stays in continuous acquisition mode between calls, so frame rate is bounded only by the device and any optional encoding step.

Available on: IDS, Webcam, Intel RealSense.

The Skill ​

python
image = camera.capture_video_color_frame()

Call repeatedly inside a loop. The first call switches the camera to continuous acquisition; subsequent calls return the next frame.

Example ​

IDS

Webcam

Intel RealSense

The Code ​

python
"""
Stream live color video from an IDS camera, with parameter tuning and JPEG
encoding to reduce bandwidth before logging to Rerun.

Run from a terminal to avoid issues with Rerun's spawn mode.
"""
import time
from loguru import logger

import cv2
import rerun as rr

from telekinesis.medulla.cameras import ids


def main():
    camera = ids.IDS(
        name="my_ids_camera",
        serial_number="4108909352",
        load_factory_defaults=False,
    )
    try:
        rr.init("IDS_Example", spawn=True)
        camera.connect()

        camera.set_parameter('ExposureTime', 35000.0)
        camera.set_parameter('AcquisitionFrameRate', 14.0)

        end_time = time.time() + 10
        while time.time() < end_time:
            image = camera.capture_video_color_frame()
            if image is not None:
                _, encoded = cv2.imencode('.jpg', image, [cv2.IMWRITE_JPEG_QUALITY, 80])
                rr.log(
                    "Continuous_Image_Capture",
                    rr.EncodedImage(contents=encoded, media_type="image/jpeg"),
                )
    except Exception as e:
        logger.error(f"Unable to run video capture. {type(e).__name__}: {e}")
    finally:
        camera.disconnect()


if __name__ == "__main__":
    main()
python
"""
Stream live color video from a webcam, logging each frame to Rerun.

Run from a terminal to avoid issues with Rerun's spawn mode.
"""
import time
from loguru import logger

import rerun as rr

from telekinesis.medulla.cameras import webcam


def main():
    camera = webcam.Webcam(name="my_webcam", camera_id=0)
    try:
        rr.init("Webcam_Example", spawn=True)
        camera.connect()

        end_time = time.time() + 10
        while time.time() < end_time:
            image = camera.capture_video_color_frame()
            rr.log("Continuous_Image_Capture", rr.Image(image))
    except Exception as e:
        logger.error(f"Unable to stream video. {type(e).__name__}: {e}")
    finally:
        camera.disconnect()


if __name__ == "__main__":
    main()
python
"""
Stream live color video from a RealSense camera, logging each frame to Rerun.

Run from a terminal to avoid issues with Rerun's spawn mode.
"""
import time
from loguru import logger

import rerun as rr

from telekinesis.medulla.cameras import realsense


def main():
    camera = realsense.RealSense(name="my_realsense")
    try:
        rr.init("RealSense_Video_Example", spawn=True)
        camera.connect()

        end_time = time.time() + 10
        while time.time() < end_time:
            image = camera.capture_video_color_frame()
            if image is None:
                logger.warning("Image is none")
                continue
            rr.log("Continuous_Image_Capture", rr.Image(image))
    except Exception as e:
        logger.error(f"Unable to stream video. {type(e).__name__}: {e}")
    finally:
        camera.disconnect()


if __name__ == "__main__":
    main()

The Explanation of the Code ​

The streaming loop runs for 10 seconds using a wall-clock deadline rather than a fixed frame counter, so the actual number of frames captured depends on the camera's effective frame rate.

python
end_time = time.time() + 10
while time.time() < end_time:
    image = camera.capture_video_color_frame()
    ...

Each call to capture_video_color_frame returns one frame as a NumPy array and keeps the camera in continuous acquisition mode between calls — the sensor is not stopped and re-armed per frame, which is what makes streaming efficient.

Per-camera behavior:

  • IDS — the first call switches the camera to Continuous acquisition mode. Parameters such as AcquisitionFrameRate and ExposureTime should be tuned with set_parameter before the loop starts. JPEG-encoding each frame with cv2.imencode at quality 80 dramatically reduces the data volume before logging, and Rerun's EncodedImage accepts the compressed bytes directly without an intermediate decode step.
  • Webcam — reads each frame from the OpenCV VideoCapture device in BGR order and converts to RGB before returning.
  • RealSense — keeps the colour stream running continuously and returns one frame per call, or None if no frame is ready yet. The None guard skips logging until a valid frame arrives rather than passing invalid data to Rerun.

The disconnect call lives in a finally block so the device is released regardless of how the loop exits.

Running the Example ​

bash
python capture_video_example.py

A Rerun viewer opens automatically. Frames stream under Continuous_Image_Capture for 10 seconds, then the camera disconnects cleanly.

Where to Use the Skill ​

  • Live monitoring — display continuous video for inspection, calibration, or robot teleoperation.
  • Vision pipelines that consume video — feed downstream modules such as Cornea with a live feed.
  • Local debugging — when remote streaming over BabyROS isn't needed; for that, see Publish Video with BabyROS.