pip install remotivelabs-broker

import asyncio

from remotivelabs.broker import BrokerClient


async def main() -> None:
    # Create a BrokerClient instance and list all namespaces available on that broker.
    async with BrokerClient(url="http://127.0.0.1:50051") as broker_client:
        ns = await broker_client.list_namespaces()
        print(f"Namespaces: {ns}")


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

import asyncio

from remotivelabs.broker import BrokerClient, FrameInfo


async def main() -> None:
    async with BrokerClient(url="http://127.0.0.1:50051") as broker_client:
        # list all available frames in the "DriverCan" namespace
        frame_infos: list[FrameInfo] = await broker_client.list_frame_infos("DriverCan")

        # simply print all frames and their signals
        for frame_info in frame_infos:
            print(f"{frame_info.name}")
            for signal_info in frame_info.signals.values():
                print(f"  {signal_info.name}")


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

import asyncio
from typing import AsyncIterator

from remotivelabs.broker import BrokerClient, Frame


async def main() -> None:
    async with BrokerClient(url="http://127.0.0.1:50051") as broker_client:
        # subscribe to a frame named "EngineData" (including all its signals) in the "DriverCan" namespace
        stream: AsyncIterator[Frame] = await broker_client.subscribe_frames(
            ("DriverCan", ["EngineData"]),
        )

        # will receive frames until stream is closed by broker or an error occurs
        async for frame in stream:
            print(f"Received frame: {frame}")


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

from remotivelabs.broker import FrameInfo, SignalInfo
from remotivelabs.broker.filters import (
    AllFramesFilter,
    FrameFilter,
    ReceiverFilter,
    SenderFilter,
    SignalFilter,
    filter_recursive,
    is_frame_filter,
    is_signal_filter,
)

# assume frames and signals are populated elsewhere
frames: list[FrameInfo] = []
signals: list[SignalInfo] = []

# Example 1: Include all frames
all_frames_filter = AllFramesFilter()
filtered_frames = list(filter(all_frames_filter, frames))

# Example 2: Include all frames but exclude a specific frame
frame_exclude_filter = FrameFilter(frame_name="Frame1", exclude=True)
filtered_frames = list(filter(frame_exclude_filter, frames))

# Example 3: Filter frames sent by a specific ECU
sender_filter = SenderFilter(ecu_name="ECU1")
filtered_frames = list(filter(sender_filter, frames))

# Example 4: Filter frames received by a specific ECU, excluding one signal
receiver_filter = ReceiverFilter(ecu_name="ECU2")
signal_exclude_filter = SignalFilter(signal_name="SignalA", exclude=True)
filtered_frames = list(filter(receiver_filter, frames))
filtered_signals = list(filter(signal_exclude_filter, signals))

# Example 5: Chaining filters
frame_include = FrameFilter(frame_name="FrameA")
frame_exclude = FrameFilter(frame_name="FrameB", exclude=True)
filtered_frames = list(filter(frame_exclude, filter(frame_include, frames)))

# Example 6: Combining inclusion and exclusion with AllFramesFilter and FrameFilter
all_frames_filter = AllFramesFilter()
frame_exclude_filter = FrameFilter(frame_name="FrameC", exclude=True)
filtered_frames = list(filter(frame_exclude_filter, filter(all_frames_filter, frames)))

# Example 7: Recursive filtering of signals (SignalInfos) in frames (FrameInfos)
filtered_frames = [
    filtered_frame
    for frame in frames
    if (filtered_frame := filter_recursive(frame, filters=[all_frames_filter, signal_exclude_filter])) is not None
]

# Example 10: Type checking filters
assert is_frame_filter(all_frames_filter)  # works with frames
assert not is_signal_filter(all_frames_filter)  # doesn't work with signals

assert not is_frame_filter(signal_exclude_filter)  # doesn't work with frames
assert is_signal_filter(signal_exclude_filter)  # works with signals

assert is_frame_filter(sender_filter)  # works with frames
assert is_signal_filter(sender_filter)  # works with signals

import asyncio
from typing import AsyncIterator

from remotivelabs.broker import BrokerClient, Frame, FrameInfo
from remotivelabs.broker.filters import ReceiverFilter


async def main() -> None:
    async with BrokerClient(url="http://127.0.0.1:50051") as broker_client:
        # list frames in the "DriverCan" namespace
        frame_infos: list[FrameInfo] = await broker_client.list_frame_infos("DriverCan")

        # filter according to some critera
        filtered_frame_infos = list(filter(ReceiverFilter(ecu_name="ECU2"), frame_infos))

        # subscribe using the filtered frame infos
        stream: AsyncIterator[Frame] = await broker_client.subscribe_frames(
            ("DriverCan", filtered_frame_infos),
        )

        # will receive frames until stream is closed by broker or an error occurs
        async for frame in stream:
            print(f"Received frame: {frame}")


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

import asyncio
from typing import AsyncIterator

from remotivelabs.broker import BrokerClient, Frame
from remotivelabs.broker.filters import ReceiverFilter


async def main() -> None:
    async with BrokerClient(url="http://127.0.0.1:50051") as broker_client:
        # subscribe using the ReceiverFilter to only receive frames from ECU named "ECU2" in the "DriverCan" namespace
        stream: AsyncIterator[Frame] = await broker_client.subscribe_frames(
            ("DriverCan", [ReceiverFilter(ecu_name="ECU2")]),
        )

        # will receive frames until stream is closed by broker or an error occurs
        async for frame in stream:
            print(f"Received frame: {frame}")


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

import asyncio

from remotivelabs.broker import BrokerClient
from remotivelabs.broker.restbus import RestbusFrameConfig, RestbusSignalConfig


async def main() -> None:
    async with BrokerClient(url="http://127.0.0.1:50051") as broker_client:
        # Start the restbus on the "DriverCan" namespace with one frame configuration
        await broker_client.restbus.add(
            ("DriverCan", [RestbusFrameConfig(name="EngineData")]),
            start=True,
        )

        # now update some of its signals
        signal_configs: list[RestbusSignalConfig] = [
            RestbusSignalConfig.set(name="EngineData.EngineRpm", value=1500),
            RestbusSignalConfig.set(name="EngineData.EngineTemp", value=90),
        ]
        await broker_client.restbus.update_signals(
            ("DriverCan", signal_configs),
        )


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

import asyncio

from remotivelabs.broker.recording_session import RecordingSessionClient


async def main() -> None:
    # Create a recording session client
    async with RecordingSessionClient(url="http://127.0.0.1:50051") as rs_client:
        # List available recording files on the broker
        files = await rs_client.list_recording_files()
        print(f"Available recording files: {files}")

        # open a recording session with the first file
        rs_ref = rs_client.get_session(str(files[0]))
        async with rs_ref as session:
            # start playback
            status = await session.play()
            print(f"Playback started with status: {status}")

        # exiting the async with block will close the session, but playback will continue if other sessions are open


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

import asyncio

from remotivelabs.broker import BrokerClient
from remotivelabs.broker.secoc import SecocFreshnessValue


async def main() -> None:
    async with BrokerClient(url="http://127.0.0.1:50051") as broker_client:
        prop = SecocFreshnessValue(fv=b"\x00\x00\x12\x34")
        await broker_client.set_secoc_property(namespace="MyNamespace", property=prop)


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

import logging

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