Core Module

PyRbs Main Class

class pyrbs.PyRbs(config=None, log_all=False, run_on_start=True, exit_on_stop=True, timeseriesdb=True, dashboard=True)

Bases: object

Parameters:

config (str | BytesIO | None)
log_all (bool)
run_on_start (bool)
timeseriesdb (bool)
dashboard (bool)

on_pre_start()

Callback event that is executed right befor the actual start of the Restbus-Simulation

@rbs.on_pre_start():
def on_pre_start_event(self)
    with open("large_config.toml") as f:
        config = rtoml.load(f)

Parameters:: self (PyRbs)

on_start()

Callback event that is executed right at start of the Restbus-Simulation

@rbs.on_pre_start():
def on_start_event(self)
    default_can_trace.start()

Parameters:: self (PyRbs)

on_stop()

Callback event that is executed right at stop of the Restbus-Simulation

@rbs.on_pre_start():
def on_start_event(self)
    default_can_trace.stop()

Parameters:: self (PyRbs)

on_post_stop()

Callback event that is executed right after the stop of the Restbus-Simulation and befor actual process shutdown

@rbs.on_pre_start():
def on_pre_start_event(self)
    with open("large_config.toml") as f:
        rtoml.dump(f, some_result_or_state)

Parameters:: self (PyRbs)

stop()

stops the Restbus-Simulation

@rbs.timer.on("test_runtime")
def test_runtime_end(self):
    rbs.stop()

Return type:: None

Handlers

Bus Logger Handler

class pyrbs.buslog.buslog_handler.BuslogHandler(_rbs)

Bases: object

Parameters:: _rbs (PyRbs)

add(name, logpath, channels=[], sysvar=[], mode='complete')

Adds a new BusLogger to the simulation

from pyrbs import PyRbs, BusLogger
rbs = PyRbs()

new_trace: BusLogger = rbs.buslog.add(
    "new_trace", Path(f"/default/{dt.now().strftime('%Y-%m-%d_%H-%M-%S')}.mf4")
)

Parameters:

name (str) – Identifier for this BusLogger
logpath (Union[str, Path]) – Base path for the log file, root for this logger is /basepath_traces/{project_name}@{profile_name}/
channels (List, optional) – List of channels to consider (empty uses all available channels). Defaults to [].
sysvar (List[str], optional) – List of sysvars to consider in trace file (empty uses all available sysvars). Defaults to [].
mode (Literal["complete", "toggle"], optional) – Obsolete since pause was introduced. Defaults to “complete”.

Return type:

BusLogger

get(name)

get a predefined BusLogger

Parameters:: name (str) – Identifier for BusLogger
Return type:: BusLogger

@rbs.sysvar.on_change("testbench_logging_trigger")
def change_mode(self: SysVar):
    if self.value == "pause"
        tb_log = rbs.buslog.get("testbench_log")
        tb_log.pause()

snapshot(name, logpath, pre_trigger, post_trigger, channels=[], sysvar=[])

Create a snapshot logger that captures data around trigger events.

A Snapshot is created the moment this function is called so it must be embedded into the triggercondition. The file will be created at the end of the post_trigger time.

For the Snapshot to work the cantimedb parameter of PyRbs must be True

Parameters:

name (str) – Identifier for this snapshot
logpath (str) – Base path for exported trace files (relative to /opt/pycando/store/traces)
pre_trigger (float) – Seconds of data to retain before trigger
post_trigger (float) – Seconds of data to capture after trigger
channels ([int]) – CAN channels to capture. Defaults to [].
sysvar ([str]) – System variables to capture. Defaults to [].

Returns:

The created snapshot instance

Return type:

Snapshot

stop_all(): stops all bus logger

db_execute(query, params=())

Parameters:

query (str)
params (tuple)

Return type:

None

db_fetch_one(query, params=())

Parameters:

query (str)
params (tuple)

db_fetch_all(query, params=())

Parameters:

query (str)
params (tuple)

Return type:

list[Row]

db_commit()

Return type:: None

queue_export(record_uuid, stop_time=None, legacy_status='stopped')

Parameters:

record_uuid (str)
stop_time (int | None)
legacy_status (str)

Return type:

None

wake_export_worker(record_uuid)

Parameters:: record_uuid (str)
Return type:: None

