Video Stream API

Media clients (compatibility facade).

The implementation lives in frame (frame types and pixel-format helpers), encoded (encoded-stream UDS client) and fd_client (zero-copy DMA-BUF fd client). This module re-exports the full historical surface — including the private wire-protocol names tests and internal callers patch — so from .media import X keeps working unchanged.

FdMediaClient

class neoruntime_ipc_sdk.FdMediaClient(socket_path=None)

Bases: object

Zero-copy media client using DMA-BUF FD passing over Unix Domain Socket.

__init__(socket_path=None)

get_frame(stream_id, timeout_ms=5000, *, keep_fd=False)

Receive one frame.

With keep_fd=True the frame’s dma-buf fds are retained (zero-copy handoff; see FrameHandle) instead of copied, and the daemon-side buffer release is deferred until frame.release() / GC / client close.

subscribe_raw(stream_id, skip_frames=True, keep_fd=False)

subscribe(stream_id, skip_frames=True, keep_fd=False)

on_frame(stream_id, callback)

close()

get_encoded_stream(stream_id='main', socket_dir=None)

Return an EncodedStreamClient for the given encoded stream.

Parameters:

• stream_id (str) – Stream name (e.g. "main", "sub").

• socket_dir (str | None) – Directory containing EncodedPublisher UDS sockets (default /run/aipc/encoded, or ENCODED_SOCK_DIR).

Returns:

A connected EncodedStreamClient reading from {socket_dir}/{stream_id}.sock.

Return type:

EncodedStreamClient

list_streams()

List available raw stream IDs by scanning the camera socket.

Returns common stream IDs. For detailed status use CameraClient.get_stream_status.

get_rtsp_url(stream_id='main', host='192.0.2.72', port=8554)

Return an RTSP URL for the given stream.

Note: RTSP must be enabled on the device first (via CameraClient or REST API).

Data Types

Frame

class neoruntime_ipc_sdk.Frame(sequence: 'int', timestamp_ns: 'int', width: 'int', height: 'int', format: 'str', image: 'np.ndarray | None', metadata: 'dict[str, Any]'=<factory>, handle: 'FrameHandle | None' = None)

sequence: int

timestamp_ns: int

width: int

height: int

format: str

image: ndarray | None

metadata: dict[str, Any]

handle: FrameHandle | None = None

property data: ndarray | None

Alias for image, returns raw frame data as flat numpy array.

to_array()

Return the raw pixel buffer, materializing a retained fd.

For keep-fd frames this maps the dma-buf planes once (with the DMA_BUF_IOCTL_SYNC read fences) and caches the copy in image; later calls return the cache without re-mapping.

release()

Release a retained fd frame back to the daemon (idempotent).

No-op for frames that were copied on receive.

to_rgb()

crop(x, y, width, height)

Return a new Frame cropped to the given pixel rectangle.

NV12/NV21 require even x, y, width, height (chroma subsampling). The original Frame is left untouched.

resize(width, height, mode='letterbox', pad_value=114)

Return a new Frame resized to width x height.

Modes:

• “letterbox”: fit inside, preserve aspect ratio, pad with pad_value (NV12 pads luma with pad_value and chroma with neutral 128). Default.

• “stretch”: fill exactly, aspect ratio not preserved.

• “crop”: scale to cover, center-crop the overflow.

NV12/NV21 require even target dimensions. Frames received with keep_fd=True are scaled on the DSP without materializing their dma-bufs first (falling back to the CPU path when the DSP service is unavailable). The fast path respects the accel router’s policy: SOFTWARE_ONLY skips the DSP attempt, HARDWARE_ONLY raises on a failed attempt instead of silently degrading, and the default PREFER_HARDWARE records the fallback in the router’s health counters. cv2 accelerates the CPU path when available; a pure-numpy nearest-neighbour path is the fallback.

to_jpeg_bytes(quality=85)

Encode the frame as JPEG bytes, hardware-first for keep-fd frames.

Frames with a live dma-buf handle ride the accel router’s zero-copy EncodeImage leg (the daemon imports the buffer — no RGB materialization, no read-back) and degrade to the CPU cv2/Pillow encode when the daemon cannot serve them. In-memory frames stay on the CPU encode outright: the daemon encoder is N-threaded libjpeg behind an RPC, so its win is the zero-copy import, not raw speed — tight loops (e.g. MjpegStream.push_frame) must not pay pool-alloc + copy + RPC per frame (S-3 record).

