加速路由 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)[source]

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

set_policy(policy)[source]

Switch the routing policy at runtime (also accepts its value name).

register(op, software=None, hardware=None, note='')[source]

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

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

add_probe(name, probe)[source]

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

property policy: RoutePolicy: The active policy (fixed at construction).

route(op)[source]

Decide which backend serves op under the current policy.

run(op, *args, **kwargs)[source]

Execute op through the leg chosen by route().

note_degradation(op, reason)[source]

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

Run all registered probes → name → available.

health()[source]

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

class neoruntime_ipc_sdk.accel.DegradationRecord(op, reason, timestamp)[source]

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

Bases: RuntimeError

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

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

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

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

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.

neoruntime_ipc_sdk.accel.shared_dsp_call(method, *args, **kwargs)[source]

Run one DspClient method on the process-resident client.

The same resident machinery the router legs use, exposed for SDK fast paths that live outside the route table (Frame.resize’s DSP arm). On any error the resident client is retired — the next call reconnects — and the original exception propagates so the caller keeps its own fallback policy. Retirement defers the close until no shared call is in flight and no live pool remains, so a failure in one thread cannot cut the transport under a concurrent call or kill a ref another thread is holding. The resident is never retired on success.

类与函数

AccelRouter

class neoruntime_ipc_sdk.AccelRouter(policy=RoutePolicy.PREFER_HARDWARE)[source]

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

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

set_policy(policy)[source]

Switch the routing policy at runtime (also accepts its value name).

register(op, software=None, hardware=None, note='')[source]

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

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

add_probe(name, probe)[source]

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

property policy: RoutePolicy: The active policy (fixed at construction).

route(op)[source]

Decide which backend serves op under the current policy.

run(op, *args, **kwargs)[source]

Execute op through the leg chosen by route().

note_degradation(op, reason)[source]

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

Run all registered probes → name → available.

health()[source]

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

RoutePolicy

class neoruntime_ipc_sdk.RoutePolicy(value)[source]

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

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

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

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.

使用示例

默认路由器：硬件优先，自动降级

from neoruntime_ipc_sdk import get_default_router

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

# DSP 不可用时自动回落 numpy/cv2，并记录一次降级
health = router.health()
print(health["ops"]["resize_nv12"]["backend"])
print(health["recent_degradations"])

颜色转换同样走硬件腿

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)  # 反方向同腿
# 路由器以 cpu_fallback=False 调用 DSP——真降级时只记一次，
# 软件腿恰好执行一遍（不会先在客户端内部悄悄跑一遍 CPU）。

# keep-fd 源（Frame/FrameHandle）直接透传给 *_hw：dma-buf 在
# DspClient 内部零拷贝导入，路由腿不再 ascontiguousarray 拷贝
#（dsp-offload P2）。数组源行为不变。

JPEG 编码：daemon 一发 RPC，CPU 兜底

router = get_default_router()
jpeg = router.run("encode_jpeg", rgb, quality=85)
# 硬件腿 = camera-daemon EncodeImage（DspClient.encode_jpeg_hw，
# DSP 核上 N 线程 libjpeg——hailo15 无专用 JPEG 编码块），
# 软件腿 = cv2/Pillow。daemon 未暴露该 RPC 时自动降级并记账，
# health()["ops"]["encode_jpeg"]["backend"] 如实反映当前后端。

检测标注：NV12 走 DSP blend，RGB 走软件光栅

router = get_default_router()
annotated = router.run("draw_detections", nv12, result)
# 硬件腿把标注渲染成最小 RGBA 画布（draw.render_overlay_rgba），
# 一次 DspClient.blend_hw 合成回 NV12——输入 NV12 返回 NV12。
# RGB 数组留在软件腿（draw_detections 光栅）：把 RGB 折腾两次
# 色彩转换再合成，比省下的光栅还贵。空检测列表直接返回拷贝，
# 不碰 DSP。

降级事件转发到事件总线

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

只信硬件：失败即抛错

from neoruntime_ipc_sdk import AccelRouter, RoutePolicy

# HARDWARE_ONLY 路由器不降级——硬件失败直接抛 HardwareUnavailable，
# 适合零 CPU 占用比可用性更重要的调用点
strict = AccelRouter(policy=RoutePolicy.HARDWARE_ONLY)
strict.register("resize_nv12", hardware=my_dsp_resize)

能力探测

print(get_default_router().probe())
# {'cv2': True, 'dsp': False}  ← dsp=False 表示服务当前不可达