AI 叠加层 API

AI Overlay Control Client

Control the platform’s AI overlay system that draws detection boxes, labels, and confidence scores directly on NV12 frames before encoding. Zero CPU cost — drawing happens in camera-daemon before encoding.

Two halves:

  • configuration — enable / disable / configure / apply via the UpdateAiOverlay RPC;

  • content — annotate / annotate_result push detection boxes (or classifications / landmarks / OCR lines) through the event bus; camera-daemon’s overlay subscriber renders them with its HAL draw ops.

Uses gRPC over Unix domain socket, consistent with other SDK clients.

class neoruntime_ipc_sdk.overlay.OverlayConfig(enabled=True, show_label=True, show_confidence=True, line_thickness=2, box_color=0, label_color=0, font_size=0, strict_frame_lock=None, strict_wait_cap_ms=None)[source]

Bases: object

AI overlay configuration.

strict_frame_lock / strict_wait_cap_ms mirror the optional wire fields: None (the default) leaves them unset, so the daemon keeps its current (yaml) strict settings.

enabled: bool = True
show_label: bool = True
show_confidence: bool = True
line_thickness: int = 2
box_color: int = 0
label_color: int = 0
font_size: int = 0
strict_frame_lock: bool | None = None
strict_wait_cap_ms: int | None = None
to_proto()[source]
__init__(enabled=True, show_label=True, show_confidence=True, line_thickness=2, box_color=0, label_color=0, font_size=0, strict_frame_lock=None, strict_wait_cap_ms=None)
class neoruntime_ipc_sdk.overlay.OverlayClient(*args, **kwargs)[source]

Bases: GrpcClient

AI Overlay Control Client

Uses gRPC to communicate with camera-daemon’s CameraControl service.

Usage:

from neoruntime_ipc_sdk import OverlayClient

oc = OverlayClient()

# Enable overlay with default settings
oc.enable()

# Customize appearance
oc.configure(
    show_label=True,
    show_confidence=True,
    line_thickness=3,
)

# Disable overlay
oc.disable()
Environment variables:
CAMERA_CONTROL_ENDPOINT: Camera control gRPC endpoint