save(path)

__init__(sequence, timestamp_ns, width, height, format, image, metadata=<factory>, handle=None)

FrameHandle

class neoruntime_ipc_sdk.FrameHandle(fds, strides, plane_sizes, frame_id, on_release=None, width=0, height=0, format='')

Retained dma-buf backing store for one received frame (SDK-1).

In keep-fd mode the per-plane dma-buf fds the daemon passed with the FRAME message are kept open here and the RELEASE message — the daemon’s buffer-recycling ticket — is deferred until close(). CPU access to the pixels must go through Frame.to_array(), which applies the required DMA_BUF_IOCTL_SYNC read fences. The fds can also be handed to a DspClient job as-is for a zero-copy hardware path (resize_hw(frame, ...)); the geometry carried here is what the daemon needs to import them.

__init__(fds, strides, plane_sizes, frame_id, on_release=None, width=0, height=0, format='')

property closed: bool

True once close() ran (fds gone, frame released).

property fd: int

Convenience accessor for the first plane’s fd.

close()

Close the fds and send the deferred RELEASE (idempotent).

StreamInfo

class neoruntime_ipc_sdk.StreamInfo(stream_id: 'str', width: 'int', height: 'int', format: 'str', fps: 'float', buffer_count: 'int')

stream_id: str

width: int

height: int

format: str

fps: float

buffer_count: int

__init__(stream_id, width, height, format, fps, buffer_count)

PixelFormat

class neoruntime_ipc_sdk.PixelFormat(value)

NV12 = 0

NV21 = 1

RGB = 2

BGR = 3

RGBA = 4

BGRA = 5

GRAY8 = 6

YUYV = 7

EncodedStreamClient

class neoruntime_ipc_sdk.EncodedStreamClient(socket_path=None, *, stream_id='main', socket_dir=None)

Read encoded video frames from an EncodedPublisher UDS socket.

Connects to sockets like /run/aipc/encoded/main.sock and yields EncodedFrame objects containing H.264/H.265 NAL units.

Usage:

client = EncodedStreamClient()                    # main stream
client = EncodedStreamClient(stream_id="sub")     # sub stream
client = EncodedStreamClient("/run/aipc/encoded/main.sock")  # explicit
for frame in client.subscribe():
    print(f"{frame.codec_name} {frame.width}x{frame.height} "
          f"keyframe={frame.is_keyframe} {len(frame.data)}B")

__init__(socket_path=None, *, stream_id='main', socket_dir=None)

Resolve the socket path.

socket_path (explicit) wins; otherwise the path is derived as {socket_dir}/{stream_id}.sock with socket_dir defaulting to /run/aipc/encoded (overridable via ENCODED_SOCK_DIR).

EncodedFrame

class neoruntime_ipc_sdk.EncodedFrame(codec, flags, pts_ns, width, height, dts_ns, data)

Encoded video frame (H.264/H.265) from the EncodedPublisher.

codec: int

flags: int

pts_ns: int

width: int

height: int

dts_ns: int

data: bytes

property is_keyframe: bool

property codec_name: str

__init__(codec, flags, pts_ns, width, height, dts_ns, data)

Usage Examples

Getting Raw Video Stream

from neoruntime_ipc_sdk import FdMediaClient
import cv2

media = FdMediaClient()

# Get main stream (available IDs come from list_streams(), usually main / sub)
for frame in media.subscribe("main"):
    print(f"Frame {frame.sequence}: {frame.width}x{frame.height}")
    print(f"Format: {frame.format}, Timestamp: {frame.timestamp_ns}")

    # frame.image is a numpy array (H, W, C) or (H*3//2, W) for NV12
    # Can be used directly with OpenCV or other image processing libraries
    cv2.imshow("Camera", frame.image)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cv2.destroyAllWindows()

No Frame Skipping

# Get every frame (no skipping)
for frame in media.subscribe("main", skip_frames=False):
    process_frame(frame.image)

Getting a Single Frame

# Get a single frame
frame = media.get_frame("main", timeout_ms=1000)

if frame:
    print(f"Frame: {frame.width}x{frame.height}")
    print(f"Format: {frame.format}")

Getting Stream Info

# FdMediaClient has no separate stream-info API;
# read dimensions and format from the first frame
frame = media.get_frame("main", timeout_ms=5000)

if frame:
    print(f"Resolution: {frame.width}x{frame.height}")
    print(f"Format: {frame.format}")
    print(f"Available streams: {media.list_streams()}")

