Multicast module with multi-instance support and active trigger mechanism. More...

Public Member Functions
	__init__ (self, config)
	init preload some needed locals and then call onLoad() directly

	onLoad (self)
	Initialize module configuration and start the global cleanup thread.

	doWork (self, bwPacket)
	Process an incoming packet and handle multicast logic.

	onUnload (self)
	Unregister instance from the global cleanup process.

Data Fields
	instance_id

	name

Data Fields inherited from module.moduleBase.ModuleBase
	config

Protected Member Functions
	_get_packet_data (self, bwPacket)
	Safely extract data dictionary from packet.

	_filter_output (self, result)
	Apply multicastRole filtering before output.

	_combine_results (self, *results)
	Combine multiple result sources into a single list or status.

	_should_output_packet (self, multicast_role)
	Check if packet should be output based on role.

	_add_tone_ric_packet (self, freq, packet_dict)
	Add a tone-RIC to the shared buffer.

	_get_queued_packets (self)
	Pop and return all packets currently in the static queue.

	_copy_packet_dict_to_packet (self, recipient_dict, packet, index=1)
	Copy dict fields to Packet with timestamp shift for DB uniqueness.

	_distribute_complete (self, freq, text_packet_dict)
	Create full multicast packets with message content.

	_create_incomplete_multicast (self, freq, recipient_dicts)
	Generate multicast packets for timeouts (no text message).

	_enrich_normal_alarm (self, bwPacket, packet_dict)
	Enrich a standard single alarm with multicast metadata.

	_handle_delimiter (self, freq, ric, bwPacket=None)
	Handle delimiter packet and clear orphaned tone-RICs.

	_set_mcast_metadata (self, packet, mode, role, source="", count="1", index="1")
	Helper to set standard multicast fields and register wildcards.

	_apply_list_tags (self, packet, recipient_dicts)
	Helper to aggregate fields from all recipients into comma-separated lists.

	_register_wildcard_safe (self, wildcard, field)
	Register wildcard if not already globally registered.

	_perform_instance_tick (self)
	Tick-entry point for this specific instance.

	_check_all_my_frequencies (self)
	Monitor timeouts for all frequencies assigned to this instance.

	_check_instance_auto_clear (self, freq)
	Check if frequency has exceeded timeout (called from doWork).

	_send_wakeup_trigger (self, freq, fallback_ric)
	Send a loopback trigger via socket to wake up the system.

Protected Member Functions inherited from module.moduleBase.ModuleBase
	_cleanup (self)
	Cleanup routine calls onUnload() directly.

	_run (self, bwPacket)
	start an run of the module.

	_getStatistics (self)
	Returns statistical information's from last module run.

Static Protected Member Functions
	_global_cleanup_worker ()
	Static background thread that ticks all active module instances.

	_cleanup_hard_timeout_global ()
	Global failsafe for really old packets (ignores instance config).

Protected Attributes
	_my_frequencies

	_auto_clear_timeout

	_hard_timeout

	_delimiter_rics

	_text_rics

	_netident_rics

	_trigger_ric

	_trigger_ric_mode

	_trigger_host

	_trigger_port

	_block_delimiter

	_block_netident

Protected Attributes inherited from module.moduleBase.ModuleBase
	_moduleName

	_cumTime

	_moduleTime

	_runCount

	_moduleErrorCount

Static Protected Attributes
	_tone_ric_packets = defaultdict(list)

	_last_tone_ric_time = defaultdict(float)

	_processing_text_ric = defaultdict(bool)

	_processing_text_ric_started = defaultdict(float)

	_lock = threading.Lock()

	_cleanup_thread = None

bool	_running = False

	_wildcards_registered = set()

list	_packet_queue = []

	_queue_lock = threading.Lock()

list	_instances = []

str	_TRIGGER_HOST = "127.0.0.1"

int	_TRIGGER_PORT = 8080

str	_MAGIC_WAKEUP_MSG = "###_MULTICAST_WAKEUP_###"

str	_DEFAULT_TRIGGER_RIC = "9999999"

Static Protected Attributes inherited from module.moduleBase.ModuleBase
list	_modulesActive = []

Additional Inherited Members
Static Public Member Functions inherited from module.moduleBase.ModuleBase
	registerWildcard (newWildcard, bwPacketField)
	Register a new wildcard.

Detailed Description

Multicast module with multi-instance support and active trigger mechanism.

This module handles multicast alarm distribution. It manages the correlation between tone-RICs (recipients) and text-RICs (message content), ensuring reliable alarm delivery even in complex multi-frequency scenarios.

Constructor & Destructor Documentation

◆ init()

module.multicast.BoswatchModule.__init__	(	self,
		moduleName
	)

init preload some needed locals and then call onLoad() directly