sanitize_stale_active_exports()

Return type:: int

sanitize_orphaned_recordings()

Return type:: int

fetch_pending_exports()

Return type:: list[Row]

mark_export_active(record_uuid, owner)

Parameters:

record_uuid (str)
owner (str)

Return type:

None

mark_export_success(record_uuid)

Parameters:: record_uuid (str)
Return type:: None

mark_export_failure(record_uuid, exit_code, error)

Parameters:

record_uuid (str)
exit_code (int | None)
error (str | None)

Return type:

None

CAN Channel Handler

class pyrbs.canbus.channel.can_channel_handler.CanChannelHandler(_rbs)

Bases: object

Parameters:: _rbs (PyRbs)

add(config, *, register_with_core=True)

Adds a CAN channel to the rbs.

CAN database parsing and model ownership now live in pyrbs_core. The parent only mirrors configured channels and hydrates message/signal wrappers from the core after startup.

Parameters:

config (CanChannelConfig) – config will be generated from the files in /config folder
register_with_core (bool)

Return type:

None

CAN Message Handler

class pyrbs.canbus.message.can_message_handler.CanMessageHandler(_rbs)

Bases: object

Parameters:: _rbs (PyRbs)

add(ch_id, arb_id, frame_metadata=None, core_message=None)

Add or hydrate a CAN message wrapper from core-owned metadata.

Parameters:

ch_id (int)
arb_id (int)
frame_metadata (dict | None)

Return type:

CanMessage

hydrate_from_core()

Return type:: None

get(identifier)

Get a CAN message from its identifier.

Examples for identifier usage:

"CAN1.Control_01"
"ALIAS_NAME.256"
"1.256"
"1.0x100"
["CAN1", "Control_01"]
[1, 256]

Parameters:: identifier (str | list[str])
Return type:: CanMessage

on(identifier, callback_name=None, priority=100, **kwargs)

Callback for a can message event

@rbs.can_message.on("CAN1.Control_01")
def on_control_01_msg(self):
    print(f"new msg: {self}")

Parameters:

identifier (str)
callback_name (str)
priority (int)

CAN Signal Handler

class pyrbs.canbus.signal.can_signal_handler.CanSignalHandler(_rbs)

Bases: object

Parameters:: _rbs (PyRbs)

get(identifier)

Get a CAN signal from its identifier.

Examples for identifier usage:

"CAN1.Control_01.D_Value"
"ALIAS_NAME.256.D_Value"
"1.256.D_Value"
"1.0x100.D_Value"
["CAN1", "Control_01", "D_Value"]
[1, 256, "D_Value"]

from pyrbs import PyRbs, CanMessage, CanSignal, Timer
rbs = PyRbs(dashboard=True)

d_val_sig: CanSignal = rbs.can_signal.get("CAN1.Control_01.D_Value")

Return type:: CanSignal

on_update(identifier, callback_name=None, priority=100, **kwargs)

Callback for a can signal update event

@rbs.can_signal.on_update("CAN1.Control_01.D_Value")
def on_update_dval(self):
    print(f"new D_Value raw: {self.raw}, phys: {self.phys}")

Parameters:

identifier (str)
callback_name (str | None)
priority (int)

on_change(identifier, callback_name=None, priority=100, **kwargs)

Callback for a can signal change event, triggers only if the value changed in comparison to the previos signal value

@rbs.can_signal.on_change("CAN1.Control_01.D_Value")
def on_change_dval(self):
    print(f"new D_Value raw: {self.raw}, phys: {self.phys}")

Parameters:

identifier (str)
callback_name (str | None)
priority (int)

property decode_information

System Variables Handler

class pyrbs.variables.sysvar_handler.SysVarHandler(_rbs)

Bases: object

Parameters:: _rbs (PyRbs)

add(identifier, init_value)

Add a sysvar to the restbus simulation.

A "." is used for namespace indentation. Sysvars that are defined in .toml or .json files in /config follow the same pattern.

Example config shape:

{"pid_control": {"p_val": 0.5, "i_val": 0.1, "d_val": 0.2}}

This results in identifiers:

"pid_control.p_val"
"pid_control.i_val"
"pid_control.d_val"

The data type of init_value determines the type of the sysvar and cannot be changed after initialization.

