openrois-sdk

Python client SDK for OpenRoIS, an open-source middleware implementing the OMG RoIS Framework 2.0.

Connect to an OpenRoIS engine, discover the components of every connected physical robot, virtual avatar, or AI service, and drive them through the standard RoIS interfaces. Asyncio only.

Alpha, pre-1.0, unstable API.

Installation

pip

pip install openrois-sdk

From Source

git clone https://github.com/openrois/openrois.git
pip install -e ./openrois/sdk/python

wss:// has not been exercised: every run used ws://. The gateway does not authenticate connections in the alpha releases, so use it on trusted networks only.

Usage

import asyncio
from collections import defaultdict

from openrois.interfaces import CompletedParams, RoISIdentifierType
from openrois.sdk import RoISClient, component_ref, component_type


async def main() -> None:
    async with RoISClient.connect("ws://localhost:8765") as client:
        # Find a component by type, and select it by ref in later calls.
        navigation = RoISIdentifierType(authority="OMG", code="Navigation")
        [nav] = await client.search(component_type(navigation))
        target = component_ref(nav)

        # Read state.
        status = await client.query("component_status", target)
        print(status)

        # React to events.
        client.on(
            "rois.event.notify_event",
            lambda event: print(event.event_type, event.results),
        )
        await client.subscribe("reached_target", target)

        # Record how each command ends, before running any.
        statuses: dict[str, str] = {}
        ended: defaultdict[str, asyncio.Event] = defaultdict(asyncio.Event)

        def on_completed(completed: CompletedParams) -> None:
            statuses[completed.command_id] = completed.status
            ended[completed.command_id].set()

        client.on("rois.command.completed", on_completed)

        # Reserve, configure, command, wait, release.
        await client.bind(nav)
        await client.set_parameter(
            nav,
            [{"name": "target_positions", "data_type_ref": "string[]", "value": '["kitchen"]'}],
        )
        [command_id] = await client.execute([{"component_ref": nav, "command_type": "start"}])
        # Wait here, because release aborts a running command, which then ends with ABORT.
        await asyncio.wait_for(ended[command_id].wait(), timeout=30)
        print(statuses[command_id])
        await client.release(nav)


asyncio.run(main())

Run it against the Docker Compose stack at the repository root, which serves simulated components through a gateway.

Waiting for a Command

RoIS has no operation that waits for a command, so neither does the client. execute() returns as soon as the engine accepts the commands. Each command ends later, with a rois.command.completed notification that names its id. set_parameter() works the same way.

A command that ends at once can complete before the reply reaches your code, so a listener added after execute() can miss it. The recipe has three steps:

1. Add the listener before the call, once for the whole session.
2. Record each status by command_id.
3. Read the status with the ids in the reply.

The quickstart above does this. Its asyncio.Event for each id stays set, so a command that has already ended does not wait, and asyncio.wait_for limits the wait.

Wait before you release the component: release() aborts your running commands on it, and they end with ABORT. disconnect() aborts them too.

API

RoISClient has one method per operation of the RoIS method catalog. Each takes the operation's IDL parameters in IDL order, validates them and the result against the catalog models in openrois-interfaces, and returns the result's out parameters as interface models. Inputs take a model or a mapping.

A streaming component is driven with these same methods. Carrying its media is planned.

• get_profile() returns an EngineProfile with profile and component_profiles: the engine profile and the profile of every component it lists, keyed by fully qualified ref.
• execute() takes a command_unit_list: commands that run in order, and {"branch_list": [...], "delay_time": ...} groups whose branches run at the same time, each branch in order. The commands need no command_id: the engine assigns one to each, and execute() returns them in order, branch by branch.
• query() and subscribe() go to the one component that declares the query or event type and matches the condition.

One client is one session. Bindings and subscriptions end with it, and there is no automatic reconnect, because a silent reconnect would hide that loss. After close, connect again and bind and subscribe again.

Conditions

Every condition is a string in the OpenRoIS subset of CQL2-Text: = and LIKE comparisons, joined by AND, over component_ref and component_type. An empty string is no filter. The builders quote values for you:

Listening

on(name, listener) adds a listener and off(name, listener) removes it. Listeners of catalog notifications receive the validated params, as models from openrois.interfaces.

A plain function runs inline in the task that reads the socket, in the order messages arrive, so it must not block. A coroutine function runs in a task of its own, so it can await a call on the client:

async def on_person(event: NotifyEventParams) -> None:
    print(await client.get_event_detail(event.event_id))

client.on("rois.event.notify_event", on_person)

A listener that raises, in either form, never stops the client: the exception goes to the error listeners, and the other listeners still run. With no error listener, a fault goes to the openrois.sdk.transport logger at level ERROR. It is never raised.

Closing

async with calls disconnect() when the block ends, and needs no await on connect(). Without it, call disconnect() yourself:

client = await RoISClient.connect("ws://localhost:8765")
await client.search()
await client.disconnect()

Error Handling

from openrois.interfaces import ReturnCode
from openrois.sdk import RoISError

try:
    await client.bind("robot_1/navigation")
except RoISError as error:
    if error.return_code is ReturnCode.OUT_OF_RESOURCES:
        ...  # Another application holds the reservation.

Timeouts

RoISClient.connect() takes the timeouts of the transport, in seconds:

client = await RoISClient.connect(
    "ws://localhost:8765", request_timeout=5.0, connect_timeout=2.0
)

A call can also be cancelled, with asyncio.timeout or asyncio.wait_for. The client keeps working, and it drops the reply when one arrives. A cancelled connect() ends at once and leaves the connection to close by itself.

The Transport

client.transport is the WebSocketTransport under the client, for JSON-RPC methods outside the catalog. Nothing sent through it is validated.

result = await client.transport.send("vendor.custom.method", {"level": 3})

Its notification event receives each notification as a mapping with jsonrpc, method and, when the message has them, params, in or outside the catalog.

It accepts messages of up to 16 MiB, where the default of the websockets package is 1 MiB, so that a large get_profile does not close the connection. A larger message closes the connection with code 1009, and the close listeners and the failed requests carry that code and its reason.

With websockets 15 or later, the client connects through the proxy that the environment names in variables such as http_proxy and https_proxy, for a local gateway too. Add the gateway to no_proxy to reach it directly. Earlier versions of websockets and the JavaScript client on its default socket ignore these variables.

Modules

Tested Versions

The wire has no version handshake, so a mismatch shows only when a message carries a field the other side does not know. This release is tested against these versions together, which testing/conformance/frozen.txt pins:

Development

pip install -e ../../interfaces/python -e ".[dev]"
pytest
mypy
ruff check src/ tests/

tests/test_transport.py runs the client against testing/conformance/scripted_server.py, which it starts by itself, so it needs no stack.

tests/test_conformance.py runs the conformance list of testing/conformance/README.md, one test per scenario. C01 to C17 need a live gateway with the mock adapter behind it, such as the Docker Compose stack at the repository root. They run when OPENROIS_GATEWAY_URL names the gateway, and are skipped otherwise. C18 to C21 and the coverage tests need no stack, so they always run.

docker compose up --build -d   # from the repository root
OPENROIS_GATEWAY_URL=ws://127.0.0.1:8765 pytest

License

Apache-2.0. See LICENSE.