整管线 API（平台调度管线）

Whole-pipeline facade: one call wiring capture -> inference -> overlay.

Positioning (read this before reaching for InferencePipeline in pipeline.py): this module is the platform-scheduled pipeline. The camera daemon feeds frames to the model (InferenceClient.subscribe), so no pixels ever cross into the app process; results come back over the stream, get pushed to the hardware overlay via the event bus (OverlayClient), and the annotated video ships inside the encoded stream — the app only ever sees lightweight InferenceResult structs. InferencePipeline, by contrast, is the client-side convenience: it pulls frames into the app, runs pre->infer->post on the app thread, and is the right tool when the app needs the pixels (custom preprocessing, multi-model chains on the same frame, drawing with its own rasterizer).

Choose by where the pixels should live: keep them on the device and take results — StreamPipeline; need them in your process — InferencePipeline.

One instance is single-shot: start() once, stop() once; build a new instance for another run. Typical use:

zones = [{"points": [[0, 0], [1, 0], [1, 1]], "label": "yard"}]
pipe = StreamPipeline(
    "main", "yolov5n", fps=10,
    min_score=0.5, labels=["person"], polygons=zones,
    on_result=lambda r: r if r.count_by_label("person") else None,
)
pipe.start()
for seq, result in pipe.results():   # optional: results in the app too
    ...
pipe.stop()                          # clears boxes + zones on the stream
class neoruntime_ipc_sdk.stream_pipeline.StreamPipelineStatus(running, results_seen, results_annotated, annotate_errors, result_queue_drops, subscribe_dropped, last_error, last_latency_ms, avg_latency_ms, last_skew_us, avg_skew_us)[source]

Bases: object

Point-in-time health snapshot from StreamPipeline.status().

results_seen counts every result the worker accepted from the subscription, before the on_result hook runs — results the hook dropped or replaced are included. last_latency_ms/avg_latency_ms likewise time every accepted result, including ones later dropped by the hook.

running: bool
results_seen: int
results_annotated: int
annotate_errors: int
result_queue_drops: int
subscribe_dropped: int
last_error: str
last_latency_ms: float
avg_latency_ms: float
last_skew_us: int
avg_skew_us: float
__init__(running, results_seen, results_annotated, annotate_errors, result_queue_drops, subscribe_dropped, last_error, last_latency_ms, avg_latency_ms, last_skew_us, avg_skew_us)
class neoruntime_ipc_sdk.stream_pipeline.StreamPipeline(stream, model, *, fps=10, session_id='', raw_output_only=False, max_consecutive_failures=10, queue_size=100, draw=True, overlay_config=None, ttl_ms=None, min_score=0.0, labels=None, polygons=None, on_result=None, result_queue_size=100, inference=None, overlay=None)[source]

Bases: object

One-call pipeline: subscribe, draw on the hardware overlay, observe.

Parameters:

  • stream (str) – camera stream id the model runs on (e.g. "main").

  • model (str) – registered model id.

  • fps (int) – result-rate cap for the subscription.

  • session_id (str) – optional inference session id (observability/logging).

  • raw_output_only (bool) – skip post-processing, hand back raw tensors.

  • max_consecutive_failures (int | None) – consecutive failed frames tolerated before the worker gives up (0/None = unlimited); see InferenceClient.subscribe().

  • queue_size (int | None) – bound for the subscribe-side result queue.

  • draw (bool) – push results to the hardware overlay. False also suppresses static polygons — the encoded stream stays clean; results still flow to results().

  • overlay_config (OverlayConfig | None) – optional OverlayConfig applied at start (labels, colors, strict frame-lock, …).

  • ttl_ms (int | None) – positive int, per-event validity override forwarded to every annotate call; omitted = the daemon derives it from the stream fps.

  • min_score (float) – draw only detections with score >= min_score.

  • labels (list[str] | None) – draw only these labels (None = all). Filtering affects drawing only — results() yields the full result — and a fully-filtered result still publishes an empty detection list, clearing stale boxes.

  • polygons (list[dict] | None) – static zone polygons pushed once at start and cleared at stop (normalized [0,1] points; see OverlayClient.annotate()).

  • on_result (Callable[[InferenceResult], InferenceResult | None] | None) – hook called per result on the worker thread. Return a (possibly replaced) result to flow onward, or None to drop it (not drawn, not queued) — put tracking/filtering logic here. A raised exception drops that result, is recorded in status().last_error and is non-fatal.

  • result_queue_size (int) – bound for the app-facing queue behind results(); a slow app drops its oldest queued results (counted in status().result_queue_drops).

  • inference (InferenceClient | None) – inject an InferenceClient (tests / client reuse). Injected clients are NOT closed by stop().

  • overlay (OverlayClient | None) – inject an OverlayClient. Not closed on stop.

