视频流 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)[source]

Bases: object

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

__init__(socket_path=None)[source]

property retained_frames: int: Number of keep-fd frames currently held (not yet released).

get_frame(stream_id, timeout_ms=5000, *, keep_fd=False)[source]

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=1, keep_fd=False)[source]

Yield frames from stream_id, auto-reconnecting on socket errors.

skip_frames is a decimation interval: 1 (the default, and the effective value of the historical True/False flags) yields every frame; N > 1 yields every Nth frame. Decimated keep-fd frames are released back to the daemon immediately so their dma-bufs are not retained.

subscribe(stream_id, skip_frames=1, keep_fd=False)[source]

Yield frames; see subscribe_raw() for the parameters.

on_frame(stream_id, callback)[source]

close()[source]

get_encoded_stream(stream_id='main', socket_dir=None)[source]

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()[source]

List available raw stream IDs by querying the camera daemon.

Streams whose pipeline is not active are excluded. Falls back to the platform’s guaranteed streams (main/sub) when the camera control service is unreachable, unresponsive (1s probe deadline) or reports nothing, so data-plane-only setups keep working. Results are cached for _LIST_STREAM_CACHE_S seconds. For detailed status (resolution, fps, codec) use CameraClient.get_stream_status.

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

Return an RTSP URL for the given stream.

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

数据类型

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, flags: 'int' = 0)[source]

sequence: int

timestamp_ns: int

width: int

height: int

format: str

image: ndarray | None

metadata: dict[str, Any]

handle: FrameHandle | None = None

flags: int = 0

property baked_overlay: bool: True when the AI-overlay bake pass is active on this stream.

property baked_dpm: bool: True when the DPM (privacy mask) render pass is active.

property data: ndarray | None: Alias for image, returns raw frame data as flat numpy array.

to_array()[source]

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()[source]

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

No-op for frames that were copied on receive. Also invoked by with frame: on exit — the context manager makes the keep-fd ownership window explicit:

>>> with media.get_frame("sub", keep_fd=True) as frame:
...     result = pipeline.run(frame)
>>> # handle returned to the daemon here, even on exceptions

to_rgb()[source]

crop(x, y, width, height)[source]

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. The crop geometry is recorded in the new frame’s metadata['transform'] (affine, scale 1:1, origin (-x, -y)).

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

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.

The geometry is recorded in the new frame’s metadata['transform'] as a per-axis affine mapping dst = src * scale + origin (composed across chained crop/resize calls). Postprocessing maps model-input coordinates back to the source frame with src = (dst - origin) / scale — no hand-derived letterbox maths in the app.

to_jpeg_bytes(quality=85)[source]

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)[source]

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

FrameHandle

class neoruntime_ipc_sdk.FrameHandle(fds, strides, plane_sizes, frame_id, on_release=None, width=0, height=0, format='', flags=0)[source]

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='', flags=0)[source]

property closed: bool: True once close() ran (fds gone, frame released).

property fd: int: Convenience accessor for the first plane’s fd.

close()[source]

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')[source]

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)[source]

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)[source]

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)[source]

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, seq=None)[source]

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

seq: int | None = None

property is_keyframe: bool

property codec_name: str

__init__(codec, flags, pts_ns, width, height, dts_ns, data, seq=None)

使用示例

获取原始视频流

from neoruntime_ipc_sdk import FdMediaClient
import cv2

media = FdMediaClient()

# 获取主码流（可用流 ID 见 list_streams()，通常为 main / sub）
for frame in media.subscribe("main"):
    print(f"帧 {frame.sequence}: {frame.width}x{frame.height}")
    print(f"格式: {frame.format}, 时间戳: {frame.timestamp_ns}")

    # frame.image 是 numpy array (H, W, C) 或 (H*3//2, W) 对于 NV12
    # 可以直接用于 OpenCV 或其他图像处理库
    cv2.imshow("Camera", frame.image)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cv2.destroyAllWindows()

不跳过帧

# 获取每一帧（不跳过）
for frame in media.subscribe("main", skip_frames=False):
    process_frame(frame.image)

获取单帧

# 获取单帧
frame = media.get_frame("main", timeout_ms=1000)

if frame:
    print(f"帧: {frame.width}x{frame.height}")
    print(f"格式: {frame.format}")

