remotivelabs.topology.namespaces.scripted

class ScriptedNamespace(remotivelabs.topology.namespaces.namespace.Namespace[remotivelabs.broker.filters.SenderFilter | remotivelabs.broker.filters.ReceiverFilter | remotivelabs.broker.filters.FrameFilter | remotivelabs.broker.filters.AllFramesFilter | remotivelabs.broker.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 = True, decode_named_values: bool | None = None) -> 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: If True, the first broker message (an empty sync marker) is consumed internally before the iterator is returned, ensuring the broker is ready before the first real value arrives. Useful when you need to guarantee the subscription is live before publishing something.
  • decode_named_values: True will decode named values to str. Defaults to the value set in the constructor.
Returns:

An asynchronous iterator with Frames.

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

Create an input handler for this namespace to be used with a BehavioralModel.

Arguments:
  • filters: A sequence of filter objects to select relevant frames or signals.
  • callback: An async callback invoked with each matching Frame.
Returns:

A (namespace_name, handler) pair ready to pass to BehavioralModel.

Inherited Members
remotivelabs.topology.namespaces.namespace.Namespace
name