(default: unix:///run/aipc/camera-control.sock)

__init__(*args, **kwargs)[source]
close()[source]
property channel_options

gRPC channel options applied at connect time.

enable(show_label=True, show_confidence=True, line_thickness=2)[source]

Enable AI overlay with specified settings.

disable()[source]

Disable AI overlay.

configure(enabled=True, show_label=True, show_confidence=True, line_thickness=2, box_color=0, label_color=0, font_size=0, strict_frame_lock=None, strict_wait_cap_ms=None)[source]

Configure AI overlay with full control.

Parameters:

  • enabled (bool) – Enable or disable overlay

  • show_label (bool) – Show class label on detections

  • show_confidence (bool) – Show confidence score on detections

  • line_thickness (int) – Box line thickness (1-10)

  • box_color (int) – Box color in ARGB format (e.g. 0xFFFF0000 for red)

  • label_color (int) – Label color in ARGB format

  • font_size (int) – Font size (8-72)

  • strict_frame_lock (bool | None) – opt in to / opt out of strict frame-lock on identity-fed display streams (streams whose stream_map entry feeds them their own inference results). Strict mode waits — bounded by strict_wait_cap_ms — for each frame’s own inference result at the bake site, so a box lands on the frame it was computed from instead of the newest frame. None keeps the daemon’s current setting.

  • strict_wait_cap_ms (int | None) – max ms the bake site waits per frame before degrading to the freshest result (warned + counted). 0 derives the cap from the stream fps (2 frame periods, clamp [1, 500]). None keeps the current setting; only meaningful together with strict_frame_lock.

apply(config)[source]

Apply an OverlayConfig object.

annotate(stream_id, detections=None, *, polygons=None, ttl_ms=None, session_id=None, frame_sequence=None, stream_epoch=None)[source]

Draw detection boxes / zone polygons on stream_id’s video.

Parameters:

  • stream_id (str) – the camera stream the boxes belong to (e.g. "main") — camera-daemon keys overlay results by it.

  • detections (list | None) – DetectedObject items or dicts with label / score (or confidence) and bbox ({"x","y","width","height"} or [x, y, w, h]). A detection’s track_id (>= 0) rides along and renders as a #id label suffix. None leaves the stream’s previous boxes untouched, so polygons can push alone.

  • polygons (list | None) – dicts {"points": [[x, y], ...], "label": str, "closed": bool, "color": ARGB int, "thickness": px int (-1 = filled)} with only points required — points are normalized [0, 1] floats, 2..128 per polygon, at most 16 polygons per call. Any list ([] included) replaces the stream’s previous polygons; None keeps them.

  • ttl_ms (int | None) – positive int — per-event validity override for the stream’s results. Omitted: the daemon derives a default from the stream’s fps (about two frame periods).

  • session_id (str | None) – tag this event as belonging to a lifecycle session (P2-13). The daemon marks the stream’s results with the tag and a "session/end" event sweeps matching tagged polygons on disconnect; an untagged write resets the mark. None / "" leave the event untagged.

  • frame_sequence (int | None) – bind this event to that frame of the stream (positive int — InferenceResult.frame_sequence from a subscribe() iteration is the natural source). The daemon draws the layer only while the bake site is within its frame-bind slack of it, then expires it, and rejects events whose frame the pipeline already passed. None leaves the event unbound (legacy semantics).

  • stream_epoch (int | None) – the stream generation counter (positive int, from get_stream_status()’s stream_epoch) this event was made against. A ReconfigureEncoder restart bumps the epoch — the daemon rejects events carrying a stale one instead of drawing ghosts over the new generation. None skips the check.

Returns:

The published event id.

Return type:

str

Results expire ttl_ms after the last event (fps-derived default), so call this at a few Hz while detections are fresh — publish an empty detections list to clear boxes, an empty polygons list to clear zones. The overlay itself must be on: enable() first.

oc = OverlayClient()
oc.enable()
for result in inf.subscribe(stream, model):
    oc.annotate(stream, result.objects)

# zone polygons ride independently of the boxes channel
oc.annotate(stream, polygons=[{"points": ZONE, "label": "yard"}])
annotate_result(stream_id, result, *, ttl_ms=None, session_id=None, frame_sequence=None, stream_epoch=None)[source]

Push whichever section of an InferenceResult is populated.

Detections take precedence, then classifications, landmarks and OCR lines — a payload carries one kind (that is how camera-daemon’s parser discriminates them). An empty result publishes zero detections, clearing stale boxes. ttl_ms overrides the stream’s result validity window for this event. session_id tags the event with a lifecycle session (P2-13). frame_sequence / stream_epoch bind the event to the stream’s frame generation — see annotate().

OverlayClient

class neoruntime_ipc_sdk.OverlayClient(*args, **kwargs)[source]

Bases: GrpcClient

AI Overlay Control Client

Uses gRPC to communicate with camera-daemon’s CameraControl service.

Usage:

from neoruntime_ipc_sdk import OverlayClient

oc = OverlayClient()

# Enable overlay with default settings
oc.enable()

# Customize appearance
oc.configure(
    show_label=True,
    show_confidence=True,
    line_thickness=3,
)

# Disable overlay
oc.disable()
Environment variables:
CAMERA_CONTROL_ENDPOINT: Camera control gRPC endpoint

(default: unix:///run/aipc/camera-control.sock)

__init__(*args, **kwargs)[source]
close()[source]
property channel_options

gRPC channel options applied at connect time.

enable(show_label=True, show_confidence=True, line_thickness=2)[source]

Enable AI overlay with specified settings.

disable()[source]

Disable AI overlay.

configure(enabled=True, show_label=True, show_confidence=True, line_thickness=2, box_color=0, label_color=0, font_size=0, strict_frame_lock=None, strict_wait_cap_ms=None)[source]

Configure AI overlay with full control.

Parameters:

  • enabled (bool) – Enable or disable overlay

  • show_label (bool) – Show class label on detections

  • show_confidence (bool) – Show confidence score on detections

  • line_thickness (int) – Box line thickness (1-10)

  • box_color (int) – Box color in ARGB format (e.g. 0xFFFF0000 for red)

  • label_color (int) – Label color in ARGB format

  • font_size (int) – Font size (8-72)

  • strict_frame_lock (bool | None) – opt in to / opt out of strict frame-lock on identity-fed display streams (streams whose stream_map entry feeds them their own inference results). Strict mode waits — bounded by strict_wait_cap_ms — for each frame’s own inference result at the bake site, so a box lands on the frame it was computed from instead of the newest frame. None keeps the daemon’s current setting.

  • strict_wait_cap_ms (int | None) – max ms the bake site waits per frame before degrading to the freshest result (warned + counted). 0 derives the cap from the stream fps (2 frame periods, clamp [1, 500]). None keeps the current setting; only meaningful together with strict_frame_lock.

apply(config)[source]

Apply an OverlayConfig object.

annotate(stream_id, detections=None, *, polygons=None, ttl_ms=None, session_id=None, frame_sequence=None, stream_epoch=None)[source]

Draw detection boxes / zone polygons on stream_id’s video.

Parameters:

  • stream_id (str) – the camera stream the boxes belong to (e.g. "main") — camera-daemon keys overlay results by it.

  • detections (list | None) – DetectedObject items or dicts with label / score (or confidence) and bbox ({"x","y","width","height"} or [x, y, w, h]). A detection’s track_id (>= 0) rides along and renders as a #id label suffix. None leaves the stream’s previous boxes untouched, so polygons can push alone.

  • polygons (list | None) – dicts {"points": [[x, y], ...], "label": str, "closed": bool, "color": ARGB int, "thickness": px int (-1 = filled)} with only points required — points are normalized [0, 1] floats, 2..128 per polygon, at most 16 polygons per call. Any list ([] included) replaces the stream’s previous polygons; None keeps them.

  • ttl_ms (int | None) – positive int — per-event validity override for the stream’s results. Omitted: the daemon derives a default from the stream’s fps (about two frame periods).

  • session_id (str | None) – tag this event as belonging to a lifecycle session (P2-13). The daemon marks the stream’s results with the tag and a "session/end" event sweeps matching tagged polygons on disconnect; an untagged write resets the mark. None / "" leave the event untagged.

  • frame_sequence (int | None) – bind this event to that frame of the stream (positive int — InferenceResult.frame_sequence from a subscribe() iteration is the natural source). The daemon draws the layer only while the bake site is within its frame-bind slack of it, then expires it, and rejects events whose frame the pipeline already passed. None leaves the event unbound (legacy semantics).

  • stream_epoch (int | None) – the stream generation counter (positive int, from get_stream_status()’s stream_epoch) this event was made against. A ReconfigureEncoder restart bumps the epoch — the daemon rejects events carrying a stale one instead of drawing ghosts over the new generation. None skips the check.

Returns:

The published event id.

Return type:

str

Results expire ttl_ms after the last event (fps-derived default), so call this at a few Hz while detections are fresh — publish an empty detections list to clear boxes, an empty polygons list to clear zones. The overlay itself must be on: enable() first.

oc = OverlayClient()
oc.enable()
for result in inf.subscribe(stream, model):
    oc.annotate(stream, result.objects)

# zone polygons ride independently of the boxes channel
oc.annotate(stream, polygons=[{"points": ZONE, "label": "yard"}])
annotate_result(stream_id, result, *, ttl_ms=None, session_id=None, frame_sequence=None, stream_epoch=None)[source]

Push whichever section of an InferenceResult is populated.

Detections take precedence, then classifications, landmarks and OCR lines — a payload carries one kind (that is how camera-daemon’s parser discriminates them). An empty result publishes zero detections, clearing stale boxes. ttl_ms overrides the stream’s result validity window for this event. session_id tags the event with a lifecycle session (P2-13). frame_sequence / stream_epoch bind the event to the stream’s frame generation — see annotate().

OverlayConfig

class neoruntime_ipc_sdk.OverlayConfig(enabled=True, show_label=True, show_confidence=True, line_thickness=2, box_color=0, label_color=0, font_size=0, strict_frame_lock=None, strict_wait_cap_ms=None)[source]

AI overlay configuration.

strict_frame_lock / strict_wait_cap_ms mirror the optional wire fields: None (the default) leaves them unset, so the daemon keeps its current (yaml) strict settings.

enabled: bool = True
show_label: bool = True
show_confidence: bool = True
line_thickness: int = 2
box_color: int = 0
label_color: int = 0
font_size: int = 0
strict_frame_lock: bool | None = None
strict_wait_cap_ms: int | None = None
to_proto()[source]
__init__(enabled=True, show_label=True, show_confidence=True, line_thickness=2, box_color=0, label_color=0, font_size=0, strict_frame_lock=None, strict_wait_cap_ms=None)

使用示例

启用叠加层

from neoruntime_ipc_sdk import OverlayClient

overlay = OverlayClient()

# 启用硬件叠加层：在视频上绘制检测框 + 标签 + 置信度
overlay.enable(show_label=True, show_confidence=True, line_thickness=2)

# 关闭
overlay.disable()

自定义样式

# configure 一次设置全部样式参数
overlay.configure(
    enabled=True,
    show_label=True,
    show_confidence=False,
    line_thickness=3,
    box_color=0x00FF00,     # 绿色框
    label_color=0xFFFFFF,   # 白色标签
    font_size=16,
)

结构化配置

from neoruntime_ipc_sdk import OverlayClient, OverlayConfig

config = OverlayConfig(
    enabled=True,
    show_label=True,
    show_confidence=True,
    line_thickness=2,
)
OverlayClient().apply(config)

严格对帧模式（P1-6）

# 默认是"实时预览"语义：烘焙用的是有效期内的最新结果，框与画面不
# 保证同帧。严格对帧是 opt-in：configure 打开后，daemon 在烘焙点
# （编码前）为本帧等待它自己的推理结果——等待有上限，超限降级为
# 最新结果并打日志/计数，编码输出永不因等待而停住
overlay.configure(strict_frame_lock=True)              # 打开（cap 按 fps 推导）
overlay.configure(strict_frame_lock=True,
                  strict_wait_cap_ms=0)                # 0 = 按流 fps 推导（2 帧周期）
overlay.configure(strict_frame_lock=True,
                  strict_wait_cap_ms=66)               # 显式 66ms 上限
overlay.configure(strict_frame_lock=False)             # 显式退出严格模式

# 两个参数都是"可选字段"语义：不传 = 保持 daemon 当前设置；
# enable()/disable() 与不带参数的 configure() 不会改动严格模式

# 严格模式只作用于"自喂"显示流——stream_map 里 display 流的推理结果
# 来自它自己（如 main:main）。交叉喂流的流（如 third:main）保持实时
# 预览语义

# 严格模式只锁定平台结果层，不排除应用图层（annotate() 的绘制）：
# 有锁定层时，app 图层按自身 TTL 语义叠加绘制在锁定层之上；SKIP /
# 降级（本帧没有可锁定的平台结果）时，app 图层单独绘制，不再被
# 严格模式清掉。严格模式永不压制已锁定的平台层（帧同步保证正是
# 该模式的意义），也绝不放行交叉喂流的平台层；只有"既无锁定层、
# 又无在效 app 图层"才干净出帧（计入 bake_skips）

与推理结果联动（annotate）

# annotate() 把检测结果经事件总线推给 camera-daemon 的叠加渲染器：
# app 不接触视频帧，检测框在编码前由 daemon 画到码流上。结果
# 500ms 过期，按推理节奏持续调用；发空列表清屏
from neoruntime_ipc_sdk import OverlayClient

overlay = OverlayClient()
overlay.enable()

for result in inference_results:      # 如 InferenceClient.subscribe(...)
    overlay.annotate("main", result.objects)

overlay.annotate("main", [])          # 清屏

其他结果类型

# annotate_result 按 objects > classifications > landmarks >
# ocr_lines 的优先级取 InferenceResult 中已填充的那一段
overlay.annotate_result("main", result)

生命周期会话标记（P2-13）

# annotate / annotate_result / 多边形写入都接受 session_id：同一会话
# 的绘制带同一标记。推理会话结束（含进程被 SIGKILL 后连接断开）时，
# daemon 按 session/end 事件把带该标记的所有 polygon 一次性清掉；
# 检测框本身随结果 TTL 自然过期
overlay.annotate("main", result.objects, session_id="app-1")
overlay.annotate("main", polygons=zones, session_id="app-1")

# 不传 session_id = 运维写入：无条件生效、不参与会话清理。
# StreamPipeline 内部即按此区分——结果绘制带会话标记，启动时的
# 静态 zone 与 stop() 的清屏写入不带标记

两种模式与视频副作用契约

SDK 有两条互不干扰的取数路径，契约上明确区分：

只取结果（零视频副作用）：InferenceClient.subscribe(...) 只 返回推理结果，不触碰视频通路。平台上自己的模型（ai-runtime / auto-infer 发布的结果）默认也**不会**改变码流——除非运维在 daemon 配置里显式绑定（见下节）。annotate() 主动画框不算副作用违规： 它是应用显式下发的绘制指令。

自取帧自画（keep_fd / copy）：FdMediaClient 的 keep_fd / copy 模式取到的是**干净帧**（未经叠加层烘焙）。应用在自己的缓冲上组 合：copy 帧直接改写即可；keep_fd 帧经 DSP_IMPORT 进应用自有的 DSP 池做 RESIZE→BLEND（P1-9 池驻留后每调用开销可接受）。 keep_fd 帧永不作为 DSP 写目标或 blend base——平台侧按只读 契约执法（写入即拒绝），测试 dsp-import-write-target-test 守卫。

平台结果绑定（daemon 配置）

平台模型的检测结果默认不画（行为解耦）：一个裸 subscribe() 不 应该改变任何视频输出。需要平台框时，在 daemon 配置 yaml 的 [ai_overlay] 段显式绑定：

[ai_overlay]
# 推理流:显示流，方向与 stream_map 相同；逗号分隔多个绑定
bindings: "main:main,third:main"
# 迁移开关：恢复旧行为（所有平台结果自动画 + 默认 stream_map 回填）
legacy_auto_bind: 0

迁移说明：legacy_auto_bind 默认关闭。升级后若发现平台框 消失，两种恢复方式任选——(1) 按上例把需要的流加进 bindings （推荐：只开需要的流）；(2) 设 legacy_auto_bind: 1 完整恢复 旧行为（所有平台结果自动绑定，含默认 stream_map 回填）。应用经 annotate() 下发的绘制不受此开关影响，永远生效。

帧同步绑定（frame_sequence / stream_epoch）

# annotate / annotate_result 接受两个可选的帧同步参数（都走事件
# metadata，不进 payload；不传 = 不绑定，保持既有语义）
#
# frame_sequence：把这次绘制绑定到该帧——只在烘焙点走到该帧附近
#   （+2 帧容差）内绘制，过后过期；帧已被流水线越过的指令会被
#   拒收（overlay_late_commands 计数）。subscribe() 结果里的
#   result.frame_sequence 就是天然来源
#   注意：frame_sequence 是 HAL 的共享帧计数——同一 media context
#   下所有流（main/sub/...）共用一个计数器，序号同尺可比；但拿
#   别条流的结果当来源会引入跨流延迟（推理耗时落在计数器刻度
#   里），可能吃掉绑定窗口。推荐用目标显示流自己 subscribe()
#   出的结果做序号来源；跨流来源可用但可见帧数更少。绑定窗口
#   按"显示流帧数"自适应换算（多流共享计数器下每帧跨多步）
# stream_epoch：流代计数（get_stream_status() 可读）。重配置
#   （ReconfigureEncoder / 分辨率、变换全量重建）会 bump 代数并
#   清空持有的图层；携带旧代数的事件被拒收（overlay_epoch_rejects
#   计数），不再化作幽灵框
for result in stream.subscribe(model):
    overlay.annotate_result("main", result,
                            frame_sequence=result.frame_sequence,
                            stream_epoch=epoch)

# 重配置后的自愈：读到 epoch 变了就换用新值继续发布。重配置 /
#   流水线重建时平台会 bump 代数并清空该流的序号锚点——HAL 计数
#   可能从新纪元的小值重新开始，app 侧不要沿用重建前拿到的旧
#   frame_sequence，重新 subscribe() 取新值即可
status = {s.stream_id: s for s in camera.get_stream_status()}
epoch = status["main"].stream_epoch

keep_fd 帧的只读配方

# copy 模式：帧已是应用自己的内存，直接改写
frame = media.get_frame(stream, copy=True)
draw_boxes(frame, results)               # 随便画

# keep_fd 模式：dma-buf 只读。经 DSP_IMPORT 导入应用自有的 DSP
# 池，RESIZE 到目标尺寸后作为 BLEND 的 overlay 层叠加到应用自己
# 的底图上；帧本体永不作为写目标——RESIZE 的 dst / BLEND 的
# base 若指向 keep_fd 帧都会被平台拒绝（平台侧执法，非约定）
dsp.import_buffer(frame.fds, frame_id=frame.frame_id)
dsp.schedule(op=DSP_OP_RESIZE, src=imported, dst=scratch)    # 自有 scratch
dsp.schedule(op=DSP_OP_BLEND, base=composed, overlay=scratch)  # 自有底图
# 简言之：keep_fd 帧 = BLEND 的 overlay 输入或 RESIZE 的 src，
#           自有缓冲 = 一切写目标

上下文管理器

with OverlayClient() as overlay:
    overlay.enable()