Bus Manager Module

exception mtf.network_port.bus_manager.BusNotFoundError

__init__(bus_name)

__new__(**kwargs)

args

with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class mtf.network_port.bus_manager.FrameListenerFactory

Factory class for creating frame listeners based on bus type.

frame_listeners_factory: Dict[EthProtocolType | int, Type[CanFrameListener] | Type[LinFrameListener] | Type[FlexrayFrameListener]]

Factory lookup table.

classmethod get_frame_listener(bus_type: BusType) → Type[CanFrameListener] | Type[LinFrameListener] | Type[FlexrayFrameListener] | None

Get a frame listener class based on bus type.

Args:

bus_type (BusType): Enum value representing the type of the bus.

Returns:

Type[FrameListener]: Class representing the frame listener for the given bus type.

class mtf.network_port.bus_manager.BusListenerFactory

Factory class for creating bus listeners based on bus type.

listeners_factory: Dict[EthProtocolType | int, Type[CanBusListener] | Type[IoListener] | Type[LinBusListener] | Type[FlexrayBusListener] | Type[EthBusListener]]

Factory lookup table

classmethod get_listener(bus_type: BusType) → Type[CanBusListener] | Type[IoListener] | Type[LinBusListener] | Type[FlexrayBusListener] | Type[EthBusListener] | None

Get a bus listener class based on bus type.

Args:

bus_type: BusType enum value representing the type of the bus.

Returns:

Type[BusListener]: Class representing the bus listener for the given bus type.

class mtf.network_port.bus_manager.EthListenerFactory

Factory class for creating Ethernet bus listeners based on Ethernet protocol type.

eth_listeners_factory: Dict[EthProtocolType | int, Type[ArpListener] | Type[DhcpListener] | Type[IgmpListener] | Type[EthNmController] | Type[PtpListener] | Type[SomeipListener] | Type[DoipListener] | Type[SomeipSDWatcher] | Type[EthBusListener]]

Factory lookup table.

classmethod get_eth_listener(protocol: EthProtocolType) → Type[ArpListener] | Type[DhcpListener] | Type[IgmpListener] | Type[EthNmController] | Type[PtpListener] | Type[SomeipListener] | Type[DoipListener] | Type[SomeipSDWatcher] | Type[EthBusListener] | None

Get an Ethernet bus listener class based on Ethernet protocol type.

Parameters:

protocol – Enum value representing the Ethernet protocol type.

class mtf.network_port.bus_manager.BusTransmitterFactory

Factory class for creating bus transmitters based on bus type.

transmitters_factory: Dict[BusType, Type[CanFdTransmitter] | Type[CanTransmitter] | Type[LinTransmitter] | Type[FlexrayTransmitter] | Type[EthernetTransmitter]]

Factory lookup table

classmethod get_transmitter(bus_type: BusType) → Type[CanFdTransmitter] | Type[CanTransmitter] | Type[LinTransmitter] | Type[FlexrayTransmitter] | Type[EthernetTransmitter] | None

Get a bus transmitter class based on bus type.

Parameters:

bus_type – BusType enum value representing the type of the bus.

class mtf.network_port.bus_manager.BusManager

Singleton class for managing bus listeners and transmitters.

__init__()

load_buses_properties()

Read bus types from the YAML file from mtf-core level.

get_bus_name_from_alias(alias: str) → str

Get the bus name associated with a given alias.

Parameters:

alias – The alias to search for.

Returns:

The bus name associated with the given alias if found, otherwise the original alias.

static check_bus_availability(keyword: str)

A decorator that checks if the specified bus is available in the cache before calling the target function.

Args:

keyword (str): The keyword argument name for the bus name.

Returns:

Callable[…, Any]: The actual decorator function that takes a target function as an argument.

Example:

@check_bus_availability(1, ‘bus_name’)

def bus_listener(self, bus_name: str, bus_type: BusType = None):

…

check_if_bus_is_available(bus_name: str) → bool

Check if the bus corresponding to the given bus name is available.

Args:

bus_name (str): The name of the bus.

Returns:

bool: True if the bus is available, False otherwise.

property burst_manager

