AI Inference API

AI Inference Client

class neoruntime_ipc_sdk.inference.InferenceClient(endpoint=None)[source]

Bases: GenAiMixin

AI Inference Client

Usage:

inf = InferenceClient()

# Single inference
result = inf.infer(image, model_id="person_v1")

# Stream inference
for frame, res in inf.subscribe(stream="cam0_main", model="person_v1", fps=10):
    print(f"Detected {len(res.objects)} objects")

__init__(endpoint=None)[source]

connect()[source]

property connected: bool

close()[source]

infer(image, model_id, timeout_ms=5000, priority=4, session_id='')[source]

infer_async(image, model_id, timeout_ms=5000, priority=4, session_id='')[source]

Non-blocking infer: returns a concurrent.futures.Future that resolves to a parsed InferenceResult. The caller MUST call fut.result(timeout=…) to obtain the result (or propagate the error).

Enables depth-N pipelines: submit frame N+1 while still awaiting frame N so the NPU stays busy across the host-side gap between jobs. Schedules onto the same background asyncio loop infer() already uses. The existing blocking infer() is unchanged.

infer_batch(items, timeout_ms=10000)[source]

Submit multiple model inferences in a single batch RPC.

ai-runtime runs them in parallel on the NPU via shared VDevice ROUND_ROBIN scheduling, returning all results together.

Parameters:

items (list[BatchInferItem]) – List of (image, model_id, …) tuples.
timeout_ms (int) – Overall wall-clock timeout for the entire batch.

Returns:

List of InferenceResult, one per item, in the same order.

Return type:

list[InferenceResult]

infer_batch_async(items, timeout_ms=10000)[source]

Non-blocking infer_batch: returns a concurrent.futures.Future that resolves to List[InferenceResult] (one per item, in submission order). The caller MUST call fut.result(timeout=…). Symmetric to infer_batch(); enables depth-N pipelines on the dual-model (pose+detect) path. The existing blocking infer_batch() is unchanged.

infer_with_tensors(model_id, inputs, input_names=None, timeout_ms=5000)[source]

subscribe(stream, model, fps=10, session_id='', raw_output_only=False, max_consecutive_failures=10)[source]

Yield (frame_sequence, InferenceResult) for a camera stream subscription.

Failed frames are skipped with a warning. If max_consecutive_failures frames fail in a row (default 10), a RuntimeError is raised instead of yielding nothing forever. Pass 0 or None to disable the limit.

register_model(model_path, model_id=None, owner_id=None, model_type=None, model_variant=None, inputs=None, outputs=None)[source]

unregister_model(model_id)[source]

list_models()[source]

get_model_info(model_id)[source]

get_stats()[source]

create_session(session_id, app_id='', allowed_models=None, max_qps=0, max_concurrent=0, priority=4)[source]

destroy_session(session_id)[source]

update_postprocess_config(model_id, config_json)[source]

Update postprocess configuration for a model at runtime.

For CLIP models, config_json can contain:

{“prompts”: [“a person”, “a car”], “score_threshold”: 0.3}

For detection models, the numeric postprocess keys are accepted:

{“detection_threshold”: 0.38, “iou_threshold”: 0.45,: “max_boxes”: 80}

Applicability, verified on-device (hailo15, 2026-09):

detection_threshold is honored at runtime only when the model’s postprocess resolves to a family function (hailo_yolov8n/hailo_yolov8s/hailo_yolov8m — the default for detection models registered without a backend_function in their variant JSON). Generic plugin exports (e.g. yolov5m_vehicles) hardcode their thresholds and ignore JSON tuning entirely.
iou_threshold / max_boxes are accepted by the chain but have no behavioral effect: suppression and box capping happen in the HEF’s compile-time integrated NMS on the accelerator, so they cannot be moved after compilation.

Unknown keys are rejected server-side (the RPC raises) — push only keys the model’s postprocess schema knows.

Returns True on success.

InferenceClient

class neoruntime_ipc_sdk.InferenceClient(endpoint=None)[source]

Bases: GenAiMixin

AI Inference Client

Usage:

inf = InferenceClient()

# Single inference
result = inf.infer(image, model_id="person_v1")

# Stream inference
for frame, res in inf.subscribe(stream="cam0_main", model="person_v1", fps=10):
    print(f"Detected {len(res.objects)} objects")

__init__(endpoint=None)[source]

