remotivelabs.broker.restbus

class Restbus:
Restbus(channel: grpc.aio._base_channel.Channel, client_name: str)
async def close(self, *namespace: str) -> None:

Removes all configured signals and stops the Restbus on the specified namespace.

Arguments:
  • *namespace: One or more namespaces to close
async def add( self, *frames: tuple[str, list[RestbusFrameConfig]], start: bool = False):

Adds one or more frames to the Restbus with optional start flag.

Arguments:
  • *frames: One or more tuples, each containing namespace and list of frame configurations which should be added to the restbus
  • start: If True, starts the frames after adding. Defaults to False.

Note: The start flag affects all frames running on the namespace, even if they are added since before.

async def start(self, *namespace: str):

Starts Restbus signal publishing for the specified namespaces.

Arguments:
  • *namespace: One or more namespaces to start
async def stop(self, *namespace: str):

Stops Restbus signal publishing for the specified namespaces.

Arguments:
  • *namespace: One or more namespaces to stop. A stopped restbus can be started again.
async def remove(self, *frames: tuple[str, list[str]]):

Removes specific frames from the Restbus.

Arguments:
  • *frames: One or more tuples, each containing namespace and list of frame names to remove from the restbus
async def update_signals( self, *signals: tuple[str, list[RestbusSignalConfig]]) -> None:

Updates the configured signals on the Restbus with new values.

Arguments:
  • *signals: One or more tuples, each containing namespace and list of signal configurations to apply to the restbus
async def reset_signals(self, *signals: tuple[str, list[str]]) -> None:

Resets specified signals to their default configured values.

Arguments:
  • *signals: One or more tuples, each containing namespace and list of signal names to reset
async def reset_namespaces(self, *namespace: str) -> None:

Resets all configured namespaces and their associated signals to default values.

Arguments:
  • *namespace: One or more namespaces to reset
@dataclass
class RestbusSignalConfig:

This class defines how a specific signal should behave when emitted by Restbus. A signal can have:

Attributes:
  • name: The name of the signal
  • loop: Values emitted in order after the initial sequence
  • initial: Optional values emitted once before the loop starts
RestbusSignalConfig( name: str, loop: Sequence[Union[int, float, bytes, str]], initial: Sequence[Union[int, float, bytes, str]] = <factory>)
name: str
loop: Sequence[Union[int, float, bytes, str]]
initial: Sequence[Union[int, float, bytes, str]]
@classmethod
def set( cls, name: str, value: Union[int, float, bytes, str]) -> RestbusSignalConfig:

Create a SignalConfig with a constant value.

Arguments:
  • name: Name of the signal
  • value: Value to set in Restbus
@classmethod
def set_update_bit( cls, name: str) -> RestbusSignalConfig:

Create a SignalConfig for an update-bit pattern (sends 1 once, then constant 0).

Arguments:
  • name: Name of the signal
@dataclass
class RestbusFrameConfig:

Configuration for a frame in the Restbus.

Attributes:
  • name: The name of the frame to configure.
  • cycle_time: Optional cycle time override for the frame. If None, the default from the broker's database is used.
RestbusFrameConfig(name: str, cycle_time: float | None = None)
name: str
cycle_time: float | None = None