Reimplemented from module.moduleBase.ModuleBase.

    def __init__(self, config):
        super().__init__(__name__, config)
 

Member Function Documentation

◆ onLoad()

module.multicast.BoswatchModule.onLoad ( self )

Initialize module configuration and start the global cleanup thread.

Parameters

None

Returns: None

Reimplemented from module.moduleBase.ModuleBase.

    def onLoad(self):
        r"""!Initialize module configuration and start the global cleanup thread.
 
        @param None
        @return None"""
        self._my_frequencies = set()
        self.instance_id = hex(id(self))[-4:]
        self.name = f"MCAST_{self.instance_id}"
 
        self._auto_clear_timeout = int(self.config.get("autoClearTimeout", default=10))
        self._hard_timeout = self._auto_clear_timeout * 3
 
        def parse_list(key):
            val = self.config.get(key)
            if val:
                return [x.strip() for x in str(val).split(",") if x.strip()]
            return []
 
        self._delimiter_rics = parse_list("delimiterRics")
        self._text_rics = parse_list("textRics")
        self._netident_rics = parse_list("netIdentRics")
 
        trigger_ric_cfg = self.config.get("triggerRic")
        if trigger_ric_cfg:
            self._trigger_ric = str(trigger_ric_cfg).strip()
            self._trigger_ric_mode = "explicit"
        else:
            self._trigger_ric = None
            self._trigger_ric_mode = "dynamic"
 
        self._trigger_host = self.config.get("triggerHost", default=self._TRIGGER_HOST)
        self._trigger_port = int(self.config.get("triggerPort", default=self._TRIGGER_PORT))
 
        self._block_delimiter = bool(self._delimiter_rics)
        self._block_netident = bool(self._netident_rics)
 
        logging.info("[%s] Multicast module loaded", self.name)
 
        with BoswatchModule._lock:
            if self not in BoswatchModule._instances:
                BoswatchModule._instances.append(self)
 
            if not BoswatchModule._running:
                BoswatchModule._running = True
                BoswatchModule._cleanup_thread = threading.Thread(
                    target=BoswatchModule._global_cleanup_worker, daemon=True
                )
                BoswatchModule._cleanup_thread.start()
                logging.info("Global multicast cleanup thread started")
 
# ============================================================
# MAIN PROCESSING
# ============================================================
 

◆ doWork()

module.multicast.BoswatchModule.doWork	(	self,
		bwPacket
	)

Process an incoming packet and handle multicast logic.

Parameters

bwPacket A BOSWatch packet instance or list of packets

Returns: bwPacket, a list of packets, or False if blocked

Reimplemented from module.moduleBase.ModuleBase.

    def doWork(self, bwPacket):
        r"""!Process an incoming packet and handle multicast logic.
 
        @param bwPacket: A BOSWatch packet instance or list of packets
        @return bwPacket, a list of packets, or False if blocked"""
        if isinstance(bwPacket, list):
            result_packets = []
            for single_packet in bwPacket:
                processed = self.doWork(single_packet)
                if processed is None:
                    result_packets.append(single_packet)
                elif processed is not False:
                    if isinstance(processed, list):
                        result_packets.extend(processed)
                    else:
                        result_packets.append(processed)
            return result_packets if result_packets else None
 
        packet_dict = self._get_packet_data(bwPacket)
        msg = packet_dict.get("message")
        ric = packet_dict.get("ric")
        freq = packet_dict.get("frequency", "default")
        mode = packet_dict.get("mode")
 
        if msg == BoswatchModule._MAGIC_WAKEUP_MSG:
            if self._trigger_ric and ric != self._trigger_ric:
                pass
            else:
                logging.debug("[%s] Wakeup trigger received (RIC=%s)", self.name, ric)
                queued = self._get_queued_packets()
                return queued if queued else False
 
        if mode != "pocsag":
            queued = self._get_queued_packets()
            return queued if queued else None
 
        self._my_frequencies.add(freq)
 
        is_text_ric = False
        if self._text_rics:
            is_text_ric = ric in self._text_rics and msg and msg.strip()
        else:
            with BoswatchModule._lock:
                is_text_ric = msg and msg.strip() and len(BoswatchModule._tone_ric_packets[freq]) > 0
 
        if is_text_ric:
            with BoswatchModule._lock:
                BoswatchModule._processing_text_ric[freq] = True
                BoswatchModule._processing_text_ric_started[freq] = time.time()
            logging.debug("[%s] Text-RIC %s detected", self.name, ric)
 
        queued_packets = self._get_queued_packets()
        incomplete_packets = None if is_text_ric else self._check_instance_auto_clear(freq)
 
        if self._netident_rics and ric in self._netident_rics:
            self._set_mcast_metadata(bwPacket, "control", "netident")
            result = self._combine_results(incomplete_packets, queued_packets, [bwPacket])
            return self._filter_output(result)
 
        if self._delimiter_rics and ric in self._delimiter_rics:
            delimiter_incomplete = self._handle_delimiter(freq, ric, bwPacket)
            result = self._combine_results(delimiter_incomplete, incomplete_packets, queued_packets)
            return self._filter_output(result)
 
        if not msg or not msg.strip():
            self._add_tone_ric_packet(freq, packet_dict)
            result = self._combine_results(incomplete_packets, queued_packets, False)
            return self._filter_output(result)
 
        if is_text_ric and msg:
            logging.info("[%s] Text-RIC received: RIC=%s", self.name, ric)
            alarm_packets = self._distribute_complete(freq, packet_dict)
            with BoswatchModule._lock:
                BoswatchModule._processing_text_ric[freq] = False
                BoswatchModule._processing_text_ric_started.pop(freq, None)
 
            if not alarm_packets:
                logging.warning("[%s] No tone-RICs for text-RIC=%s", self.name, ric)
                normal = self._enrich_normal_alarm(bwPacket, packet_dict)
                result = self._combine_results(normal, incomplete_packets, queued_packets)
            else:
                result = self._combine_results(alarm_packets, incomplete_packets, queued_packets)
            return self._filter_output(result)
 
        if msg:
            normal = self._enrich_normal_alarm(bwPacket, packet_dict)
            result = self._combine_results(normal, incomplete_packets, queued_packets)
            return self._filter_output(result)
 
        result = self._combine_results(incomplete_packets, queued_packets)
        return self._filter_output(result)
 
