Accel Routing API

Hardware-first capability router.

The SDK ships loose components (散件), not a pipeline; apps compose them. Most components have (or will have) two implementations: a hardware leg executed by a daemon (DSP color convert / resize, ai-runtime NMS, codec JPEG, camera-daemon overlay) and a numpy/cv2 software leg that always works. Today several hardware legs are not reachable because the service layer does not expose them yet — see docs/proposals/sdk-hardware-routing.md for the platform ask list.

This router makes that situation explicit and uniform:

• one policy per app — prefer hardware, software-only, or hardware-only;

• one route table — which leg serves each operation, and why the other leg is not used (“pending platform exposure”, “probe failed”, “runtime error”);

• degradation tracking — every hardware→software fallback is recorded and surfaced through health() (and an optional on_degradation hook an app can forward to the event bus as a health event).

from neoruntime_ipc_sdk.accel import get_default_router

router = get_default_router()
small = router.run("resize_nv12", nv12, (1920, 1080), (640, 384))
if router.health()["ops"]["resize_nv12"]["backend"] == "software":
    ...  # hardware path not available on this platform build

class neoruntime_ipc_sdk.accel.AccelRouter(policy=RoutePolicy.PREFER_HARDWARE)

Bases: object

Route-table dispatcher between hardware and software providers.

Operations are registered with register(); run() executes through the leg chosen by the policy, recording any fallback. get_default_router() returns a singleton pre-registered with the operations the SDK can serve today.

__init__(policy=RoutePolicy.PREFER_HARDWARE)

register(op, software=None, hardware=None, note='')

Map an operation name to its software/hardware providers.

Either leg may be None: a missing hardware leg means the capability awaits platform exposure (state it in note), a missing software leg means there is no fallback.

use_hardware(op, hardware)

Attach (or replace) the hardware leg of an operation.

add_probe(name, probe)

Register a cheap availability probe (e.g. service reachable).

property policy: RoutePolicy

The active policy (fixed at construction).

route(op)

Decide which backend serves op under the current policy.

run(op, *args, **kwargs)

Execute op through the leg chosen by route().

note_degradation(op, reason)

Record a fallback that happened outside run().

Frame.resize keeps its own direct DSP fast path (its geometry contract is richer than the router op’s NV12 signature); when that attempt fails under PREFER_HARDWARE it reports here so health() and the fallback counters stay honest. Fires on_degradation exactly like a run fallback does.

probe()

Run all registered probes → name → available.

health()

Snapshot of routing state, per-op counters and recent fallbacks.

class neoruntime_ipc_sdk.accel.DegradationRecord(op, reason, timestamp)

Bases: object

One hardware→software fallback, kept for health reporting.

op: str

reason: str

timestamp: float

__init__(op, reason, timestamp)

exception neoruntime_ipc_sdk.accel.HardwareUnavailable

Bases: RuntimeError

A hardware provider could not run (service down, op not exposed).

class neoruntime_ipc_sdk.accel.RouteDecision(op, backend, provider, reason)

Bases: object

Why a given backend serves an operation right now.

op: str

backend: str

provider: str

reason: str

__init__(op, backend, provider, reason)

class neoruntime_ipc_sdk.accel.RoutePolicy(value)

Bases: Enum

How the router picks a leg for each operation.

PREFER_HARDWARE uses the hardware leg when it is registered and healthy, falling back to software otherwise. SOFTWARE_ONLY never touches hardware — for benchmarks and reproducible tests. HARDWARE_ONLY raises instead of degrading — when the zero-CPU guarantee matters more than uptime.

PREFER_HARDWARE = 'prefer_hardware'

SOFTWARE_ONLY = 'software_only'

HARDWARE_ONLY = 'hardware_only'

neoruntime_ipc_sdk.accel.get_default_router()

Return the pre-registered router singleton.

The public convenience layer rides this singleton: color.rgb_to_nv12() / color.nv12_to_rgb(), draw.draw_detections() (NV12 arrays) and Frame.to_jpeg_bytes() route through run; Frame.resize() keeps its direct DSP fast path but consults route/policy and reports its fallbacks via note_degradation so health stays honest.

Operations served today: resize_nv12, rgb_to_nv12 and nv12_to_rgb (DSP when reachable, numpy otherwise), encode_jpeg (camera-daemon EncodeImage when reachable, cv2/Pillow otherwise), draw_detections (DSP blend when reachable and the frame is NV12, draw.py raster otherwise), nms (software only — suppression already runs in the HEF’s integrated hardware NMS before the app sees boxes, and its iou_threshold / max_boxes are compile-time there; the runtime-tunable detection_threshold lives in InferenceClient.update_postprocess_config(), honored for family-function postprocess models. See docs/proposals/sdk-hardware-routing.md S-2). OverlayClient.annotate is hardware-first already and needs no routing.

Classes and Functions

AccelRouter

class neoruntime_ipc_sdk.AccelRouter(policy=RoutePolicy.PREFER_HARDWARE)

Route-table dispatcher between hardware and software providers.

Operations are registered with register(); run() executes through the leg chosen by the policy, recording any fallback. get_default_router() returns a singleton pre-registered with the operations the SDK can serve today.

__init__(policy=RoutePolicy.PREFER_HARDWARE)

on_degradation: Callable[[DegradationRecord], None] | None

register(op, software=None, hardware=None, note='')

Map an operation name to its software/hardware providers.

Either leg may be None: a missing hardware leg means the capability awaits platform exposure (state it in note), a missing software leg means there is no fallback.

use_hardware(op, hardware)

Attach (or replace) the hardware leg of an operation.

add_probe(name, probe)

Register a cheap availability probe (e.g. service reachable).

property policy: RoutePolicy

The active policy (fixed at construction).