connect()[source]

property connected: bool

close()[source]

infer(image, model_id, timeout_ms=5000, priority=4, session_id='')[source]

infer_async(image, model_id, timeout_ms=5000, priority=4, session_id='')[source]

Non-blocking infer: returns a concurrent.futures.Future that resolves to a parsed InferenceResult. The caller MUST call fut.result(timeout=…) to obtain the result (or propagate the error).

Enables depth-N pipelines: submit frame N+1 while still awaiting frame N so the NPU stays busy across the host-side gap between jobs. Schedules onto the same background asyncio loop infer() already uses. The existing blocking infer() is unchanged.

infer_batch(items, timeout_ms=10000)[source]

Submit multiple model inferences in a single batch RPC.

ai-runtime runs them in parallel on the NPU via shared VDevice ROUND_ROBIN scheduling, returning all results together.

Parameters:

items (list[BatchInferItem]) – List of (image, model_id, …) tuples.
timeout_ms (int) – Overall wall-clock timeout for the entire batch.

Returns:

List of InferenceResult, one per item, in the same order.

Return type:

list[InferenceResult]

infer_batch_async(items, timeout_ms=10000)[source]

Non-blocking infer_batch: returns a concurrent.futures.Future that resolves to List[InferenceResult] (one per item, in submission order). The caller MUST call fut.result(timeout=…). Symmetric to infer_batch(); enables depth-N pipelines on the dual-model (pose+detect) path. The existing blocking infer_batch() is unchanged.

infer_with_tensors(model_id, inputs, input_names=None, timeout_ms=5000)[source]

subscribe(stream, model, fps=10, session_id='', raw_output_only=False, max_consecutive_failures=10)[source]

Yield (frame_sequence, InferenceResult) for a camera stream subscription.

Failed frames are skipped with a warning. If max_consecutive_failures frames fail in a row (default 10), a RuntimeError is raised instead of yielding nothing forever. Pass 0 or None to disable the limit.

register_model(model_path, model_id=None, owner_id=None, model_type=None, model_variant=None, inputs=None, outputs=None)[source]

unregister_model(model_id)[source]

list_models()[source]

get_model_info(model_id)[source]

get_stats()[source]

create_session(session_id, app_id='', allowed_models=None, max_qps=0, max_concurrent=0, priority=4)[source]

destroy_session(session_id)[source]

update_postprocess_config(model_id, config_json)[source]

Update postprocess configuration for a model at runtime.

For CLIP models, config_json can contain:

{“prompts”: [“a person”, “a car”], “score_threshold”: 0.3}

For detection models, the numeric postprocess keys are accepted:

{“detection_threshold”: 0.38, “iou_threshold”: 0.45,: “max_boxes”: 80}

Applicability, verified on-device (hailo15, 2026-09):

detection_threshold is honored at runtime only when the model’s postprocess resolves to a family function (hailo_yolov8n/hailo_yolov8s/hailo_yolov8m — the default for detection models registered without a backend_function in their variant JSON). Generic plugin exports (e.g. yolov5m_vehicles) hardcode their thresholds and ignore JSON tuning entirely.
iou_threshold / max_boxes are accepted by the chain but have no behavioral effect: suppression and box capping happen in the HEF’s compile-time integrated NMS on the accelerator, so they cannot be moved after compilation.

Unknown keys are rejected server-side (the RPC raises) — push only keys the model’s postprocess schema knows.

Returns True on success.

Data Types

InferenceResult

class neoruntime_ipc_sdk.InferenceResult(frame_sequence: 'int', timestamp_ns: 'int', objects: 'list[DetectedObject]' = <factory>, classifications: 'list[Classification]' = <factory>, landmarks: 'list[LandmarkSet]' = <factory>, masks: 'list[SegmentationMask]' = <factory>, ocr_lines: 'list[OcrLine]' = <factory>, embeddings: 'list[Embedding]' = <factory>, depth_maps: 'list[DepthMap]' = <factory>, raw_outputs: 'list[np.ndarray] | None' = None, infer_time_us: 'int' = 0, queue_time_us: 'int' = 0, hw_infer_time_us: 'int' = 0, status_message: 'str' = '')[source]

frame_sequence: int

timestamp_ns: int

objects: list[DetectedObject]

classifications: list[Classification]

landmarks: list[LandmarkSet]

masks: list[SegmentationMask]