# ============================================================
# PACKET PROCESSING HELPERS (called by doWork)
# ============================================================
 

◆ _get_packet_data()

module.multicast.BoswatchModule._get_packet_data	(	self,
		bwPacket
	)

protected

Safely extract data dictionary from packet.

Parameters

bwPacket Packet instance or dict

Returns: dict: The packet data dictionary

    def _get_packet_data(self, bwPacket):
        r"""!Safely extract data dictionary from packet.
 
        @param bwPacket: Packet instance or dict
        @return dict: The packet data dictionary"""
        if isinstance(bwPacket, dict):
            return bwPacket
        try:
            if hasattr(bwPacket, 'data'):
                return bwPacket.data
            return {}
        except Exception:
            return {}
 

◆ _filter_output()

module.multicast.BoswatchModule._filter_output	(	self,
		result
	)

protected

Apply multicastRole filtering before output.

Parameters

result Single packet, list of packets, None or False

Returns: Final packet(s) or False if blocked

    def _filter_output(self, result):
        r"""!Apply multicastRole filtering before output.
 
        @param result: Single packet, list of packets, None or False
        @return Final packet(s) or False if blocked"""
        if result is None or result is False:
            return result
 
        if isinstance(result, list):
            filtered = [p for p in result if self._should_output_packet(p.get("multicastRole"))]
            if not filtered:
                logging.debug("All packets filtered out by multicastRole")
                return False
            return filtered if len(filtered) > 1 else filtered[0]
        else:
            if self._should_output_packet(result.get("multicastRole")):
                return result
            logging.debug("Packet filtered out: multicastRole=%s", result.get("multicastRole"))
            return False
 

◆ _combine_results()

module.multicast.BoswatchModule._combine_results	(		self,
		*	results
	)

protected

Combine multiple result sources into a single list or status.

Parameters

results Multiple packet objects, lists, or booleans

Returns: combined list, False or None

    def _combine_results(self, *results):
        r"""!Combine multiple result sources into a single list or status.
 
        @param results: Multiple packet objects, lists, or booleans
        @return combined list, False or None"""
        combined = []
        has_false = False
        for result in results:
            if result is False:
                has_false = True
                continue
            if result is None:
                continue
            if isinstance(result, list):
                combined.extend(result)
            else:
                combined.append(result)
        if combined:
            return combined
        return False if has_false else None
 

◆ _should_output_packet()

module.multicast.BoswatchModule._should_output_packet	(	self,
		multicast_role
	)

protected

Check if packet should be output based on role.

Parameters

multicast_role The role string to check

Returns: bool: True if allowed

    def _should_output_packet(self, multicast_role):
        r"""!Check if packet should be output based on role.
 
        @param multicast_role: The role string to check
        @return bool: True if allowed"""
        if self._block_delimiter and multicast_role == "delimiter":
            return False
        if self._block_netident and multicast_role == "netident":
            return False
        return True
 
# ============================================================
# TONE-RIC BUFFER MANAGEMENT
# ============================================================
 

◆ _add_tone_ric_packet()

module.multicast.BoswatchModule._add_tone_ric_packet	(	self,
		freq,
		packet_dict
	)

protected

Add a tone-RIC to the shared buffer.

Parameters

freq	Frequency identifier
packet_dict	Dictionary containing packet data

