Connector API

This page specifies the connector base classes you subclass to build connectors.

FleetConnector

inorbit_connector.connector.FleetConnector is the base class for connectors that manage multiple robots.

Constructor kwargs

The following keyword arguments can be passed to FleetConnector.__init__():

  • register_user_scripts (bool): Register user scripts automatically. Default: False

  • default_user_scripts_dir (str): Default user scripts directory path. Default: ~/.inorbit_connectors/connector-{class_name}/local/

  • create_user_scripts_dir (bool): Create user scripts directory if it doesn’t exist. Default: False

  • register_custom_command_handler (bool): Register custom command handler. Default: True

  • publish_connector_system_stats (bool): When True, publish the connector host’s system stats (CPU, RAM, HDD) as default values for robots that don’t provide their own stats. Requires psutil to be installed (pip install inorbit-connector[system-stats]). If psutil is not available, falls back to zeroed defaults with a warning. Default: False

_connect()

Override.

Called once on startup (inside the connector’s background thread / event loop), before robot sessions are initialized.

Typical responsibilities:

  • Connect to your fleet manager / backend services.

  • Optionally fetch the fleet membership and call update_fleet() before sessions are created.

  • Start background polling tasks that keep fresh state for all robots. Schedule them with _create_supervised_task() rather than a bare asyncio.create_task, so a crash is logged and the loop restarts instead of dying silently.

_execution_loop()

Override.

Called repeatedly until stopped. The loop is rate-limited by config.update_freq, but it will never run faster than the body can execute.

Typical responsibilities:

  • For each robot in self.robot_ids, fetch the latest state (often from background polling state) and publish data via the publish_robot_* methods.

  • Handle exceptions inside the loop when possible (the framework logs and continues on exceptions).

_disconnect()

Override.

Called once during shutdown, after Edge SDK sessions are disconnected. Use this to stop polling tasks, close sockets, and release resources. (Tasks scheduled via _create_supervised_task() / _spawn_logged_task() are cancelled automatically — you only need to clean up tasks you started with a bare asyncio.create_task, which is discouraged.)

_create_supervised_task(name, coro_factory, restart_delay=5.0) / _spawn_logged_task(name, coro)

Callable (advanced).

Schedule background work that runs alongside _execution_loop — for example a fast pose loop and a slower key-value loop, each on its own cadence. Prefer these over a bare asyncio.create_task: a fire-and-forget task that raises has its exception stored on a task nobody awaits, so it is never logged, never restarted, and the datasource it fed silently freezes until the process restarts (while unrelated datasources keep working, masking the failure).

  • _create_supervised_task(name, coro_factory, restart_delay=5.0) — for long-lived loops. Runs coro_factory() and, if it ever returns or raises, logs it (with a traceback, tagged with name) and restarts it after restart_delay seconds. coro_factory is a zero-argument callable returning a fresh coroutine (it is re-called on each restart). asyncio.CancelledError is propagated so the task stops cleanly on shutdown.

  • _spawn_logged_task(name, coro) — for one-shot work. Runs coro once; if it raises, the failure is logged (with a traceback, tagged with name) via a done-callback instead of being silently swallowed. It is not restarted.

Call these from _connect() (they require the connector’s event loop to be running). Tasks scheduled this way are tracked by the framework and cancelled automatically during shutdown, so they do not need to be stopped in _disconnect().

_inorbit_robot_command_handler(robot_id, command_name, args, options)

Override.

Called when a command arrives from InOrbit for a specific robot in the fleet.

Contract:

  • options["result_function"](...) must be called to report success/failure, or you may raise CommandFailure for structured failure reporting.

See Commands Handling for the full command result contract.

fetch_robot_map(robot_id, frame_id) -> MapConfigTemp | None

Optional override.

If publish_robot_pose() references a frame_id that is not present in config.maps, the framework schedules an async fetch:

  • Calls your fetch_robot_map() coroutine.

  • If you return a MapConfigTemp containing image bytes and metadata, the framework writes the image to a temporary file and inserts a corresponding MapConfig into config.maps.

  • Then it publishes the map via publish_robot_map().

Return None if the map can’t be fetched.

_is_fleet_robot_online(robot_id) -> bool

Optional override.

The Edge SDK uses this callback to determine if a robot should be considered online. It is invoked when InOrbit sends a get_state request, which happens automatically when the robot is marked as offline but system stats are still being received. Default implementation returns True.

start() / join() / stop()

Callable.

  • start() creates a background thread and runs the connector lifecycle in an asyncio event loop.

  • join() blocks until the background thread exits.

  • stop() signals the event loop to stop and waits briefly for shutdown.

robot_ids / update_fleet(fleet)

Callable.

  • robot_ids is a cached list of robot IDs from config.fleet.

  • update_fleet() updates config.fleet and refreshes robot_ids.

This is designed to be used during _connect() when your fleet membership comes from an external system.

publish_robot_pose(robot_id, x, y, yaw, frame_id=None, **kwargs)

Callable.

Publishes pose for one robot. If the frame_id differs from the last published frame_id for that robot, the framework triggers publish_robot_map(..., is_update=True).

publish_robot_map(robot_id, frame_id, is_update=False)

Callable.

  • If frame_id exists in config.maps, publishes the configured map to InOrbit.

  • Otherwise schedules an async fetch via fetch_robot_map() (see above).

publish_robot_odometry(robot_id, **kwargs)

Callable. Publishes odometry data for one robot.

publish_robot_key_values(robot_id, **kwargs)

Callable. Publishes key-value telemetry for one robot.

publish_robot_system_stats(robot_id, **kwargs)

Callable. Stores system stats for one robot to be published at the end of the execution loop.

If no stats are stored for a robot during the loop iteration, default values are published automatically. By default, zeroed values are used. To use the connector host’s actual system stats as defaults, set publish_connector_system_stats=True in the constructor.

If immediate publishing is required, use _get_robot_session(robot_id) to access the underlying RobotSession and call publish_system_stats() directly.

_get_robot_session(robot_id) -> RobotSession

Callable (advanced).

Returns the underlying Edge SDK session for robot_id. Use this if you need Edge SDK functionality that is not wrapped by this package.

Connector

inorbit_connector.connector.Connector is a single-robot specialization of FleetConnector. It selects a single robot out of the fleet config and provides convenience wrappers that omit robot_id.

Lifecycle hooks

Override. Same intent as fleet:

  • _connect()

  • _execution_loop()

  • _disconnect()

_inorbit_command_handler(command_name, args, options)

Override.

Single-robot command handler. It is called through the fleet-level handler internally, but without requiring you to accept robot_id.

fetch_map(frame_id) -> MapConfigTemp | None

Optional override.

Single-robot convenience for map fetching. The framework uses it by delegating fetch_robot_map() to fetch_map().

_is_robot_online() -> bool

Optional override.

Single-robot convenience for online status. The fleet-level online check delegates to this method. Called when InOrbit requests state due to a discrepancy between the robot’s offline status and incoming system stats.

Publishing wrappers

Callable. Single-robot wrappers over the fleet publishing methods:

  • publish_pose(...) / publish_map(...)

  • publish_odometry(...)

  • publish_key_values(...)

  • publish_system_stats(...)

_get_session() -> RobotSession

Callable (advanced).

Returns the underlying Edge SDK session for the current robot.