E2E Verification Module

class mtf.network_port.e2e_verification.E2EVerificationListener

This class is responsible for Starting and Stopping the E2E verification process

__init__(path: str, someip_check: bool = False)

someip_e2e_check

start_verifying()

Start the E2E verification listener

Returns:: True if the E2E Verification Listener is started. False otherwise.

stop_verifying(): Stop the E2E verification listener

register_callback(callback: Callable[[str, int, E2ECheckStatus], None] | Callable[[str, mtf_timestamp, E2EReceiverState], None]) → None

Register a callback

Parameters:: callback – E2ECallback
Returns:: None

unregister_callback(callback: Callable[[str, int, E2ECheckStatus], None] | Callable[[str, mtf_timestamp, E2EReceiverState], None]) → None

Unregister a callback

Parameters:: callback – E2ECallback
Returns:: None

start_listening(keep_cache: bool = False, silent_mode: bool = True) → bool

Start the E2E verification listener

Returns:

True if the E2E Verification Listener is started. False otherwise.

Examples:

>>> path = "channel_name::frame_id::pdu_name" # the path could be for LegacyPDU
>>> path = "ecu_name::npdu_name" # the path could be for NPDU
>>> path = "ecu_name::npdu_id"  # the path could be for NPDU
 # Create an instance of E2EVerificationListener
>>> e2e_verif_listener = E2EVerificationListener(path)
>>> e2e_verif_listener.start_verifying()

stop_listening()

Stop the E2E verification listener

Examples:

>>> path = channel_name::frame_id::pdu_name # the path could be for LegacyPDU
>>> path = ecu_name::npdu_name # the path could be for NPDU
>>> path = ecu_name::npdu_id  # the path could be for NPDU
# Create an instance of E2EVerificationListener
>>> e2e_verif_listener = E2EVerificationListener(path)
>>> e2e_verif_listener.stop_verifying()

get_e2e_monitoring_status()

Get the E2E monitoring status

Examples:

>>> path = channel_name::frame_id::pdu_name # the path could be for LegacyPDU
>>> path = ecu_name::npdu_name # the path could be for NPDU
>>> path = ecu_name::npdu_id  # the path could be for NPDU
# Create an instance of E2EVerificationListener
>>> e2e_verif_listener = E2EVerificationListener(path)
>>> e2e_monitoring_status = e2e_verif_listener.get_e2e_monitoring_status()

reset(): cleanup the queue

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.

adapt_queue_elements() → Queue[BaseEvent]: ensure the object.timestamp, object.payload

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.

get_queue() → Queue: Get 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