Returns: None

    def _add_tone_ric_packet(self, freq, packet_dict):
        r"""!Add a tone-RIC to the shared buffer.
 
        @param freq: Frequency identifier
        @param packet_dict: Dictionary containing packet data
        @return None"""
        with BoswatchModule._lock:
            stored_packet = packet_dict.copy()
            stored_packet['_multicast_timestamp'] = time.time()
            BoswatchModule._tone_ric_packets[freq].append(stored_packet)
            BoswatchModule._last_tone_ric_time[freq] = stored_packet['_multicast_timestamp']
            logging.info("[%s] Tone-RIC added: RIC=%s (total: %d on %s)", self.name, stored_packet.get('ric'), len(BoswatchModule._tone_ric_packets[freq]), freq)
 

◆ _get_queued_packets()

module.multicast.BoswatchModule._get_queued_packets ( self )

protected

Pop and return all packets currently in the static queue.

Parameters

None

Returns: list: List of packets or None

    def _get_queued_packets(self):
        r"""!Pop and return all packets currently in the static queue.
 
        @param None
        @return list: List of packets or None"""
        with BoswatchModule._queue_lock:
            if BoswatchModule._packet_queue:
                packets = BoswatchModule._packet_queue[:]
                BoswatchModule._packet_queue.clear()
                return packets
        return None
 
# ============================================================
# MULTICAST PACKET CREATION
# ============================================================
 

◆ _copy_packet_dict_to_packet()

module.multicast.BoswatchModule._copy_packet_dict_to_packet	(	self,
		recipient_dict,
		packet,
		index = `1`
	)

protected

Copy dict fields to Packet with timestamp shift for DB uniqueness.

Parameters

recipient_dict	Source dictionary
packet	Target Packet object
index	Packet index (1-based) - shifts timestamp by milliseconds

Returns: None

    def _copy_packet_dict_to_packet(self, recipient_dict, packet, index=1):
        r"""!Copy dict fields to Packet with timestamp shift for DB uniqueness.
 
        @param recipient_dict: Source dictionary
        @param packet: Target Packet object
        @param index: Packet index (1-based) - shifts timestamp by milliseconds
        @return None"""
        for k, v in recipient_dict.items():
            if k.startswith('_'):
                continue
            if k == 'timestamp' and index > 1:
                try:
                    dt = datetime.datetime.strptime(str(v), '%d.%m.%Y %H:%M:%S')
                    dt_shifted = dt + datetime.timedelta(milliseconds=index - 1)
                    packet.set(k, dt_shifted.strftime('%d.%m.%Y %H:%M:%S'))
                except:
                    packet.set(k, str(v))
            else:
                packet.set(k, str(v))
 

◆ _distribute_complete()

module.multicast.BoswatchModule._distribute_complete	(	self,
		freq,
		text_packet_dict
	)

protected

Create full multicast packets with message content.

Parameters

freq	Frequency identifier
text_packet_dict	Data of the message-carrying packet

Returns: list: List of fully populated Packet instances

    def _distribute_complete(self, freq, text_packet_dict):
        r"""!Create full multicast packets with message content.
 
        @param freq: Frequency identifier
        @param text_packet_dict: Data of the message-carrying packet
        @return list: List of fully populated Packet instances"""
        with BoswatchModule._lock:
            recipient_dicts = BoswatchModule._tone_ric_packets[freq].copy()
            BoswatchModule._tone_ric_packets[freq].clear()
            BoswatchModule._last_tone_ric_time.pop(freq, None)
 
        if not recipient_dicts:
            return []
        text_ric = text_packet_dict.get("ric")
        message_text = text_packet_dict.get("message")
        alarm_packets = []
 
        for idx, recipient_dict in enumerate(recipient_dicts, 1):
            p = Packet()
            self._copy_packet_dict_to_packet(recipient_dict, p, idx)
            p.set("message", message_text)
            self._apply_list_tags(p, recipient_dicts)
            self._set_mcast_metadata(p, "complete", "recipient", text_ric, len(recipient_dicts), idx)
            alarm_packets.append(p)
 
        logging.info("[%s] Generated %d complete multicast packets for text-RIC %s", self.name, len(alarm_packets), text_ric)
        return alarm_packets
 

◆ _create_incomplete_multicast()

module.multicast.BoswatchModule._create_incomplete_multicast	(	self,
		freq,
		recipient_dicts
	)

protected

Generate multicast packets for timeouts (no text message).

Parameters

freq	Frequency identifier
recipient_dicts	List of recipient data dictionaries

