robonix_api.lifecycle

Driver lifecycle gRPC server + per-contract Servicer resolution.

Every managed Robonix provider uses the shared robonix/lifecycle/driver capability that rbnx boot calls Driver(CMD_INIT, config_json) on. The wire shape is fixed by lib/lifecycle/srv/Driver.srv (uint8 command + string config_json → bool ok + string state + string error).

An exact legacy manifest may still select <provider namespace>/driver. When current generated code contains only the shared service, that legacy selection may upgrade to the shared runtime Driver with a migration warning. Shared selections never downgrade to a legacy runtime Driver.

Users can also declare arbitrary rpc-mode contracts (e.g. primitive/chassis/move) which generate PrimitiveChassisMoveServicer with method Move. We resolve both via the same contract_id_to_pascal() mapping.

Functions

bind_user_handler(servicer_cls, method_name, fn)

Build a dynamic subclass overriding method_name to call fn.

build_lifecycle_servicer(namespace, ...[, ...])

Build (without starting a server) the lifecycle Servicer instance.

coerce_response(response_cls, ret)

Allow handlers to return the response_cls directly OR a dict like {ok, state, error}.

contract_id_to_pascal(contract_id)

Mirror of robonix_codegen::contract_id_to_service_name.

driver_pascal_for_namespace(namespace)

primitive/lidarPrimitiveLidarDriver (the driver Pascal name).

parse_cfg(request)

resolve_servicer(contract_id, ...)

Find the generated Servicer class + add-to-server fn + canonical method name for a contract.

robonix_api.lifecycle.bind_user_handler(servicer_cls: type, method_name: str, fn: Callable) type[source]

Build a dynamic subclass overriding method_name to call fn. Adapts to handler arity: 1-arg signatures get (request), 2-arg get (request, context).

robonix_api.lifecycle.build_lifecycle_servicer(namespace: str, contracts_grpc_module, response_cls, *, on_init=None, on_activate=None, on_deactivate=None, on_shutdown=None, on_shutdown_complete=None, on_state_change=None, log_tag: str = 'robonix_api', requested_contract_id: str | None = None, allow_old_artifact_fallback: bool = False)[source]

Build (without starting a server) the lifecycle Servicer instance.

Returns (instance, add_to_server_fn, pascal_base, method_name='Driver', contract_id) when codegen emitted the shared lifecycle Servicer (preferred) or a legacy <namespace>/driver Servicer. An exact legacy manifest may use shared generated stubs only when the launcher supplied its compatibility marker. A provider is never allowed to continue without one usable lifecycle Driver.

on_state_change(state, detail) is invoked AFTER each handler returns ok=true and is the framework’s hook for pushing state transitions to atlas. State strings: “inactive” / “active” / “error”. Capability layer wires this; lower-level callers can leave it None.

robonix_api.lifecycle.coerce_response(response_cls, ret) Any[source]

Allow handlers to return the response_cls directly OR a dict like {ok, state, error}. Capability’s ready/error/deferred helpers return dicts.

robonix_api.lifecycle.contract_id_to_pascal(contract_id: str) str[source]

Mirror of robonix_codegen::contract_id_to_service_name. Uniform PascalCase, no prefix stripping. robonix/primitive/chassis/twist_inRobonixPrimitiveChassisTwistIn. mycomp/a/b/cMycompABC.

robonix_api.lifecycle.driver_pascal_for_namespace(namespace: str) str[source]

primitive/lidarPrimitiveLidarDriver (the driver Pascal name).

robonix_api.lifecycle.generated_grpc_metadata(service_base: str, method: str) tuple[str, str][source]

Return Atlas metadata for a generated robonix_contracts.proto RPC.

Generated Python class names omit the protobuf package, while Atlas stores the fully qualified gRPC service and method route used on the wire.

robonix_api.lifecycle.lifecycle_contract_for_module(namespace: str, contracts_grpc_module, requested_contract_id: str | None = None, *, allow_old_artifact_fallback: bool = False) tuple[str, str] | None[source]

Select exactly one generated lifecycle service for the manifest.

Returns (contract_id, generated_service_name). A package built with current Robonix contracts selects robonix/lifecycle/driver. During incremental migration, rbnx supplies the exact Driver ID from the selected package manifest. allow_old_artifact_fallback is the historical keyword for the launcher’s positive legacy-manifest marker; it only permits an exact legacy selection to use shared runtime stubs. It never permits a shared selection to downgrade.

robonix_api.lifecycle.parse_cfg(request) dict[source]
robonix_api.lifecycle.resolve_servicer(contract_id: str, contracts_grpc_module)[source]

Find the generated Servicer class + add-to-server fn + canonical method name for a contract. Returns (servicer_class, method_name, add_fn, pascal_base) or None.