__init__(stream, model, *, fps=10, session_id='', raw_output_only=False, max_consecutive_failures=10, queue_size=100, draw=True, overlay_config=None, ttl_ms=None, min_score=0.0, labels=None, polygons=None, on_result=None, result_queue_size=100, inference=None, overlay=None)[source]
start()[source]

Subscribe, apply overlay config / static zones, start drawing.

stop()[source]

Stop the worker, clear what was drawn, close owned clients.

Idempotent, and a no-op before start(). The worker is joined first so the box-clearing annotate cannot race a final result. Safe to call from on_result (i.e. from the worker thread itself): the self-join is skipped and the teardown below runs on the caller’s thread.

results()[source]

Yield (frame_sequence, InferenceResult) post-callback.

Ends when the worker stops (stop() or a fatal stream error). The queue is bounded by result_queue_size with drop-oldest overflow — a slow consumer loses its oldest results, counted in status().result_queue_drops. The pre-start check is eager: it raises at call time, not on the first next().

status()[source]

Snapshot of pipeline health and pass-through subscribe stats.

StreamPipeline

class neoruntime_ipc_sdk.StreamPipeline(stream, model, *, fps=10, session_id='', raw_output_only=False, max_consecutive_failures=10, queue_size=100, draw=True, overlay_config=None, ttl_ms=None, min_score=0.0, labels=None, polygons=None, on_result=None, result_queue_size=100, inference=None, overlay=None)[source]

Bases: object

One-call pipeline: subscribe, draw on the hardware overlay, observe.

Parameters:

  • stream (str) – camera stream id the model runs on (e.g. "main").

  • model (str) – registered model id.

  • fps (int) – result-rate cap for the subscription.

  • session_id (str) – optional inference session id (observability/logging).

  • raw_output_only (bool) – skip post-processing, hand back raw tensors.

  • max_consecutive_failures (int | None) – consecutive failed frames tolerated before the worker gives up (0/None = unlimited); see InferenceClient.subscribe().

  • queue_size (int | None) – bound for the subscribe-side result queue.

  • draw (bool) – push results to the hardware overlay. False also suppresses static polygons — the encoded stream stays clean; results still flow to results().

  • overlay_config (OverlayConfig | None) – optional OverlayConfig applied at start (labels, colors, strict frame-lock, …).

  • ttl_ms (int | None) – positive int, per-event validity override forwarded to every annotate call; omitted = the daemon derives it from the stream fps.

  • min_score (float) – draw only detections with score >= min_score.

  • labels (list[str] | None) – draw only these labels (None = all). Filtering affects drawing only — results() yields the full result — and a fully-filtered result still publishes an empty detection list, clearing stale boxes.

  • polygons (list[dict] | None) – static zone polygons pushed once at start and cleared at stop (normalized [0,1] points; see OverlayClient.annotate()).

  • on_result (Callable[[InferenceResult], InferenceResult | None] | None) – hook called per result on the worker thread. Return a (possibly replaced) result to flow onward, or None to drop it (not drawn, not queued) — put tracking/filtering logic here. A raised exception drops that result, is recorded in status().last_error and is non-fatal.

  • result_queue_size (int) – bound for the app-facing queue behind results(); a slow app drops its oldest queued results (counted in status().result_queue_drops).

  • inference (InferenceClient | None) – inject an InferenceClient (tests / client reuse). Injected clients are NOT closed by stop().

  • overlay (OverlayClient | None) – inject an OverlayClient. Not closed on stop.

__init__(stream, model, *, fps=10, session_id='', raw_output_only=False, max_consecutive_failures=10, queue_size=100, draw=True, overlay_config=None, ttl_ms=None, min_score=0.0, labels=None, polygons=None, on_result=None, result_queue_size=100, inference=None, overlay=None)[source]
start()[source]

Subscribe, apply overlay config / static zones, start drawing.

stop()[source]

Stop the worker, clear what was drawn, close owned clients.

Idempotent, and a no-op before start(). The worker is joined first so the box-clearing annotate cannot race a final result. Safe to call from on_result (i.e. from the worker thread itself): the self-join is skipped and the teardown below runs on the caller’s thread.