route(op)

Decide which backend serves op under the current policy.

run(op, *args, **kwargs)

Execute op through the leg chosen by route().

note_degradation(op, reason)

Record a fallback that happened outside run().

Frame.resize keeps its own direct DSP fast path (its geometry contract is richer than the router op’s NV12 signature); when that attempt fails under PREFER_HARDWARE it reports here so health() and the fallback counters stay honest. Fires on_degradation exactly like a run fallback does.

probe()

Run all registered probes → name → available.

health()

Snapshot of routing state, per-op counters and recent fallbacks.

RoutePolicy

class neoruntime_ipc_sdk.RoutePolicy(value)

How the router picks a leg for each operation.

PREFER_HARDWARE uses the hardware leg when it is registered and healthy, falling back to software otherwise. SOFTWARE_ONLY never touches hardware — for benchmarks and reproducible tests. HARDWARE_ONLY raises instead of degrading — when the zero-CPU guarantee matters more than uptime.

PREFER_HARDWARE = 'prefer_hardware'

SOFTWARE_ONLY = 'software_only'

HARDWARE_ONLY = 'hardware_only'

RouteDecision / DegradationRecord

class neoruntime_ipc_sdk.RouteDecision(op, backend, provider, reason)

Why a given backend serves an operation right now.

op: str

backend: str

provider: str

reason: str

__init__(op, backend, provider, reason)

class neoruntime_ipc_sdk.DegradationRecord(op, reason, timestamp)

One hardware→software fallback, kept for health reporting.

op: str

reason: str

timestamp: float

__init__(op, reason, timestamp)

get_default_router

neoruntime_ipc_sdk.accel.get_default_router()

Return the pre-registered router singleton.

The public convenience layer rides this singleton: color.rgb_to_nv12() / color.nv12_to_rgb(), draw.draw_detections() (NV12 arrays) and Frame.to_jpeg_bytes() route through run; Frame.resize() keeps its direct DSP fast path but consults route/policy and reports its fallbacks via note_degradation so health stays honest.

Operations served today: resize_nv12, rgb_to_nv12 and nv12_to_rgb (DSP when reachable, numpy otherwise), encode_jpeg (camera-daemon EncodeImage when reachable, cv2/Pillow otherwise), draw_detections (DSP blend when reachable and the frame is NV12, draw.py raster otherwise), nms (software only — suppression already runs in the HEF’s integrated hardware NMS before the app sees boxes, and its iou_threshold / max_boxes are compile-time there; the runtime-tunable detection_threshold lives in InferenceClient.update_postprocess_config(), honored for family-function postprocess models. See docs/proposals/sdk-hardware-routing.md S-2). OverlayClient.annotate is hardware-first already and needs no routing.

Examples

Default router: hardware first, automatic fallback

from neoruntime_ipc_sdk import get_default_router

router = get_default_router()
small = router.run("resize_nv12", nv12, (1920, 1080), (640, 384))

# Falls back to numpy/cv2 when the DSP is unreachable and records
# the degradation
health = router.health()
print(health["ops"]["resize_nv12"]["backend"])
print(health["recent_degradations"])

Color conversion runs on the hardware leg too

from neoruntime_ipc_sdk import get_default_router

router = get_default_router()
nv12 = router.run("rgb_to_nv12", rgb)               # DSP convert_hw
rgb2 = router.run("nv12_to_rgb", nv12, 1920, 1080)  # same leg, reversed
# The router calls the DSP with cpu_fallback=False — a real
# degradation is recorded once and the software leg runs exactly
# once (no hidden in-client CPU pass first).

# Keep-fd sources (Frame/FrameHandle) pass straight through to the
# *_hw methods: their dma-bufs import zero-copy inside DspClient and
# the router legs skip the ascontiguousarray copy (dsp-offload P2).
# Array sources behave exactly as before.

JPEG encode: daemon one-shot RPC, CPU leg as fallback

router = get_default_router()
jpeg = router.run("encode_jpeg", rgb, quality=85)
# Hardware leg = camera-daemon EncodeImage (DspClient.encode_jpeg_hw,
# N-threaded libjpeg on the DSP core — hailo15 has no dedicated JPEG
# encode block); software leg = cv2/Pillow. When the daemon does not
# expose the RPC it degrades automatically and honestly:
# health()["ops"]["encode_jpeg"]["backend"] shows the live backend.

Detection annotation: NV12 via DSP blend, RGB on the software raster

router = get_default_router()
annotated = router.run("draw_detections", nv12, result)
# The hardware leg renders the annotation as a minimal RGBA canvas
# (draw.render_overlay_rgba) and composites it back onto NV12 in one
# DspClient.blend_hw job — NV12 in, NV12 out. RGB arrays stay on the
# software leg (the draw_detections raster): round-tripping RGB
# through two color converts would cost more than the raster it
# offloads. An empty detection list returns a copy without touching
# the DSP.

Forward degradations to the event bus

from neoruntime_ipc_sdk import EventClient, get_default_router

bus = EventClient()
router = get_default_router()

def report(record):
    bus.publish("app/health/degradation", {
        "op": record.op,
        "reason": record.reason,
    })

router.on_degradation = report

Hardware-only: fail instead of degrading

from neoruntime_ipc_sdk import AccelRouter, RoutePolicy

# A HARDWARE_ONLY router never falls back — a hardware failure
# raises HardwareUnavailable. For call sites where the zero-CPU
# guarantee matters more than uptime.
strict = AccelRouter(policy=RoutePolicy.HARDWARE_ONLY)
strict.register("resize_nv12", hardware=my_dsp_resize)

Capability probes

print(get_default_router().probe())
# {'cv2': True, 'dsp': False}  <- dsp=False: service unreachable now