ocr_lines: list[OcrLine]

embeddings: list[Embedding]

depth_maps: list[DepthMap]

raw_outputs: list[ndarray] | None = None

infer_time_us: int = 0

queue_time_us: int = 0

hw_infer_time_us: int = 0

status_message: str = ''

has_person()[source]

count_by_label(label)[source]

get_objects_by_label(label)[source]

__init__(frame_sequence, timestamp_ns, objects=<factory>, classifications=<factory>, landmarks=<factory>, masks=<factory>, ocr_lines=<factory>, embeddings=<factory>, depth_maps=<factory>, raw_outputs=None, infer_time_us=0, queue_time_us=0, hw_infer_time_us=0, status_message='')

BatchInferItem

class neoruntime_ipc_sdk.BatchInferItem(image, model_id, timeout_ms=5000, priority=4)[source]

A single inference request within a batch.

image: ndarray

model_id: str

timeout_ms: int = 5000

priority: int = 4

__init__(image, model_id, timeout_ms=5000, priority=4)

DetectedObject

class neoruntime_ipc_sdk.DetectedObject(label: 'str', score: 'float', bbox: 'BoundingBox', class_id: 'int' = 0, track_id: 'int | None' = None)[source]

label: str

score: float

bbox: BoundingBox

class_id: int = 0

track_id: int | None = None

__init__(label, score, bbox, class_id=0, track_id=None)

BoundingBox

class neoruntime_ipc_sdk.BoundingBox(x: 'float', y: 'float', width: 'float', height: 'float')[source]

x: float

y: float

width: float

height: float

to_xyxy()[source]

to_xywh()[source]

__init__(x, y, width, height)

LandmarkPoint

class neoruntime_ipc_sdk.LandmarkPoint(x: 'float', y: 'float', confidence: 'float' = 1.0)[source]

x: float

y: float

confidence: float = 1.0

__init__(x, y, confidence=1.0)

LandmarkSet

class neoruntime_ipc_sdk.LandmarkSet(type: 'str', points: 'list[LandmarkPoint]' = <factory>)[source]

type: str

points: list[LandmarkPoint]

__init__(type, points=<factory>)

SegmentationMask

class neoruntime_ipc_sdk.SegmentationMask(class_id: 'int', label: 'str', confidence: 'float', bbox: 'BoundingBox', mask_rle: 'bytes', mask_width: 'int', mask_height: 'int')[source]

class_id: int

label: str

confidence: float

bbox: BoundingBox

mask_rle: bytes

mask_width: int

mask_height: int

to_numpy_mask()[source]

Decode RLE to HxW bool numpy array.

__init__(class_id, label, confidence, bbox, mask_rle, mask_width, mask_height)

OcrLine

class neoruntime_ipc_sdk.OcrLine(text: 'str', confidence: 'float', bbox: 'BoundingBox')[source]

text: str

confidence: float

bbox: BoundingBox

__init__(text, confidence, bbox)

Embedding

class neoruntime_ipc_sdk.Embedding(dim: 'int', data: 'list[float]')[source]

dim: int

data: list[float]

__init__(dim, data)

DepthMap

class neoruntime_ipc_sdk.DepthMap(width: 'int', height: 'int', data: 'np.ndarray')[source]

width: int

height: int

data: ndarray

__init__(width, height, data)

Classification

class neoruntime_ipc_sdk.Classification(type: 'str', class_id: 'int', label: 'str', confidence: 'float')[source]

type: str

class_id: int

label: str

confidence: float

__init__(type, class_id, label, confidence)

ModelInfo

class neoruntime_ipc_sdk.ModelInfo(model_id: 'str', model_path: 'str', version: 'str' = '', inputs: 'list[dict]' = <factory>, outputs: 'list[dict]' = <factory>, estimated_tops: 'float' = 0.0, estimated_memory: 'int' = 0, load_timestamp: 'int' = 0)[source]

model_id: str

model_path: str

version: str = ''

inputs: list[dict]

outputs: list[dict]

estimated_tops: float = 0.0

estimated_memory: int = 0

load_timestamp: int = 0

__init__(model_id, model_path, version='', inputs=<factory>, outputs=<factory>, estimated_tops=0.0, estimated_memory=0, load_timestamp=0)

Usage Examples

Single-shot Inference

from neoruntime_ipc_sdk import InferenceClient
import numpy as np