static create_burst_can_frame(frame_id: int, payload: List[int], fd: bool, data_flags: int)

static create_burst_flexray_frame(frame_id: int, payload: List[int], comm_channel: FRCommunicationChannel, error_flags: int)

static create_burst_ethernet_frame(frame_id: int, payload: List[int])

check_if_bus_is_available_from_cache(bus_name: str) → bool

Check in the python cache if the bus corresponding to the given bus name is available.

Args:

bus_name (str): The name of the bus.

Returns:

bool: True if the bus is available, False otherwise.

get_ethernet_bus_interface_name(bus_name: str) → str | None

Get the Ethernet bus interface name associated with the given bus name.

Args:

bus_name (str): The name of the bus.

Returns:

str or None: The Ethernet bus interface name associated with the given bus name, or None if the bus name is not found in the configuration.

get_all_available_busses() → List[str]

Get the list of all available buses.

Returns:

List[str]: The list of bus names of all available buses.

get_all_available_busses_with_properties()

Get the list of all available buses with properties. Properties are bus_name, bus_type, alias

Returns:

Dict[str, Dict[str, List[Any[int,str,float]]]]: The list of bus names of all available buses, each buses have a dictionary containing its properties

get_all_available_channels_with_properties()

Get the list of all available channels with properties. Properties are bus_name, bus_type, alias

Returns:

Dict[str, Dict[str, List[Any[int,str,float]]]]: The list of bus names of all available buses, each buses have a dictionary containing its properties

get_tracer_properties(tracer_name: str)

Get the BaseFileName ad Extension from a given tracer_name.

Args:

tracer_name (str): The name of the tracer.

Returns:

dict: Tracer properties as a dictionary.

replay_pcap_file(pcap_file_path: str, channel_name: str)

Replay pcap file on the given channel

Args:

pcap_file_path (str): full path of pcap_file. channel_name (str) : channel within it pcap replay will be applied

Returns: True if the PCAP replay was successful; False otherwise.

bus_listener(bus_name: str, bus_type: BusType | None = None) → BusListener

Get or create a bus listener instance for the specified bus name.

This method supports all legacy busses (CAN, LIN, FlexRay) and Ethernet busses using only ETHERNET_II protocol. If you want to use other Ethernet protocols, please use the ‘ethernet_listener’ method.

Args:

bus_name (str): Name of the bus.

bus_type (BusType, optional): Type of the bus. If not provided, it will be determined automatically

based on the bus name. Default is None.

Returns:

BusListener: Bus listener instance for the specified bus name.

ethernet_listener(bus_name: str, protocol: EthProtocolType) → EthBusListener

Get or create an Ethernet bus listener instance for the specified Ethernet protocol.

Args:

bus_name (str): Name of the Ethernet bus. protocol (EthProtocolType): Enum value representing the Ethernet protocol type.

Returns:

EthBusListener: An instance of the Ethernet bus listener for the specified bus name and protocol.

create_ethernet_message(protocol: EthProtocolType)

bus_transmitter(bus_type: BusType) → BusTransmitter

Get or create a bus transmitter instance for the specified bus type.

Args:

bus_type (BusType): The type of the bus.

Returns:

BusTransmitter: The bus transmitter instance for the specified bus type.

frame_listener(bus_name: str, frame_id: str, bus_type: BusType | None = None)

Get or create a frame listener instance for the specified bus name and frame ID.

This method supports different bus types, and the frame ID should be provided as a string.

Args:

bus_name (str): Name of the bus. frame_id (str): ID of the frame. The format depends on the bus type.

Returns:

FrameListener: Frame listener instance for the specified bus name and frame ID.

get_custom_iso_tp(config: IsoTpConfig) → CustomFrTp | CustomCanTp

create_iso_tp_config(channel_name: str, rx_id: str, tx_id: str, protocol: str, pdu_length: int, source_adr: int = 0, target_adr: int = 0, padding_value: int = 255, flow_timeout: int = 1, is_padding_activated: bool = True)

cleanup_all_listeners() → None

Stop all bus listeners and perform cleanup.

This method stops all bus listeners and performs cleanup operations, such as reseting queues, associated with the listeners.