Npdu Manager Module

class mtf.network_port.npdu_manager.NpduManager

Manager for configuring and transmitting NPDUs

__init__()

Creates a new instance of NpduManager.

Returns:

NpduManager: The newly created instance.

property vlan_assign

Gets the vlan_assign flag.

set_up(ecu_name: str, adapter_name: str, ecu_local_ip: str, brcf_path: str = None, drcf_path: str = None, drcf_channels: [<class 'str'>] = None, input_local_port: int = 0, listener_channel_name: str = None)

Sets up the npdu manager to be configured with global properties.

Args:

ecu_name (str): The ECU name (from the BRCF file). adapter_name (str): The adapter name from the YAML configuration file. ecu_local_ip (str): The local IP address of the ECU. brcf_path (str, optional): The BRCF path where all its channels will be configured. Defaults to None. drcf_path (str, optional): The DRCF file in which channels specified by the drcf_channels paramter are configured.

Defaults to None.

drcf_channels ([str], optional): DRCF channels to be configured. Defaults to None. input_local_port (int, optional): The input local port to use. Defaults to Config.get_local_npdu_port(). listener_channel_name (str, optional): The channel name to listen over. This parameter is used in case there is a need

to listen over a specific channel. Defaults to None.

Returns:

bool: True on success, False otherwise.

Note:

Note

If a DRCF file is passed, the appropriate BRCF path must be passed accordingly, as well as ecu_name and drcf_channels.

add_input_pdu_tunnel(channel_name: str, multi_cast_ip_v4_addresses: list, multi_cast_port: int, remote_ip_v4_address: str, local_port: int = 0, with_vlan_assign: bool = True, force_add: bool = False, adapter_name: str = '')

Adds an input tunnel to the Npdu manager’s configuration.

Args:

channel_name (str): The Tunnel name from BRCF file. multi_cast_ip_v4_addresses (list[str]): The multicast IPv4 addresses to use. multi_cast_port (int): The multicast port to use. remote_ip_v4_address (str): The remote IPv4 address to use. local_port (int): The local port to assign. Defaults to Config.get_local_npdu_port(). with_vlan_assign (bool): Flag to allow adding the Vlan. Defaults to True. force_add (bool): Whether to enforce the addition of the tunnel. Defaults to False. adapter_name (str): The local adapter name. Defaults to “”.

add_output_pdu_tunnel(channel_name: str, multi_cast_ip_v4_addresses: list, multi_cast_port: int, transmission_timeout: int, local_ip: str, local_port: int = 0, force_add: bool = False, adapter_name: str = '')

Adds an output tunnel to the Npdu manager’s configuration.

Args:

channel_name (str): The Tunnel name from BRCF file. multi_cast_ip_v4_addresses (list[str]): The multicast IPv4 addresses to use. multi_cast_port (int): The multicast port to use. transmission timeout (int): The transmission cycle time in second. local_ip (str): The remote IPv4 address to use. local_port (int): The local port to assign. Defaults to Config.get_local_npdu_port() force_add (bool): Whether to enforce the addition of the tunnel. Defaults to False. adapter_name (str): The local adapter name. Defaults to “”.

start_npdu_listener(channel_name: str, silent_mode: bool = True, npdu_filter: mtf_eth_npdu_filter = None) → bool

Starts listening on a given tunnel.

Args:

channel_name (str): The tunnel to listen on. silent_mode (bool): Whether to trigger the callback function on reception. Defaults to True. npdu_filter (mtf_eth_npdu_filter): The filter to use. Defaults to None.

Returns:

bool: True on success; False otherwise.

Note:

Note

Before the tunnel can be used, it must be configured through the NPDU Manager setup or by adding a custom tunnel.

stop_npdu_listener(channel_name: str, npdu_filter: mtf_eth_npdu_filter = None) → bool

Stops listening on the given tunnel.

Args:

channel_name (str): The tunnel to stop listening on. npdu_filter (mtf_eth_npdu_filter): The filter to use. Defaults to None.

Returns:

bool: True on success, False otherwise.

get_received_pdus(channel_name: str, pdu_id: int, reset: bool = True, npdu_filter: mtf_eth_npdu_filter = None)

Gets the list of payloads of the received NPDUs with the given ID.

Args:

channel_name (str): The tunnel to retrieve the payloads from. pdu_id (int): The NPDU ID to look for. reset (bool): Reset the message queue after retrieval. Defaults to True.. npdu_filter (mtf_eth_npdu_filter): The NPDU filter to apply. Defaults to None.

Returns:

list[bytes]: The payloads of the matching NPDUs.

Note:

Warning

Before the tunnel can be used, it must be configured through the NPDU Manager setup or by adding a custom tunnel.