results()[source]

Yield (frame_sequence, InferenceResult) post-callback.

Ends when the worker stops (stop() or a fatal stream error). The queue is bounded by result_queue_size with drop-oldest overflow — a slow consumer loses its oldest results, counted in status().result_queue_drops. The pre-start check is eager: it raises at call time, not on the first next().

status()[source]

Snapshot of pipeline health and pass-through subscribe stats.

StreamPipelineStatus

class neoruntime_ipc_sdk.StreamPipelineStatus(running, results_seen, results_annotated, annotate_errors, result_queue_drops, subscribe_dropped, last_error, last_latency_ms, avg_latency_ms, last_skew_us, avg_skew_us)[source]

Point-in-time health snapshot from StreamPipeline.status().

results_seen counts every result the worker accepted from the subscription, before the on_result hook runs — results the hook dropped or replaced are included. last_latency_ms/avg_latency_ms likewise time every accepted result, including ones later dropped by the hook.

running: bool
results_seen: int
results_annotated: int
annotate_errors: int
result_queue_drops: int
subscribe_dropped: int
last_error: str
last_latency_ms: float
avg_latency_ms: float
last_skew_us: int
avg_skew_us: float
__init__(running, results_seen, results_annotated, annotate_errors, result_queue_drops, subscribe_dropped, last_error, last_latency_ms, avg_latency_ms, last_skew_us, avg_skew_us)

使用示例

一次启动：订阅推理 + 硬件叠加 + 应用侧结果

from neoruntime_ipc_sdk import StreamPipeline

zones = [{"points": [[0, 0], [1, 0], [1, 1]], "label": "yard"}]
pipe = StreamPipeline(
    "main", "yolov5n", fps=10,
    min_score=0.5, labels=["person"], polygons=zones,
    on_result=lambda r: r if r.objects else None,  # 空结果不画不进队列
).start()

# 可选：应用侧同时消费结果（告警、二次处理、转发）。
# 叠加画面不经过这里 —— 框已由硬件叠加层烘进编码流。
for seq, result in pipe.results():
    print(seq, [(o.label, round(o.score, 2)) for o in result.objects])
    if seq >= 100:
        break

st = pipe.status()          # results_seen / annotated / drops / skew ...
pipe.stop()                 # 清除检测框与静态多边形，编码画面恢复干净

语义要点

  • 与 InferencePipeline（pipeline.py）的定位区分：StreamPipeline 是 平台调度管线 —— 帧由 camera daemon 喂给模型（subscribe），像素 不进应用进程，结果经 OverlayClient 推到硬件叠加层并烘进编码流； InferencePipeline 是客户端便捷层 —— 帧拉进应用、pre→infer→post 在 应用线程同步执行，适合需要像素的场景（自定义预处理、同帧多模型链、 自绘）。按”像素应该住哪”选择。

  • 单发生命周期：start() 一次、stop() 一次；再次运行须新建 实例。stop() 幂等、start 前调用为 no-op；先 join 工作线程再清 框/清静态多边形，只关闭自己创建的客户端（注入的 inference= / overlay= 不关闭）。

  • 过滤只影响绘制：min_score / labels 仅过滤画到叠加层的 内容，results() 仍返回完整结果；全被滤掉的结果发布空检测列表， 以清除滞留旧框。

  • on_result 钩子：在工作线程逐结果调用，返回（可替换的）结果继续 流动，返回 None 丢弃；抛异常只丢该结果并记入 status().last_error，非致命。

  • 两段有界队列、丢最旧：subscribe 侧 queue_size （subscribe_dropped 透出）与应用侧 result_queue_size （result_queue_drops 计数）；叠加失败计入 annotate_errors 且不中断管线。

  • draw=False 完全不触碰叠加层：静态 polygons 也不推，编码流保持 干净；结果仍进 results()。

  • 会话标记（P2-13）：session_id= 贯穿推理与绘制 —— 订阅请求 与结果绘制带同一标记；进程意外退出（含 SIGKILL）后，推理会话结束 事件触发 daemon 把带该标记的多边形一并清掉。启动时的静态 zone 与 stop() 的清屏写入不带标记，属于运维写入、无条件生效。

  • status() 透传订阅观测面：last/avg latency_ms 与 last/avg skew_us （结果时间戳相对帧时间戳的偏移 —— 实时预览与严格对帧两种模式共用 的验收指标）。