# -*- coding: utf-8 -*- """ RegistrationManager für die Verwaltung von Satellite-Registrierungen. Satellites müssen vor der Verbindung registriert werden. Die Registrierung hat ein Timeout von 60 Sekunden. """ import asyncio import json from datetime import datetime, timedelta from pathlib import Path from typing import TYPE_CHECKING from trixy_core.satellite.satellite import Satellite from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn if TYPE_CHECKING: from trixy_core.application import IApplication class PendingRegistration: """Eine ausstehende Registrierung.""" def __init__(self, mac_address: str, room: str, alias: str) -> None: self.mac_address = mac_address self.room = room self.alias = alias self.created_at = datetime.now() self.expires_at = datetime.now() + timedelta(seconds=60) def is_expired(self) -> bool: """Prüft, ob die Registrierung abgelaufen ist.""" return datetime.now() > self.expires_at class RegistrationManager: """ Verwaltet Satellite-Registrierungen. Registrierungsmodi: - Zeitfenster: 60-Sekunden Registrierungsfenster (Standard) - Permanent: --auto-regist Argument, dauerhaft offen - Datei-Trigger: regist.txt Datei, nächster Satellite wird registriert - Speicherung als JSON-Dateien nach MAC-Adresse - Automatisches Laden registrierter Satellites beim Start """ REGISTRATION_TIMEOUT = 60 # Sekunden DEFAULT_DIRECTORY = "satellites" TRIGGER_FILE = "regist.txt" def __init__( self, application: "IApplication", directory: str | Path = DEFAULT_DIRECTORY ) -> None: """ Initialisiert den RegistrationManager. Args: application: Referenz zur Hauptanwendung directory: Verzeichnis für Registrierungsdateien """ self._application = application self._directory = Path(directory) self._pending: dict[str, PendingRegistration] = {} self._registration_mode = False self._permanent_mode = False self._file_trigger_enabled = False self._file_trigger_active = False # Durch Datei-Trigger aktiviert self._mode_task: asyncio.Task | None = None @property def registration_mode(self) -> bool: """Ist der Registrierungsmodus aktiv?""" return self._registration_mode or self._permanent_mode @property def is_permanent_mode(self) -> bool: """Ist der permanente Registrierungsmodus aktiv?""" return self._permanent_mode @property def is_file_trigger_enabled(self) -> bool: """Ist der Datei-Trigger aktiviert?""" return self._file_trigger_enabled @property def pending_count(self) -> int: """Anzahl ausstehender Registrierungen.""" return len(self._pending) def enable_registration_mode(self, timeout_seconds: int = 60) -> None: """ Aktiviert den Registrierungsmodus. Args: timeout_seconds: Zeit bis zur automatischen Deaktivierung """ self._registration_mode = True pinfo(f"Registrierungsmodus aktiviert ({timeout_seconds}s)") # Bestehenden Task abbrechen if self._mode_task and not self._mode_task.done(): self._mode_task.cancel() # Neuen Timeout-Task starten (nur wenn Event-Loop läuft) async def timeout() -> None: await asyncio.sleep(timeout_seconds) self.disable_registration_mode() try: loop = asyncio.get_running_loop() self._mode_task = loop.create_task(timeout()) except RuntimeError: # Kein laufender Event-Loop (z.B. in Tests) self._mode_task = None def disable_registration_mode(self) -> None: """Deaktiviert den Registrierungsmodus.""" self._registration_mode = False self._pending.clear() if self._mode_task and not self._mode_task.done(): self._mode_task.cancel() self._mode_task = None pinfo("Registrierungsmodus deaktiviert") def enable_permanent_registration_mode(self) -> None: """ Aktiviert den permanenten Registrierungsmodus. Server registriert automatisch alle neuen Satellites ohne Timeout. Wird über --auto-regist Argument aktiviert. """ self._permanent_mode = True self._registration_mode = True pinfo("Permanenter Auto-Registrierungsmodus aktiviert") def disable_permanent_registration_mode(self) -> None: """Deaktiviert den permanenten Registrierungsmodus.""" self._permanent_mode = False self._registration_mode = False pinfo("Permanenter Auto-Registrierungsmodus deaktiviert") def enable_file_trigger(self) -> None: """ Aktiviert den Datei-Trigger für Registrierungen. Wenn eine regist.txt Datei im aktuellen Verzeichnis existiert, wird der nächste unregistrierte Satellite automatisch registriert und die Datei gelöscht. """ self._file_trigger_enabled = True pdebug(f"Datei-Trigger aktiviert ({self.TRIGGER_FILE})") def disable_file_trigger(self) -> None: """Deaktiviert den Datei-Trigger.""" self._file_trigger_enabled = False pdebug("Datei-Trigger deaktiviert") def check_file_trigger(self) -> bool: """ Prüft ob der Datei-Trigger aktiv ist. Wenn die Datei existiert, wird sie gelöscht und ein einmaliger Registrierungsmodus für den nächsten Satellite aktiviert. Returns: True wenn regist.txt existiert (und gelöscht wurde) """ if not self._file_trigger_enabled: return False trigger_path = Path(self.TRIGGER_FILE) if trigger_path.exists(): try: trigger_path.unlink() pinfo(f"Registrierungs-Trigger erkannt ({self.TRIGGER_FILE})") # Aktiviere einmaligen Registrierungsmodus self._registration_mode = True self._file_trigger_active = True return True except Exception as e: perror(f"Fehler beim Löschen der Trigger-Datei: {e}") return False return False def should_auto_register(self) -> bool: """ Prüft ob automatische Registrierung erlaubt ist. Berücksichtigt: - Permanenter Modus (--auto-regist) - Zeitfenster-Modus (enable_registration_mode) - Datei-Trigger (regist.txt) Returns: True wenn Registrierung erlaubt """ # Permanenter Modus hat höchste Priorität if self._permanent_mode: return True # Zeitfenster-Modus (inkl. durch Datei-Trigger aktiviert) if self._registration_mode: return True # Datei-Trigger prüfen (aktiviert _registration_mode wenn gefunden) if self.check_file_trigger(): return True return False def begin_registration( self, mac_address: str, room: str, alias: str ) -> bool: """ Startet eine neue Registrierung. Args: mac_address: MAC-Adresse des Satellites room: Raum-Kennung alias: Anzeigename Returns: True wenn erfolgreich gestartet """ if not self.should_auto_register(): pwarn("Registrierungsversuch ohne aktiven Registrierungsmodus") return False # Bestehende abgelaufene Registrierungen entfernen self._cleanup_expired() # Neue Registrierung erstellen self._pending[mac_address] = PendingRegistration(mac_address, room, alias) pdebug(f"Registrierung gestartet: {alias} ({mac_address})") return True def complete_registration(self, mac_address: str) -> Satellite | None: """ Schließt eine Registrierung ab. Args: mac_address: MAC-Adresse des Satellites Returns: Der registrierte Satellite oder None """ pending = self._pending.pop(mac_address, None) if pending is None: perror(f"Keine ausstehende Registrierung für: {mac_address}") return None if pending.is_expired(): perror(f"Registrierung abgelaufen: {mac_address}") return None # Satellite erstellen satellite = Satellite( room_id=pending.room, mac_address=pending.mac_address, alias=pending.alias ) # In Datei speichern self._save_registration(satellite) # Wenn durch Datei-Trigger aktiviert, Modus zurücksetzen if self._file_trigger_active: self._registration_mode = False self._file_trigger_active = False pdebug("Datei-Trigger-Modus zurückgesetzt") pinfo(f"Satellite registriert: {satellite.alias} ({satellite.mac_address})") return satellite def cancel_registration(self, mac_address: str) -> bool: """ Bricht eine ausstehende Registrierung ab. Args: mac_address: MAC-Adresse Returns: True wenn abgebrochen """ if mac_address in self._pending: del self._pending[mac_address] pdebug(f"Registrierung abgebrochen: {mac_address}") return True return False def is_registered(self, mac_address: str) -> bool: """ Prüft, ob ein Satellite registriert ist. Args: mac_address: MAC-Adresse Returns: True wenn registriert """ file_path = self._get_file_path(mac_address) return file_path.exists() def load_registered(self, mac_address: str) -> Satellite | None: """ Lädt einen registrierten Satellite. Args: mac_address: MAC-Adresse Returns: Satellite oder None """ file_path = self._get_file_path(mac_address) if not file_path.exists(): return None try: with open(file_path, "r", encoding="utf-8") as f: data = json.load(f) satellite = Satellite( satellite_id=data.get("id"), room_id=data.get("room_id", ""), mac_address=data.get("mac_address", mac_address), alias=data.get("alias", "") ) pdebug(f"Satellite geladen: {satellite.alias}") return satellite except Exception as e: perror(f"Fehler beim Laden von {mac_address}: {e}") return None def load_all(self) -> list[Satellite]: """ Lädt alle registrierten Satellites. Returns: Liste aller registrierten Satellites """ satellites: list[Satellite] = [] if not self._directory.exists(): return satellites for file_path in self._directory.glob("*.json"): mac_address = file_path.stem satellite = self.load_registered(mac_address) if satellite: satellites.append(satellite) pinfo(f"{len(satellites)} registrierte Satellites geladen") return satellites def delete_registration(self, mac_address: str) -> bool: """ Löscht eine Registrierung. Args: mac_address: MAC-Adresse Returns: True wenn gelöscht """ file_path = self._get_file_path(mac_address) if file_path.exists(): file_path.unlink() pinfo(f"Registrierung gelöscht: {mac_address}") return True return False def _save_registration(self, satellite: Satellite) -> None: """Speichert eine Registrierung in eine Datei.""" self._directory.mkdir(parents=True, exist_ok=True) file_path = self._get_file_path(satellite.mac_address) data = { "id": satellite.id, "room_id": satellite.room_id, "mac_address": satellite.mac_address, "alias": satellite.alias, "registered_at": datetime.now().isoformat() } with open(file_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) pdebug(f"Registrierung gespeichert: {file_path}") def _get_file_path(self, mac_address: str) -> Path: """Gibt den Dateipfad für eine MAC-Adresse zurück.""" # MAC-Adresse für Dateinamen bereinigen safe_mac = mac_address.replace(":", "-").replace(".", "-") return self._directory / f"{safe_mac}.json" def _cleanup_expired(self) -> None: """Entfernt abgelaufene Registrierungen.""" expired = [mac for mac, reg in self._pending.items() if reg.is_expired()] for mac in expired: del self._pending[mac] pdebug(f"Abgelaufene Registrierung entfernt: {mac}")