get_received_pdus_elemets(channel_name: str, pdu_id: int, reset=True, npdu_filter: mtf_eth_npdu_filter = None)

Gets the received NPDU elements with the given ID.

Args:

channel_name (str): The tunnel to retrieve the payloads from. pdu_id (int): The NPDU ID to look for. reset (bool): Reset the message queue after retrieval. Defaults to True. npdu_filter (mtf_eth_npdu_filter): The NPDU filter to apply. Defaults to None.

Returns:

list: The matching NPDU elements.

Note:

Warning

Before the tunnel can be used, it must be configured through the NPDU Manager setup or by adding a custom tunnel.

Note:

Note

Each NpduElement has the following attributes:

• pdu_id

• pdu_index

• payload

• direction

• tci (VlanTci object, optional)

get_received_pdus_timestamp(channel_name: str, pdu_id: int, reset=True, npdu_filter: mtf_eth_npdu_filter = None)

Gets the timestamps of the received NPDUs with the given ID.

Args:

channel_name (str): The tunnel to retrieve the payloads from. npdu_id (int): The NPDU ID to look for. reset (bool): Reset the message queue after retrieval. Defaults to True. npdu_filter (mtf_eth_npdu_filter): The NPDU filter to apply. Defaults to None.

Returns:

list[int]: The timestamps of the matching NPDUs.

Note:

Warning

Before the tunnel can be used, it must be configured through the NPDU Manager setup or by adding a custom tunnel.

clean_up() → bool

Cleans up the manager.

Returns:

bool: True on success, False otherwise.

Note:

Note

This method is already called in the Teardown of the Testcase.

This method does the following cleanups for each configured tunnel:

• Stop all listeners.

• Clean up the queue.

• Remove tunnel from configuration.

• Remove all listener instances.

transmit(channel_name, npdu, trig_mode=0)

Transmits a given NPDU on a specific tunnel.

Args:

channel_name (str): The tunnel to transmit over. npdu (NpduElement): The NPDU to be transmitted. trig_mode (int): The triggering mode (0 = buffered, 1 = direct). Defaults to 0.

Note:

Warning

Before the tunnel can be used, it must be configured through the NPDU Manager setup or by adding a custom tunnel.

Note

Each NpduElement has the following attributes:

• pdu_id

• pdu_index

• payload

• direction

• tci (VlanTci object, optional)

get_all_received_pdus(channel_name: str, reset=True, npdu_filter: mtf_eth_npdu_filter = None)

Gets a list of tuples (pdu_id, payload) of all the received pdus.

Args:

channel_name (str): The tunnel to retrieve the payloads from. reset (bool): Reset the message queue after retrieval. Defaults to True. npdu_filter (mtf_eth_npdu_filter): The NPDU filter to apply. Defaults to None.

Returns:

list[(npdu_element.pdu_id, npdu_element.payload)]: list of tuples containing the NPDU ID and its payload.

Note:

Warning

Before the tunnel can be used, it must be configured through the NPDU Manager setup or by adding a custom tunnel.

get_all_received_npdu_data(channel_name: str, reset=True, npdu_filter: mtf_eth_npdu_filter = None)

Gets all received NPDU elements on the given tunnel.

Args:

channel_name (str): The tunnel to retrieve the payloads from. reset (bool): Reset the message queue after retrieval. Defaults to True. npdu_filter (mtf_eth_npdu_filter): The NPDU filter to apply. Defaults to None.

Returns:

list: The received NPDU elements.

Note:

Warning

Before the tunnel can be used, it must be configured through the NPDU Manager setup or by adding a custom tunnel.

transmit_npdus(channel_name, frames: list)

Transmits NPDU elements on a given tunnel.

Args:

channel_name (str): The tunnel to transmit over. frames (list): The NPDUs to transmit.

Note:

Warning

• Before the tunnel can be used, it must be configured through the NPDU Manager setup or by adding a custom tunnel.

• Trigger mode is by default buffered in this case.

Note

Each NpduElement has the following attributes:

• pdu_id

• pdu_index

• payload

• direction

• tci (VlanTci object, optional)

set_npdu_listener_filter(channel_name: str, filter: dict)

Sets a filter on a specific tunnel.

Args:

channel_name (str): The name of tunnel. filters (dict): A dictionary with the names of the elements to be filtered as keys and their corresponding values.

Note:

Note

Before the tunnel can be used, it must be configured through the NPDU Manager setup or by adding a custom tunnel.

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.

clear_npdu_listener_filter(channel_name: str)

Clears the filter of a specific tunnel.

Args:

channel_name (str): The tunnel name.

Note:

Note

Before the tunnel can be used, it must be configured through the NPDU Manager setup or by adding a custom tunnel.

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.