diff --git a/boswatch/router/route.py b/boswatch/router/route.py
index 65eaa89b..a743cbf9 100644
--- a/boswatch/router/route.py
+++ b/boswatch/router/route.py
@@ -10,11 +10,10 @@
by Bastian Schroll
@file: route.py
-@date: 04.03.2019
+@date: 30.08.2026
@author: Bastian Schroll
@description: Class for a single BOSWatch packet router route point
"""
-
import logging
logging.debug("- %s loaded", __name__)
@@ -22,15 +21,22 @@
class Route:
r"""!Class for single routing points"""
- def __init__(self, name, callback, statsCallback=None, cleanupCallback=None):
+ def __init__(self, name, callback, statsCallback=None, cleanupCallback=None, isRouter=False):
r"""!Create a instance of an route point
@param name: name of the route point
@param callback: instance of the callback function
@param statsCallback: instance of the callback to get statistics (None)
@param cleanupCallback: instance of the callback to run a cleanup method (None)
+ @param isRouter: True if this route point delegates to another Router instance
+ (type: router in the config). Used by _process_route_recursive()
+ to distinguish a sub-router that filtered all packets internally
+ (parent should continue to the next route point) from a
+ module/plugin that explicitly stopped the route
+ (parent should stop immediately). (False)
"""
self.name = name
self.callback = callback
self.statistics = statsCallback
self.cleanup = cleanupCallback
+ self.isRouter = isRouter
diff --git a/boswatch/router/router.py b/boswatch/router/router.py
index 1374e961..d45901da 100644
--- a/boswatch/router/router.py
+++ b/boswatch/router/router.py
@@ -10,8 +10,8 @@
by Bastian Schroll
@file: router.py
-@date: 01.03.2019
-@author: Bastian Schroll
+@date: 14.10.2026
+@author: Bastian Schroll, Claus Schichl
@description: Class for the BOSWatch packet router
"""
import logging
@@ -51,32 +51,89 @@ def runRouter(self, bwPacket):
r"""!Run the router
@param bwPacket: instance of Packet class
- @return a instance of Packet class
+ @return an instance of Packet class, a list of packets, or False
"""
self._runCount += 1
tmpTime = time.time()
logging.debug("[%s] started", self.name)
- for routeObject in self.routeList:
+ # Start processing the route list recursively from index 0
+ final_result = self._process_route_recursive(bwPacket, 0)
+
+ logging.debug("[%s] finished", self.name)
+
+ self._routerTime = time.time() - tmpTime
+ self._cumTime += self._routerTime
+
+ return final_result
+
+ def _process_route_recursive(self, bwPacket, start_index):
+ r"""!Recursively process the route to support list branching (e.g. for multicast)
+
+ @param bwPacket: current packet to process
+ @param start_index: index of the routeList to start from
+ @return processed bwPacket, list of packets, or False
+ """
+ # Catch incoming lists (e.g. from cascaded routers passing a list via runRouter)
+ if isinstance(bwPacket, list):
+ logging.debug("[%s] Received a list of %d packets. Branching immediately.", self.name, len(bwPacket))
+ return self._process_packet_list(bwPacket, start_index)
+
+ # Normal path for a single packet starts here
+ current_packet = bwPacket
+
+ for i in range(start_index, len(self.routeList)):
+ routeObject = self.routeList[i]
logging.debug("[%s] -> run route: %s", self.name, routeObject.name)
- bwPacket_tmp = routeObject.callback(copy.deepcopy(bwPacket)) # copy bwPacket to prevent edit the original
- if bwPacket_tmp is None: # returning None doesnt change the bwPacket
+ # State Isolation: pass a deep copy to prevent plugins from corrupting the RAM state of subsequent list items
+ bwPacket_tmp = routeObject.callback(copy.deepcopy(current_packet))
+
+ if bwPacket_tmp is None:
+ # returning None doesnt change the current_packet
continue
- if bwPacket_tmp is False: # returning False stops the router immediately
- logging.debug("[%s] stopped", self.name)
- break
+ if bwPacket_tmp is False:
+ if routeObject.isRouter:
+ # A nested router filtered/stopped internally for this packet.
+ # That's just an empty branch - the parent router continues normally.
+ logging.debug("[%s] sub-router '%s' returned False - continuing with next route point", self.name, routeObject.name)
+ continue
- bwPacket = bwPacket_tmp
- logging.debug("[%s] bwPacket returned", self.name)
- logging.debug("[%s] finished", self.name)
+ # returning False stops the route immediately for this specific packet branch
+ logging.debug("[%s] stopped at route %s", self.name, routeObject.name)
+ return False
- self._routerTime = time.time() - tmpTime
- self._cumTime += self._routerTime
+ if isinstance(bwPacket_tmp, list):
+ # Branching logic: A module returned a list of packets
+ logging.debug("[%s] route %s returned a list. Branching for %d packets.", self.name, routeObject.name, len(bwPacket_tmp))
+ # every packet runs through the rest of the route on its own - this finishes the route for all branches
+ return self._process_packet_list(bwPacket_tmp, i + 1)
- return bwPacket
+ # Normal single packet path: update the packet for the next iteration
+ current_packet = bwPacket_tmp
+
+ return current_packet
+
+ def _process_packet_list(self, packets, start_index):
+ r"""!Process every packet of a list on its own from the given index
+
+ @param packets: list of packets
+ @param start_index: index of the routeList to start from
+ @return list of the remaining packets, or False if no packet is left
+ """
+ results = []
+ for single_packet in packets:
+ res = self._process_route_recursive(single_packet, start_index)
+ if res is None or res is False:
+ # branch was stopped - nothing to collect
+ continue
+ if isinstance(res, list):
+ results.extend(res)
+ else:
+ results.append(res)
+ return results if results else False
def _getStatistics(self):
r"""!Returns statistical information's from last router run
diff --git a/boswatch/router/routerManager.py b/boswatch/router/routerManager.py
index c78f112d..aa7f02bd 100644
--- a/boswatch/router/routerManager.py
+++ b/boswatch/router/routerManager.py
@@ -10,7 +10,7 @@
by Bastian Schroll
@file: routerManager.py
-@date: 04.03.2019
+@date: 30.08.2026
@author: Bastian Schroll
@description: Class for the BOSWatch packet router manager class
"""
@@ -89,7 +89,7 @@ def buildRouters(self, config):
loadedClass._cleanup))
elif routeType == "router":
- routerDict_tmp[routerName].addRoute(Route(routeName, routerDict_tmp[routeRes].runRouter))
+ routerDict_tmp[routerName].addRoute(Route(routeName, routerDict_tmp[routeRes].runRouter, isRouter=True))
else:
logging.error("unknown type '%s' in %s", routeType, route)
diff --git a/docu/docs/config-editor.html b/docu/docs/config-editor.html
index aa6db3c5..b50a0c3e 100644
--- a/docu/docs/config-editor.html
+++ b/docu/docs/config-editor.html
@@ -569,6 +569,21 @@ Für Entwickler: neue Module/Plugins hinzufügen
]
},
+ "multicast": {
+ kind: "module",
+ title: "Multicast",
+ creator: "BW3 Dev Team",
+ fields: [
+ { key: "autoClearTimeout", type: "number", label: "autoClearTimeout", default: 10, required: false },
+ { key: "delimiterRics", type: "text", label: "delimiterRics", required: false },
+ { key: "textRics", type: "text", label: "textRics", required: false },
+ { key: "netIdentRics", type: "text", label: "netIdentRics", required: false },
+ { key: "triggerRic", type: "text", label: "triggerRic", required: false },
+ { key: "triggerHost", type: "text", label: "triggerHost", default: "127.0.0.1", required: false },
+ { key: "triggerPort", type: "number", label: "triggerPort", default: 8080, required: false },
+ ]
+ },
+
// Plugins
"http": {
kind: "plugin",
diff --git a/docu/docs/develop/ModulPlugin.md b/docu/docs/develop/ModulPlugin.md
index 69511720..d29edcdf 100644
--- a/docu/docs/develop/ModulPlugin.md
+++ b/docu/docs/develop/ModulPlugin.md
@@ -89,7 +89,8 @@ Mögliche Rückgabewerte eines Moduls:
- `return bwPacket` Gibt das modifizierte bwPacket an den Router zurück (Paket Modifikation)
- `return None` Der Router fährt mit dem unveränderten bwPacket fort (Input = Output)
-- `return False` Der Router stopt sofort die Ausführung (zB. in Filtern verwendet)
+- `return False` Der Router stoppt sofort die Ausführung für dieses Paket (z. B. in Filtern verwendet).
+- `return [bwPacket1, bwPacket2, ...]` (Liste von Paketen) Der Router startet eine automatische Verzweigung ("Fan-out" bzw. "Auffächerung"). Für jedes Paket in der Liste wird der nachfolgende Rest der Route unabhängig und isoliert durchlaufen.
### Rückgabewert bei Plugins
Plugins geben keine Pakete mehr zurück. Sie fungieren ausschließlich als Endpunkt.
diff --git a/docu/docs/information/router.md b/docu/docs/information/router.md
index b360d9ea..ae2b3521 100644
--- a/docu/docs/information/router.md
+++ b/docu/docs/information/router.md
@@ -29,6 +29,15 @@ Nachfolgender Ablauf soll am Beispiel eines Alarms mit einem Pocsag Paket erklä
Jetzt sind alle Routenpunkte abgearbeitet und die Alarmierung damit abgeschlossen.
+---
+## Verzweigung / Listen-Verarbeitung (Fan-out)
+
+Der Routing-Mechanismus kann auch mit der Aufspaltung von Alarmen umgehen. Wenn ein Modul statt eines einzelnen Alarms eine **Liste von Paketen** zurückgibt (wie z. B. das `multicast`-Modul bei Sammelalarmen), teilt der Router den Pfad automatisch auf:
+
+- Jedes Paket aus der Liste läuft ab diesem Punkt ganz unabhängig den **verbleibenden Rest der Route** ab.
+- Dabei wird im Hintergrund sichergestellt, dass sich die einzelnen Alarme nicht gegenseitig beeinflussen oder Daten überschreiben.
+- Diese Aufteilung funktioniert auch problemlos über mehrere, ineinander verschachtelte Router hinweg.
+
---
## Konfiguration
diff --git a/docu/docs/modul/multicast.md b/docu/docs/modul/multicast.md
new file mode 100644
index 00000000..fc169712
--- /dev/null
+++ b/docu/docs/modul/multicast.md
@@ -0,0 +1,473 @@
+# Multicast
+---
+
+## Beschreibung
+Das Multicast-Modul verarbeitet komplexe Alarmsequenzen, bei denen eine Nachricht (Text-RIC) an eine Liste zuvor gesendeter Empfänger (Tone-RICs) verteilt wird. Es sorgt dafür, dass jeder Empfänger ein individuelles Paket mit dem Alarmtext erhält.
+
+Das Modul filtert keine inhaltlich relevanten Pakete. Alle Pakete mit Alarminhalt werden mit `multicastRole` markiert und weitergereicht. Die Filterung nach Bedarf erfolgt nachgelagert, z.B. mit `filter.regexFilter`.
+
+Das Modul unterstützt:
+
+- **Multi-Instance Support:** Vollständige Isolation bei parallelem Betrieb in verschiedenen Routen.
+- **Frequenz-Trennung:** Verhindert die Vermischung von Alarmen auf unterschiedlichen Kanälen.
+- **Active Trigger System:** Nutzt TCP-Loopback, um auch bei Inaktivität des Funkkanals Timeouts sicher zu verarbeiten.
+- **Dynamische Listen:** Generiert aggregierte Listenfelder (z. B. {RIC_LIST}) für Sammel-Alarmierungen.
+- **Metadaten-Enrichment:** Markiert Pakete präzise für nachgelagerte Filter (z. B. RegEx).
+
+### Funktionsweise
+Multicast-Alarme funktionieren in zwei bis vier Phasen:
+**Wichtig:** Das Modul arbeitet verzögert bei der Ausgabe der Text-RICs, um die Pakete anzureichern.
+
+1. **Delimiter-Phase (Optional)**: Ein spezieller Delimiter-RIC markiert den Start eines neuen Multicast-Blocks. Er wird als technisches Paket (`multicastRole: delimiter`) **immer sofort** durchgereicht und leert intern den RAM-Puffer für neue Empfänger. Liegen dort noch Tone-RICs eines unvollständigen Blocks, werden zuerst diese als `incomplete` ausgegeben, danach der Delimiter. Diese Phase ist optional - ohne Delimiter werden alle leeren Nachrichten als Tone-RICs behandelt. Downstream-Filter (z.B. filter.regexFilter) können ihn bei Bedarf ausfiltern.
+
+2. **Tone-RIC-Phase**: Eingehende leere Nachrichten werden **nicht direkt als Pakete ausgegeben**, sondern im RAM zwischengespeichert. Das Modul gibt hier `False` zurück, wodurch der Router die Verarbeitung für dieses spezifische Paket vorerst pausiert.
+
+3. **Text-RIC**: Ein spezieller Message-RIC empfängt die eigentliche Alarmnachricht. Sobald eine Text-RIC empfangen wird, "kopiert" das Modul diesen Text in jedes einzelne der gespeicherten Tone-RIC-Pakete. Diese werden dann als **Liste von Paketen** gesammelt an den Router als `multicastMode: complete` übergeben. Falls keine Tone-RICs im Puffer liegen (z.B. Einzelalarm), wird die Text-RIC als `multicastMode: single` ausgegeben.
+**Wichtig:** Die Text-RIC (Message-RIC) wird **nicht als separates Paket ausgegeben**. Sie dient nur als Nachrichtenträger, der seinen Text an alle gesammelten Tone-RICs verteilt. Ausnahme: Einzelalarm (Single)
+
+4. **Timeout-Phase (Auto-Clear - Optional):** Läuft der `autoClearTimeout` ab, ohne dass ein Text-RIC eintrifft, werden die gepufferten RICs als `multicastMode: incomplete` (ohne Text) ausgegeben.
+
+## Unterstützte Alarmtypen
+- POCSAG
+
+Pakete anderer Modi (z.B. FMS, ZVEI) werden unverändert durchgereicht.
+
+## Resource
+`multicast`
+
+## Konfiguration
+
+|Feld|Beschreibung|Default|
+|----|------------|-------|
+|autoClearTimeout|Zeit in Sekunden, nach der Tone-RICs ohne Text-Eingang als `incomplete` ausgegeben werden|10|
+|delimiterRics|Komma-getrennte Liste von Startmarkern (leert Puffer, `multicastRole: delimiter`)|leer|
+|textRics|Komma-getrennte Liste von RICs, die den Alarmtext tragen|leer|
+|netIdentRics|Komma-getrennte Liste von Netzwerk-Identifikations-RICs (`multicastRole: netident`)|leer|
+|triggerRic|RIC für das Wakeup-Trigger-Paket (optional, bei leer: dynamisch = erste Tone-RIC)|leer|
+|triggerHost|IP-Adresse für Loopback-Trigger|127.0.0.1|
+|triggerPort|Port für Loopback-Trigger (entspricht meist Server-Port)|8080|
+
+**Hinweis:** Zahlen mit führenden Nullen müssen in Anführungszeichen gesetzt werden, z.B. `'0012345'`.
+
+### Konfigurationsbeispiel 1: Automatische Delimiter-Erkennung (oder nicht verfügbar im Netzwerk) (= Minimalkonfiguration)
+```yaml
+- type: module
+ res: multicast
+ name: Multicast Handler
+ config:
+ textRics: '0299001,0310001'
+```
+In diesem Modus werden **alle leeren Nachrichten** als toneRics behandelt (keine `delimiterRics` angegeben).
+
+### Konfigurationsbeispiel 2: Mit Delimiter-Trenner (empfohlen)
+```yaml
+- type: module
+ res: multicast
+ name: Multicast Handler
+ config:
+ autoClearTimeout: 10
+ delimiterRics: '0988988'
+ textRics: '0299001,0310001'
+```
+In diesem Modus wird **0988988 als Trenner (= Delimiter)** behandelt und **alle anderen leeren Nachrichten als Empfänger**.
+
+### Konfigurationsbeispiel 3: Mit expliziter Trigger-RIC
+```yaml
+- type: module
+ res: multicast
+ name: Multicast Handler
+ config:
+ autoClearTimeout: 10
+ delimiterRics: '0988988'
+ textRics: '0299001,0310001'
+ triggerRic: '9999999'
+ triggerHost: '127.0.0.1'
+ triggerPort: 8080
+```
+Verwendet eine feste RIC (9999999) für das interne Wakeup-Trigger-Paket.
+
+### Konfigurationsbeispiel 4: Mit Netzident-Filterung
+```yaml
+- type: module
+ res: multicast
+ name: Multicast Handler
+ config:
+ autoClearTimeout: 10
+ delimiterRics: '0988988'
+ textRics: '0299001,0310001'
+ netIdentRics: '0000001'
+```
+Markiert Netzident-Pakete (RIC 0000001) mit multicastRole: netident. Downstream-Filter können sie gezielt ausfiltern (z.B. RegEx-Filter).
+
+---
+## Modul Abhängigkeiten
+- keine
+
+---
+## Externe Abhängigkeiten
+- keine
+
+---
+## Paket Modifikationen
+
+### Hinzugefügte Felder
+- `multicastMode`(string): Beschreibt das Ergebnis der Multicast-Verarbeitung, besitzt einen der Werte:
+ - `complete`: Vollständiges Multicast-Packet
+ - `incomplete`: Unvollständiges Multicast-Packet (meist fehlt die Text-RIC) (Timeout)
+ - `single`: Einzelner, "normaler" Alarm (Tone-RIC = Text-RIC)
+ - `control`: Technisches Paket (Delimiter, Netzident oder Trigger)
+- `multicastRole`:
+ - `delimiter`
+ - `netident`
+ - `recipient` (Empfänger)
+ - `single`
+ - `trigger` (Wakeup-Trigger, siehe [Active Trigger System](#das-active-trigger-system-verlustfreie-paketauslieferung))
+- `multicastRecipientIndex` (string): Der Index zählt nur Alarm-Empfänger:
+ - Bei **recipient**: Zählt hoch (z.B. 1 von 5, 2 von 5...)
+ - Bei **single**: Immer **1**
+ - Bei **delimiter / netident / trigger**: Immer **0**, da sie keine Alarm-Empfänger sind
+- `multicastRecipientCount` (string): Gesamtanzahl der Empfänger des Multicasts (**0** bei technischen Paketen)
+- `_list` (string): Liste von Werten aus allen Empfänger-RICs für jedes Originalfeld (z.B. `ric_list`, `subric_list`)
+
+### Ergänzte Felder (von Text-RIC an Tone-RIC):
+- `message`: Der Text wird aus der Text-RIC übernommen und in die Empfänger-Pakete eingefügt (Bei incomplete-Modus leer)
+- `multicastSourceRic` (string): RIC des ursprünglichen Message-RICs
+
+### Erhaltene Felder (Tone-RIC):
+Diese Felder bleiben **unverändert** bestehen, damit die Zuordnung zum Endgerät korrekt bleibt:
+- `ric`
+- `subric`
+- alle bereits zuvor hinzugefügten Felder (z.B. Descriptor-Modul)
+
+### Rückgabewerte:
+- **False**: Paket wurde intern konsumiert (z.B. Tone-RIC wurde in den Buffer aufgenommen), Router stoppt Verarbeitung für dieses Paket (Verhindert die Ausgabe leerer Nachrichten). Allerdings: Das Paket wird im RAM geparkt.
+- **Liste von Paketen**: Tritt ein, sobald eine Text-RIC die Verteilung auslöst oder ein Timeout abläuft. Der Router verarbeitet jedes Element der Liste (die nun angereicherten Tone-RICs) als eigenständigen Alarm.
+- **None**: Der Router verarbeitet das Original-Paket normal weiter.
+
+---
+
+## Zusätzliche Wildcards
+
+Folgende Wildcards stehen in allen nachfolgenden Plugins zur Verfügung:
+
+|Wildcard|Beschreibung|Beispiel|
+|--------|------------|--------|
+|{MCAST_SOURCE}|RIC des ursprünglichen Message-RICs|0299001|
+|{MCAST_COUNT}|Gesamtanzahl der Empfänger dieses Multicasts.|3|
+|{MCAST_INDEX}|Index des Empfängers (1-basiert für Recipients, 1 für Single, 0 für Control-Pakete)|0, 1, 2, 3, ...|
+|{MCAST_MODE}|Art der Multicast-Verarbeitung durch das Modul|complete, incomplete, single, control|
+|{MCAST_ROLE}|Rolle des Pakets im Multicast-Ablauf|recipient, single, delimiter, netident, trigger|
+
+### Erweiterung der Listen-Wildcards
+Das Modul generiert Wildcards für alle gesammelten Felder (RICs, SubRICs, etc.) in Listenform. Diese sind besonders nützlich, um eine kombinierte Ausgabe (z.B. in Telegram) zu erstellen. Im Folgenden ein paar Beispiele:
+
+|Wildcard|Beschreibung|Zugrundeliegendes Feld|Beispiel|
+|--------|------------|--------|--------|
+|{RIC_LIST}|Liste aller RICs der Empfänger (durch Komma getrennt).|ric_list|"0299001, 0299002"|
+|{SUBRIC_LIST}|Liste aller SubRICs der Empfänger|subric_list|"4, 3"|
+|{DESCRIPTION_LIST}|Liste aller (deskriptiven) Namen der Empfänger (BEISPIEL! **NUR** bei vorher durchlaufenen Descriptor-Modul)|description_list|"FF Musterstadt, BF Beispiel"|
+|{_LIST}|Liste der Werte für jedes Originalfeld aus dem Paket|_list|{FREQUENCY_LIST}, {BITRATE_LIST}|
+
+**Wichtig:** Verwende die **originalen Feldnamen** (z.B. `frequency_list`), nicht die Wildcard-Namen (z.B. ~~`FREQ_list`~~).
+
+### Verwendungsbeispiel in Plugins, z.B. Telegram-Plugin:
+```yaml
+- type: plugin
+ res: telegram
+ config:
+ message_pocsag: |
+ {CNAME}
+ {MSG}
+ RIC: {RIC} / SubRIC: {SRIC}
+ Multicast: {MCAST_INDEX}/{MCAST_COUNT} (Quelle: {MCAST_SOURCE})
+ {TIME}
+```
+
+---
+# Funktionsweise im Detail
+
+## Grundsätzliche Funktion
+
+**Beispiel:**
+```
+10:31:16 - RIC: 0123456 SubRIC: 1 Message: (leer) → Delimiter-RIC
+10:31:16 - RIC: 0234567 SubRIC: 4 Message: (leer) → Empfänger 1
+10:31:16 - RIC: 0345678 SubRIC: 3 Message: (leer) → Empfänger 2
+10:31:17 - RIC: 0456789 SubRIC: 1 Message: "B3 WOHNHAUS" → Message-RIC
+
+Generierte Alarme:
+→ RIC: 0234567 SubRIC: 4 Message: "B3 WOHNHAUS" (behält SubRIC 4!)
+→ RIC: 0345678 SubRIC: 3 Message: "B3 WOHNHAUS" (behält SubRIC 3!)
+```
+
+**Wichtig:** Jeder Empfänger behält seine ursprüngliche SubRIC, da diese oft unterschiedliche Alarmtypen oder Prioritäten repräsentiert.
+
+### Logik der hinzugefügten Felder
+
+Um die Logik der Felder multicastMode, multicastRole, etc. zu verstehen, hilft eine tabellarische Gegenüberstellung:
+
+**1) Szenario Der "Feldstärke-Alarm" (Netident/Delimiter)**
+
+|Paket-Typ|RIC|multicastMode|multicastRole|sourceRic|Index|Count|
+|--------|-------|-------|---------|------|--|--|
+|Delimiter|0288088|control|delimiter|0288088|0|0|
+|Netzident|0000001|control|netident|0000001|0|0|
+
+Hinweis: Diese Pakete dienen der Systemsteuerung. Index und Count sind immer 0, da sie keine Alarm-Empfänger sind. Dadurch fallen sie bei einer Filterung auf `multicastRecipientIndex: ^1$` (ein Paket pro Alarm) automatisch heraus.
+Beide RIC werden unmittelbar nach der Verarbeitung weitergereicht, d.h. es wird nicht auf die Netzident-RIC gewartet, um die Delmitier-RIC weiterzureichen.
+
+**2) Szenario Echter Multicast-Alarm (Vollständig)**
+Hier sieht man den Ablauf:
+- Der Delimiter leert den Speicher und wird mit den ergänzenden Feldern angereichert und sofort weitergegeben.
+- Zwei Tone-RICs sammeln sich an
+- Die Text-RIC löst die Verteilung aus
+
+**Beachte:** Die Text-RIC (0456789) dient als Nachrichtenträger und erscheint nicht als eigenes Paket im Output.
+
+|Phase|Paket-Typ|RIC|multicastMode|multicastRole|sourceRic|Index|Count|
+|-----|---------|---|-------------|-------------|---------|-----|-----|
+|Start|Delimiter|0288088|control|delimiter|0288088|0|0|
+|Sammler|Tone-RIC 1|0234567|-|(interner Buffer)|-|-|-|
+|Sammler|Tone-RIC 2|0345678|-|(interner Buffer)|-|-|-|
+|Auslöser|Text-RIC|0456789|(wird verteilt - keine Ausgabe)|(Nachrichtenträger)|-|-|-|
+|Output 1|Alarm-Paket|0234567|complete|recipient|0456789|1|2|
+|Output 2|Alarm-Paket|0345678|complete|recipient|0456789|2|2|
+
+**3) Szenario Unvollständiger Alarm (Incomplete / Timeout)**
+In diesem Fall fehlt die Text-RIC. Das System wartet bis zum Timeout und schickt dann die Empfänger mit leerer Nachricht raus (getriggert durch das Active Trigger System).
+
+|Phase|Paket-Typ|RIC|multicastMode|multicastRole|sourceRic|Index|Count|
+|-----|---------|---|-------------|-------------|---------|-----|-----|
+|Start|Delimiter|0288088|control|delimiter|0288088|0|0|
+|Sammler|Tone-RIC 1|0234567|-|(interner Buffer)|-|-|-|
+|Sammler|Tone-RIC 2|0345678|-|(interner Buffer)|-|-|-|
+|Event|Timeout|Kein Text|Auto-Clear nach standardmäßig 10s|-|-|-|-|
+|Output 1|Incomplete|0234567|incomplete|recipient|0234567|1|2|
+|Output 2|Incomplete|0345678|incomplete|recipient|0234567|2|2|
+
+Der Delimiter wird mit den ergänzenden Feldern angereichert und sofort weitergegeben.
+
+---
+## Integration in Router-Konfiguration
+
+Das Multicast-Modul muss **vor** den Plugins platziert werden, damit die generierten Alarme korrekt verarbeitet werden:
+
+```yaml
+- name: Router POCSAG
+ route:
+ - type: module
+ res: filter.modeFilter
+ name: Filter POCSAG
+ config:
+ allowed:
+ - pocsag
+
+ # Multicast-Modul hier einfügen
+ - type: module
+ res: multicast
+ name: Multicast Handler
+ config:
+ textRics: '0299001,0310001'
+ delimiterRics: '0288088'
+ autoClearTimeout: 10
+
+ # Weitere Module und Plugins
+ - type: plugin
+ res: mysql
+ config:
+ # ...
+
+ - type: plugin
+ res: telegram
+ config:
+ # ...
+```
+
+## Beispielhafte Verwendung in Router-Konfigurationen
+Das Multicast-Modul gibt für jedes RIC ein eigenes Paket aus UND generiert für konsistente Verarbeitung Listenfelder.
+Dies ermöglicht es, entweder jede RIC einzeln zu verarbeiten oder die Listenfelder für eine gesammelte Ausgabe zu verwenden. Vor der weiteren Verarbeitung in Plugins empfiehlt sich eventuell eine Filterung mittels [RegEx-Filter](regex_filter.md).
+Die folgenden Beispiele dienen zur Veranschaulichung der Möglichkeiten des Multicast-Modul in Verbindung mit RegEx-Filter.
+
+
+### Beispiel (siehe auch "Zusätzliche Wildcards"):
+```yaml
+router:
+- name: Router POCSAG
+ route:
+ - type: module
+ res: filter.modeFilter
+ config:
+ [...]
+ - type: module
+ res: descriptor
+ config:
+ [...]
+ - type: module
+ res: multicast
+ name: Multicast
+ config:
+ autoClearTimeout: 10
+ delimiterRics: '0123456' # Start eines Multicast-Alarms
+ textRics: '9909909' # Text-RIC
+ - type: module
+ res: filter.doubleFilter
+ config:
+ [...]
+ - type: router
+ res: RouterMySQL
+ - type: router
+ res: RouterTelegram
+
+- name: RouterMySQL
+ route:
+ - type: module
+ res: filter.regexFilter
+ name: Filter MySQL
+ config:
+ - name: "Multicast Mode complete or single"
+ checks:
+ - field: multicastMode
+ regex: ^(complete|single)$
+ - type: plugin
+ res: mysql
+ config:
+ [...]
+
+- name: RouterTelegram
+ route:
+ - type: module
+ res: filter.regexFilter
+ name: Multicast Recipient Index Filter # 1. Paket, da ist alles drin für einen kombinierten Alarm und ist immer vorhanden
+ config:
+ - name: "Multicast 1 Paket pro Alarm-Paket"
+ checks:
+ - field: multicastRecipientIndex
+ regex: ^1$
+ - type: plugin
+ res: telegram
+ config:
+ message_pocsag: |
+ {CNAME}
+ {MSG}
+ Alarmierte Einheiten [{MCAST_COUNT}]: {DESCRIPTION_LIST}
+ RICs: {RIC_LIST}
+ {TIME}
+ [...]
+
+```
+
+---
+
+## Das Active Trigger System (Verlustfreie Paketauslieferung)
+BOSWatch arbeitet **synchron**. Das bedeutet: Der Router "schläft", wenn kein Funk-Paket von außen eingeht. Ein Timeout im Hintergrund-Thread des Moduls kann den schlafenden Router nicht von alleine aufwecken, um die im RAM wartenden Pakete (`incomplete`) herauszuschieben.
+
+**Lösung:**
+Das Modul verwendet ein aktives Trigger-System, um sicherzustellen, dass **keine Multicast-Pakete verloren gehen**
+
+**Ausführung**
+Das Modul sendet über via TCP (Loopback) ein minimales Trigger-Paket an den eigenen BOSWatch-Server. Dieser empfängt es wie einen normalen Funk-Alarm, weckt den Router auf und das Modul kann die wartenden Alarme (`incomplete`) sicher ausliefern.
+
+Das Trigger-Paket wird wie die anderen technischen Pakete gekennzeichnet (`multicastMode: control`, `multicastRole: trigger`, Index und Count `0`) und trägt den festen Client-Namen `MulticastTrigger`. Im Router mit dem Multicast-Modul wird es vom Modul verbraucht und nie an nachfolgende Module oder Plugins weitergereicht.
+
+**Wichtig - Router ohne Multicast-Modul:** Der BOSWatch-Server schickt das Trigger-Paket wie jedes andere Paket durch **alle** Router aus `alarmRouter`. Gibt es dort Router ohne Multicast-Modul, erreicht das Trigger-Paket deren Plugins. In diesen Routern sollte es ausgefiltert werden:
+```yaml
+- type: module
+ res: filter.regexFilter
+ name: Multicast-Trigger ausfiltern
+ config:
+ - name: "Kein Trigger"
+ checks:
+ - field: clientName
+ regex: ^(?!MulticastTrigger$)
+```
+Gefiltert wird auf `clientName`, weil jedes Paket dieses Feld besitzt. Ein Filter auf `multicastRole` würde in Routern ohne Multicast-Modul **alle** Pakete verwerfen, da der RegEx-Filter Pakete ohne das geprüfte Feld nicht durchlässt.
+
+Router, die bereits auf `multicastRecipientIndex: ^1$` oder `multicastMode: ^(complete|single)$` filtern, schließen das Trigger-Paket automatisch aus.
+
+**Reihenfolge im Router:** Das Multicast-Modul sollte **vor** einem `filter.doubleFilter` stehen. Sonst kann der doubleFilter das Trigger-Paket als Duplikat verwerfen, und die wartenden Pakete werden erst mit dem nächsten Funk-Paket ausgeliefert.
+
+**Trigger-RIC Auswahl** (3-stufige Fallback-Kette):
+- **Explizit**: Wenn `triggerRic` konfiguriert ist, wird diese RIC verwendet
+- **Dynamisch**: Wenn nicht konfiguriert, wird die erste Tone-RIC der aktuellen Gruppe verwendet
+- **Fallback**: Falls keine Tone-RICs vorhanden sind (sollte nicht vorkommen), wird "9999999" verwendet
+
+**Beispiel-Ablauf bei Auto-Clear:**
+```
+10:31:16 - Tone-RIC empfangen (0234567)
+10:31:16 - Tone-RIC empfangen (0345678)
+10:31:26 - Auto-Clear-Timeout (10s) erreicht
+ → Incomplete-Pakete in Queue gespeichert
+ → Trigger-Paket via Loopback gesendet (RIC: 0234567)
+10:31:26 - Trigger-Paket durch BOSWatch-Server verarbeitet
+ → Queue-Flush: Incomplete-Pakete werden ausgegeben
+```
+
+**Vorteile:**
+- Keine Pakete gehen verloren, auch bei hoher Systemlast
+- Saubere Trennung von Verarbeitung und Ausgabe
+- Ermöglicht zeitlich versetzte Ausgabe ohne Race Conditions
+
+### Zeitbasierte Verarbeitung
+
+1. **Tone-RIC-Sammlung**: Tone-RICs (meist leere Nachrichten) werden empfangen und gespeichert
+2. **Text-RIC-Verteilung**: Sobald ein Text-RIC empfangen wird, erfolgt die sofortige Verteilung an alle gesammelten Tone-RICs
+3. **Auto-Clear**: Nach `autoClearTimeout` Sekunden ohne Text-RIC werden die Tone-RICs als incomplete ausgegeben (via Trigger-System)
+4. **Hard-Timeout-Cleanup**: Nach 3x `autoClearTimeout` werden veraltete Pakete aus dem Speicher gelöscht (Failsafe)
+
+### Frequenz-Trennung
+
+Das Modul trennt Multicast-Listen nach Frequenzen. Dies verhindert Vermischung von Alarmen verschiedener Sender.
+
+**Beispiel:**
+```
+Frequenz 173.050 MHz: Tone-RICs [0234567, 0345678]
+Frequenz 173.075 MHz: Tone-RICs [0456789, 0567890]
+→ Werden getrennt verarbeitet, keine Vermischung möglich (wichtig für Multi-Client mit Single-Server)
+```
+
+## Paketmarkierung statt interner Filterung
+
+Das Modul filtert keine inhaltlich relevanten Pakete.
+Alle Pakete werden mit `multicastRole` markiert und weitergereicht. Die Filterung nach Bedarf erfolgt nachgelagert, z.B. mit `filter.regexFilter`.
+
+Eine Ausnahme bilden **Tone-RICs** (leere Nachrichten): Diese werden zuerst intern im Buffer gesammelt und bei einem complete-Alarm (und incomplete) in die Listenfelder aggregiert. Die Listenfelder werden an alle **Tone-RICs** angehängt und anschließend jede **Tone-RIC** angereichert ausgegeben.
+
+Pakete und ihre Rollen:
+
+- **Delimiter-Pakete**: Erhalten `multicastRole: delimiter`
+- **Netzident-Pakete**: Erhalten `multicastRole: netident`
+- **Empfänger-Pakete**: Erhalten `multicastRole: recipient`
+- **Einzelalarme**: Erhalten `multicastRole: single`
+- **Trigger-Pakete**: Erhalten `multicastRole: trigger`
+
+Beispiel-Filter um Delimiter und Netident auszublenden:
+```yaml
+- type: module
+ res: filter.regexFilter
+ config:
+ - name: "Nur echte Alarme"
+ checks:
+ - field: multicastRole
+ regex: ^(recipient|single)$
+```
+
+
+---
+
+## Zusammenfassung: Was passiert mit den Daten?
+
+- Der Delimiter: Wird immer sofort markiert und als technisches Paket weitergereicht. Er sorgt dafür, dass keine "Leichen" von alten, abgebrochenen Alarmen im Speicher liegen.
+- Die Tone-RICs: Diese werden vom Modul "geschluckt" (return False). Sie verschwinden kurzzeitig aus dem Datenfluss und warten im RAM.
+- Die Text-RIC: Wenn sie eintrifft, nimmt das Modul ihren Text (B3 WOHNHAUS) und kopiert ihn in die Tone-RIC-Pakete im RAM. Anschließend erfolgt die Ausgabe der Tone-RICs.
+- Die Metadaten: Erst beim Erzeugen der Output-Pakete werden Felder wie multicastRecipientIndex berechnet, damit nachfolgende Plugins wissen, dass diese Pakete zusammengehören.
+
+Technischer Hinweis:
+Da die Text-RIC im complete-Fall "verbraucht" wird, um die Tone-RICs zu füllen, wird sie nicht als separates zusätzliches Paket ausgegeben. Das verhindert Dopplungen in der Datenbank. Nur wenn gar keine Tone-RICs da sind, wird die Text-RIC als `single` ausgegeben.
+In diesem Fall (Single) ist es völlig egal, ob die Text-RIC mit Delimiter oder ohne empfangen wird - die Delimiter-RIC wird als Delimiter gekennzeichnet, das Text-RIC als Single (in `multicastMode` sowie in `multicastRole`, `multicastRecipientIndex: 1`, `multicastRecipientCount: 1`).
+
+### Multi-Instanz-Betrieb
+Das Modul unterstützt unbegrenzte parallele Instanzen durch vollständige Isolation:
+
+- **Encapsulated State:** Jede Instanz verwaltet ihren eigenen Tone-RIC-Speicher. Es gibt keine Vermischung zwischen verschiedenen Routen.
+- **Isolated Cleanup:** Jede Instanz startet einen eigenen, internen Cleanup-Thread für präzises Timeout-Management.
+- **Instance IDs:** Zur besseren Nachverfolgung im Log erhält jede Instanz eine eindeutige ID (z.B. MCAST_a1b2).
diff --git a/docu/mkdocs.yml b/docu/mkdocs.yml
index c2602dd7..73d112e5 100644
--- a/docu/mkdocs.yml
+++ b/docu/mkdocs.yml
@@ -20,10 +20,11 @@ nav:
- Changelog: changelog.md
- Module:
- Descriptor: modul/descriptor.md
+ - Double Filter: modul/double_filter.md
- Geocoding: modul/geocoding.md
- Mode Filter: modul/mode_filter.md
+ - Multicast: modul/multicast.md
- Regex Filter: modul/regex_filter.md
- - Double Filter: modul/double_filter.md
- Plugins:
- Http: plugin/http.md
- Telegram: plugin/telegram.md
diff --git a/module/multicast.py b/module/multicast.py
new file mode 100644
index 00000000..2ebde5aa
--- /dev/null
+++ b/module/multicast.py
@@ -0,0 +1,579 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+r"""!
+ ____ ____ ______ __ __ __ _____
+ / __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
+ / __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
+ / /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ /
+/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/
+ German BOS Information Script
+ by Bastian Schroll
+
+@file: multicast.py
+@date: 14.10.2026
+@author: Claus Schichl
+@description: multicast module
+"""
+
+import logging
+import time
+import threading
+from collections import defaultdict
+import socket
+from module.moduleBase import ModuleBase
+from boswatch.packet import Packet
+from boswatch import wildcard
+
+logging.debug("- %s loaded", __name__)
+
+
+class BoswatchModule(ModuleBase):
+ r"""!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.
+ """
+
+ # Trigger defaults
+ _TRIGGER_HOST = "127.0.0.1"
+ _TRIGGER_PORT = 8080
+ _MAGIC_WAKEUP_MSG = "###_MULTICAST_WAKEUP_###"
+ _DEFAULT_TRIGGER_RIC = "9999999"
+
+# ============================================================
+# LIFECYCLE METHODS
+# ============================================================
+
+ def __init__(self, config):
+ super().__init__(__name__, config)
+
+ def onLoad(self):
+ r"""!Initialize module configuration and start the cleanup thread of this instance.
+
+ @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
+# ============================================================
+
+ 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
+ @return a list of packets, False to stop the route, or None if unchanged"""
+ 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 False # wrong RIC - stop
+ logging.debug("[%s] Wakeup trigger received (RIC=%s)", self.name, ric)
+ queued = self._get_queued_packets()
+ return queued if queued else False # no queued packets - stop
+
+ # Only process POCSAG - other packets (ZVEI, FMS, MSG) pass unchanged, together with waiting packets
+ if mode != "pocsag":
+ queued = self._get_queued_packets()
+ return self._combine_results(queued, [bwPacket]) 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, "0", "0")
+ 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 (text but no configured text-RICs) ===
+ 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)
+# ============================================================
+
+ 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"""
+ if isinstance(bwPacket, dict):
+ return bwPacket.copy()
+
+ # Packet instance - the fields are stored in _packet
+ if hasattr(bwPacket, '_packet'):
+ return bwPacket._packet.copy()
+
+ # fallback for other objects
+ 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 {}
+
+ 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
+# ============================================================
+
+ def _add_tone_ric_packet(self, freq, packet_dict):
+ r"""!Add a tone-RIC to the buffer of this instance.
+
+ @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)
+
+ def _get_queued_packets(self):
+ r"""!Pop and return all packets currently waiting in the 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
+# ============================================================
+
+ 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:
+ # UNIX timestamp: shift by 1 ms per recipient for unique database entries
+ v = float(v) + 0.001 * (index - 1)
+ except (ValueError, TypeError):
+ pass # keep the original value
+ packet.set(k, v)
+
+ 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
+
+ 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
+
+ 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]
+
+ 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)
+ incomplete = self._create_incomplete_multicast(freq, orphaned)
+ if bwPacket is not None:
+ self._set_mcast_metadata(bwPacket, "control", "delimiter", ric, "0", "0")
+ return incomplete + [bwPacket] # pass incomplete packets AND the delimiter itself
+ return incomplete
+ if bwPacket is not None:
+ self._set_mcast_metadata(bwPacket, "control", "delimiter", ric, "0", "0")
+ return [bwPacket]
+ return None
+
+# ============================================================
+# PACKET METADATA HELPERS
+# ============================================================
+
+ 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, trigger)
+ @param source: The originating RIC
+ @param count: Total number of recipients (0 for control packets)
+ @param index: Current recipient index (1-based, 0 for control packets)
+ @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, newWildcard) in mapping.items():
+ packet.set(key, val)
+ self._register_wildcard_safe(newWildcard, key)
+
+ 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)
+
+ def _register_wildcard_safe(self, newWildcard, bwPacketField):
+ r"""!Register a wildcard only once - also if another multicast instance registered it already.
+
+ @param newWildcard: The wildcard string (e.g. {MCAST_MODE})
+ @param bwPacketField: The packet field name
+ @return None"""
+ if newWildcard not in self._wildcards_registered:
+ if newWildcard not in wildcard._additionalWildcards:
+ self.registerWildcard(newWildcard, bwPacketField)
+ self._wildcards_registered.add(newWildcard)
+
+# ============================================================
+# CLEANUP & 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()
+
+ 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:
+ threading.Thread(
+ target=self._send_wakeup_trigger, args=(freq, safe_ric), daemon=True).start()
+
+ 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
+
+ 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
+# ============================================================
+
+ def _send_wakeup_trigger(self, freq, fallback_ric):
+ r"""!Send a loopback trigger using a direct socket.
+
+ Uses a raw socket instead of TCPClient to avoid polluting the
+ process-wide default timeout via socket.setdefaulttimeout(),
+ which would stall the server's router main loop.
+
+ @param freq: Frequency identifier
+ @param fallback_ric: RIC to use if no explicit triggerRic 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": self._MAGIC_WAKEUP_MSG,
+ "clientName": "MulticastTrigger",
+ "inputSource": "loopback",
+ "frequency": freq,
+ # marked like the other control packets, so it can be filtered in routers without multicast module
+ "multicastMode": "control",
+ "multicastRole": "trigger",
+ "multicastSourceRic": trigger_ric,
+ "multicastRecipientCount": "0",
+ "multicastRecipientIndex": "0"
+ }
+ data = str(payload).encode("utf-8")
+ header = str(len(data)).ljust(10).encode("utf-8")
+
+ # Direct socket with scoped timeout - does NOT affect socket.getdefaulttimeout()
+ sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ sock.settimeout(2)
+ try:
+ sock.connect((self._trigger_host, self._trigger_port))
+ sock.sendall(header + data)
+ sock.recv(1024) # drain ACK to prevent connection reset by peer
+ logging.debug("[%s] Wakeup trigger sent and acknowledged (RIC=%s)", self.name, trigger_ric)
+ except Exception as e:
+ logging.error("[%s] Could not connect to local server for wakeup: %s", self.name, e)
+ finally:
+ sock.close()
+
+ except Exception as e:
+ logging.error("[%s] Failed to send wakeup trigger: %s", self.name, e)
+
+# ============================================================
+# LIFECYCLE (End)
+# ============================================================
+
+ def onUnload(self):
+ r"""!Stop the cleanup thread of this instance.
+
+ @param None
+ @return None"""
+ self._running = False
+ logging.debug("[%s] Multicast instance unloaded", self.name)