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 all fields from packet as a dictionary.

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

	_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.

	_cleanup_worker (self)
	Per-instance background thread for timeout management.

	_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).

	_cleanup_hard_timeout (self)
	Failsafe for really old packets.

	_send_wakeup_trigger (self, freq, fallback_ric)
	Send a loopback trigger using the standard TCPClient class.

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.

Protected Attributes
	_my_frequencies

	_auto_clear_timeout

	_hard_timeout

	_delimiter_rics

	_text_rics

	_netident_rics

	_trigger_ric

	_trigger_host

	_trigger_port

	_tone_ric_packets

	_last_tone_ric_time

	_processing_text_ric

	_processing_text_ric_started

	_wildcards_registered

	_packet_queue

	_lock

	_queue_lock

	_running

	_cleanup_thread

	_MAGIC_WAKEUP_MSG

Protected Attributes inherited from module.moduleBase.ModuleBase
	_moduleName

	_cumTime

	_moduleTime

	_runCount

	_moduleErrorCount

Static Protected Attributes
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()
        else:
            self._trigger_ric = None
 
        self._trigger_host = self.config.get("triggerHost", default=self._TRIGGER_HOST)
        self._trigger_port = int(self.config.get("triggerPort", default=self._TRIGGER_PORT))
 
        # --- Per-instance state (replaces all former class-variables) ---
        # Key: frequency string (e.g. "85.125M")
        self._tone_ric_packets = defaultdict(list)  # buffered tone-RICs per frequency
        self._last_tone_ric_time = defaultdict(float)  # last arrival time per frequency
        self._processing_text_ric = defaultdict(bool)  # text-RIC currently being processed?
        self._processing_text_ric_started = defaultdict(float)  # when did processing start?
        self._wildcards_registered = set()  # avoid double-registering wildcards
        self._packet_queue = []  # deferred packets waiting for trigger
 
        # --- Locks (only needed within this instance, no cross-instance sharing) ---
        self._lock = threading.Lock()
        self._queue_lock = threading.Lock()
 
        # --- Per-instance cleanup thread ---
        self._running = True
        self._cleanup_thread = threading.Thread(target=self._cleanup_worker, daemon=True)
        self._cleanup_thread.start()
 
        logging.info("[%s] Multicast module loaded", self.name)
 
# ============================================================
# MAIN PROCESSING
# ============================================================
 

◆ doWork()

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

Process an incoming packet and handle multicast logic.

Enriches packets with multicast metadata (mode, role, source). Does NOT filter - all packets pass through, downstream modules handle filtering.

Parameters

bwPacket A BOSWatch packet instance or list of packets

Returns: bwPacket, a list of packets, or None if no processing

Reimplemented from module.moduleBase.ModuleBase.

    def doWork(self, bwPacket):
        r"""!Process an incoming packet and handle multicast logic.
 
        Enriches packets with multicast metadata (mode, role, source).
        Does NOT filter - all packets pass through, downstream modules handle filtering.
 
        @param bwPacket: A BOSWatch packet instance or list of packets
        @return bwPacket, a list of packets, or None if no processing"""
        if isinstance(bwPacket, list):
            result_packets = []
            for single_packet in bwPacket:
                processed = self.doWork(single_packet)
                if processed is not None and 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")
 
        # Handle wakeup triggers
        if msg == BoswatchModule._MAGIC_WAKEUP_MSG:
            if self._trigger_ric and ric != self._trigger_ric:
                return None
            logging.debug("[%s] Wakeup trigger received (RIC=%s)", self.name, ric)
            queued = self._get_queued_packets()
            return queued if queued else None
 
        # Only process POCSAG
        if mode != "pocsag":
            queued = self._get_queued_packets()
            return queued if queued else None
 
        self._my_frequencies.add(freq)
 
        # Determine if this is a text-RIC
        is_text_ric = False
        if self._text_rics:
            is_text_ric = ric in self._text_rics and msg and msg.strip()
        else:
            with self._lock:
                is_text_ric = msg and msg.strip() and len(self._tone_ric_packets[freq]) > 0
 
        if is_text_ric:
            with self._lock:
                self._processing_text_ric[freq] = True
                self._processing_text_ric_started[freq] = time.time()
 
        queued_packets = self._get_queued_packets()
        incomplete_packets = None if is_text_ric else self._check_instance_auto_clear(freq)
 
        # === CONTROL PACKETS (netident, delimiter) ===
        # Mark and pass through - no filtering!
 
        if self._netident_rics and ric in self._netident_rics:
            self._set_mcast_metadata(bwPacket, "control", "netident", ric)
            return self._combine_results(incomplete_packets, queued_packets, [bwPacket])
 
        if self._delimiter_rics and ric in self._delimiter_rics:
            delimiter_incomplete = self._handle_delimiter(freq, ric, bwPacket)
            return self._combine_results(delimiter_incomplete, incomplete_packets, queued_packets)
 
        # === TONE-RICs (no message) ===
        if not msg or not msg.strip():
            self._add_tone_ric_packet(freq, packet_dict)
            return self._combine_results(incomplete_packets, queued_packets, False)
 
        # === TEXT-RICs (with message) ===
        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 self._lock:
                self._processing_text_ric[freq] = False
                self._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)
                return self._combine_results(normal, incomplete_packets, queued_packets)
            else:
                return self._combine_results(alarm_packets, incomplete_packets, queued_packets)
 
        # === SINGLE ALARM (message but no text-RICs configured) ===
        if msg:
            normal = self._enrich_normal_alarm(bwPacket, packet_dict)
            return self._combine_results(normal, incomplete_packets, queued_packets)
 
        return self._combine_results(incomplete_packets, queued_packets)
 
