remotivelabs.topology.namespaces.scripted

class ScriptedNamespace(remotivelabs.topology.namespaces.namespace.Namespace[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 = True, 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: 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.
Returns:

An asynchronous iterator with Frames.

def create_input_handler( self, filters: Sequence[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