remotivelabs.topology.behavioral_model

class BehavioralModel:

A BehavioralModel is used to emulate some behavior instead of a real ECU.

It manages lifecycle operations for namespaces (e.g., CanNamespace, SomeIPNamespace), handles inputs, routes control requests, and provides a unified interface for testing setups.

BehavioralModel( name: str, broker_client: remotivelabs.broker.BrokerClient, namespaces: list[remotivelabs.topology.namespaces.namespace.Namespace] | None = None, input_handlers: list[tuple[str, remotivelabs.topology.namespaces.input_handlers.InputHandler]] | None = None, control_handlers: list[tuple[str, remotivelabs.topology.control.Handler]] | None = None)

Initialize the BehavioralModel instance.

Arguments:
  • name: Identifier for the ECU stub instance, then name which receives control messages.
  • broker_client: The client used for communication with the broker.
  • namespaces: list of Namespace instances (CanNamespace, SomeIPNamespace, etc.).
  • input_handlers: Optional list of (namespace, handler list) pairs to receive callbacks on inputs. It is advised to create these using the namespace's create_input_handler method.
  • control_handlers: Optional list of (command, handler) pairs for routing control messages.
Note:

Start the instance using a context manager:

async with BehavioralModel(...) as bm:
    ...
    await bm.run_forever()

Or use the start/stop methods directly:

bm = BehavioralModel(...)
await bm.start()
# ...
await bm.stop()

async def start(self) -> None:

Start the behavioral model, open all namespaces, and initialize input handlers. This is an idempotent operation - calling it multiple times has no additional effect.

async def stop(self) -> None:

Stop the behavioral model, close all namespaces, and clean up resources. This is an idempotent operation - calling it multiple times has no additional effect.

async def run_forever(self) -> None:

Run the BehavioralModel indefinitely, processing inputs and control requests.

@dataclass
class PingRequest(remotivelabs.topology.control.request.ControlRequest):

Control request to check if the BehavioralModel is alive and responsive.

Use remotivelabs.topology.control.ControlClient to send control requests.

PingRequest(type: str = 'ping_v1', argument: 'Any | None' = None)
type: str = 'ping_v1'
@dataclass
class RebootRequest(remotivelabs.topology.control.request.ControlRequest):

Control request to reset all namespace restbus to default values.

Use remotivelabs.topology.control.ControlClient to send control requests.

RebootRequest(type: str = 'reboot_v1', argument: 'Any | None' = None)
type: str = 'reboot_v1'