# ============================================================
# PACKET PROCESSING HELPERS (called by doWork)
# ============================================================
 

◆ _get_packet_data()

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

protected

Safely extract all fields from packet as a dictionary.

Handles both dict objects and Packet instances. Dynamically extracts all fields including those added by other modules.

Parameters

bwPacket Packet instance or dict

Returns: dict: Complete dictionary of all packet fields

    def _get_packet_data(self, bwPacket):
        r"""!Safely extract all fields from packet as a dictionary.
 
        Handles both dict objects and Packet instances.
        Dynamically extracts all fields including those added by other modules.
 
        @param bwPacket: Packet instance or dict
        @return dict: Complete dictionary of all packet fields"""
        # 1. Fall: Es ist bereits ein Dictionary
        if isinstance(bwPacket, dict):
            return bwPacket.copy()
 
        # 2. Fall: Es ist ein Packet-Objekt (Daten liegen in _packet)
        if hasattr(bwPacket, '_packet'):
            return bwPacket._packet.copy()
 
        # 3. Fallback: Falls es ein anderes Objekt ist, versuche __dict__ ohne '_' Filter für 'packet'
        try:
            return {k: v for k, v in bwPacket.__dict__.items() if not k.startswith('_')}
        except Exception as e:
            logging.warning("[%s] Error: %s", self.name, e)
            return {}
 

◆ _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
 