inf = InferenceClient()

# Prepare image (numpy array)
image = np.zeros((1080, 1920, 3), dtype=np.uint8)

# Execute inference
result = inf.infer(image, model_id="person_v1")

# Process results
for obj in result.objects:
    print(f"{obj.label}: {obj.score:.2f}")
    print(f"  Position: ({obj.bbox.x}, {obj.bbox.y})")
    print(f"  Size: {obj.bbox.width}x{obj.bbox.height}")

# Convenience methods
if result.has_person():
    print("Person detected")

person_count = result.count_by_label("person")
print(f"Person count: {person_count}")

persons = result.get_objects_by_label("person")

Streaming Inference

# Subscribe to video stream inference results
for frame_seq, result in inf.subscribe(
    stream="main",
    model="person_v1",
    fps=15
):
    print(f"Frame {frame_seq}: detected {len(result.objects)} object(s)")

    for obj in result.objects:
        if obj.score > 0.8:
            print(f"  High confidence: {obj.label} ({obj.score:.2f})")

Tensor Inference

import numpy as np

inf = InferenceClient()

# Prepare input tensors
input1 = np.random.randn(1, 3, 224, 224).astype(np.float32)
input2 = np.random.randn(1, 3, 112, 112).astype(np.float32)

# Execute inference
outputs = inf.infer_with_tensors(
    model_id="custom_model",
    inputs=[input1, input2],
    input_names=["input_main", "input_sub"]
)

# Process output tensors
for i, output in enumerate(outputs):
    print(f"Output {i}: shape={output.shape}")

Model Management

# List all models
models = inf.list_models()
for model in models:
    print(f"Model ID: {model.model_id}")
    print(f"Path: {model.model_path}")
    print(f"Version: {model.version}")
    print(f"Inputs: {model.inputs}")
    print(f"Outputs: {model.outputs}")
    print(f"Estimated TOPS: {model.estimated_tops}")
    print(f"Estimated Memory: {model.estimated_memory} bytes")

# Get model details
info = inf.get_model_info("person_v1")
if info:
    print(f"Model ID: {info.model_id}")
    print(f"Path: {info.model_path}")
    print(f"Version: {info.version}")

# Register new model
model_id = inf.register_model(
    model_path="/opt/models/custom.hef",
    model_id="custom_v1"
)
print(f"Registered model ID: {model_id}")

# Unregister model
inf.unregister_model("custom_v1")

Getting Statistics

stats = inf.get_stats()

print(f"Device utilization: {stats['device_utilization']}%")
print(f"Device temperature: {stats['device_temperature']}°C")
print(f"Total memory: {stats['total_memory_bytes']} bytes")
print(f"Used memory: {stats['used_memory_bytes']} bytes")

for model_stat in stats['model_stats']:
    print(f"Model: {model_stat['model_id']}")
    print(f"  Total inferences: {model_stat['total_inferences']}")
    print(f"  Total errors: {model_stat['total_errors']}")
    print(f"  Average latency: {model_stat['avg_latency_us']}us")
    print(f"  Current QPS: {model_stat['current_qps']}")
    print(f"  Queue depth: {model_stat['queue_depth']}")

Session Management

# Create session
session_id = inf.create_session(
    session_id="my_session",
    app_id="my_app",
    allowed_models=["person_v1", "car_v1"],
    max_qps=10,
    max_concurrent=2,
    priority=4
)
print(f"Session ID: {session_id}")

# Use session for inference
result = inf.infer(image, model_id="person_v1", session_id=session_id)

# Destroy session
inf.destroy_session(session_id)

Handling Different Result Types

result = inf.infer(image, model_id="face_detection")

# Detection results
for obj in result.objects:
    # Bounding box
    x1, y1, x2, y2 = obj.bbox.to_xyxy()
    print(f"Bounding box: ({x1}, {y1}) - ({x2}, {y2})")

# Classification results
for cls in result.classifications:
    print(f"Classification: {cls.type} - {cls.label}: {cls.confidence:.2f}")

# Landmarks
for lm_set in result.landmarks:
    print(f"Landmark set type: {lm_set.type}")
    for point in lm_set.points:
        print(f"  Point: ({point.x}, {point.y}), confidence: {point.confidence}")

# Raw outputs
if result.raw_outputs:
    print(f"Raw output count: {len(result.raw_outputs)}")