Listing Available Streams

# List all available streams
streams = media.list_streams()
for stream_id in streams:
    print(f"Stream: {stream_id}")

Frame Callback

def handle_frame(frame):
    print(f"Received frame: {frame.sequence}")

# Use callback for frame processing
thread = media.on_frame("main", handle_frame)

# Keep running
import time
while True:
    time.sleep(1)

Image Processing

import cv2

media = FdMediaClient()

for frame in media.subscribe("main"):
    # Convert to RGB
    rgb = frame.to_rgb()

    # Grayscale
    gray = cv2.cvtColor(rgb, cv2.COLOR_RGB2GRAY)

    # Edge detection
    edges = cv2.Canny(gray, 100, 200)

    # Display results
    cv2.imshow("Original", rgb)
    cv2.imshow("Edges", edges)

    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

Saving Images

media = FdMediaClient()

for frame in media.subscribe("main"):
    # Save as image
    frame.save("frame.jpg")
    break

Getting Encoded Stream

get_encoded_stream() returns an EncodedStreamClient (not an iterator) yielding H.264/H.265 Annex-B packets, which the recording module can write to disk directly:

from neoruntime_ipc_sdk import FdMediaClient

media = FdMediaClient()

# get_encoded_stream() returns an EncodedStreamClient, not an iterator
client = media.get_encoded_stream("main")

for packet in client.subscribe():
    print(f"{packet.codec_name()} packet: {len(packet.data)} bytes")
    if packet.is_keyframe():
        print("  keyframe")

Multi-Stream Processing

import threading

def process_main_stream():
    media = FdMediaClient()
    for frame in media.subscribe("main"):
        # Process main stream (high resolution)
        process_high_res(frame.image)

def process_sub_stream():
    media = FdMediaClient()
    for frame in media.subscribe("sub"):
        # Process sub stream (low resolution)
        process_low_res(frame.image)

# Process multiple streams in parallel
t1 = threading.Thread(target=process_main_stream)
t2 = threading.Thread(target=process_sub_stream)

t1.start()
t2.start()

t1.join()
t2.join()

Frame Rate Control

import time

media = FdMediaClient()
target_fps = 10
frame_interval = 1.0 / target_fps

last_time = time.time()

for frame in media.subscribe("main"):
    current_time = time.time()
    elapsed = current_time - last_time

    if elapsed >= frame_interval:
        # Process frame
        process_frame(frame.image)
        last_time = current_time

Saving Video

To save the H.264/H.265 encoded stream, prefer reusing it directly with the recording module (TsWriter / HlsWriter) — no decode/re-encode needed. The example below decodes frames and saves them with OpenCV:

import cv2

media = FdMediaClient()

# Read the resolution from the first frame
# (FdMediaClient has no separate stream-info API)
first = media.get_frame("main", timeout_ms=5000)
if first is None:
    raise RuntimeError("no frame received")

# Create video writer (fill in the fps of your actual stream, 30 here)
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter(
    'output.mp4',
    fourcc,
    30.0,
    (first.width, first.height)
)

frame_count = 0
max_frames = 300  # Record 10 seconds (30fps)

for frame in media.subscribe("main"):
    rgb = frame.to_rgb()
    bgr = rgb[:, :, ::-1]  # RGB to BGR
    out.write(bgr)
    frame_count += 1

    if frame_count >= max_frames:
        break

out.release()
print(f"Saved {frame_count} frames to output.mp4")

Context Manager

# Use context manager for automatic resource management
with FdMediaClient() as media:
    for frame in media.subscribe("main"):
        process_frame(frame.image)

Zero-Copy Access

# The default receive path copies pixel data on arrival
# (frame.image is immediately usable). keep_fd=True retains the
# dma-buf fds for true zero-copy: the buffer is only returned to
# the daemon on frame.release() / GC / client close

media = FdMediaClient()

for frame in media.subscribe("main", keep_fd=True):
    # frame.handle holds the dma-buf fds
    # frame.to_array() maps them once on first call and caches
    result = inference_engine.process(frame.to_array())

    # Return the buffer as early as possible (idempotent, optional)
    frame.release()

Error Handling

media = FdMediaClient()

try:
    for frame in media.subscribe("invalid_stream"):
        process_frame(frame.image)
except Exception as e:
    print(f"Stream access failed: {e}")
except KeyboardInterrupt:
    print("User interrupted")
finally:
    media.close()