AI Box Library

Python library for the NXP Edge AI Industrial Platform.

Overview

ai-box-lib provides thin clients for component-to-component communication:

• DataCollectorClient for publishing data collector payloads
• PreProcessorClient for subscribing data collector payloads and publishing features
• PostProcessorClient for subscribing ML model results and publishing processed results
• ContextEngineClient for publishing context messages
• ChannelClient for generic pub/sub communication between any components

Installation

Install from PyPI:

pip install ai-box-lib

Quick Start

1) Publish data from a data collector

from ai_box_lib.data_collector_client import DataCollectorClient

client = DataCollectorClient[dict]()
client.connect()
client.publish_timestream({"random_number": "42"})

2) Subscribe and publish from a pre-processor

from ai_box_lib.pre_processor_client import PreProcessorClient

client = PreProcessorClient[dict, dict]()
client.connect()

def handle_raw(message: dict) -> None:
	processed = {"feature_a": [1.0, 2.0, 3.0]}
	client.publish_data(processed)

unsubscribe = client.subscribe_timestream(handle_raw)

3) Reduce model output in a post-processor

from ai_box_lib.post_processor_client import PostProcessorClient

client = PostProcessorClient[dict, dict]()
client.connect()

def handle_inference(message: dict) -> None:
	client.publish_object_detection([[10, 20, 30, 40, 0.9, 2]])  # x, y, w, h, score, classIndex

unsubscribe = client.subscribe_inference_result(handle_inference)

The task type of the pipeline's model is available as client.model_task_type. It is a label: what arrives in the handler is whatever the model emitted, which for object detection is typically a raw detection head that this component exists to reduce. Parse it yourself.

The model version's metadata is available as client.model_meta_data, typed as the variant matching model_task_type — ObjectDetectionModelMetaData, ClassificationModelMetaData, and so on — each declaring only the fields its task type uses. It is None when the version declares no metadata, or the box is too old to send it, so guard the object rather than each field. See the "Prepare a vision model" platform guide for the per-task-type fields.

4) Publish and subscribe to context data

from ai_box_lib.context_engine_client import ContextEngineClient

client = ContextEngineClient[dict]()
client.connect()
client.publish_data({"state": "ok"})

In a multi-asset setup, a single context engine is often responsible for all assets. Pass custom_asset_id to publish to a different asset's context engine topic:

client.publish_data({"state": "ok"}, custom_asset_id="asset-abc123")

from ai_box_lib.context_engine_client import ContextEngineClient
client = ContextEngineClient[dict]()
client.connect()
client.subscribe()

client.context  # Access the latest context value at any time

def handle_context(message: dict) -> None:
	print(f"Context update: {message}")

client.subscribe(handle_context) # Subscribe with a handler to receive real-time updates

Optionally you can pass custom_asset_id to subscribe to a Context Engine on a different asset, as long as it's connected to the same Box.

5) Communicate over a generic channel

A channel lets any two components exchange messages without being tied to a specific pipeline stage. You define the channel by providing a channel_id string. Both publisher and subscriber must use the same channel_id.

Messages are delivered on the topic {asset_id}/channel/{channel_id}.

Publisher:

from ai_box_lib.channel_client import ChannelClient

client: ChannelClient[None, dict] = ChannelClient("my-alerts")
client.connect()
client.publish({"severity": "high", "value": 42.0})

Subscriber:

from ai_box_lib.channel_client import ChannelClient

client: ChannelClient[dict, None] = ChannelClient("my-alerts")
client.connect()

def handle_alert(message: dict) -> None:
    print(f"Alert received: {message}")

unsubscribe = client.subscribe(handle_alert)
# call unsubscribe() when done

A single client instance can both publish and subscribe on the same channel.

Correlation IDs

Every message carries a correlation id so you can trace one piece of data through the pipeline (e.g. link an ML result back to its source frame). It is minted automatically at the origin and forwarded automatically when you publish from inside a subscription handler — no extra code needed.

Read the id of the message being handled with current_correlation_id():

def handle_raw(message: dict) -> None:
    frame_id = client.current_correlation_id()
    client.publish_data(process(message))  # inherits frame_id automatically

Automatic forwarding only works when you publish from within the handler. If you publish later — from another thread, a timer, or an external trigger such as an API callback — capture the id and pass it back explicitly. Every publish* method accepts an optional correlation_id:

client.publish_data(process(message), correlation_id=frame_id)

API Summary

All clients must call connect() before any publish or subscribe operation.

• DataCollectorClient.connect()
• DataCollectorClient.publish_timestream(data, correlation_id?)
• DataCollectorClient.publish_audio(data, correlation_id?)
• DataCollectorClient.publish_image(data, correlation_id?)
• PreProcessorClient.connect()
• PreProcessorClient.subscribe_timestream(handler)
• PreProcessorClient.subscribe_audio(handler)
• PreProcessorClient.subscribe_image(handler)
• PreProcessorClient.publish_data(data, correlation_id?)
• PostProcessorClient.connect()
• PostProcessorClient.subscribe_inference_result(handler)
• PostProcessorClient.publish_anomaly_detection(score, correlation_id?)
• PostProcessorClient.publish_classification(scores, correlation_id?)
• PostProcessorClient.publish_regression(values, correlation_id?)
• PostProcessorClient.publish_object_detection(detections, correlation_id?)
• PostProcessorClient.publish_custom(data, correlation_id?) — for a CUSTOM task type, or any payload the publishers above do not fit
• PostProcessorClient.model_task_type — task type declared on the pipeline's model (read-only)
• PostProcessorClient.model_meta_data — metadata declared on the pipeline's model version (read-only)
• ContextEngineClient.connect()
• ContextEngineClient.publish_data(data, custom_asset_id?, correlation_id?)
• ContextEngineClient.subscribe(handler?, custom_asset_id?)
• ContextEngineClient.context — latest received context value (read-only)
• ChannelClient(channel_id).connect()
• ChannelClient(channel_id).publish(data, correlation_id?)
• ChannelClient(channel_id).subscribe(handler)
• current_correlation_id() — id of the message currently being handled (any client; None outside a handler)

Validation and Limits

• DataCollectorClient validates message keys against CHANNELS.
• PreProcessorClient validates feature keys and feature shapes against FEATURES.
• PostProcessorClient does not validate published payloads. The task type is a label, not a contract; the per-task-type publishers build a payload the platform can interpret, and publish_custom publishes whatever it is given.
• Maximum publish payload size is 2 MB.

License

BSD-3-Clause. See LICENSE.