NeoRuntime Platform Python SDK
Python SDK for NeoRuntime EdgeCam AI Platform
中文接口使用手册:SDK_API_ZH.md
No device yet? The device onboarding page in the docs covers buying an NE503 and first-time login.
Installation
Install from PyPI:
python -m pip install neoruntime-ipc-sdk
Pre-release builds are staged on TestPyPI for validation (dependencies still come from PyPI):
python -m pip install --index-url https://test.pypi.org/simple/ \
--extra-index-url https://pypi.org/simple/ neoruntime-ipc-sdk
Install from source:
git clone https://github.com/camthink-ai/neoruntime-sdks.git
cd neoruntime-sdks
python -m pip install -e ./python
Or build and install a local wheel:
cd neoruntime-sdks/python
python -m pip install --upgrade build
python -m build --wheel
python -m pip install dist/neoruntime_ipc_sdk-*.whl
Learn it hands-on
sdk-teaching-demo is an installable interactive lesson for this SDK:
five live stations against real camera streams (subscribe + infer + draw,
platform overlay, hardware routing, StreamPipeline, events), each showing
the real code behind it plus the error lesson it prevents (DMA -2811,
overlay None vs [], exit codes). Download
sdk-teaching-demo-latest-arm64.neoapp
from the apps repository, or browse
its source.
Quick Start
1. AI Inference
from neoruntime_ipc_sdk import InferenceClient
# Create inference client
inf = InferenceClient()
# Single inference
import numpy as np
image = np.zeros((1080, 1920, 3), dtype=np.uint8)
result = inf.infer(image, model_id="person_v1")
print(f"Detected {len(result.objects)} objects")
# Subscribe to video stream inference results
for frame_seq, result in inf.subscribe(stream="cam0_main", model="person_v1", fps=10):
print(f"Frame {frame_seq}: Detected {len(result.objects)} objects")
for obj in result.objects:
print(f" - {obj.label}: {obj.score:.2f} @ [{obj.bbox.x:.2f}, {obj.bbox.y:.2f}]")
2. Event Bus
from neoruntime_ipc_sdk import EventClient
events = EventClient()
# Publish event
events.publish("app/alert", {
"type": "person_detected",
"zone": "A",
"confidence": 0.95
})
# Subscribe to events (supports wildcards)
for event in events.subscribe("model/*/detections"):
print(f"Received event: {event.topic}")
print(f"Data: {event.payload}")
# Subscribe with callback
def on_alert(event):
print(f"Alert: {event.payload}")
events.on_event("app/alert", on_alert)
3. Device Control
from neoruntime_ipc_sdk import DeviceClient, IrCutMode
dev = DeviceClient()
# Light control
dev.set_white_light(80) # White light brightness 80%
dev.set_ir_led(True) # Turn on IR LED
dev.set_ircut(IrCutMode.NIGHT) # Night vision mode
# PTZ control
dev.pan_left(speed=50)
dev.tilt_up(speed=30)
dev.ptz_stop()
dev.save_preset(1) # Save preset
dev.call_preset(1) # Call preset
# Zoom and focus
dev.zoom_in(speed=50)
dev.zoom_out(speed=50)
dev.set_zoom_level(0.5) # Set zoom to 50%
dev.set_focus_level(0.5) # Set focus to 50%
dev.focus_auto(True)
dev.lens_init() # Initialize lens module
dev.lens_reset_zero(zoom=True, focus=True) # Reset both axes
dev.oneshot_autofocus() # One-shot autofocus
dev.set_lens_limits(zoom_limit={"min_pos": 0, "max_pos": 1000}) # Set lens limits
status = dev.get_lens_status() # Get lens status dict
dev.lens_goto_ratio_distance(2.0, 3.0) # Zoom+focus linked move
# GPIO
dev.gpio_set(pin=10, value=True)
value = dev.gpio_get(pin=11)
# Get device status
status = dev.get_device_status()
print(f"SoC Temperature: {status.soc_temp_c}C")
print(f"White light level: {status.white_light_level}")
4. Video Stream Access
from neoruntime_ipc_sdk import FdMediaClient
media = FdMediaClient()
# List available streams
streams = media.list_streams()
print(f"Available streams: {streams}") # ['main', 'sub']
# Get single frame
frame = media.get_frame("main")
if frame:
print(f"Frame size: {frame.width}x{frame.height}, format: {frame.format}")
rgb_image = frame.to_rgb() # Convert to RGB format
# Subscribe to video stream
for frame in media.subscribe("main"):
# frame.image is the decoded numpy array
process_frame(frame.image)
# Encoded stream (H.264/H.265): get_encoded_stream() returns a client
for packet in media.get_encoded_stream("main").subscribe():
print(f"{packet.codec_name()} packet: {len(packet.data)} bytes")
# Use callback
def process(frame):
print(f"Frame: {frame.sequence}")
media.on_frame("main", process)
5. Complete Example: AI + Device Linkage
from neoruntime_ipc_sdk import InferenceClient, DeviceClient, EventClient
# Initialize clients
inf = InferenceClient()
dev = DeviceClient()
events = EventClient()
# Subscribe to person detection results
for frame_seq, result in inf.subscribe(stream="cam0_main", model="person_v1"):
# Person detected
if result.has_person():
# Turn on white light
dev.set_white_light(100)
# Publish alert event
events.publish("app/perimeter_alert", {
"person_count": result.count_by_label("person"),
"objects": [
{"label": obj.label, "score": obj.score}
for obj in result.objects
]
})
else:
# Turn off white light
dev.set_white_light(0)
7. Inference Pipeline (client-side pre/post)
For custom models whose postprocess is not handled server-side, or apps
that need their own thresholds — InferencePipeline composes
preprocessing (derived from the model's input spec, DSP-scaled for
keep-fd frames), inference, and a client-side YOLO decoder whose
thresholds are yours:
from neoruntime_ipc_sdk import (
FdMediaClient, InferencePipeline, YoloV8Postprocessor,
)
pipe = InferencePipeline.from_model(
"custom_yolov8n",
postprocessor=YoloV8Postprocessor(labels=["person", "car"], score_threshold=0.3),
)
for frame in FdMediaClient().subscribe("sub", keep_fd=True):
out = pipe.run(frame) # out.objects in source-frame coordinates
Preprocessor also works standalone: tensor, meta = pre(frame) with
meta.to_source(x, y) mapping model-input coordinates back to the frame.
When a family model fits, InferenceClient.subscribe remains the cheaper
path — see PERFORMANCE.md for the decision table.
For long-running apps, PipelineRunner productises the whole loop — a
reader thread pulling frames into a latest-wins queue, a worker thread
running the pipeline and handing results to a sink, keep-fd handles
released on consumption, drops/latency/errors visible in stats:
from neoruntime_ipc_sdk import FdMediaClient, InferencePipeline, PipelineRunner
media = FdMediaClient()
runner = PipelineRunner(
source=media.subscribe("sub", keep_fd=True),
pipeline=InferencePipeline.from_model("person_v1"),
sink=lambda out, frame: print(out.objects),
queue_size=1, # latest-wins: always infer the newest frame
)
runner.start()
print(runner.stats) # {"fps": ..., "dropped": ..., "recent_latency_ms": ...}
runner.stop() # or: with PipelineRunner(...) as runner: ...
Sinks receive (result, frame) on the worker thread and must not retain
the frame beyond the call — the runner releases it right after.
8. App Toolkit: Drawing, Recording, and Web Streaming
Detection visualization on a live frame (pure numpy/PIL, cv2 optional):
from neoruntime_ipc_sdk import FdMediaClient, draw_detections
frame = FdMediaClient().get_frame("main")
img = frame.crop(0, 0, 1920, 1080).resize(960, 540).to_rgb()
for seq, result in inf.subscribe(stream="cam0_main", model="person_v1"):
annotated = draw_detections(img, result) # boxes + labels + scores
break
HLS recording from the encoded stream — no ffmpeg on the device needed:
from neoruntime_ipc_sdk import EncodedStreamClient, HlsWriter
with HlsWriter("/srv/hls", segment_seconds=4.0, window=5) as hls:
for frame in EncodedStreamClient("/run/aipc/encoded/main.sock").subscribe():
hls.write(frame) # cuts on keyframes
# serve /srv/hls/ over HTTP and point hls.js at index.m3u8
MJPEG preview page — one WSGI app (mount in Flask) or a standalone server:
from neoruntime_ipc_sdk import FdMediaClient, MjpegServer, MjpegStream
source = MjpegStream()
MjpegServer(port=8080, source=source).start() # http://device:8080/
for frame in FdMediaClient().subscribe("sub"):
source.push_frame(frame.resize(640, 360)) # slow clients just drop frames
API Reference
InferenceClient
AI inference client for model inference and streaming inference subscription.
Methods:
| Method | Parameters | Returns | Description |
|---|---|---|---|
connect() |
- | - | Connect to service |
close() |
- | - | Close connection |
infer(image, model_id, timeout_ms, priority, session_id) |
ndarray, str, int, int, str | InferenceResult | Single inference |
infer_with_tensors(model_id, inputs, input_names, timeout_ms) |
str, List[ndarray], List[str], int | List[ndarray] | Multi-tensor inference |
subscribe(stream, model, fps, session_id, raw_output_only, max_consecutive_failures, queue_size) |
str, str, int, str, bool, Optional[int], Optional[int] | Iterator[Tuple[int, InferenceResult]] | Streaming inference; failed frames are skipped with a warning and a RuntimeError is raised after 10 consecutive failures (0/None disables). The client-side queue is bounded (queue_size, default 100): a slow consumer drops its oldest queued results — counted on the iterator's dropped attribute and reported in a throttled warning — instead of growing memory without bound (0/None = unbounded) |
register_model(model_path, model_id, inputs, outputs) |
str, str, List[Dict], List[Dict] | str | Register model |
unregister_model(model_id) |
str | - | Unregister model |
list_models() |
- | List[ModelInfo] | List models |
get_model_info(model_id) |
str | ModelInfo | Get model info |
get_stats() |
- | Dict | Get statistics |
create_session(session_id, ...) |
str, ... | str | Create session |
destroy_session(session_id) |
str | - | Destroy session |
Data Classes:
BoundingBox: x, y, width, heightDetectedObject: label, score, bbox, class_id, track_idInferenceResult: frame_sequence, timestamp_ns, objects, classifications, landmarks, raw_outputsModelInfo: model_id, model_path, version, inputs, outputs
EventClient
Event bus client for publishing and subscribing to events.
Methods:
| Method | Parameters | Returns | Description |
|---|---|---|---|
publish(topic, payload, persistent, ttl_ms, metadata) |
str, dict, bool, int, dict | str | Publish event |
publish_batch(events, persistent) |
List[dict], bool | - | Batch publish |
subscribe(topic, filters, queue_size, drop_old) |
str, dict, int, bool | Iterator[Event] | Subscribe to events |
on_event(topic, callback, filters) |
str, Callable, dict | Thread | Callback subscription |
unsubscribe(topic) |
str | - | Unsubscribe |
list_topics() |
- | List[TopicInfo] | List topics |
get_topic_info(topic) |
str | TopicInfo | Get topic info |
get_stats() |
- | Dict | Get statistics |
Data Classes:
Event: topic, payload, source, event_id, timestamp_ns, metadataTopicInfo: topic, subscriber_count, total_messages, last_message_ts
DeviceClient
Device control client for controlling camera peripherals.
Light Control:
set_white_light(level: int)- Set white light brightness (0-100)set_ir_led(on: bool)- IR LED switchset_ircut(mode: IrCutMode)- IR-Cut mode
PTZ Control:
pan_left(speed: int)/pan_right(speed: int)- Pan left/righttilt_up(speed: int)/tilt_down(speed: int)- Tilt up/downptz_stop()- Stop PTZsave_preset(preset_id: int)/call_preset(preset_id: int)- Preset operations
Lens Control:
zoom(speed: int)- Zoom (-100 ~ 100)zoom_in(speed: int)/zoom_out(speed: int)- Zoom in/outset_zoom_level(level: float)- Set zoom level (0-1)focus(speed: int)- Focus (-100 ~ 100)focus_in(speed: int)/focus_out(speed: int)- Focus in/outset_focus_level(level: float)- Set focus level (0-1)focus_auto(enable: bool)- Auto focusoneshot_autofocus(timeout: float)- One-shot autofocus (enable → wait → disable)lens_init()- Initialize lens modulelens_reset_zero(zoom: bool, focus: bool)- Reset lens axes to zeroset_lens_limits(zoom_limit, focus_limit)- Set lens axis position limitslens_goto_ratio_distance(zoom_ratio: float, focus_distance_m: float)- Zoom+focus linked movecontrol_iris(open: bool)- Open/close irisset_iris_target(target: int)- Set iris target valueget_lens_status()- Get lens status dict (positions, states, limits)
Autofocus (native jobs):
start_oneshot_af()- Start a one-shot AF job, returnsAfJobstart_zoom_follow(ratio: float)- Start continuous AF following a zoom ratio, returnsAfJobget_autofocus_status()- Poll the AF engine, returnsAfStatus(state, progress, busy, ...)cancel_autofocus(job_id: int = 0)- Cancel a running AF jobset_af_windows(enabled, windows, stream_id: str = "main")- Restrict AF metering to windowsget_af_measurement()- Read AF statistics, returnsAfMeasurement(focus_energy, mean_luma, ...)
Known device limitation: on current firmware (verified on a test device, 2026-08) the lens HAL bridge rejects
set_af_windowsandget_af_measurementwith "not yet supported". The SDK-side plumbing is complete; both calls will start working once the device firmware adds bridge support. Oneshot, zoom-follow, status, and cancel all work today.
GPIO:
gpio_set(pin: int, value: bool)- GPIO outputgpio_get(pin: int)- GPIO input
Wiegand:
set_wiegand_out(channel: int, enable: bool)- Wiegand output controlget_wiegand_out(channel: int)- Wiegand output state query
RS-485:
rs485_init(baudrate: int, config: str)- RS-485 initializationrs485_deinit()- RS-485 deinitializationrs485_tx(data: bytes)- RS-485 data transmission
Status Query:
get_device_status()- Get device statussubscribe_events()- Subscribe to device events
Enums:
IrCutMode: AUTO, DAY, NIGHT
AppClient
Application container management client.
Lifecycle:
install_app(manifest_path, image_path)- Install applicationstart_app(app_id)- Start applicationstop_app(app_id, timeout_seconds)- Stop applicationrestart_app(app_id, timeout_seconds)- Restart application (stop + start)uninstall_app(app_id, keep_logs)- Uninstall application
Query:
list_apps()- List all applicationsget_app(app_id)- Get application infoget_app_stats(app_id)- Get application statisticsget_logs(app_id, max_lines, follow)- Stream application logsget_logs_text(app_id, max_lines, follow)- Stream logs as text
Other:
register_web_url(path)- Register web access path
FdMediaClient (media)
Video stream client receiving frames over UDS (dma-buf fds, decoded on receive by default).
Methods:
| Method | Parameters | Returns | Description |
|---|---|---|---|
subscribe(stream_id, skip_frames, keep_fd) |
str, int/bool, bool | Iterator[Frame] | Subscribe to video stream; skip_frames is a decimation interval — 1 (or the historical True/False) yields every frame, N>1 yields every Nth (decimated keep-fd frames are released back to the daemon immediately) |
subscribe_raw(stream_id, skip_frames, keep_fd) |
str, int/bool, bool | Iterator[Frame] | Same as subscribe |
get_frame(stream_id, timeout_ms, keep_fd) |
str, int, bool | Frame | None | Get single frame |
get_encoded_stream(stream_id) |
str | EncodedStreamClient | H.264/H.265 Annex-B stream client |
list_streams() |
- | List[str] | List active stream IDs from the camera daemon (falls back to main / sub when the service is unreachable) |
get_rtsp_url(stream_id, host, port) |
str, str, int | str | RTSP playback URL (needs RTSP enabled) |
on_frame(stream_id, callback) |
str, Callable | Thread | Callback subscription |
close() |
- | - | Close connection |
Data Classes:
Frame: sequence, timestamp_ns, width, height, format, image, metadata, handleframe.image/frame.to_array()- Decoded numpy array (H, W, C) or (H*3//2, W) for NV12Frame.crop(x, y, w, h)- New cropped Frame (NV12 needs even x/y/w/h)Frame.resize(width, height, mode="letterbox", pad_value=114)- New resized Frame (stretch/letterbox/crop)Frame.to_jpeg_bytes(quality=85)- JPEG bytes (cv2 fast path, PIL fallback)frame.release()- Return akeep_fd=Truebuffer to the daemon (idempotent)frame.metadata["transform"]- Affine geometry of the last crop/resize chain:dst = src * scale + originper axis (composed across chained calls). Map model-input coordinates back to the source frame withsrc = (dst - origin) / scale— no hand-derived letterbox maths in the appEncodedStreamClient/EncodedFrame: encoded stream subscription;EncodedFrame.data(Annex-B bytes),.is_keyframe(),.codec_name()StreamInfo,PixelFormat: NV12, NV21, RGB, BGR, RGBA, BGRA, GRAY8, YUYV
OverlayClient (overlay)
Draws results on the device's hardware overlay: camera-daemon bakes boxes and zone polygons into the encoded stream, so the app never rasterizes pixels (use the draw module instead when you own the pixels, e.g. your own MJPEG preview). The overlay must be enabled first.
Methods:
| Method | Parameters | Returns | Description |
|---|---|---|---|
enable(show_label, show_confidence, line_thickness) |
bool, bool, int | - | Turn the overlay on with the given appearance |
disable() |
- | - | Turn the overlay off |
configure(enabled, show_label, show_confidence, line_thickness, box_color, label_color, font_size, strict_frame_lock, strict_wait_cap_ms) |
see notes | - | Full appearance control. Colors are ARGB ints. strict_frame_lock=True binds each box to the frame it was computed from (the bake site waits, bounded by strict_wait_cap_ms); None keeps the daemon's current setting |
apply(config) |
OverlayConfig | - | Apply an OverlayConfig object |
annotate(stream_id, detections, polygons, ttl_ms, ...) |
str, list, list, int | str | Push detections and/or zone polygons for one stream (event id) |
annotate_result(stream_id, result, ttl_ms, ...) |
str, InferenceResult | str | Push whichever section of an InferenceResult is populated (detections, then classifications, landmarks, OCR lines) |
close() |
- | - | Close connection |
Data Classes:
OverlayConfig: enabled, show_label, show_confidence, line_thickness, box_color, label_color, font_size, strict_frame_lock, strict_wait_cap_ms
Results expire ~two frame periods after the last event, so call annotate at a few Hz while detections are fresh. The None-vs-empty distinction matters: detections=None leaves the stream's previous boxes untouched (polygons can ride alone), while detections=[] clears them — same for polygons. Polygon points are normalized [0, 1] floats (2-128 points each, at most 16 polygons per call); ttl_ms must be a positive int (omitted = derived from the stream fps).
from neoruntime_ipc_sdk import OverlayClient
oc = OverlayClient()
oc.enable()
for result in inference.subscribe("main", "person_v1"):
oc.annotate("main", result.objects) # fresh boxes every frame
oc.annotate("main", detections=[]) # done: clear stale boxes
oc.annotate("main", polygons=[{"points": ZONE, "label": "yard"}])
StreamPipeline (stream_pipeline)
One-call pipeline: subscribe a registered model to a camera stream, draw every result on the hardware overlay, and yield the same results to the app. Single-shot (create a new instance per run) and usable as a context manager.
from neoruntime_ipc_sdk import StreamPipeline
pipe = StreamPipeline(
"main", "person_v1", fps=10,
polygons=[{"points": [[0, 0], [1, 0], [1, 1], [0, 1]], "label": "zone"}],
)
pipe.start()
for sequence, result in pipe.results():
... # drawing already happened on the overlay
pipe.stop() # clears boxes and zones
Methods:
| Method | Parameters | Returns | Description |
|---|---|---|---|
start() |
- | StreamPipeline | Subscribe, apply overlay config / static zones, start drawing |
stop() |
- | - | Stop the worker, clear what was drawn, close owned clients. Idempotent, safe to call from on_result |
results() |
- | Iterator[(int, InferenceResult)] | Yield (frame_sequence, result); bounded drop-oldest queue (counted in status()) |
status() |
- | StreamPipelineStatus | Health snapshot incl. pass-through subscribe stats |
Constructor notes: draw=False keeps the encoded stream clean (results still flow to results()); min_score / labels filter drawing only — a fully-filtered result still publishes an empty detection list, clearing stale boxes; on_result (worker thread) may replace or drop each result; overlay_config accepts an OverlayConfig; ttl_ms overrides per-event validity; the subscription runs at stream geometry, so register the model for the stream's dimensions.
Data Classes:
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
Diagnostics
diagnostics() (never raises) snapshots the SDK's execution environment — the first stop when an app feels slow:
from neoruntime_ipc_sdk import diagnostics
report = diagnostics()
report["optional_deps"]["cv2"] # {"available": bool, "version": ...}
report["services"]["camera_control"] # {"path", "exists", "connectable"}
report["media"]["encoded_dir"]["streams"]
report["accel_router"]["ops"]["nv12_to_rgb"]["backend"] # "hardware" | "software"
Without cv2 installed, conversions/resize/JPEG run on pure-numpy fallbacks that are 10-50x slower at frame rate — the SDK warns once per operation on that path.
diagnostics.check() grades the snapshot against an app's requirements — a deployment gate:
from neoruntime_ipc_sdk.diagnostics import check
check(["inference", "camera_sock"]).raise_if_unhealthy() # RuntimeError listing failures
# live model/stream probes are opt-in:
check(["model_registered", "stream_active"],
inference=InferenceClient(), model_id="person_v1",
camera=CameraClient(), stream_id="main").raise_if_unhealthy()
Hardware/software routing is one app-wide knob (see AccelRouter.health() for what is actually serving each operation):
from neoruntime_ipc_sdk import set_route_policy
set_route_policy("software_only") # baseline/benchmark; "prefer_hardware" (default), "hardware_only" (refuse silent CPU fallback)
Architecture and stability
The SDK is layered, and the middle layer is a stability contract:
- Daemon clients —
InferenceClient,FdMediaClient,CameraClient,DeviceClient,EventClient,AppClient,OverlayClient,AudioClient: communication, timeouts, reconnection, resource release. - Data & algorithm primitives (stable contract) —
Frame/FrameHandle,PreprocessMeta,DetectedObject, and the callable protocolspreprocessor(frame) -> (tensor, meta)/postprocessor(raw_outputs, meta) -> objects. The built-inPreprocessorandYoloV5/V8Postprocessorimplement them; custom decoders (RetinaFace, OCR, pose...) can replace them freely. Breaking changes here are gated behind a minor-version bump with deprecation warnings. - App pipeline —
InferencePipeline(single pass) andPipelineRunner(long-running): composition, backpressure (latest-wins with drop counters), latency/error stats, frame ownership.
Recording (recording)
Pure-python MPEG-TS muxing of EncodedFrame Annex-B payloads — no ffmpeg required.
TsWriter(path, codec="h264")- Single-file .ts event clipHlsWriter(out_dir, segment_seconds=6.0, window=5)- Keyframe-aligned HLS segments + liveindex.m3u8PrerollBuffer(seconds=10.0)- Ring buffer;push(frame),dump("event.ts")writes "seconds before the event"
Web Streaming (web)
MJPEG helpers for app preview pages.
MjpegStream()- Thread-safe latest-frame holder;push_frame(frame, quality=85)/push_jpeg(data)mjpeg_wsgi_app(source, fps=15)- WSGI callable — mount straight into FlaskMjpegServer(port=8080, host="0.0.0.0", source=...)- Standalone threaded HTTP server
Drawing (draw)
Detection visualization on RGB numpy arrays (returns new arrays, input untouched).
draw_boxes(image, boxes, labels=None, scores=None, color=(0,255,0), thickness=2)draw_text(image, text, xy, color=(255,255,255), font_scale=0.5, thickness=1)draw_detections(image, result_or_objects, color=None)- AcceptsInferenceResult/DetectedObject/ raw(x1,y1,x2,y2)tuples
Audio (audio, audio_stream)
Audio HAL via camera-daemon: device enumeration, capture/playback control, and two-way-talk PCM streaming (AudioClient); captured frames arrive over UDS (AudioStreamClient).
Methods (AudioClient):
| Method | Parameters | Returns | Description |
|---|---|---|---|
list_capture_devices() / list_playback_devices() |
- | List[AudioDevice] | Enumerate audio devices |
start_capture(device, sample_rate, channels, codec, bitrate) |
str, int, int, str, int | - | Start capture (0/empty = daemon defaults) |
stop_capture() |
- | - | Stop capture |
start_playback(device, sample_rate, channels) |
str, int, int | - | Start playback |
stop_playback() |
- | - | Stop playback |
get_status() |
- | AudioStatus | Current audio status |
set_config(volume, mute, ...) |
float, bool | - | Update volume/mute/codec (defaults keep current values) |
stream_pcm(pcm_iter, sample_rate, channels, fmt) |
Iterator[bytes], int, int, str | - | Stream PCM chunks to the device (two-way talk) |
stream_pcm_file(path, chunk_size, ...) |
str, int | - | Stream a raw PCM file |
Data Classes:
AudioDevice: name, descriptionAudioStatus: capturing, playing, device, sample_rate, channels, codec, volume, muteAudioFrame(fromAudioStreamClient): codec (0=pcm, 1=aac, 2=g711a, 3=g711u), flags (bit0 = keyframe), pts_ns, sample_rate, channels, bits_per_sample,databytes; helpersis_keyframe(),codec_name(),duration_ms()AudioStreamClient(socket_path=None):subscribe()→ Iterator[AudioFrame],on_frame(cb),close(); default socket/run/aipc/encoded/audio_capture.sock(envAUDIO_CAPTURE_SOCK_PATH)
from neoruntime_ipc_sdk import AudioClient, AudioStreamClient
audio = AudioClient()
audio.start_capture(codec="aac", sample_rate=48000)
for frame in AudioStreamClient().subscribe():
print(frame.codec_name, len(frame.data))
audio.stop_capture()
DSP Offload (dsp)
Hardware image ops on the camera-daemon DSP service: resize, crop, convert, blend, and JPEG encode without burning CPU. Apps usually reach this indirectly through the accel router (set_route_policy / AccelRouter.health()); use DspClient directly for ops the router does not cover, or to pre-allocate buffer pools in hot loops.
DspClient.resize_hw(src, width, height, fmt=None, ...)- HW resize (bilinear;stretchscaling)DspClient.crop_hw(src, x, y, width, height, dst_width=None, dst_height=None, ...)- HW crop, optional resize in the same jobDspClient.multi_crop_hw(src, rects, ...)- Many crop/resize windows in one jobDspClient.convert_hw(src, dst_fmt, ...)-nv12/rgb24/gray8conversion, dimensions kept; the supported pair matrix is device-dependent, and a pair the firmware refuses falls back to CPU whencpu_fallback=True(the default)DspClient.blend_hw(base, overlays, ...)- Composite ARGB32 overlays onto an NV12 baseDspClient.encode_jpeg_hw(src, quality=85)- One JPEG frame, returns bytesDspClient.alloc_buffers(width, height, fmt="nv12", count=1)→DspBufferPool- Pre-allocate daemon-side buffers so hot-loop calls only write, submit, readDspClient.import_frame(handle)/release_buffer(buffer_id)- Import aFrameHandle's dma-bufs into the daemon registry / free an id
Every *_hw method takes cpu_fallback=True by default: when the daemon lacks the DSP surface, array-source calls warn and compute on CPU; pass False to make unavailability raise instead. With wait=False a call returns a PendingDspJob (done(timeout_s), wait_result(), wait()) instead of blocking.
Data Classes:
DspBufferPool:write(index, arr)/read(index)/release()- Pre-allocated daemon-side buffersDspBufferRef: buffer_id, width, height, fmt;read()/release()- Zero-copy result handle (out="ref")PendingDspJob:done(timeout_s)/wait_result()/wait()- Async job handle (wait=False)DspError- DSP failures (allocation, import, firmware refusals)
from neoruntime_ipc_sdk import DspClient
dsp = DspClient()
small = dsp.resize_hw(frame.image, 640, 384, fmt="nv12")
# hot loop: allocate once, then every call only writes / submits / reads
pool = dsp.alloc_buffers(frame.width, frame.height, fmt="nv12", count=2)
Config
Configuration management, reads from environment variables.
Static Methods:
get_app_id()- Get application IDget_inference_endpoint()- AI Runtime endpointget_event_bus_endpoint()- Event Bus endpointget_device_control_endpoint()- Device Control endpointget_shm_base_path()- SHM base pathis_debug()- Debug modeget_log_level()- Log level
Environment Variables
SDK automatically reads configuration from environment variables:
| Variable | Default | Description |
|---|---|---|
APP_ID |
unknown | Application ID |
AI_RUNTIME_ENDPOINT |
unix:///run/aipc/ai-runtime.sock | AI Runtime endpoint |
EVENT_BUS_ENDPOINT |
unix:///run/aipc/event-bus.sock | Event Bus endpoint |
DEVICE_CONTROL_ENDPOINT |
unix:///run/aipc/device-control.sock | Device Control endpoint |
SHM_BASE_PATH |
/run/aipc/shm | SHM base path |
DEBUG |
0 | Debug mode |
LOG_LEVEL |
INFO | Log level |
Development
Protobuf Stubs
The generated protobuf stubs in neoruntime_ipc_sdk/proto/ (*_pb2.py / *_pb2_grpc.py)
are committed to the repo so the SDK imports cleanly on a fresh clone, editable
install, and inside packaged wheels. They are re-included via sdk/python/.gitignore
and do not affect the global "no generated artifacts" policy for Go services.
If you change any .proto source, regenerate and re-commit them:
make sdk-proto # regenerate stubs (inference/event/device/app/camera)
make sdk-proto-check # verify committed stubs match .proto sources
git add sdk/python/neoruntime_ipc_sdk/proto/*_pb2*.py
Run Tests
cd sdk/python
pip install -e ".[dev]"
pytest tests/
Build Package
python setup.py build
Build Wheel
The recommended way to build a distributable wheel is:
python -m pip install --upgrade build
python -m build --wheel
ls dist/*.whl
The generated wheel is written to dist/, for example:
pip install dist/neoruntime_ipc_sdk-*.whl
For older tooling, this also works:
python setup.py bdist_wheel
Do not commit files from dist/; publish them as release artifacts instead.
Automated Wheel Builds
In the public neoruntime-sdks repository, GitHub Actions builds and tests
the wheel and sdist for pull requests, pushes to main, and manual workflow
runs, uploading them as a workflow artifact named python-sdk-dist.
Pushing a v* tag (or running the workflow manually with publishing enabled)
creates a GitHub Release with the distributions attached and publishes to
TestPyPI and then PyPI:
git tag v0.8.1
git push origin v0.8.1
The release tag version must match the package version in setup.py and
neoruntime_ipc_sdk/__init__.py.
License
MIT License. See the repository LICENSE file.
Metadata
Release files for neoruntime-ipc-sdk 0.8.1
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| neoruntime_ipc_sdk-0.8.1.tar.gz | 372.5 kB | With provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| neoruntime_ipc_sdk-0.8.1-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 609.6 kB
Release files / neoruntime_ipc_sdk-0.8.1.tar.gz
| Download URL | neoruntime_ipc_sdk-0.8.1.tar.gz |
|---|---|
| Size | 372.5 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
752cfdb5c0471f6a1947afb8c4468349604cc24ccb440fd7d44d5801f12b6f6f
|
|
BLAKE2b-256 checksum How to use checksums |
938dfd0dff4c0467d8f4a44761b0e0773fb5fec5df79d80e2241fc01911810ab
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 15, 2026.
Transparency logRelease files / neoruntime_ipc_sdk-0.8.1-py3-none-any.whl
| Download URL | neoruntime_ipc_sdk-0.8.1-py3-none-any.whl |
|---|---|
| Size | 237.1 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
667d285ff0eedf471e2ec4cb9f443ddf4588ad65811ceb86d349cb84c26359d0
|
|
BLAKE2b-256 checksum How to use checksums |
b6cf0f371170f27f3606c9b97b868a517a8d72e316f2f58edec192753caf0a4e
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 15, 2026.
Transparency logSecurity for neoruntime-ipc-sdk 0.8.1
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
Full provenance
All 2 files in this release have been uploaded with provenance metadata.
Attestation for neoruntime_ipc_sdk-0.8.1.tar.gz
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 15, 2026.
Transparency logA note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.