remotivelabs.topology.control

class ControlClient:

Client for sending control messages in a Remotive topology.

A ControlClient is used to send abstract or synthetic control messages that do not exist in a real vehicle but are useful to manipulate or initialize models into a specific state.

ControlClient(client: remotivelabs.broker.BrokerClient)

Control message client.

Arguments:
  • client: BrokerClient instance.
Note:

Start the instance using a context manager:

async with ControlClient(...) as client:
    ...
async def send( self, target_ecu: str, request: ControlRequest, timeout: float = 10.0, retries: int = 0) -> ControlResponse:

Send a control request to the target ECU, retrying on timeout.

Arguments:
  • target_ecu: The name of the target ECU.
  • request: The control message request.
  • timeout: Timeout per attempt in seconds.
  • retries: Number of retries on timeout. Defaults to 0.
Returns:

Control response.

@dataclass
class ControlRequest:

Message structure and serialization for control requests.

Attributes:
  • type: The type of control message.
  • argument (Any | None): Optional argument to the control message. Must be JSON-serializable if provided.
ControlRequest(type: str, argument: Optional[Any] = None)
type: str
argument: Optional[Any] = None
@dataclass
class ControlResponse:

Message structure and serialization for control responses.

ControlResponse(status: str = 'ok', data: Any = None)
status: str = 'ok'
data: Any = None
@runtime_checkable
class Handler(typing.Protocol):

A callable protocol for handling control messages asynchronously.

Implementations must define an async __call__ method that takes a ControlRequest and returns a ControlResponse.

Handler(*args, **kwargs)