# -*- coding: utf-8 -*- """ SyncService — Standalone-seitiger Sync-Orchestrator. Probe-Loop: Alle probe_interval_seconds prüfen ob Server erreichbar. Bei Verbindung: Initialer Sync, danach periodischer Sync alle sync_interval_seconds. """ from __future__ import annotations import asyncio from datetime import datetime, timezone from typing import TYPE_CHECKING from trixy_core.sync.connection import SyncConnection, SyncConnectionConfig from trixy_core.sync.models import SyncItem, _now_iso from trixy_core.network.cmd.sync import SyncPush, SyncRequest, SyncResponse, SyncAck from trixy_core.network.protocol import ProtocolMessage from trixy_core.network.client_connection import get_mac_address from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn if TYPE_CHECKING: from trixy_core.config.datasets.standalone import SyncConfig from trixy_core.standalone import StandaloneApplication class SyncService: """ Orchestriert die Datensynchronisation zwischen Standalone und Server. - Probe-Loop: Regelmaessig versuchen, den Server zu erreichen - Bei Verbindung: Handshake → initialer Sync → periodischer Sync - Bei Verbindungsverlust: Zurueck zum Probe-Loop """ def __init__(self, application: StandaloneApplication, config: SyncConfig) -> None: """ Initialisiert den SyncService. Args: application: Die Standalone-Anwendung config: Sync-Konfiguration """ self._app = application self._config = config self._mac_address = get_mac_address() # Verbindung conn_config = SyncConnectionConfig( host=config.server_host, port=config.server_port, probe_interval_seconds=config.probe_interval_seconds, sync_interval_seconds=config.sync_interval_seconds, connect_timeout_seconds=config.connect_timeout_seconds, ) self._connection = SyncConnection( config=conn_config, room=application.room, alias=application.alias, mac_address=self._mac_address, ) self._connection.on_connected(self._on_connected) self._connection.on_disconnected(self._on_disconnected) self._connection.on_message(self._on_message) # Zustand self._running = False self._probe_task: asyncio.Task | None = None self._sync_task: asyncio.Task | None = None self._sync_in_progress = False # Fuer Antwort-Handling self._pending_response: asyncio.Future[SyncResponse] | None = None async def start(self) -> None: """Startet den SyncService (Probe-Loop).""" if not self._config.server_host: pwarn("SyncService: Kein Server-Host konfiguriert, Sync deaktiviert") return self._running = True self._probe_task = asyncio.create_task(self._probe_loop()) pinfo(f"SyncService: Gestartet (Probe alle {self._config.probe_interval_seconds}s zu {self._config.server_host}:{self._config.server_port})") async def stop(self) -> None: """Stoppt den SyncService.""" self._running = False # Tasks beenden for task in [self._probe_task, self._sync_task]: if task and not task.done(): task.cancel() try: await task except asyncio.CancelledError: pass # Verbindung trennen if self._connection.is_connected: await self._connection.disconnect("SyncService gestoppt") pinfo("SyncService: Gestoppt") async def sync_now(self) -> bool: """ Fuehrt einen manuellen Sync durch (z.B. per Sprachbefehl). Returns: True bei Erfolg """ if not self._connection.is_connected: pwarn("SyncService: Nicht verbunden, Sync nicht moeglich") return False if self._sync_in_progress: pdebug("SyncService: Sync bereits aktiv") return False return await self._perform_sync() # ========================================================================= # Probe-Loop # ========================================================================= async def _probe_loop(self) -> None: """Versucht regelmaessig, den Server zu erreichen.""" while self._running: try: if not self._connection.is_connected: pdebug(f"SyncService: Probe zu {self._config.server_host}:{self._config.server_port}...") connected = await self._connection.connect() if connected: # Initialer Sync wird in _on_connected ausgeloest # Periodischen Sync starten self._sync_task = asyncio.create_task(self._periodic_sync_loop()) await asyncio.sleep(self._config.probe_interval_seconds) except asyncio.CancelledError: break except Exception as e: perror(f"SyncService: Probe-Fehler: {e}") await asyncio.sleep(self._config.probe_interval_seconds) async def _periodic_sync_loop(self) -> None: """Fuehrt periodische Syncs durch solange verbunden.""" while self._running and self._connection.is_connected: try: await asyncio.sleep(self._config.sync_interval_seconds) if not self._running or not self._connection.is_connected: break await self._perform_sync() except asyncio.CancelledError: break except Exception as e: perror(f"SyncService: Periodischer Sync Fehler: {e}") # ========================================================================= # Sync-Ablauf # ========================================================================= async def _perform_sync(self) -> bool: """ Fuehrt einen vollstaendigen Sync-Zyklus durch. 1. Lokale Aenderungen an Server pushen 2. Server-Aenderungen anfragen 3. Server-Items lokal mergen 4. last_sync_time aktualisieren """ store = self._app.sync_store if store is None: perror("SyncService: Kein SyncStore verfuegbar") return False self._sync_in_progress = True try: last_sync = store.get_last_sync_time() pdebug(f"SyncService: Starte Sync (seit: {last_sync or 'nie'})") # 1. Lokale Aenderungen pushen local_changes = store.get_changes_since(last_sync) if last_sync else [] if local_changes: push_msg = SyncPush( items=[item.to_dict() for item in local_changes] ) if not await self._connection.send_command(push_msg): perror("SyncService: SyncPush senden fehlgeschlagen") return False pdebug(f"SyncService: {len(local_changes)} lokale Aenderungen gepusht") # 2. Server-Aenderungen anfragen request_msg = SyncRequest(since=last_sync) # Future fuer Antwort erstellen loop = asyncio.get_running_loop() self._pending_response = loop.create_future() if not await self._connection.send_command(request_msg): perror("SyncService: SyncRequest senden fehlgeschlagen") self._pending_response = None return False # 3. Auf Antwort warten (mit Timeout) try: response = await asyncio.wait_for(self._pending_response, timeout=30.0) except asyncio.TimeoutError: perror("SyncService: Timeout beim Warten auf SyncResponse") self._pending_response = None return False finally: self._pending_response = None # 4. Server-Items lokal mergen merged_count = 0 for item_dict in response.items: item = SyncItem.from_dict(item_dict) if store.merge_incoming(item): merged_count += 1 # 5. last_sync_time aktualisieren server_time = response.server_time or _now_iso() store.set_last_sync_time(server_time) pinfo(f"SyncService: Sync abgeschlossen ({len(local_changes)} gepusht, {merged_count} gemerged)") # 6. Event emittieren await self._app.events.emit("sync_completed", { "pushed_count": len(local_changes), "merged_count": merged_count, "server_time": server_time, }) return True except Exception as e: perror(f"SyncService: Sync-Fehler: {e}") return False finally: self._sync_in_progress = False # ========================================================================= # Callbacks # ========================================================================= async def _on_connected(self) -> None: """Callback bei Verbindungsaufbau.""" pinfo("SyncService: Mit Server verbunden, starte initialen Sync...") await self._perform_sync() async def _on_disconnected(self, reason: str) -> None: """Callback bei Verbindungsabbruch.""" pinfo(f"SyncService: Verbindung verloren: {reason}") # Periodischen Sync stoppen if self._sync_task and not self._sync_task.done(): self._sync_task.cancel() try: await self._sync_task except asyncio.CancelledError: pass async def _on_message(self, message: ProtocolMessage) -> None: """Callback fuer eingehende Nachrichten vom Server.""" class_name = message.class_name if class_name == "SyncResponse": # Antwort auf SyncRequest data = message.data if isinstance(data, SyncResponse): response = data elif isinstance(data, dict): response = SyncResponse( items=data.get("items", []), server_time=data.get("server_time", ""), ) else: perror(f"SyncService: Ungueltiges SyncResponse-Format: {type(data)}") return if self._pending_response and not self._pending_response.done(): self._pending_response.set_result(response) elif class_name == "SyncAck": pdebug("SyncService: SyncAck empfangen") elif class_name == "SyncPush": # Server pusht Aenderungen (z.B. nach Aenderung durch anderen Satellite) store = self._app.sync_store if store is None: return data = message.data items: list[dict] = [] if isinstance(data, SyncPush): items = data.items elif isinstance(data, dict): items = data.get("items", []) merged_count = 0 for item_dict in items: item = SyncItem.from_dict(item_dict) if store.merge_incoming(item): merged_count += 1 if merged_count > 0: pdebug(f"SyncService: {merged_count} Server-Push Items gemerged") await self._app.events.emit("sync_push_received", { "merged_count": merged_count, })