remotivelabs.topology

RemotiveTopology framework

RemotiveTopology framework is a Python library for modelling automotive ECUs and testing automotive network communication using real protocols (CAN, SOME/IP) via RemotiveBroker. It is built on top of remotivelabs-broker, which provides the low-level gRPC client, signal types, and protocol primitives.

Installation

pip install remotivelabs-topology

# with optional FMU support
pip install remotivelabs-topology[fmu]

# with the optional mapping module (message forwarding, signal transforms)
pip install remotivelabs-topology[mapping]

Project Links

Usage

The framework has two primary use-cases:

Testing: Write test cases that interact with a running topology to verify correct behavior.
Behavioral Models: Create ECU stubs that model real ECU behavior on a running RemotiveTopology instance, sending and receiving network messages over real protocols.

Testing

Capturing frames

Use remotivelabs.topology.testing.frames.capture_frames to subscribe to CAN frames from a namespace and assert on the received frames and signal values:

from remotivelabs.broker import BrokerClient

from remotivelabs.topology.testing.frames import capture_frames


async def test_hazard_light_sequence(broker_client: BrokerClient) -> None:
    # Subscribe to frames published by the HazardLightControlUnit behavioral model
    async with capture_frames(
        (broker_client, "HazardLightControlUnit-DriverCan0"),
        frames=["HazardLightButton"],
    ) as cap:
        # Assert that the signal transitions 0 → 1 → 0 (button press and release)
        await cap.wait_for_signal_values(
            "HazardLightButton",
            "HazardLightButton.HazardLightButton",
            values=[0, 1, 0],
            timeout=5.0,
        )

Capturing SOME/IP events

Use remotivelabs.topology.testing.some_ip.capture_events to subscribe to SOME/IP events:

from remotivelabs.broker import BrokerClient

from remotivelabs.topology.testing.some_ip import capture_events

MY_SERVICE = "MyTestService"
SPEED_EVENT = (MY_SERVICE, "SpeedEvent")


async def test_speed_event_sequence(broker_client: BrokerClient) -> None:
    async with capture_events(
        (broker_client, "consuming_service", 99),
        events=[SPEED_EVENT],
    ) as cap:
        # Wait for the speed to reach 50, then drop back to 0
        await cap.wait_for_event_parameter_values(SPEED_EVENT, "speed", values=[50, 0], timeout=5.0)

See remotivelabs.topology.testing for full documentation.

Behavioral Models

remotivelabs.topology.behavioral_model.BehavioralModel instances in RemotiveTopology run on top of RemotiveBroker, allowing them to use real network protocols such as CAN buses or SOME/IP networks.

The example below shows the simplest possible behavioral model - it connects to a broker and handles built-in control messages (ping, reboot), but performs no other work:

import asyncio

from remotivelabs.broker import BrokerClient

from remotivelabs.topology.behavioral_model import BehavioralModel


async def main():
    async with BrokerClient(url="http://127.0.0.1:50051") as broker_client:
        async with BehavioralModel(
            "BodyCanModule",
            broker_client=broker_client,
        ) as bm:
            await bm.run_forever()


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

See remotivelabs.topology.behavioral_model for full documentation and more examples.

Namespaces

A namespace maps to a single namespace in a RemotiveBroker topology, providing protocol-specific access to frames and signals. See remotivelabs.broker for more details.

Available namespace types:

remotivelabs.topology.namespaces.can — CAN bus. Sends and receives frames; drives the restbus.
remotivelabs.topology.namespaces.lin — LIN bus. Sends and receives frames; schedule-driven, no restbus.
remotivelabs.topology.namespaces.some_ip — SOME/IP. Sends requests and subscribes to events.
remotivelabs.topology.namespaces.scripted — Enables subscribing to frames transformed by scripts.

The Restbus - Sending Periodic Network Messages

