antioch-py 2.0.2

The Antioch Python SDK

Antioch Python

Python client library for the Antioch middleware platform.

Installation

Authenticate your local environment with GCP via the Google Cloud CLI:

gcloud auth login                                       # Login to Google Cloud
gcloud config set project proof-of-concept-staging-9072 # Set the correct project
gcloud auth application-default login                   # Create app credentials

This project uses uv, a fast Python package and project manager. Install it and set up authentication:

# Install uv
curl -LsSf https://astral.sh/uv/install.sh | sh

# Install keyring authentication globally for uv
uv tool install keyring --with keyrings.google-artifactregistry-auth

# Install all default groups in one step
uv sync

Install the pre-commit hooks to auto-run uv sync and ruff format on Git commit:

uv run pre-commit install

Usage

The Antioch Python client builds both Docker images that can be used and a Python package.

Docker Images

We have 2 Docker images currently

Standard Image (antioch-py:latest)

• Base: python:3.12-slim
• Sidecar binary at /usr/local/bin/sidecar
• antioch-py package pre-installed
• Python 3.12 virtual environment with pip
• Runs as root for maximum compatibility

Build locally using

./build-image.sh

ROS Image (antioch-py-ros:latest)

• Base: osrf/ros:jazzy-desktop
• All features of the standard image
• Full ROS 2 Jazzy desktop environment
• ROS 2 Python packages and tools pre-installed
• Ideal for robotics applications requiring ROS integration

NOTE: not included in cloudbuild currently

Build locally using

./build-image.sh --variant ros-jazzy

Usage

For standard Python modules:

FROM antioch-py:latest

# Install any additional packages you need
RUN pip install opencv-python-headless numpy

# Copy your module entrypoint
# The sidecar will automatically run your module.py file
COPY module.py .

For ROS-enabled modules:

FROM antioch-py-ros:latest

# Install ROS-specific packages if needed
RUN apt-get update && apt-get install -y ros-jazzy-cv-bridge

# You will probably need to source your ROS modules / install
# To ensure your ROS environment is available to your module, source the ROS setup script before running the sidecar.
# For example, you can add the following to your Dockerfile:
RUN echo "source /opt/ros/jazzy/setup.bash" >> /etc/profile.d/ros.sh
SHELL ["/bin/bash", "-c"]

# If you have a custom ROS workspace, source its setup script as well:
# RUN echo "source /workspace/install/setup.bash" >> /etc/profile.d/ros.sh

# The sidecar will automatically run your module.py file in the ROS environment.
COPY module.py .

Testing Modules

Antioch provides a test framework for testing modules in isolation without requiring a full deployment. The ModuleTestHarness runs your module in a separate process and communicates with it using the standard Antioch protocol.

Basic Example

from antioch.module import Module, NodeExecution
from antioch.module.test import ModuleTestHarness, TestNode, TestInput, TestOutput
from common.message import Message

# Define your message types
class InputData(Message):
    _type = "input_data"
    value: int

class OutputData(Message):
    _type = "output_data"
    result: int

# Create your module
class MyModule(Module):
    def init(self):
        self.add_node("process", self.process)

    def process(self, exec: NodeExecution):
        data = exec.input("data").data(InputData)
        if data:
            exec.output("result").set(OutputData(result=data.value * 2))

    def cleanup(self):
        pass

# Test your module
def test_my_module():
    with ModuleTestHarness(MyModule) as harness:
        # Configure the module
        harness.configure(
            module_name="my-module",
            nodes=[
                TestNode(
                    name="process",
                    budget=50,  # 50ms time budget
                    inputs=[TestInput(name="data", message_type="input_data")],
                    outputs=[TestOutput(name="result", message_type="output_data")],
                )
            ],
        )

        # Execute the node
        result = harness.execute(
            "process",
            inputs={"data": InputData(value=21)},
        )

        # Check the output
        output = result.output("result", OutputData)
        assert output.result == 42

The test framework handles module lifecycle, message serialization, and provides helpful error messages for common issues like budget overruns.

Testing Modules with Hardware

For modules that use hardware (cameras, IMUs, actuator groups), provide mock hardware data:

from antioch.module import Module, NodeExecution
from antioch.module.test import ModuleTestHarness
from antioch.message import Image, ImageEncoding

class CameraModule(Module):
    def init(self):
        self.add_node("process_frame", self.process_frame)
        self.camera = self.get_rgb_camera("my_camera")

    def process_frame(self, exec: NodeExecution):
        frame = self.camera.get_frame()
        # Process frame...
        exec.output("processed").set(...)

def test_camera_module():
    with ModuleTestHarness(CameraModule, module_yaml_path="tests/camera.yaml") as harness:
        # Execute node with mock camera data
        result = harness.execute_node(
            node_name="process_frame",
            hardware_data={
                "my_camera": Image(
                    width=640,
                    height=480,
                    encoding=ImageEncoding.RGB8,
                    data=b"mock_rgb_data",
                )
            }
        )

        # Verify outputs
        processed = result.output("processed", ProcessedData)
        assert processed is not None

        # For actuator groups, you can also verify hardware writes
        # write = result.get_hardware_write("my_actuator_group")

The test harness automatically mocks the Sim RPC layer, so hardware reads and writes work exactly as they would in production, but without requiring Isaac Sim to be running.