# Performance info
print(f"Inference time: {result.infer_time_us}us")
print(f"Queue time: {result.queue_time_us}us")

Context Manager

# Use context manager for automatic connection management
with InferenceClient() as inf:
    result = inf.infer(image, model_id="person_v1")
    print(f"Detected {len(result.objects)} object(s)")

Error Handling

from grpc import RpcError

try:
    result = inf.infer(image, model_id="nonexistent_model")
except RpcError as e:
    print(f"Inference failed: {e.details()}")
except RuntimeError as e:
    print(f"Runtime error: {e}")

Segmentation Results

result = inf.infer(image, model_id="segmentation_v1")

for mask in result.masks:
    print(f"Mask: {mask.label} (confidence: {mask.confidence:.2f})")
    print(f"  BBox: ({mask.bbox.x}, {mask.bbox.y}, {mask.bbox.width}, {mask.bbox.height})")

    # Decode RLE mask to numpy bool array (H x W)
    np_mask = mask.to_numpy_mask()
    print(f"  Mask shape: {np_mask.shape}, pixels: {np_mask.sum()}")

OCR Results

result = inf.infer(image, model_id="ocr_v1")

for line in result.ocr_lines:
    print(f"Text: '{line.text}' (confidence: {line.confidence:.2f})")
    print(f"  Position: ({line.bbox.x}, {line.bbox.y})")

Embedding (CLIP Image)

result = inf.infer(image, model_id="clip_vit_b32")

for emb in result.embeddings:
    print(f"Embedding dim: {emb.dim}")
    # Use for similarity search, etc.
    import numpy as np
    vec = np.array(emb.data)
    print(f"  L2 norm: {np.linalg.norm(vec):.4f}")

CLIP Text Encoding

# Encode text to CLIP embedding via NPU
embedding = inf.encode_text("a person walking in the park")
print(f"Embedding length: {len(embedding)}")

# Encode multiple texts and compute similarity
import numpy as np
emb1 = np.array(inf.encode_text("a cat"))
emb2 = np.array(inf.encode_text("a dog"))
similarity = np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2))
print(f"Similarity: {similarity:.4f}")

Depth Estimation

result = inf.infer(image, model_id="depth_v1")

for dm in result.depth_maps:
    print(f"Depth map: {dm.width}x{dm.height}")
    print(f"  Min depth: {dm.data.min():.2f}, Max: {dm.data.max():.2f}")

Update Postprocess Config

import json

# Update CLIP text prompts at runtime
config = json.dumps({
    "prompts": ["a person", "a car", "a bicycle"],
    "score_threshold": 0.3
})
inf.update_postprocess_config("clip_vit_b32", config)

# Tune a detection model's score gate at runtime (no re-registration)
inf.update_postprocess_config("yolov8n", json.dumps({
    "detection_threshold": 0.5
}))

detection_threshold is honored at runtime when the model’s postprocess resolves to a family function (hailo_yolov8n/s/m — the default for detection models registered without a backend_function in their variant JSON); generic plugin exports ignore JSON tuning. iou_threshold and max_boxes are accepted but have no behavioral effect: suppression and capping run in the HEF’s compile-time integrated NMS on the accelerator. Unknown keys make the RPC raise with the server’s rejection (-2801).

GenAI (LLM/VLM)

import json

# Create a GenAI session
session_id = inf.genai_create_session(
    hef_path="/opt/models/llm.hef",
    kind="llm"
)

# Stream generated tokens
messages = [json.dumps({"role": "user", "content": "Hello, who are you?"})]
full_response = ""
for token in inf.genai_generate(
    session_id=session_id,
    messages=messages,
    max_tokens=256,
    temperature=0.7,
    do_sample=True
):
    print(token, end="", flush=True)
    full_response += token

# VLM: generate with image input
with open("image.jpg", "rb") as f:
    img_data = f.read()

vlm_session = inf.genai_create_session("/opt/models/vlm.hef", kind="vlm")
messages = [json.dumps({"role": "user", "content": "Describe this image."})]
for token in inf.genai_generate(
    session_id=vlm_session,
    messages=messages,
    images=[img_data]
):
    print(token, end="", flush=True)

# Abort ongoing generation
inf.genai_abort(session_id)

# Clean up
inf.genai_destroy_session(session_id)
inf.genai_destroy_session(vlm_session)