Signal Module

This module provides all functionality for signal handling.

class mtf.network_port.signals.signal.SignalListener

This class implements the functionality for listening to and processing various signals.

__init__(channel_name: str, frame_id: int, pdu_name: str, signal_name: str)

Creates a new instance of SignalListener.

Args:

channel_name (str): The name of the bus/channel to listen on. frame_id (int): The frame ID to listen on. pdu_name (str): The PDU name to listen on. signal_name (str): The signal name to listen on.

Returns:

SignalListener: The newly created instance.

adapt_queue_elements() → Queue[BaseEvent]

Gets the BaseEvent queue.

Returns:

Queue[BaseEvent]: The event queue containing the BaseEvent elements.

get_queue() → Queue

Gets the queue.

Returns:

Queue: The event queue.

start_listening(keep_cache: bool = False, silent_mode: bool = True, on_change: bool = False)

Starts listening for the signal.

Args:

keep_cache (bool): The flag indicating whether the queue cache should be kept after restarting the listener.

When enabled, previously queued items remain available. Defaults to False.

silent_mode (bool): If True, the listener operates in silent mode and callbacks are ignored.

This way captured frames can be retrieved using get_queue(). If False, the callback function is required and must be provided by calling the register_callback method. Defaults to True.

on_change (bool): Whether or not to only trigger the listener on signal changes. Defaults to False.

Returns:

bool: True if the listener was started successfully; False otherwise.

stop_listening() → bool

Stops listening for the signal.

Returns:

bool: True if the listener was stopped successfully, False otherwise.

register_callback(callback_fnc: Callable[[BaseEvent], None])

Registers the given callback function.

Args:

callback_fnc (SignalCallback): The callback function to register.

Returns:

bool: True on success, False otherwise.

Note:

Warning

The callback function should be lightweight without logs. Otherwise, it will reduce the performance dramatically. The object that is passed to callback functions is mutable. So the user must be careful.

Examples:

Setting a callback

# ...
from mtf.network_port.listeners.listener_base import BaseEvent
# ...

def my_callback(msg: BaseEvent):
    # implement callback logic

# create the listener
# register our callback
listener.register_callback(my_callback)

unregister_callback(callback_fnc: Callable[[BaseEvent], None])

Unregisters the given callback function.

Args:

callback_fnc (SignalCallback): The callback function to unregister.

Returns:

bool: True on success, False otherwise.

static __new__(cls, *args, **kwargs)

Creates a new instance.

Args:

*args: The variable arguments list. **wargs: The keyword arguments list.

Returns:

ControllersBase: The newly created instance.

Note:

Note

The instance will be automatically added to the list of controllers to be cleaned up later. It guarantees that all the controller instances are followed up and cleaned up when the associate method ‘controllers_cleanup(cls)’ is called.

Important

Do not override this method in subclasses to maintain this functionality.

clean_up()

This method cleans up resources used by the listener by stopping any ongoing operations and resetting the listener’s internal state to ensure it is in a clean and ready state for future use.

Returns:

bool: Returns True if the cleanup is successfully done. False if an exception occurs.

classmethod controllers_cleanup()

Cleans up all controller instances.

It calls clean_up() for each controller in the list. If a cleanup fails, an error is raised, but the method continues processing the remaining controllers.

Returns:

bool: True indicating that all controllers were cleaned up successfully.

Otherwise, False is returned when an error occurs during the cleanup process.

reset()

cleanup the queue

set_queue_size(max_size: int)

This call of set queue size will clear the queue and resize it better to call it at the beginning : before start listening if max_size <= 0 the queue size is infinite

class mtf.network_port.signals.signal.Signal

static __new__(cls, channel_name: str, frame_id: str, pdu_name: str, signal_name: str)

Creates a new instance of Signal class.

Args:

channel_name (str): The bus/channel to listen on. frame_id (str): The frame ID to listen for. pdu_name (str): The PDU name to listen for. signal_name (str): The signal name to listen for.

Returns:

Signal: The newly created instance associated with the tuple (channel_name, frame_id, pdu_name, signal_name).

__init__(channel_name: str, frame_id: str, pdu_name: str, signal_name: str)

Initializes a new instance.

Args:

channel_name (str): The bus/channel to listen on. frame_id (str): The frame ID to listen for. pdu_name (str): The PDU name to listen for. signal_name (str): The signal name to listen for.

Returns:

Signal: The newly created instance associated with the defined channel_name, frame_id, pdu_name and signal_name.

property channel_name: str

Gets the channel name.

Returns:

str: The channel name associated with the Signal instance.

property frame_id: str

Gets the frame ID.

Returns:

str: The frame ID associated with the Signal instance.

property pdu_name: str

Gets the PDU name.

Returns:

str: The PDU name associated with the Signal instance.

property signal_name: str

Gets the signal name.

Returns:

str: The signal name associated with the Signal instance.

property listener

Gets the associated listener.

Returns:

SignalListener: The SignalListener instance associated with the signal.

property frame_id_int

Gets the associated frame ID.

Returns:

int: The integer representation of the frame ID associated with the signal instance.

set_signal_raw_value(signal_value: int | list[int], update_and_send: bool = True) → bool

Sets the signal to the given raw value.

Args:

signal_value (Union[int, list[int]]): The raw value of the signal. update_and_send (bool): A flag indicating whether the update and send operations should be performed together.

If False, the signal value is updated but not transmitted until a subsequent set_signal_raw_value call with update_and_send=True occurs. Defaults to True.

Returns:

bool: True if the raw value was set successfully, False otherwise.

set_signal_physical_value(signal_value: float, update_and_send: bool = True) → bool

Sets the signal to the given physical value.

Args:

signal_value (float): The physical value of the signal. update_and_send (bool): A flag indicating whether the update and send operations should be performed together.

If False, the signal value is updated but not transmitted until a subsequent set_signal_physical_value call with update_and_send=True occurs. Defaults to True.

Returns:

bool: True on success; False otherwise.

get_signal_last_value() → int | list[int]

Retrieves the last value of the signal.

Returns:

Union[int, list[int]]: The last value of the signal.

get_signal_last_physical_value() → int | list[int]

Retrieves the last physical value of the signal.

Returns:

Union[int, list[int]]: The last physical value of the signal.

last_signal_value_retrieved() → bool

Checks if the last value of the signal has been retrieved.

Returns:

bool: True if the last value has been retrieved; False otherwise.

get_signal_path() → str

Gets the signal path.

Returns:

str: The path of the signal.

is_from_dynamic_pdu_signal() → bool

Checks if the signal is from a dynamic PDU.

Returns:

bool: True if the signal is from a dynamic PDU; False otherwise.

classmethod clear_registered_signals()

Clears the instances dictionary of registered signals.