auto-trainer-api 0.9.24

API for interfacing with the core acquisition process via platform and language agnostic message queues.

Autotrainer API: Python Integration

The python auto-trainer-api module is intended to provide an efficient means to emit information that is needed for local or remote management of applications running locally on the device and to receive commands from those sources.

The exposed API is intended to be agnostic to the underlying transport layer. The current implementation uses ZeroMQ. The reasons for this decision include:

• Relatively low overhead for the acquisition application
• Does not require either side to manage connections and know when the other side is available or changes availability
• Does not require an additional process or service to maintain a persistent message queue (e.g., RabbitMQ)
  • Persistent data is managed elsewhere

Client Integration

There are two points of integration available for clients to support the remote interface. The first allows publishing "events" for state and property changes that occur in the client. The second is a "command" interface for the client to receive command requests from remote sources.

Both interfaces are provided through an instance of the RpcService class. An instance can be obtained via create_api_service(...) which constructs the specific concrete implementation. After creation, the service must be explicitly started (start(...)) and can be stopped (stop(...)). Once stopped, an instance can not be restarted. If a connection should be reestablished after stopping an instance, a new instance should be created and started.

Events

Events are published through the send_event_dict(message) or send_event(message) methods on the RpcService instance.

send_event_dict accepts a dictionary with the following entries:

• kind - an ApiEventKind value
• when - a wall-clock value of time
• index - monotonically increasing timestamp w/units of nanoseconds (typically time.perf_counter_ns())
• context - an object whose contents depend on the ApiEventKind; may be None for some event kinds

send_event accepts an ApiEvent dataclass instance with the same fields.

Emergency events (ApiEventKind.emergencyStop, ApiEventKind.emergencyResume) are automatically published on both the standard event channel and a dedicated emergency channel for subscribers that only monitor critical events.

Commands

Commands from external sources are supported by registering a CommandRequestDelegate with the RpcService instance via the command_request_delegate property. The delegate receives an instance of ApiCommandRequest and must return an instance of ApiCommandRequestResponse.

The primary property of the ApiCommandRequest is command which is an ApiCommand value. Depending on the command, there may also be a dictionary in the data property with arguments or other information relevant to the command. The nonce property can be ignored if the command is handled synchronously. For commands that send an asynchronous result after completion (see below), the nonce must be stored to associate with the result (along with the command value).

The returned ApiCommandRequestResponse object contains one required field:

• result - a value of ApiCommandRequestResult

And three optional fields:

• data - an optional object with results from the command beyond success/failure (often will be None)
• error_code - an integer error code value if the command is not successful or can't be initiated
• error_message - an optional string message if the command is not successful or can't be initiated

The expected contents of the data property are defined by the command, but is typically None.

The error_code property should be a non-zero value if there is an error code to report.

There are two fields on the ApiCommandRequestResponse object that are ignored as part of the returned object from the command delegate: command and nonce. See Asynchronous Commands for when these fields are required.

Asynchronous Commands

The command delegate is expected to return "immediately" (low millisecond type of time frame). If the command is not deterministically fast, it is expected to immediately return an ApiCommandRequestResponse with a result value of ApiCommandRequestResult.PENDING_WITH_NOTIFICATION.

Once the action associated with the command is complete, the client should call send_command_result(response) on the RpcService instance. The response argument is an ApiCommandRequestResponse instance with the command and nonce properties set to the values received in the original ApiCommandRequest (the client is responsible for storing these values until needed). Note that those two properties are ignored for synchronous command handling, but required for asynchronous responses.

Bridge

The autotrainer.api.bridge module provides a Socket.IO bridge that allows web-based clients to interact with the RPC service without a direct ZeroMQ connection. It exposes a FastAPI/Socket.IO application that:

• Forwards command requests from the RPC service to connected Socket.IO clients
• Accepts command results from Socket.IO clients and relays them back to the RPC service
• Allows Socket.IO clients to publish events via the sendApiEvent message

The bridge can be started as a standalone process:

python -m autotrainer.api.bridge --host 0.0.0.0 --port 8000

Or created programmatically via create_bridge_app(options, command_timeout, cors_allowed_origins) for embedding into a larger application. Requires the bridge optional dependency group (pip install auto-trainer-api[bridge]).

Publishing

Publishing is handled automatically via GitHub Actions. Pushing a version tag (e.g., v0.9.18) triggers the CI workflow which runs the test suite, builds the package, and publishes to PyPI using trusted publishing (OIDC).

Pushes to the workflows-edit branch publish to TestPyPI for verification.

Local TestPyPI Publishing

To publish a dev build to TestPyPI from a local checkout (this is separate from the CI flow above):

cd packages/auto-trainer-api
uv run --script publish-testpypi.py

Each run appends a fresh .devN suffix to the version, where N is one past the highest dev release already on TestPyPI (0 if there are none). It builds with uv and uploads with twine, both pulled in on demand via the script's inline dependencies, so nothing is installed into the project by default. Set TESTPYPI_TOKEN to a TestPyPI API token to upload non-interactively; otherwise you are prompted to paste it:

TESTPYPI_TOKEN=pypi-... uv run --script publish-testpypi.py

Installation

The package is published to the PyPI package index and can be installed with standard pip commands.

pip install auto-trainer-api

Optional dependency groups:

pip install auto-trainer-api[bridge]     # Socket.IO bridge dependencies
pip install auto-trainer-api[telemetry]  # OpenTelemetry support