#!/usr/bin/env python3 """ WebTG SVXLink Monitor Überwacht SVXLink-Logs und sendet Statusupdates per MQTT """ import re import json import time import paho.mqtt.client as mqtt from datetime import datetime from threading import Thread, Lock import subprocess import sys # ========== KONFIGURATION ========== CONFIG = { 'mqtt_broker': 'localhost', 'mqtt_port': 1883, 'mqtt_topic_base': 'webtg', 'num_channels': 10, 'svxlink_log': '/var/log/syslog', # oder /var/log/svxlink/svxlink.log 'scan_interval': 0.5, # Sekunden 'timeout_inactive': 5, # Sekunden bis TG als inaktiv gilt } # ========== GLOBALER STATE ========== class TGState: def __init__(self): self.slots = {} self.lock = Lock() # Initialisiere alle Slots for i in range(1, CONFIG['num_channels'] + 1): self.slots[i] = { 'tg_number': None, 'caller': None, 'mount': f'/tg{i}', 'last_update': None, 'active': False } def update_slot(self, slot_id, tg_number, caller=None): """Aktualisiere Slot-Information""" with self.lock: if slot_id not in self.slots: return self.slots[slot_id]['tg_number'] = tg_number self.slots[slot_id]['caller'] = caller self.slots[slot_id]['last_update'] = time.time() self.slots[slot_id]['active'] = True def clear_slot(self, slot_id): """Lösche Slot-Information""" with self.lock: if slot_id in self.slots: self.slots[slot_id]['tg_number'] = None self.slots[slot_id]['caller'] = None self.slots[slot_id]['active'] = False def get_slot(self, slot_id): """Hole Slot-Information (thread-safe)""" with self.lock: if slot_id in self.slots: return self.slots[slot_id].copy() return None def check_timeouts(self): """Prüfe auf inaktive Slots""" now = time.time() timed_out = [] with self.lock: for slot_id, data in self.slots.items(): if data['active'] and data['last_update']: if now - data['last_update'] > CONFIG['timeout_inactive']: timed_out.append(slot_id) return timed_out # ========== MQTT CLIENT ========== class MQTTPublisher: def __init__(self): self.client = mqtt.Client() self.connected = False self.client.on_connect = self.on_connect self.client.on_disconnect = self.on_disconnect def on_connect(self, client, userdata, flags, rc): if rc == 0: self.connected = True print(f"[MQTT] Verbunden mit {CONFIG['mqtt_broker']}:{CONFIG['mqtt_port']}") self.publish_status(True) else: print(f"[MQTT] Verbindung fehlgeschlagen: {rc}") def on_disconnect(self, client, userdata, rc): self.connected = False print(f"[MQTT] Verbindung getrennt") def connect(self): try: self.client.connect(CONFIG['mqtt_broker'], CONFIG['mqtt_port'], 60) self.client.loop_start() return True except Exception as e: print(f"[MQTT] Verbindungsfehler: {e}") return False def publish_status(self, online): """Sende Scanner-Status""" topic = f"{CONFIG['mqtt_topic_base']}/status" payload = json.dumps({'online': online}) self.client.publish(topic, payload, retain=True) def publish_slot(self, slot_id, tg_number, caller, mount): """Sende Slot-Update""" topic = f"{CONFIG['mqtt_topic_base']}/tg/{slot_id}/status" payload = json.dumps({ 'tg_number': tg_number, 'caller': caller, 'mount': mount }) self.client.publish(topic, payload, retain=True) print(f"[MQTT] Slot {slot_id}: TG {tg_number}, Caller: {caller or 'None'}") def clear_slot(self, slot_id): """Lösche Slot""" topic = f"{CONFIG['mqtt_topic_base']}/tg/{slot_id}/status" payload = json.dumps({ 'tg_number': None, 'caller': None, 'mount': f'/tg{slot_id}' }) self.client.publish(topic, payload, retain=True) print(f"[MQTT] Slot {slot_id}: Gelöscht") # ========== LOG PARSER ========== class SVXLinkLogParser: """ Parser für SVXLink-Logs Passt diese Regex-Patterns an deine SVXLink-Log-Ausgaben an! """ # Beispiel-Patterns (müssen an deine Logs angepasst werden!) PATTERNS = { # TalkGroup-Aktivierung 'tg_active': re.compile(r'TG\s*#?(\d+).*activated', re.IGNORECASE), # Caller-Information 'caller': re.compile(r'(?:from|caller|call)\s*:?\s*([A-Z0-9]{3,10})', re.IGNORECASE), # TalkGroup-Deaktivierung 'tg_inactive': re.compile(r'TG\s*#?(\d+).*(?:deactivated|closed|end)', re.IGNORECASE), # PTT/Transmission 'ptt': re.compile(r'PTT\s+(?:on|start|begin)', re.IGNORECASE), } def parse_line(self, line): """ Parse eine Log-Zeile Returns: dict mit 'type', 'tg_number', 'caller' oder None """ result = None # TalkGroup aktiviert? match = self.PATTERNS['tg_active'].search(line) if match: tg_number = int(match.group(1)) result = {'type': 'tg_active', 'tg_number': tg_number, 'caller': None} # Caller suchen caller_match = self.PATTERNS['caller'].search(line) if caller_match: result['caller'] = caller_match.group(1).upper() # TalkGroup deaktiviert? if not result: match = self.PATTERNS['tg_inactive'].search(line) if match: tg_number = int(match.group(1)) result = {'type': 'tg_inactive', 'tg_number': tg_number} return result # ========== LOG MONITOR ========== def monitor_logs(state, mqtt_pub, parser): """Überwache SVXLink-Logs""" print(f"[LOG] Überwache {CONFIG['svxlink_log']}...") # tail -f auf Log-Datei try: proc = subprocess.Popen( ['tail', '-F', '-n', '0', CONFIG['svxlink_log']], stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True ) # Slot-Zuordnung: TG -> Slot tg_to_slot = {} next_slot = 1 for line in proc.stdout: line = line.strip() if not line: continue # Parse Log-Zeile event = parser.parse_line(line) if not event: continue # TalkGroup aktiviert if event['type'] == 'tg_active': tg = event['tg_number'] caller = event['caller'] # Finde freien Slot oder benutze existierenden if tg in tg_to_slot: slot_id = tg_to_slot[tg] else: # Suche freien Slot slot_id = next_slot next_slot = (next_slot % CONFIG['num_channels']) + 1 tg_to_slot[tg] = slot_id # Update State state.update_slot(slot_id, tg, caller) # MQTT publish slot_data = state.get_slot(slot_id) if slot_data: mqtt_pub.publish_slot( slot_id, slot_data['tg_number'], slot_data['caller'], slot_data['mount'] ) # TalkGroup deaktiviert elif event['type'] == 'tg_inactive': tg = event['tg_number'] if tg in tg_to_slot: slot_id = tg_to_slot[tg] state.clear_slot(slot_id) mqtt_pub.clear_slot(slot_id) del tg_to_slot[tg] except Exception as e: print(f"[LOG] Fehler beim Überwachen: {e}") import traceback traceback.print_exc() # ========== TIMEOUT CHECKER ========== def timeout_checker(state, mqtt_pub): """Prüfe regelmäßig auf Timeouts""" while True: time.sleep(1) timed_out = state.check_timeouts() for slot_id in timed_out: print(f"[TIMEOUT] Slot {slot_id} ist abgelaufen") state.clear_slot(slot_id) mqtt_pub.clear_slot(slot_id) # ========== MAIN ========== def main(): print("=" * 60) print("WebTG SVXLink Monitor") print("=" * 60) print() # Initialisiere Komponenten state = TGState() mqtt_pub = MQTTPublisher() parser = SVXLinkLogParser() # Verbinde MQTT print("[MQTT] Verbinde...") if not mqtt_pub.connect(): print("[FEHLER] MQTT-Verbindung fehlgeschlagen!") return 1 # Warte auf Verbindung for i in range(10): if mqtt_pub.connected: break time.sleep(0.5) if not mqtt_pub.connected: print("[FEHLER] MQTT-Timeout!") return 1 print() print("✓ Bereit!") print() # Starte Timeout-Checker timeout_thread = Thread(target=timeout_checker, args=(state, mqtt_pub), daemon=True) timeout_thread.start() # Starte Log-Monitor (blocking) try: monitor_logs(state, mqtt_pub, parser) except KeyboardInterrupt: print("\n[EXIT] Beende...") mqtt_pub.publish_status(False) return 0 if __name__ == '__main__': sys.exit(main())