Returns: list: List of incomplete Packet instances

    def _create_incomplete_multicast(self, freq, recipient_dicts):
        r"""!Generate multicast packets for timeouts (no text message).
 
        @param freq: Frequency identifier
        @param recipient_dicts: List of recipient data dictionaries
        @return list: List of incomplete Packet instances"""
        if not recipient_dicts:
            return []
        first_ric = recipient_dicts[0].get("ric", "unknown")
        incomplete_packets = []
        for idx, recipient_dict in enumerate(recipient_dicts, 1):
            p = Packet()
            self._copy_packet_dict_to_packet(recipient_dict, p, idx)
            p.set("message", "")
            self._apply_list_tags(p, recipient_dicts)
            self._set_mcast_metadata(p, "incomplete", "recipient", first_ric, len(recipient_dicts), idx)
            incomplete_packets.append(p)
        return incomplete_packets
 

◆ _enrich_normal_alarm()

module.multicast.BoswatchModule._enrich_normal_alarm	(	self,
		bwPacket,
		packet_dict
	)

protected

Enrich a standard single alarm with multicast metadata.

Parameters

bwPacket	Target Packet object
packet_dict	Source data dictionary

Returns: list: List containing the enriched packet

    def _enrich_normal_alarm(self, bwPacket, packet_dict):
        r"""!Enrich a standard single alarm with multicast metadata.
 
        @param bwPacket: Target Packet object
        @param packet_dict: Source data dictionary
        @return list: List containing the enriched packet"""
        self._copy_packet_dict_to_packet(packet_dict, bwPacket, index=1)
        self._apply_list_tags(bwPacket, [packet_dict])
        self._set_mcast_metadata(bwPacket, "single", "single", packet_dict.get("ric", ""), "1", "1")
        return [bwPacket]
 

◆ _handle_delimiter()

module.multicast.BoswatchModule._handle_delimiter	(	self,
		freq,
		ric,
		bwPacket = `None`
	)

protected

Handle delimiter packet and clear orphaned tone-RICs.

Parameters

freq	Frequency identifier
ric	Delimiter RIC
bwPacket	Optional delimiter packet instance

Returns: list: Incomplete packets or delimiter control packet

    def _handle_delimiter(self, freq, ric, bwPacket=None):
        r"""!Handle delimiter packet and clear orphaned tone-RICs.
 
        @param freq: Frequency identifier
        @param ric: Delimiter RIC
        @param bwPacket: Optional delimiter packet instance
        @return list: Incomplete packets or delimiter control packet"""
        with BoswatchModule._lock:
            orphaned = BoswatchModule._tone_ric_packets[freq].copy()
            BoswatchModule._tone_ric_packets[freq].clear()
            BoswatchModule._last_tone_ric_time.pop(freq, None)
            BoswatchModule._processing_text_ric[freq] = False
 
        if orphaned:
            age_seconds = time.time() - orphaned[0].get('_multicast_timestamp', time.time())
            logging.info("[%s] Delimiter RIC=%s cleared %d orphaned tone-RICs on freq %s: RICs=[%s], age=%.1fs → Creating INCOMPLETE multicast packets for forwarding", self.name, ric, len(orphaned), freq, ', '.join([packet.get('ric', 'unknown') for packet in orphaned]), age_seconds)
            return self._create_incomplete_multicast(freq, orphaned)
        if bwPacket is not None:
            self._set_mcast_metadata(bwPacket, "control", "delimiter", ric, "0", "0")
            return [bwPacket]
        return None
 
# ============================================================
# PACKET METADATA HELPERS
# ============================================================
 

◆ _set_mcast_metadata()

module.multicast.BoswatchModule._set_mcast_metadata	(	self,
		packet,
		mode,
		role,
		source = `""`,
		count = `"1"`,
		index = `"1"`
	)

protected

Helper to set standard multicast fields and register wildcards.

Parameters

packet	The Packet instance to modify
mode	multicastMode (complete, incomplete, single, control)
role	multicastRole (recipient, single, delimiter, netident)
source	The originating RIC
count	Total number of recipients
index	Current recipient index

Returns: None

    def _set_mcast_metadata(self, packet, mode, role, source="", count="1", index="1"):
        r"""!Helper to set standard multicast fields and register wildcards.
 
        @param packet: The Packet instance to modify
        @param mode: multicastMode (complete, incomplete, single, control)
        @param role: multicastRole (recipient, single, delimiter, netident)
        @param source: The originating RIC
        @param count: Total number of recipients
        @param index: Current recipient index
        @return None"""
        mapping = {
            "multicastMode": (mode, "{MCAST_MODE}"),
            "multicastRole": (role, "{MCAST_ROLE}"),
            "multicastSourceRic": (source, "{MCAST_SOURCE}"),
            "multicastRecipientCount": (str(count), "{MCAST_COUNT}"),
            "multicastRecipientIndex": (str(index), "{MCAST_INDEX}")
        }
        for key, (val, wildcard) in mapping.items():
            packet.set(key, val)
            self._register_wildcard_safe(wildcard, key)
 