# ============================================================
# 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 self._lock:
            stored_packet = packet_dict.copy()
            stored_packet['_multicast_timestamp'] = time.time()
            self._tone_ric_packets[freq].append(stored_packet)
            self._last_tone_ric_time[freq] = stored_packet['_multicast_timestamp']
            logging.debug("[%s] Tone-RIC added: RIC=%s (total: %d on %s)", self.name, stored_packet.get('ric'), len(self._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 self._queue_lock:
            if self._packet_queue:
                packets = self._packet_queue[:]
                self._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':
                try:
                    # use native float (UNIX-timestamp)
                    ts_float = float(v)
 
                    # shifting for database-unique-keys (only for index >=2)
                    if index > 1:
                        ts_float += 0.001 * (index - 1)
 
                    # back to paket - float (UNIX-timestamp)
                    packet.set(k, ts_float)
                except (ValueError, TypeError):
                    # failsafe, if no float
                    packet.set(k, 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 self._lock:
            recipient_dicts = self._tone_ric_packets[freq].copy()
            logging.debug("Text RIC found. Matching against %d stored RICs", len(recipient_dicts))
            self._tone_ric_packets[freq].clear()
            self._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")
        logging.debug("Creating single-alarm for RIC %s", packet_dict.get('ric'))
        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 self._lock:
            orphaned = self._tone_ric_packets[freq].copy()
            self._tone_ric_packets[freq].clear()
            self._last_tone_ric_time.pop(freq, None)
            self._processing_text_ric[freq] = False
 
        if orphaned:
            age_seconds = time.time() - orphaned[0].get('_multicast_timestamp', time.time())
 
            logging.debug("[%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"""
        logging.debug("setting Metadata - Mode: %s, Role: %s, Index: %s of %s for RIC: %s", mode, role, index, count, source)
        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 self._wildcards_registered:
            self.registerWildcard(wildcard, field)
            self._wildcards_registered.add(wildcard)
 
# ============================================================
# CLEANUP & TIMEOUT MANAGEMENT
# ============================================================
 

◆ _cleanup_worker()

module.multicast.BoswatchModule._cleanup_worker ( self )

protected

Per-instance background thread for timeout management.

    def _cleanup_worker(self):
        r"""!Per-instance background thread for timeout management."""
        logging.info("[%s] Cleanup thread started", self.name)
        while self._running:
            time.sleep(1)
            try:
                self._check_all_my_frequencies()
            except Exception as e:
                logging.error("[%s] Error in cleanup thread: %s", self.name, e)
            if int(time.time()) % 60 == 0:
                self._cleanup_hard_timeout()
 

◆ _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 self._lock:
            current_time = time.time()
            for freq in list(self._my_frequencies):
                if freq not in self._tone_ric_packets or not self._tone_ric_packets[freq]:
                    continue
 
                if self._processing_text_ric.get(freq, False):
                    flag_age = current_time - self._processing_text_ric_started.get(freq, current_time)
                    if flag_age > 2:
                        self._processing_text_ric[freq] = False
                        self._processing_text_ric_started.pop(freq, None)
                    else:
                        continue
 
                last_time = self._last_tone_ric_time.get(freq, 0)
                if current_time - last_time > self._auto_clear_timeout:
                    recipient_dicts = self._tone_ric_packets[freq].copy()
                    safe_ric = recipient_dicts[0].get('ric', self._DEFAULT_TRIGGER_RIC)
                    trigger_data.append((freq, safe_ric))
                    self._tone_ric_packets[freq].clear()
                    self._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 self._queue_lock:
                self._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 self._lock:
            if freq not in self._tone_ric_packets or not self._tone_ric_packets[freq]:
                return None
            last_time = self._last_tone_ric_time.get(freq, 0)
            if time.time() - last_time > self._auto_clear_timeout:
                recipient_dicts = self._tone_ric_packets[freq].copy()
                self._tone_ric_packets[freq].clear()
                self._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()

module.multicast.BoswatchModule._cleanup_hard_timeout ( self )

protected

Failsafe for really old packets.

    def _cleanup_hard_timeout(self):
        r"""!Failsafe for really old packets."""
        with self._lock:
            current_time = time.time()
            for freq in list(self._tone_ric_packets.keys()):
                self._tone_ric_packets[freq] = [
                    p for p in self._tone_ric_packets[freq]
                    if current_time - p.get('_multicast_timestamp', 0) < self._hard_timeout
                ]
                # cleaning empty frequencies
                if not self._tone_ric_packets[freq]:
                    del self._tone_ric_packets[freq]
 
# ============================================================
# TRIGGER SYSTEM
# ============================================================
 

◆ _send_wakeup_trigger()

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

protected

Send a loopback trigger using the standard TCPClient class.

    def _send_wakeup_trigger(self, freq, fallback_ric):
        r"""!Send a loopback trigger using the standard TCPClient class."""
        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": self._MAGIC_WAKEUP_MSG,
                "clientName": "MulticastTrigger",
                "inputSource": "loopback",
                "frequency": freq
            }
            json_str = json.dumps(payload)
 
            # using BOSWatch-Architecture
            client = TCPClient(timeout=2)
            if client.connect(self._trigger_host, self._trigger_port):
                # 1. Send
                client.transmit(json_str)
 
                # 2. Recieve (getting [ack] and prevents connection reset)
                client.receive(timeout=1)
 
                client.disconnect()
                logging.debug("[%s] Wakeup trigger sent and acknowledged (RIC=%s)", self.name, trigger_ric)
            else:
                logging.error("[%s] Could not connect to local server for wakeup", self.name)
 
        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"""
        self._running = False
        logging.debug("[%s] Multicast instance unloaded", self.name)

Field Documentation

◆ _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 [1/2]

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_host

module.multicast.BoswatchModule._trigger_host

protected

◆ _trigger_port

module.multicast.BoswatchModule._trigger_port

protected

◆ _tone_ric_packets

module.multicast.BoswatchModule._tone_ric_packets

protected

◆ _last_tone_ric_time

module.multicast.BoswatchModule._last_tone_ric_time

protected

◆ _processing_text_ric

module.multicast.BoswatchModule._processing_text_ric

protected

◆ _processing_text_ric_started

module.multicast.BoswatchModule._processing_text_ric_started

protected

◆ _wildcards_registered

module.multicast.BoswatchModule._wildcards_registered

protected

◆ _packet_queue

module.multicast.BoswatchModule._packet_queue

protected

◆ _lock

module.multicast.BoswatchModule._lock

protected

◆ _queue_lock

module.multicast.BoswatchModule._queue_lock

protected

◆ _running

module.multicast.BoswatchModule._running

protected

◆ _cleanup_thread

module.multicast.BoswatchModule._cleanup_thread

protected

◆ _MAGIC_WAKEUP_MSG [2/2]

module.multicast.BoswatchModule._MAGIC_WAKEUP_MSG

protected

Public Member Functions

Data Fields

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()

◆ _combine_results()

◆ _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()

◆ _cleanup_worker()

◆ _check_all_my_frequencies()

◆ _check_instance_auto_clear()

◆ _cleanup_hard_timeout()

◆ _send_wakeup_trigger()

◆ onUnload()

Field Documentation

◆ _TRIGGER_HOST

◆ _TRIGGER_PORT

◆ _MAGIC_WAKEUP_MSG [1/2]

◆ _DEFAULT_TRIGGER_RIC

◆ _my_frequencies

◆ instance_id

◆ name

◆ _auto_clear_timeout

◆ _hard_timeout

◆ _delimiter_rics

◆ _text_rics

◆ _netident_rics

◆ _trigger_ric

◆ _trigger_host

◆ _trigger_port

◆ _tone_ric_packets

◆ _last_tone_ric_time

◆ _processing_text_ric

◆ _processing_text_ric_started

◆ _wildcards_registered

◆ _packet_queue

◆ _lock

◆ _queue_lock

◆ _running

◆ _cleanup_thread

◆ _MAGIC_WAKEUP_MSG [2/2]

◆ init()