获取流信息

# FdMediaClient 没有独立的流信息接口，从首帧读取尺寸与格式
frame = media.get_frame("main", timeout_ms=5000)

if frame:
    print(f"分辨率: {frame.width}x{frame.height}")
    print(f"格式: {frame.format}")
    print(f"可用流: {media.list_streams()}")

列出可用流

# 列出所有可用的流
streams = media.list_streams()
for stream_id in streams:
    print(f"流: {stream_id}")

帧处理回调

def handle_frame(frame):
    print(f"收到帧: {frame.sequence}")

# 使用回调处理帧
thread = media.on_frame("main", handle_frame)

# 保持运行
import time
while True:
    time.sleep(1)

图像处理

import cv2

media = FdMediaClient()

for frame in media.subscribe("main"):
    # 转换为 RGB
    rgb = frame.to_rgb()

    # 灰度图
    gray = cv2.cvtColor(rgb, cv2.COLOR_RGB2GRAY)

    # 边缘检测
    edges = cv2.Canny(gray, 100, 200)

    # 显示结果
    cv2.imshow("Original", rgb)
    cv2.imshow("Edges", edges)

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

保存图像

media = FdMediaClient()

for frame in media.subscribe("main"):
    # 保存为图片
    frame.save("frame.jpg")
    break

获取编码流

get_encoded_stream() 返回 EncodedStreamClient`（不是迭代器），产出 H.264/H.265 Annex-B 码流包，配合 ``recording` 模块可直接落盘：

from neoruntime_ipc_sdk import FdMediaClient

media = FdMediaClient()

# get_encoded_stream() 返回 EncodedStreamClient，不是迭代器
client = media.get_encoded_stream("main")

for packet in client.subscribe():
    print(f"{packet.codec_name()} 包: {len(packet.data)} 字节")
    if packet.is_keyframe():
        print("  关键帧")

多流处理

import threading

def process_main_stream():
    media = FdMediaClient()
    for frame in media.subscribe("main"):
        # 处理主码流（高分辨率）
        process_high_res(frame.image)

def process_sub_stream():
    media = FdMediaClient()
    for frame in media.subscribe("sub"):
        # 处理子码流（低分辨率）
        process_low_res(frame.image)

# 并行处理多个流
t1 = threading.Thread(target=process_main_stream)
t2 = threading.Thread(target=process_sub_stream)

t1.start()
t2.start()

t1.join()
t2.join()

帧率控制

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(frame.image)
        last_time = current_time

保存视频

如需保存 H.264/H.265 编码流，更推荐直接复用 recording 模块的 TsWriter / HlsWriter，无需解码重编码。下面的示例演示解码后用 OpenCV 保存原始帧：

import cv2

media = FdMediaClient()

# 从首帧获取分辨率（FdMediaClient 没有独立的流信息接口）
first = media.get_frame("main", timeout_ms=5000)
if first is None:
    raise RuntimeError("未收到视频帧")

# 创建视频写入器（fps 需按实际码流配置填写，这里以 30 为例）
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter(
    'output.mp4',
    fourcc,
    30.0,
    (first.width, first.height)
)

frame_count = 0
max_frames = 300  # 录制 10 秒 (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"已保存 {frame_count} 帧到 output.mp4")

上下文管理器

# 使用上下文管理器自动管理资源
with FdMediaClient() as media:
    for frame in media.subscribe("main"):
        process_frame(frame.image)

零拷贝访问

# 默认接收路径在收到帧时拷贝像素数据（frame.image 立即可用）
# keep_fd=True 保留 dma-buf fd，实现真正的零拷贝：
# 缓冲区延迟到 frame.release() / GC / 客户端关闭时才归还 daemon

media = FdMediaClient()

for frame in media.subscribe("main", keep_fd=True):
    # frame.handle 持有 dma-buf fd
    # frame.to_array() 首次调用时映射一次并缓存像素数据
    result = inference_engine.process(frame.to_array())

    # 处理完尽早归还缓冲区（幂等，可省略）
    frame.release()

错误处理

media = FdMediaClient()

try:
    for frame in media.subscribe("invalid_stream"):
        process_frame(frame.image)
except Exception as e:
    print(f"流访问失败: {e}")
except KeyboardInterrupt:
    print("用户中断")
finally:
    media.close()