Npdu Listener Module

class mtf.network_port.npdu_listener.NpduListener

A singleton class for managing NPDU listeners

static __new__(cls, name: str, npdu_filter=None) NpduListener
classmethod remove_all_instances()

Remove all instances of NpduListener.

classmethod get_instance(name: str, npdu_filter=None) NpduListener

Retrieves or creates a singleton instance of NpduListener.

Args:

name (str): The name of the NPDU listener. npdu_filter (optional): The NPDU filter to be used. If None, the default filter is used.

Returns:

NpduListener: A singleton instance of the NpduListener class.

__init__(channel_name: str, npdu_filter) None
start(silent_mode: bool = False) bool

Starts the listener with the option to enable silent mode.

Args:
silent_mode (bool, optional): A flag indicating whether the listener

should run in silent mode (default is False).

Returns:

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

set_listener_filter(filter: dict) None

Sets a filter for the listener.

Args:

filter (dict): A dictionary containing the filter configuration for the listener.

Returns:

None: This method does not return any value.

clear_listener_filter() None

Clears the filter of the listener.

Returns:

None: This method does not return any value.

stop() bool

Stops the listener.

Returns:
bool: True if the listener was successfully stopped, False if the listener

was not found or could not be stopped.

set_filters(filters) None

Sets filters for the listener.

Args: filters (dict[str, list[int]]): The filters to be applied to the listener.

Returns:

None: This method does not return any value.

get_message_queue(reset=True) list[NpduListenerElement]

Retrieves the message queue.

Args:
reset (bool, optional): A flag indicating whether the message queue should be reset

(default is True).

Returns:
list[NpduListenerElement]: The current message queue, which is a list of

NpduListenerElement objects.

create_npdu_listener_element(channel_id, timestamp, npduelement) NpduListenerElement

Creates a new NpduListenerElement from the provided parameters.

Args:

channel_id (str): The identifier of the channel. timestamp (float): The timestamp when the event occurs. npduelement (NpduElement): The NpduElement instance from which to extract data

Returns:

NpduListenerElement: The newly created NpduListenerElement containing the provided data.

cleanup_queue() None

Cleans up the message queue.

Returns:

None: This method does not return any value.