Whole-Pipeline API (Platform-Scheduled)

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)

Examples

One call: subscribe + hardware overlay + app-side results

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,  # skip empty results
).start()

# Optional: consume results app-side too (alerts, forwarding, extra
# processing). The annotated video never passes through here — boxes
# are baked into the encoded stream by the hardware overlay.
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()                 # clears boxes + static zones; stream is clean

Semantics

  • Positioning vs InferencePipeline (pipeline.py): StreamPipeline is the platform-scheduled pipeline — the camera daemon feeds frames to the model (subscribe), no pixels enter the app process, results go to the hardware overlay via OverlayClient and are baked into the encoded stream. InferencePipeline is the client-side convenience — frames are pulled into the app and pre→infer→post runs synchronously on the app thread; reach for it when the app needs the pixels (custom preprocessing, multi-model chains on one frame, its own rasterizer). Choose by where the pixels should live.

  • Single-shot lifecycle: start() once, stop() once; build a new instance for another run. stop() is idempotent (no-op before start()), joins the worker before clearing boxes/static polygons, and closes only the clients it created (injected inference= / overlay= are left open).

  • Filtering affects drawing only: min_score / labels filter what reaches the overlay; results() still yields the full result. A fully-filtered result publishes an empty detection list so stale boxes clear.

  • on_result hook runs per result on the worker thread: return a (possibly replaced) result to keep it flowing, or None to drop it; a raised exception drops only that result and is recorded in status().last_error — never fatal.

  • Two bounded queues, drop-oldest: the subscribe side (queue_size, surfaced as subscribe_dropped) and the app side (result_queue_size, counted in result_queue_drops); annotate failures increment annotate_errors without stopping the pipeline.

  • draw=False never touches the overlay (static polygons included) — the encoded stream stays clean; results still flow to results().

  • Session tag (P2-13): session_id= spans inference and drawing — the subscription and every result drawing carry the same tag; when the process dies without cleanup (SIGKILL included), the inference session-end event makes camera-daemon sweep every polygon carrying that tag. The static zones at start and the clear writes at stop() carry no tag — operator writes that take effect unconditionally.

  • status() passes the subscribe observability through: last/avg latency_ms and last/avg skew_us (result-timestamp minus frame-timestamp — the acceptance metric shared by the preview and strict frame-lock modes).