FMU models

A Functional Mock-up Unit (FMU), following the Functional Mock-up Interface (FMI) standard, can run as a behavioral model. RemotiveTopology writes signals from the bus to the FMU's inputs, steps the FMU at a fixed rate, and passes its outputs to your code, which typically puts them on the bus with a restbus.

For a complete, runnable example, see the FMU use case in the examples repository. The API is described in the fmu module reference.

Requirements

• The fmu optional dependency of remotivelabs-topology. It installs fmpy and numpy, which are large and not needed for other models:

  pip install remotivelabs-topology[fmu]

  Importing remotivelabs.topology.fmu without it raises an ImportError.

• The FMU must be:

  • FMI 2.0 or 3.0, with Co-Simulation support,
  • configured with a fixed step size and no internal solver,
  • carrying a defaultExperiment in its model description.

  Loading an FMU without co-simulation support or a defaultExperiment raises UnsupportedFMUSimulationError.

Minimal model

The turn signals of the RemotiveCar BCM: the turn stalk on DriverCan feeds the FMU, and the FMU's turn light outputs go on BodyCan.

import asyncio
from pathlib import Path

from remotivelabs.broker import BrokerClient
from remotivelabs.topology.cli.behavioral_model import BehavioralModelArgs
from remotivelabs.topology.fmu import FMUBehavioralModel, FMUSimulation, FMUVars
from remotivelabs.topology.namespaces import filters
from remotivelabs.topology.namespaces.can import CanNamespace, RestbusConfig


FMU_PATH = Path(__file__).parent / "BCM.fmu"


async def main(args: BehavioralModelArgs) -> None:
    async with BrokerClient(url=args.url, auth=args.auth) as broker_client:
        driver_can = CanNamespace("BCM-DriverCan", broker_client)
        body_can = CanNamespace(
            "BCM-BodyCan",
            broker_client,
            restbus_configs=[RestbusConfig([filters.SenderFilter(ecu_name="BCM")])],
        )

        async def on_output(outputs: FMUVars) -> None:
            await body_can.restbus.update_signals(
                ("TurnLightControl.LeftTurnLightRequest", int(outputs["left_turn_light"])),
                ("TurnLightControl.RightTurnLightRequest", int(outputs["right_turn_light"])),
            )

        fmu = FMUSimulation(FMU_PATH, on_output=on_output)

        async with FMUBehavioralModel(
            "BCM",
            fmu=fmu,
            input_mapping={
                "BCM-DriverCan": {
                    "TurnStalk": {"TurnStalk.TurnSignal": "turn_stalk"},
                },
            },
            broker_client=broker_client,
            namespaces=[driver_can, body_can],
        ) as bm:
            await bm.run_forever()


if __name__ == "__main__":
    asyncio.run(main(BehavioralModelArgs.parse()))

In the instance file, the model is a container like any other Python behavioral model (see ECUs), but make sure to include remotivelabs-topology[fmu]. If the FMU has no binaries for your host platform, set the container's platform, for example to linux/amd64, and Docker emulates it.

FMUSimulation

Loads, instantiates, and steps the FMU. It doesn't need a broker, so you can also use it on its own to try out an FMU.

Parameter	Description
path	Path to the .fmu file.
on_output	Optional async callback, called after every step with the outputs.
before_step	Optional async callback, called before every step with the inputs. It returns the inputs to use for the step.
step_size	Optional step size in seconds. Defaults to the step size in the defaultExperiment.

Member	Description
print()	Prints the model information and variables. Useful to find the variable names.
step_size	The step size in seconds.
input_var_names, output_var_names	The names of the FMU's input and output variables.
inputs, outputs	The inputs and outputs of the last step.
await step()	Applies the inputs, advances one step and returns the outputs.
await wait_for_step()	Waits for the next step and returns its outputs.
set_input_value(name, value)	Sets an input for all following steps. Raises ValueError for an unknown name.
reset()	Resets the FMU to its initial state.

FMUBehavioralModel

Connects an FMUSimulation to the topology. run_forever() steps the FMU at its step size and handles inputs and control requests.

Parameter	Description
model_name	The name of the model, which receives control requests.
fmu	The FMUSimulation to run.
input_mapping	Which signals feed which FMU inputs, see Input mapping.
broker_client	The BrokerClient to use.
namespaces	Optional namespaces, for example a CanNamespace with a restbus for the outputs. A namespace in input_mapping that isn't listed is created automatically.

When it receives a RebootRequest through the control API, the model resets the FMU and its restbuses. Tests use this to start from a known state:

async with ControlClient(broker_client) as cc:
    await cc.send(target_ecu="BCM", request=RebootRequest())

Input mapping

InputMapping is a nested dictionary: namespace → frame → signal → FMU input variable.

input_mapping: InputMapping = {
    "BCM-DriverCan": {                             # namespace, <ECU>-<Channel>
        "TurnStalk": {                             # frame
            "TurnStalk.TurnSignal": "turn_stalk",  # signal → FMU input
        },
    },
}

• Namespaces are named <ECU>-<Channel>, see Namespaces.
• Signal names include their frame: <Frame>.<Signal>.
• When a mapped frame arrives, each mapped signal's value is written to its FMU input. Values that aren't numbers, or that are outside the signal's min and max in the signal database, are ignored.

Outputs

Outputs aren't mapped automatically: handle them in on_output, usually by updating restbus signals. Output values have the Python type of the FMU variable:

FMU type	Python type
Boolean	bool
Int32 (FMI 3.0), Integer (FMI 2.0)	int
Float64 (FMI 3.0), Real (FMI 2.0)	float
String	str

Convert where the types don't match the signal, for example int(...) for a Boolean that drives an on/off signal, like the turn lights in the minimal model.

Troubleshooting

• No signals reach the FMU: check the namespace names with remotive broker signals namespaces --url http://localhost:50051, and that signal names include the frame.
• Find the FMU's variable names: run FMUSimulation(path).print().