remotivelabs.topology.namespaces.scripted

class ScriptedNamespace(remotivelabs.topology.namespaces.namespace.Namespace[typing.Union[remotivelabs.topology.namespaces.filters.SenderFilter, remotivelabs.topology.namespaces.filters.ReceiverFilter, remotivelabs.topology.namespaces.filters.FrameFilter, remotivelabs.topology.namespaces.filters.AllFramesFilter, remotivelabs.topology.namespaces.filters.SignalFilter], collections.abc.Callable[[remotivelabs.broker.frame.Frame], collections.abc.Awaitable[None]]]):

ScriptedNamespace provides access to frames that have been transformed by scripts in the broker.

ScriptedNamespace( name: str, broker_client: remotivelabs.broker.BrokerClient, decode_named_values: bool = False)

Initialize the scripted namespace client.

Arguments:
  • name: The namespace name to operate in.
  • broker_client: The client used to communicate with the broker.
  • decode_named_values: True will decode named values to str.
Note:

Use together with a BehavioralModel or start the instance using a context manager:

async with ScriptedNamespace(...) as broker_client:
    ...
async def open(self) -> typing_extensions.Self:

Open the namespace. This is an idempotent operation - calling it multiple times has no additional effect.

Returns:

The namespace

Raises:
  • ValueError: If the namespace is not of type 'scripted'.
async def close(self) -> None:

Will close the namespace. This is an idempotent operation - calling it multiple times has no additional effect.

async def list_frame_infos(self) -> list[remotivelabs.broker.FrameInfo]:

Return all available frame infos in the namespace.

async def get_frame_info(self, name: str) -> remotivelabs.broker.FrameInfo | None:

Return information about a specific frame by name.

async def list_signal_infos(self) -> list[remotivelabs.broker.SignalInfo]:

Return all available signal infos in the namespace.

async def get_signal_info(self, name: str) -> remotivelabs.broker.SignalInfo | None:

Return information about a specific signal by name.

async def subscribe( self, *name: str, on_change: bool = False) -> AsyncIterator[list[remotivelabs.broker.Signal]]:

Subscribe to a list of signals.

Arguments:
  • *signals: One or more signals to subscribe to.
  • on_change: Whether to receive updates only on change.

Note: does not support decoding enums

Returns:

An asynchronous iterator of lists of Signal objects.

async def subscribe_frames( self, *frames: remotivelabs.broker.FrameSubscription, on_change: bool = False, initial_empty: bool = False, decode_named_values: bool = False) -> AsyncIterator[remotivelabs.broker.Frame]:

Subscribe to a Frames.

Arguments:
  • *frames: One or more frames to subscribe to.
  • on_change: Whether to receive updates only on change.
  • initial_empty: True will wait until the broker has sent an initial message
  • decode_named_values: True will decode named values to str.
Returns:

An asynchronous iterator with Frames.

def create_input_handler( self, filters: Sequence[typing.Union[remotivelabs.topology.namespaces.filters.SenderFilter, remotivelabs.topology.namespaces.filters.ReceiverFilter, remotivelabs.topology.namespaces.filters.FrameFilter, remotivelabs.topology.namespaces.filters.AllFramesFilter, remotivelabs.topology.namespaces.filters.SignalFilter]], callback: Callable[[remotivelabs.broker.Frame], Awaitable[None]]) -> tuple[str, remotivelabs.topology.namespaces.input_handlers.InputHandler]:

Creates a list of input handlers for the given namespace to be used with a BehavioralModel.

Each handler defines a filter and a corresponding async callback for processing matching frames.

Arguments:
  • - filters: A sequence of filter objects to select relevant frames or signals.
  • - callback: An async callback function that receives and processes a Frame.
Inherited Members
remotivelabs.topology.namespaces.namespace.Namespace
name