The communication on CAN buses is often sent periodically, several times a second. remotivelabs.topology.namespaces.generic.Restbus handles this by publishing configured frames at their database cycle time. Configure it by passing a list of remotivelabs.topology.namespaces.generic.RestbusConfig objects with filters that select which frames to send:

import asyncio

from remotivelabs.broker import BrokerClient

from remotivelabs.topology.namespaces import filters
from remotivelabs.topology.namespaces.can import CanNamespace, RestbusConfig


async def main():
    async with (
        BrokerClient(url="http://127.0.0.1:50051") as broker_client,
        CanNamespace(
            "HazardLightControlUnit-DriverCan0",
            broker_client,
            restbus_configs=[RestbusConfig([filters.SenderFilter(ecu_name="HazardLightControlUnit")])],
        ) as hlcu_can,
    ):
        # start the restbus with signal database defaults and wait until cancelled
        await hlcu_can.restbus.start()
        await asyncio.Future()


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

Signal values can be updated at any time via remotivelabs.topology.namespaces.generic.Restbus.update_signals:

import asyncio

from remotivelabs.broker import BrokerClient

from remotivelabs.topology.namespaces import filters
from remotivelabs.topology.namespaces.can import CanNamespace, RestbusConfig


async def main():
    async with (
        BrokerClient(url="http://127.0.0.1:50051") as broker_client,
        CanNamespace(
            "HazardLightControlUnit-DriverCan0",
            broker_client,
            restbus_configs=[RestbusConfig([filters.SenderFilter(ecu_name="HazardLightControlUnit")])],
        ) as hlcu_can,
    ):
        # update signals in restbus before starting it
        await hlcu_can.restbus.update_signals(
            ("HazardLightButton.HazardLightButton", 1),
        )

        # start the restbus and loop until cancelled
        await hlcu_can.restbus.start()
        await asyncio.Future()


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

See remotivelabs.topology.namespaces.generic.RestbusConfig for timing options (cycle_time_millis, delay_multiplier) and remotivelabs.topology.namespaces.generic.Restbus for the full API.

Mapping

Signal and message routing and transforms for the RemotiveTopology Python framework.

The module consolidates the pattern of subscribing on one channel, optionally transforming, and republishing on another. It is built around a declarative mapping file format that can be generated from ARXML or hand-authored for non-AUTOSAR use cases. A mapping file is loaded and executed by a remotivelabs.topology.mapping.MappingModel — including custom sources and targets registered on it — or wired into a remotivelabs.topology.behavioral_model.BehavioralModel when the ECU also needs its own callbacks.

Requires the mapping extra:

pip install remotivelabs-topology[mapping]

See remotivelabs.topology.mapping for full documentation.

Logging

This library uses Python's standard logging module. By default, the library does not configure any logging handlers, allowing applications to fully control their logging setup.

import logging

logging.basicConfig(level=logging.INFO)
logging.getLogger("remotivelabs.topology").setLevel(logging.DEBUG)

For more advanced configurations, refer to the Python logging documentation.