◆ _apply_list_tags()

module.multicast.BoswatchModule._apply_list_tags	(	self,
		packet,
		recipient_dicts
	)

protected

Helper to aggregate fields from all recipients into comma-separated lists.

Parameters

packet	The target Packet instance
recipient_dicts	List of dictionaries of all recipients in this group

Returns: None

    def _apply_list_tags(self, packet, recipient_dicts):
        r"""!Helper to aggregate fields from all recipients into comma-separated lists.
 
        @param packet: The target Packet instance
        @param recipient_dicts: List of dictionaries of all recipients in this group
        @return None"""
        all_fields = set()
        for r in recipient_dicts:
            all_fields.update(k for k in r.keys() if not k.startswith('_'))
 
        for f in sorted(all_fields):
            list_val = ", ".join([str(r.get(f, "")) for r in recipient_dicts])
            list_key = f"{f}_list"
            packet.set(list_key, list_val)
            self._register_wildcard_safe("{" + f.upper() + "_LIST}", list_key)
 

◆ _register_wildcard_safe()

module.multicast.BoswatchModule._register_wildcard_safe	(	self,
		wildcard,
		field
	)

protected

Register wildcard if not already globally registered.

Parameters

wildcard	The wildcard string (e.g. {MCAST_MODE})
field	The packet field name

Returns: None

    def _register_wildcard_safe(self, wildcard, field):
        r"""!Register wildcard if not already globally registered.
 
        @param wildcard: The wildcard string (e.g. {MCAST_MODE})
        @param field: The packet field name
        @return None"""
        if wildcard not in BoswatchModule._wildcards_registered:
            self.registerWildcard(wildcard, field)
            BoswatchModule._wildcards_registered.add(wildcard)
 
# ============================================================
# CLEANUP & TIMEOUT MANAGEMENT
# ============================================================
 

◆ _global_cleanup_worker()

module.multicast.BoswatchModule._global_cleanup_worker ( )

staticprotected

Static background thread that ticks all active module instances.

Parameters

None

Returns: None

    def _global_cleanup_worker():
        r"""!Static background thread that ticks all active module instances.
 
        @param None
        @return None"""
        logging.info("Global multicast cleanup ticker active")
        while BoswatchModule._running:
            time.sleep(1)
            with BoswatchModule._lock:
                active_instances = BoswatchModule._instances[:]
            for instance in active_instances:
                try:
                    instance._perform_instance_tick()
                except Exception as e:
                    logging.error("Error in instance cleanup: %s", e)
            if int(time.time()) % 60 == 0:
                BoswatchModule._cleanup_hard_timeout_global()
 

◆ _perform_instance_tick()

module.multicast.BoswatchModule._perform_instance_tick ( self )

protected

Tick-entry point for this specific instance.

Parameters

None

Returns: None

    def _perform_instance_tick(self):
        r"""!Tick-entry point for this specific instance.
 
        @param None
        @return None"""
        self._check_all_my_frequencies()
 

◆ _check_all_my_frequencies()

module.multicast.BoswatchModule._check_all_my_frequencies ( self )

protected

Monitor timeouts for all frequencies assigned to this instance.

Parameters

None

Returns: None

    def _check_all_my_frequencies(self):
        r"""!Monitor timeouts for all frequencies assigned to this instance.
 
        @param None
        @return None"""
        incomplete_packets = []
        trigger_data = []
 
        with BoswatchModule._lock:
            current_time = time.time()
            for freq in list(self._my_frequencies):
                if freq not in BoswatchModule._tone_ric_packets or not BoswatchModule._tone_ric_packets[freq]:
                    continue
 
                if BoswatchModule._processing_text_ric.get(freq, False):
                    flag_age = current_time - BoswatchModule._processing_text_ric_started.get(freq, current_time)
                    if flag_age > 2:
                        BoswatchModule._processing_text_ric[freq] = False
                        BoswatchModule._processing_text_ric_started.pop(freq, None)
                    else:
                        continue
 
                last_time = BoswatchModule._last_tone_ric_time.get(freq, 0)
                if current_time - last_time > self._auto_clear_timeout:
                    recipient_dicts = BoswatchModule._tone_ric_packets[freq].copy()
                    safe_ric = recipient_dicts[0].get('ric', self._DEFAULT_TRIGGER_RIC)
                    trigger_data.append((freq, safe_ric))
                    BoswatchModule._tone_ric_packets[freq].clear()
                    BoswatchModule._last_tone_ric_time.pop(freq, None)
 
                    logging.info("[%s] Auto-clear: %d tone-RICs on %s (Timeout %ds)", self.name, len(recipient_dicts), freq, self._auto_clear_timeout)
                    packets = self._create_incomplete_multicast(freq, recipient_dicts)
                    if packets:
                        incomplete_packets.extend(packets)
 
        if incomplete_packets:
            with BoswatchModule._queue_lock:
                BoswatchModule._packet_queue.extend(incomplete_packets)
            for freq, safe_ric in trigger_data:
                self._send_wakeup_trigger(freq, safe_ric)
 

