Skip to main content

Behavioral Models

A behavioral model in the RemotiveTopology Framework is a simulation component that mimics the behavior of an ECU. Like a real ECU, it is typically triggered by network messages, performs logic, and then sends new network messages to other components. Behavioral models in RemotiveTopology will run on top of the RemotiveBroker, making them use real network protocols in their communication.

As with all nodes in a RemotiveTopology, a behavioral model communicates using real network traffic over protocols such as CAN buses or SOME/IP networks. For a full list of supported network types, refer to the RemotiveBroker documentation.

The Simulation​

The Simulation is the logic that keeps the RemotiveTopology Framework together. It has several responsibilities, all related to network traffic to and from the ECU:

  • It handles the rest bus, used to send cyclic network messages.
  • It implements the RemotiveTopology control messages, making it possible to configure and reboot/reset the simulation.
  • It manages subscriptions. A list of incoming network messages that will generate a callbacks to the model.
from remotivelabs.sim import Simulation

async def startUp(brokerUrl):
    simulation = Simulation(
        'HazardLightControlUnit',
        brokerUrl
    )
    await simulation.await_ready()
    # ... do things with the simulation
    
    await simulation.stop()

The two parameters are used to communicate with the RemotiveBroker. The name is typically describing the ECU or the service being simulated. It needs to be unique for the broker instance, but you won't be using it anywhere else.

The simulation is used to drive logic that is reacting to network input and sending network output.

The Rest Bus - Sending Periodic Network Messages​

The communication on CAN buses, in particular, is often sent periodically, several times a second. In a Simulation, the Rest Bus can be used to accomplish this. You initiate the Simulation with a list of messages that you would like to send. The cycle time and the default value is normally taken from the signal database, but can be changed later on.

The rest bus is configured by passing in a list of filters that will be applied to the signal databases. They will produce a list of frames that should be sent by the rest bus. In the example below, we use a filter that will match "all frames that can be sent on the DriverCan0-HLCU namespace by the HazardLightControlUnit". There are several filters to choose from. See more in the section about filters below.

from remotivelabs.sim import Simulation, config, filters

async def startUp(brokerUrl):
    async with Simulation(
        'HazardLightControlUnit',
        brokerUrl,
        rest_bus_config=RestBus(
            [filters.Sender("HazardLightControlUnit-DriverCan0", "HazardLightControlUnit")],
        ),
    ) as simulation:
        await simulation.await_ready()

        # Simulate pressing the hazard light button
        async with simulation.getContext() as ctx:
            ctx.rest_bus_context["HazardLightButton.HazardLightButton"].update([1])

We can modify the values using the rest bus context at any time. In this example, we update the "HazardLightButton" signal within the "HazardLightButton" frame to 1.

The rest bus also supports assigning an array of values to a signal. When configured with multiple values, the rest bus will cycle through them sequentially, sending one value per tick, and repeat the sequence until reconfigured with new values.

By default, the rest bus uses cycle times from the signal database. However, you can explicitly configure timing parameters using either cycle_time or delay_multiplier:

...
    rest_bus_config=RestBus(
        [filters.Sender("HazardLightControlUnit-DriverCan0", "HazardLightControlUnit")],
        cycle_time=0.02,    # Fixed cycle time of 0.02 seconds (50Hz)
        # or/and
        delay_multiplier=2  # Scale database cycle times by factor 2
    ),

These timing parameters serve different purposes:

  • cycle_time: Sets a fixed cycle time for all signals matched by the filters, including non-cyclic ones. This allows you to send non-cyclic signals at a fixed rate.
  • delay_multiplier: Scales the cycle times of the signals by the specified factor. Use this to slow down the rest bus (saving CPU) or speed up testing. Note that it is the cycle time that is scaled, so a larger value (> 1) will result slow things down and a small value (< 1) will speed things up.

Subscriptions – Reacting to Incoming Messages​

When starting the simulation, you can subscribe to messages and react to them. In the previous section, the Hazard Light Control Unit sent frames indicating whether the hazard light button was pressed.

Below is an example of another ECU subscribing to that signal and printing a message whenever its value changes:

from remotivelabs.sim import Context, Simulation, filters, message, subscribe

def _on_hazard_button_pressed(signal: message.Signal, ctx: Context) -> None:
    print(f"New hazard light value: {signal.payload}")

async def startUp(brokerUrl):
    async with Simulation(
        'BodyCanModule',
        brokerUrl,
        subscriptions=[
            subscribe.Signal(
                [filters.Signal("BodyCanModule-DriverCan0", "HazardLightButton.HazardLightButton")],
                _on_hazard_button_pressed,
                on_change_only=True,
            ),
        ],
    ) as simulation:
        await simulation

As you can see, it's possible to set up multiple subscriptions in the subscriptions array. In the example, we are subscribing to a signal, but it is possible to subscribe to frames or SOME/IP messages as well. Each subscription will use filters to specify which messages that are of interest. More under filters below.

The subscription also specifies a callback method and a flag indicating if we want the callback to be triggered on every message or just when the message content changes.

Note that the callback function will be passed a Context object. This can be used to send new messages or change what the rest bus is sending.

Filters​

In the remotivelabs.sim.filters package you can find several filters that let you select a subset of all signals associated with an ECU. In the examples above, we have seen two examples. One where we selected everything that is sent by the ECU and one where a subscription for a single signal was created.

All filters have a boolean flag called include that specifies if the filter is used to include or exclude messages. It defaults to True.

In most cases, filters are passed in an array, meaning that you can fine tune the selection by using include and exclude filters. The order of the filters does not matter, but exclusion filters have higher priority than inclusion. Here are two examples; both will match all frames, except "Frame1".

    filters1 = [
        AllFrames(namespace=ns),
        Frame(namespace=ns, frame_name="Frame1", include=False),
    ]
    filters2 = [
        Frame(namespace=ns, frame_name="Frame1", include=False),
        AllFrames(namespace=ns),
    ]

Behavior Model Implementation Alternatives​

The behavior model is plain python code and can be implemented in various ways, possibly with the help of other libraries. In the RemotiveTopology examples, you can find examples of different implementations.

  • In the "lighting and steering example" a more complete version of the BCM is implemented with a state machine.
  • The "simple FMU example" demonstrates how you can use an FMU to drive the behavior model.