View Source

  1'''
  2# RemotiveTopology framework
  3
  4RemotiveTopology framework is a Python library for modelling automotive ECUs and testing automotive network
  5communication using real protocols (CAN, SOME/IP) via RemotiveBroker.
  6It is built on top of [remotivelabs-broker](https://pypi.org/project/remotivelabs-broker/), which
  7provides the low-level gRPC client, signal types, and protocol primitives.
  8
  9## Installation
 10
 11```bash
 12pip install remotivelabs-topology
 13
 14# with optional FMU support
 15pip install remotivelabs-topology[fmu]
 16
 17# with the optional mapping module (message forwarding, signal transforms)
 18pip install remotivelabs-topology[mapping]
 19```
 20
 21## Project Links
 22
 23- [Documentation](https://docs.remotivelabs.com/)
 24- [Examples](https://github.com/remotivelabs/remotivelabs-topology-examples)
 25- [Issues](mailto:support@remotivelabs.com)
 26
 27## Usage
 28
 29The framework has two primary use-cases:
 30
 311. [Testing](#testing): Write test cases that interact with a running topology to verify correct behavior.
 322. [Behavioral Models](#behavioral-models): Create ECU stubs that model real ECU behavior on a running RemotiveTopology instance, sending
 33   and receiving network messages over real protocols.
 34
 35### Testing
 36
 37#### Capturing frames
 38
 39.. include:: testing/__init__.py
 40    :start-after: <!-- start-include -->
 41    :end-before: <!-- end-include -->
 42
 43#### Capturing SOME/IP events
 44
 45.. include:: testing/__init__.py
 46    :start-after: <!-- start-include2 -->
 47    :end-before: <!-- end-include2 -->
 48
 49See `remotivelabs.topology.testing` for full documentation.
 50
 51#### Behavioral Models
 52
 53.. include:: behavioral_model/__init__.py
 54    :start-after: <!-- start-include -->
 55    :end-before: <!-- end-include -->
 56
 57See `remotivelabs.topology.behavioral_model` for full documentation and more examples.
 58
 59#### Namespaces
 60
 61.. include:: namespaces/__init__.py
 62    :start-line: 1
 63    :end-before: """
 64
 65#### The Restbus - Sending Periodic Network Messages
 66
 67The communication on CAN buses is often sent periodically, several times a second.
 68`remotivelabs.topology.namespaces.generic.Restbus` handles this by publishing configured frames at
 69their database cycle time. Configure it by passing a list of `remotivelabs.topology.namespaces.generic.RestbusConfig` objects with filters
 70that select which frames to send:
 71
 72```python
 73.. include:: _docs/snippets/restbus_namespace.py
 74```
 75
 76Signal values can be updated at any time via `remotivelabs.topology.namespaces.generic.Restbus.update_signals`:
 77
 78```python
 79.. include:: _docs/snippets/restbus_namespace_set_signals.py
 80```
 81
 82See `remotivelabs.topology.namespaces.generic.RestbusConfig` for timing options (`cycle_time_millis`,
 83`delay_multiplier`) and `remotivelabs.topology.namespaces.generic.Restbus` for the full API.
 84
 85#### Mapping
 86
 87.. include:: mapping/__init__.py
 88    :start-after: <!-- start-include -->
 89    :end-before: <!-- end-include -->
 90
 91See `remotivelabs.topology.mapping` for full documentation.
 92
 93#### Logging
 94
 95This library uses Python's standard `logging` module. By default, the library does not configure any
 96logging handlers, allowing applications to fully control their logging setup.
 97
 98```python
 99import logging
100
101logging.basicConfig(level=logging.INFO)
102logging.getLogger("remotivelabs.topology").setLevel(logging.DEBUG)
103```
104
105For more advanced configurations, refer to the
106[Python logging documentation](https://docs.python.org/3/library/logging.html).
107
108'''
109# Imports in this file affect import paths and documentation
110
111import logging
112from importlib.util import find_spec
113
114from remotivelabs.topology import behavioral_model, cli, control, ecu_mock, metrics, namespaces, testing, time
115
116# Disable library logging by default
117_logger = logging.getLogger("remotivelabs.topology")
118_logger.addHandler(logging.NullHandler())
119
120__all__ = [
121    "behavioral_model",
122    "namespaces",
123    "control",
124    "ecu_mock",
125    "testing",
126    "time",
127    "cli",
128    "metrics",
129]
130
131# Optional submodules gated behind extras (`fmu`, `mapping`). Their public API imports the extra's
132# dependencies, so they can't be imported eagerly like the modules above. List them in __all__ only
133# when their dependencies are present — so pdoc documents them (docs are built with all extras) and
134# `from remotivelabs.topology import *` exposes them — without requiring the extra for a plain
135# `import remotivelabs.topology`.
136for _name, _deps in (("fmu", ("fmpy",)), ("mapping", ("pydantic", "yaml"))):
137    if all(find_spec(_dep) for _dep in _deps):
138        __all__.append(_name)