◆ _check_instance_auto_clear()

module.multicast.BoswatchModule._check_instance_auto_clear	(	self,
		freq
	)

protected

Check if frequency has exceeded timeout (called from doWork).

Parameters

freq	Frequency identifier

Returns: list: Incomplete packets if timeout exceeded, else None

    def _check_instance_auto_clear(self, freq):
        r"""!Check if frequency has exceeded timeout (called from doWork).
 
        @param freq: Frequency identifier
        @return list: Incomplete packets if timeout exceeded, else None"""
        with BoswatchModule._lock:
            if freq not in BoswatchModule._tone_ric_packets or not BoswatchModule._tone_ric_packets[freq]:
                return None
            last_time = BoswatchModule._last_tone_ric_time.get(freq, 0)
            if time.time() - last_time > self._auto_clear_timeout:
                recipient_dicts = BoswatchModule._tone_ric_packets[freq].copy()
                BoswatchModule._tone_ric_packets[freq].clear()
                BoswatchModule._last_tone_ric_time.pop(freq, None)
                logging.warning("[%s] Auto-clear (doWork): %d packets", self.name, len(recipient_dicts))
                return self._create_incomplete_multicast(freq, recipient_dicts)
        return None
 

◆ _cleanup_hard_timeout_global()

module.multicast.BoswatchModule._cleanup_hard_timeout_global ( )

staticprotected

Global failsafe for really old packets (ignores instance config).

Parameters

None

Returns: None

    def _cleanup_hard_timeout_global():
        r"""!Global failsafe for really old packets (ignores instance config).
 
        @param None
        @return None"""
        with BoswatchModule._lock:
            current_time = time.time()
            max_hard_timeout = 120
            if BoswatchModule._instances:
                max_hard_timeout = max(inst._hard_timeout for inst in BoswatchModule._instances)
            for freq in list(BoswatchModule._tone_ric_packets.keys()):
                BoswatchModule._tone_ric_packets[freq] = [
                    p for p in BoswatchModule._tone_ric_packets[freq]
                    if current_time - p.get('_multicast_timestamp', 0) < max_hard_timeout
                ]
                if not BoswatchModule._tone_ric_packets[freq]:
                    del BoswatchModule._tone_ric_packets[freq]
 
# ============================================================
# TRIGGER SYSTEM
# ============================================================
 

◆ _send_wakeup_trigger()

module.multicast.BoswatchModule._send_wakeup_trigger	(	self,
		freq,
		fallback_ric
	)

protected

Send a loopback trigger via socket to wake up the system.

Parameters

freq	Frequency identifier
fallback_ric	RIC to use if no explicit trigger RIC is configured

Returns: None

    def _send_wakeup_trigger(self, freq, fallback_ric):
        r"""!Send a loopback trigger via socket to wake up the system.
 
        @param freq: Frequency identifier
        @param fallback_ric: RIC to use if no explicit trigger RIC is configured
        @return None"""
        try:
            trigger_ric = self._trigger_ric if self._trigger_ric else fallback_ric
            payload = {
                "timestamp": time.time(),
                "mode": "pocsag",
                "bitrate": "1200",
                "ric": trigger_ric,
                "subric": "1",
                "subricText": "a",
                "message": BoswatchModule._MAGIC_WAKEUP_MSG,
                "clientName": "MulticastTrigger",
                "inputSource": "loopback",
                "frequency": freq
            }
            json_str = json.dumps(payload)
            header = f"{len(json_str):<10}"
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
                sock.settimeout(1.0)
                sock.connect((self._trigger_host, self._trigger_port))
                sock.sendall(header.encode('utf-8'))
                sock.sendall(json_str.encode('utf-8'))
                logging.debug("[%s] Wakeup trigger sent for freq %s (RIC=%s)", self.name, freq, trigger_ric)
        except Exception as e:
            logging.error("[%s] Failed to send wakeup trigger: %s", self.name, e)
 
# ============================================================
# LIFECYCLE (End)
# ============================================================
 

◆ onUnload()

module.multicast.BoswatchModule.onUnload ( self )

Unregister instance from the global cleanup process.

Parameters

None

Returns: None

Reimplemented from module.moduleBase.ModuleBase.

    def onUnload(self):
        r"""!Unregister instance from the global cleanup process.
 
        @param None
        @return None"""
        with BoswatchModule._lock:
            if self in BoswatchModule._instances:
                BoswatchModule._instances.remove(self)
        logging.debug("[%s] Multicast instance unloaded", self.name)

Field Documentation