from pyrbs import PyRbs, SysVar
rbs = PyRbs(dashboard=True)

p_val_sys: SysVar = rbs.sysvar.add("P_Value", 0.1)

Parameters:

identifier (str)
init_value (int | float | str)

Return type:

SysVar

get(identifier)

Gets a sysvar by its identifier

from pyrbs import PyRbs, SysVar
rbs = PyRbs(dashboard=True)

d_val_sys: SysVar = rbs.sysvar.get("D_Value", 0.1)

Parameters:: identifier (str)
Return type:: SysVar

on_update(identifier, callback_name=None, priority=100, **kwargs)

Callback for a sysvar update event

@rbs.sysvar.on_update("alive_counter")
def on_update_alive_counter(self):
    print(f"alive count: {self.value}")

Parameters:

identifier (str)
callback_name (str | None)
priority (int)

on_change(identifier, callback_name=None, priority=100, **kwargs)

Callback for a sysvar change event

@rbs.sysvar.on_change("alive_counter")
def on_change_alive_counter(self):
    print(f"alive count: {self.value}")

Parameters:

identifier (str)
callback_name (str | None)
priority (int)

Timer Handler

class pyrbs.timers.timer_handler.TimerHandler(_rbs)

Bases: object

Parameters:: _rbs (PyRbs)

add(name, time_to_elapse, execution_count=0, active_on_start=False)

Add a timer

from pyrbs import PyRbs, Timer
rbs = PyRbs()

timer_100ms: Timer = rbs.timer.add("100ms_timer", 0.1, 0, True)

Parameters:

name (str) – name of the timer
time_to_elapse (float) – duration of the timer
execution_count (int, optional) – 0 = endles repetition. 1 = 1x execution 2 = 2x… Defaults to 0.
active_on_start (bool, optional) – timer starts running at “on_start” event. Defaults to False.

Return type:

Timer

on(timer_name, callback_name=None, priority=100, **kwargs)

Callback for a due timer event

@rbs.timer.on("100ms_timer")
def on_100ms_timer(self):
    print(f"ping from timer")

Parameters:

timer_name (str)
callback_name (str)
priority (int)

get_timer_by_name(name)

Get a timer by name

Parameters:: name (str) – name of timer
Return type:: Timer

State Machine Handler

class pyrbs.statemachine.statemachine_handler.StatemachineHandler(_rbs)

Bases: object

PyRbs uses the python-statemachine package docs can be found here: https://python-statemachine.readthedocs.io/en/latest/

Parameters:: _rbs (PyRbs)

add_statemachine(name, statemachine)

Adds a statemachine to the restbus simulation

# /rbs/control_states.py
from statemachine import StateMachine, State, exceptions

class ControllerTargetStates(StateMachine):
    sleep = State("sleep", initial=True)
    standby = State("standby")
    speed = State("speed")
    torque = State("torque")

    to_standby = sleep.to(standby) | speed.to(standby) | torque.to(standby)
    to_speed = standby.to(speed)
    to_torque = standby.to(torque)
    to_sleep = standby.to(sleep)
    to_save_state = sleep.to(standby) | speed.to(standby) | torque.to(standby)

# /rbs/main.py
from pyrbs import PyRbs
from control_states import ControllerTargetStates

rbs = PyRbs()

ctrl_state: ControllerTargetStates = rbs.state.add_statemachine(
    "ctrl_state", ControllerTargetStates
)

Parameters:

name (str)
statemachine (object)

Return type:

object

on_enter(name, state, priority=100)

Callback for a on_enter state event

@rbs.state.on_enter("ctrl_state", "sleep")
def on_enter_sleep(self: ControllerTargetStates):
    select_mode.value = 0
    timer_100ms.stop()

Parameters:: priority (int)

on_exit(name, state, priority=100)

Callback for a on_exit state event

@rbs.state.on_exit("ctrl_state", "sleep")
def on_exit_sleep(self: ControllerTargetStates):
    timer_100ms.start()

Parameters:: priority (int)

before_transition(name, state, priority=100)

Callback for a before_transition state event

Parameters:: priority (int)

on_transition(name, state, priority=100)

Callback for a on_transition state event

Parameters:: priority (int)

after_transition(name, state, priority=100)

Callback for a after_transition state event

Parameters:: priority (int)