◆ _tone_ric_packets

module.multicast.BoswatchModule._tone_ric_packets = defaultdict(list)

staticprotected

◆ _last_tone_ric_time

module.multicast.BoswatchModule._last_tone_ric_time = defaultdict(float)

staticprotected

◆ _processing_text_ric

module.multicast.BoswatchModule._processing_text_ric = defaultdict(bool)

staticprotected

◆ _processing_text_ric_started

module.multicast.BoswatchModule._processing_text_ric_started = defaultdict(float)

staticprotected

◆ _lock

module.multicast.BoswatchModule._lock = threading.Lock()

staticprotected

◆ _cleanup_thread

module.multicast.BoswatchModule._cleanup_thread = None

staticprotected

◆ _running

bool module.multicast.BoswatchModule._running = False

staticprotected

◆ _wildcards_registered

module.multicast.BoswatchModule._wildcards_registered = set()

staticprotected

◆ _packet_queue

list module.multicast.BoswatchModule._packet_queue = []

staticprotected

◆ _queue_lock

module.multicast.BoswatchModule._queue_lock = threading.Lock()

staticprotected

◆ _instances

list module.multicast.BoswatchModule._instances = []

staticprotected

◆ _TRIGGER_HOST

str module.multicast.BoswatchModule._TRIGGER_HOST = "127.0.0.1"

staticprotected

◆ _TRIGGER_PORT

int module.multicast.BoswatchModule._TRIGGER_PORT = 8080

staticprotected

◆ _MAGIC_WAKEUP_MSG

str module.multicast.BoswatchModule._MAGIC_WAKEUP_MSG = "###_MULTICAST_WAKEUP_###"

staticprotected

◆ _DEFAULT_TRIGGER_RIC

str module.multicast.BoswatchModule._DEFAULT_TRIGGER_RIC = "9999999"

staticprotected

◆ _my_frequencies

module.multicast.BoswatchModule._my_frequencies

protected

◆ instance_id

module.multicast.BoswatchModule.instance_id

◆ name

module.multicast.BoswatchModule.name

◆ _auto_clear_timeout

module.multicast.BoswatchModule._auto_clear_timeout

protected

◆ _hard_timeout

module.multicast.BoswatchModule._hard_timeout

protected

◆ _delimiter_rics

module.multicast.BoswatchModule._delimiter_rics

protected

◆ _text_rics

module.multicast.BoswatchModule._text_rics

protected

◆ _netident_rics

module.multicast.BoswatchModule._netident_rics

protected

◆ _trigger_ric

module.multicast.BoswatchModule._trigger_ric

protected

◆ _trigger_ric_mode

module.multicast.BoswatchModule._trigger_ric_mode

protected

◆ _trigger_host

module.multicast.BoswatchModule._trigger_host

protected

◆ _trigger_port

module.multicast.BoswatchModule._trigger_port

protected

◆ _block_delimiter

module.multicast.BoswatchModule._block_delimiter

protected

◆ _block_netident

module.multicast.BoswatchModule._block_netident

protected

Public Member Functions

Data Fields

Protected Member Functions

Static Protected Member Functions

Protected Attributes

Static Protected Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ onLoad()

◆ doWork()

◆ _get_packet_data()

◆ _filter_output()

◆ _combine_results()

◆ _should_output_packet()

◆ _add_tone_ric_packet()

◆ _get_queued_packets()

◆ _copy_packet_dict_to_packet()

◆ _distribute_complete()

◆ _create_incomplete_multicast()

◆ _enrich_normal_alarm()

◆ _handle_delimiter()

◆ _set_mcast_metadata()

◆ _apply_list_tags()

◆ _register_wildcard_safe()

◆ _global_cleanup_worker()

◆ _perform_instance_tick()

◆ _check_all_my_frequencies()

◆ _check_instance_auto_clear()

◆ _cleanup_hard_timeout_global()

◆ _send_wakeup_trigger()

◆ onUnload()

Field Documentation

◆ _tone_ric_packets

◆ _last_tone_ric_time

◆ _processing_text_ric

◆ _processing_text_ric_started

◆ _lock

◆ _cleanup_thread

◆ _running

◆ _wildcards_registered

◆ _packet_queue

◆ _queue_lock

◆ _instances

◆ _TRIGGER_HOST

◆ _TRIGGER_PORT

◆ _MAGIC_WAKEUP_MSG

◆ _DEFAULT_TRIGGER_RIC

◆ _my_frequencies

◆ instance_id

◆ name

◆ _auto_clear_timeout

◆ _hard_timeout

◆ _delimiter_rics

◆ _text_rics

◆ _netident_rics

◆ _trigger_ric

◆ _trigger_ric_mode

◆ _trigger_host

◆ _trigger_port

◆ _block_delimiter

◆ _block_netident

◆ init()