version 0.0.0.3

bridge/aria_bridge.py

@@ -1,8 +1,13 @@
 """
 ARIA Voice Bridge — Hauptmodul.
 
-Verbindet Wake-Word-Erkennung, Whisper STT und Piper TTS
-mit dem ARIA-Core Container ueber WebSocket.
+Verbindet die Android App (via RVS) mit ARIA-Core und bietet
+lokale Spracheingabe (Wake-Word + Whisper STT) und Sprachausgabe (Piper TTS).
+
+Nachrichtenfluss:
+  App  → RVS → Bridge → aria-core
+  aria-core → Bridge → RVS → App
+                     → Lautsprecher (TTS)
 
 Stimmen:
   - Ramona (de_DE-ramona-low)  — Alltag, Gespraeche
@@ -45,6 +50,10 @@ logger = logging.getLogger("aria-bridge")
 CONFIG_PATH = Path("/config/aria.env")
 VOICES_DIR = Path("/voices")
 CORE_WS_URL = os.getenv("ARIA_CORE_WS", "ws://aria:8080")
+RVS_HOST = os.getenv("RVS_HOST", "")          # z.B. rvs.hackersoft.de
+RVS_PORT = os.getenv("RVS_PORT", "443")      # Port des RVS
+RVS_TLS = os.getenv("RVS_TLS", "true")       # true = wss://, false = ws://
+RVS_TOKEN = os.getenv("RVS_TOKEN", "")       # Pairing-Token (gleich wie in der App)
 WHISPER_MODEL = os.getenv("WHISPER_MODEL", "small")
 WHISPER_LANGUAGE = os.getenv("WHISPER_LANGUAGE", "de")
 
@@ -355,11 +364,28 @@ def record_audio(duration: float = RECORD_SECONDS) -> np.ndarray:
 
 
 class ARIABridge:
-    """ARIA Voice Bridge — verbindet Sprache mit dem ARIA-Core."""
+    """ARIA Voice Bridge — verbindet App (via RVS) und Sprache mit ARIA-Core.
+
+    Drei parallele Aufgaben:
+      1. connect_to_core()  — WebSocket zu aria-core (lokal)
+      2. connect_to_rvs()   — WebSocket zum RVS (oeffentlich, fuer die App)
+      3. audio_loop()       — Wake-Word + STT (lokales Mikrofon)
+    """
 
     def __init__(self) -> None:
         self.config = load_config()
         self.ws_url = self.config.get("ARIA_CORE_WS", CORE_WS_URL)
+        # RVS-Verbindungsinfo aus Config oder Env
+        rvs_host = self.config.get("RVS_HOST", RVS_HOST)
+        rvs_port = self.config.get("RVS_PORT", RVS_PORT)
+        rvs_tls = self.config.get("RVS_TLS", RVS_TLS).lower() == "true"
+        self.rvs_token = self.config.get("RVS_TOKEN", RVS_TOKEN)
+        # URL zusammenbauen
+        if rvs_host:
+            proto = "wss" if rvs_tls else "ws"
+            self.rvs_url = f"{proto}://{rvs_host}:{rvs_port}"
+        else:
+            self.rvs_url = ""
         self.current_mode = Mode.NORMAL
         self.running = False
 
@@ -371,8 +397,9 @@ class ARIABridge:
         )
         self.wake_word = WakeWordDetector()
 
-        # WebSocket-Verbindung
-        self.ws: Optional[websockets.WebSocketClientProtocol] = None
+        # WebSocket-Verbindungen
+        self.ws_core: Optional[websockets.WebSocketClientProtocol] = None
+        self.ws_rvs: Optional[websockets.WebSocketClientProtocol] = None
 
     def initialize(self) -> None:
         """Initialisiert alle Komponenten."""
@@ -392,52 +419,55 @@ class ARIABridge:
         self.wake_word.initialize()
 
         logger.info("Alle Komponenten initialisiert")
-        logger.info("WebSocket-Ziel: %s", self.ws_url)
-        logger.info("Aktueller Modus: %s %s", self.current_mode.config.emoji, self.current_mode.config.name)
+        logger.info("aria-core:  %s", self.ws_url)
+        if self.rvs_url and self.rvs_token:
+            logger.info("RVS:        %s (Token: %s...)", self.rvs_url, self.rvs_token[:8])
+        else:
+            logger.warning("RVS nicht konfiguriert — App-Verbindung deaktiviert")
+            logger.warning("  Setze RVS_HOST, RVS_PORT, RVS_TOKEN in /config/aria.env")
+        logger.info("Modus:      %s %s", self.current_mode.config.emoji, self.current_mode.config.name)
+
+    # ── aria-core Verbindung ─────────────────────────────────
 
     async def connect_to_core(self) -> None:
-        """Stellt die WebSocket-Verbindung zu aria-core her.
-
-        Versucht bei Verbindungsverlust automatisch erneut zu verbinden.
-        """
+        """Persistente WebSocket-Verbindung zu aria-core mit Auto-Reconnect."""
         retry_delay = 2
 
         while self.running:
             try:
-                logger.info("Verbinde mit aria-core: %s", self.ws_url)
+                logger.info("[core] Verbinde: %s", self.ws_url)
                 async with websockets.connect(self.ws_url) as ws:
-                    self.ws = ws
-                    retry_delay = 2  # Reset bei erfolgreicher Verbindung
-                    logger.info("Verbunden mit aria-core")
+                    self.ws_core = ws
+                    retry_delay = 2
+                    logger.info("[core] Verbunden")
 
-                    # Nachrichten empfangen
                     async for message in ws:
                         await self._handle_core_message(message)
 
             except websockets.ConnectionClosed:
-                logger.warning("Verbindung zu aria-core verloren")
+                logger.warning("[core] Verbindung verloren")
             except ConnectionRefusedError:
-                logger.warning("aria-core nicht erreichbar")
+                logger.warning("[core] Nicht erreichbar")
             except Exception:
-                logger.exception("WebSocket-Fehler")
+                logger.exception("[core] WebSocket-Fehler")
             finally:
-                self.ws = None
+                self.ws_core = None
 
             if self.running:
-                logger.info("Neuverbindung in %d Sekunden...", retry_delay)
+                logger.info("[core] Reconnect in %ds...", retry_delay)
                 await asyncio.sleep(retry_delay)
                 retry_delay = min(retry_delay * 2, 30)
 
     async def _handle_core_message(self, raw_message: str) -> None:
-        """Verarbeitet eingehende Nachrichten von aria-core.
+        """Verarbeitet Nachrichten von aria-core.
 
-        Args:
-            raw_message: JSON-Nachricht als String.
+        - Leitet Antworten an die App weiter (via RVS)
+        - Sprachausgabe ueber TTS (wenn Modus erlaubt)
         """
         try:
             message = json.loads(raw_message)
         except json.JSONDecodeError:
-            logger.error("Ungueltige JSON-Nachricht: %s", raw_message[:100])
+            logger.error("[core] Ungueltige JSON: %s", raw_message[:100])
             return
 
         text = message.get("text", "")
@@ -445,64 +475,226 @@ class ARIABridge:
         is_critical = metadata.get("critical", False)
         requested_voice = metadata.get("voice")
 
-        logger.info("Nachricht von aria-core: '%s'", text[:80])
+        logger.info("[core] Nachricht: '%s'", text[:80])
 
         # Modus-Wechsel pruefen
         new_mode = detect_mode_switch(text)
         if new_mode is not None:
             self.current_mode = new_mode
             logger.info(
-                "Modus gewechselt: %s %s",
+                "[core] Modus → %s %s",
                 self.current_mode.config.emoji,
                 self.current_mode.config.name,
             )
+            # Modus-Aenderung auch an die App senden
+            await self._send_to_rvs({
+                "type": "mode",
+                "payload": {"mode": self.current_mode.name},
+                "timestamp": int(asyncio.get_event_loop().time() * 1000),
+            })
 
-        # Sprachausgabe nur wenn der Modus es erlaubt
+        # Antwort an die App weiterleiten (als Chat-Nachricht)
+        await self._send_to_rvs({
+            "type": "chat",
+            "payload": {
+                "text": text,
+                "sender": "aria",
+                "voice": requested_voice or self.voice_engine.select_voice(text),
+            },
+            "timestamp": int(asyncio.get_event_loop().time() * 1000),
+        })
+
+        # Sprachausgabe lokal (wenn Modus es erlaubt)
         if should_speak(self.current_mode, is_critical):
             self.voice_engine.speak(text, requested_voice)
         else:
-            logger.info(
-                "Sprachausgabe unterdrueckt (Modus: %s)",
-                self.current_mode.config.name,
-            )
+            logger.info("[core] TTS unterdrueckt (Modus: %s)", self.current_mode.config.name)
 
-    async def send_to_core(self, text: str) -> None:
-        """Sendet Text an aria-core ueber WebSocket.
-
-        Args:
-            text: Der erkannte Text vom Benutzer.
-        """
-        if self.ws is None:
-            logger.error("Keine Verbindung zu aria-core — Nachricht verworfen")
+    async def send_to_core(self, text: str, source: str = "bridge") -> None:
+        """Sendet Text an aria-core."""
+        if self.ws_core is None:
+            logger.error("[core] Nicht verbunden — Nachricht verworfen: '%s'", text[:60])
             return
 
         message = json.dumps({
-            "type": "voice_input",
+            "type": "voice_input" if source == "bridge" else "chat_input",
             "text": text,
             "mode": self.current_mode.name,
-            "source": "bridge",
+            "source": source,
         })
 
         try:
-            await self.ws.send(message)
-            logger.info("An aria-core gesendet: '%s'", text[:80])
+            await self.ws_core.send(message)
+            logger.info("[core] Gesendet (%s): '%s'", source, text[:80])
         except Exception:
-            logger.exception("Fehler beim Senden an aria-core")
+            logger.exception("[core] Sendefehler")
+
+    # ── RVS Verbindung (App-Relay) ──────────────────────────
+
+    async def connect_to_rvs(self) -> None:
+        """Persistente WebSocket-Verbindung zum RVS mit Auto-Reconnect.
+
+        Authentifiziert sich mit dem gleichen Token wie die App.
+        Nachrichten von der App werden an aria-core weitergeleitet.
+        """
+        if not self.rvs_url or not self.rvs_token:
+            logger.info("[rvs] Nicht konfiguriert — ueberspringe")
+            return
+
+        retry_delay = 2
+        url = f"{self.rvs_url}?token={self.rvs_token}"
+
+        while self.running:
+            try:
+                logger.info("[rvs] Verbinde: %s", self.rvs_url)
+                async with websockets.connect(url) as ws:
+                    self.ws_rvs = ws
+                    retry_delay = 2
+                    logger.info("[rvs] Verbunden — warte auf App-Nachrichten")
+
+                    # Heartbeat senden (RVS erwartet Ping alle 30s)
+                    heartbeat_task = asyncio.create_task(self._rvs_heartbeat())
+
+                    try:
+                        async for raw_message in ws:
+                            await self._handle_rvs_message(raw_message)
+                    finally:
+                        heartbeat_task.cancel()
+
+            except websockets.ConnectionClosed:
+                logger.warning("[rvs] Verbindung verloren")
+            except ConnectionRefusedError:
+                logger.warning("[rvs] Nicht erreichbar")
+            except Exception:
+                logger.exception("[rvs] WebSocket-Fehler")
+            finally:
+                self.ws_rvs = None
+
+            if self.running:
+                logger.info("[rvs] Reconnect in %ds...", retry_delay)
+                await asyncio.sleep(retry_delay)
+                retry_delay = min(retry_delay * 2, 30)
+
+    async def _rvs_heartbeat(self) -> None:
+        """Sendet Heartbeats an den RVS damit die Verbindung offen bleibt."""
+        while True:
+            await asyncio.sleep(25)
+            if self.ws_rvs and self.ws_rvs.open:
+                try:
+                    await self.ws_rvs.send(json.dumps({
+                        "type": "heartbeat",
+                        "timestamp": int(asyncio.get_event_loop().time() * 1000),
+                    }))
+                except Exception:
+                    break
+
+    async def _handle_rvs_message(self, raw_message: str) -> None:
+        """Verarbeitet Nachrichten von der App (via RVS).
+
+        Unterstuetzte Typen:
+          - chat:     Text-Nachricht → an aria-core weiterleiten
+          - audio:    Audio-Daten → STT → an aria-core
+          - mode:     Moduswechsel
+          - location: GPS-Daten (loggen, spaeter fuer Skills)
+          - file:     Datei-Upload (an aria-core weiterleiten)
+        """
+        try:
+            message = json.loads(raw_message)
+        except json.JSONDecodeError:
+            logger.error("[rvs] Ungueltige JSON: %s", raw_message[:100])
+            return
+
+        msg_type = message.get("type", "")
+        payload = message.get("payload", {})
+
+        if msg_type == "chat":
+            # Text von der App → an aria-core
+            text = payload.get("text", "")
+            if text:
+                logger.info("[rvs] App-Chat: '%s'", text[:80])
+                await self.send_to_core(text, source="app")
+
+        elif msg_type == "mode":
+            # Moduswechsel von der App
+            mode_name = payload.get("mode", "")
+            new_mode = detect_mode_switch(mode_name)
+            if new_mode is not None:
+                self.current_mode = new_mode
+                logger.info(
+                    "[rvs] Modus → %s %s (von App)",
+                    self.current_mode.config.emoji,
+                    self.current_mode.config.name,
+                )
+
+        elif msg_type == "location":
+            # GPS-Daten von der App
+            lat = payload.get("lat")
+            lng = payload.get("lng")
+            speed = payload.get("speed")
+            logger.info("[rvs] GPS: lat=%.4f lng=%.4f speed=%s", lat or 0, lng or 0, speed)
+            # An aria-core weiterleiten (fuer kontextbasierte Skills)
+            if self.ws_core:
+                await self.ws_core.send(raw_message)
+
+        elif msg_type == "file":
+            # Datei von der App → an aria-core
+            logger.info("[rvs] Datei empfangen: %s", payload.get("name", "?"))
+            if self.ws_core:
+                await self.ws_core.send(raw_message)
+
+        elif msg_type == "audio":
+            # Audio von der App → STT → an aria-core
+            logger.info("[rvs] Audio empfangen — TODO: STT")
+            # Spaeter: Audio decodieren, durch Whisper jagen, Ergebnis an core
+
+        else:
+            logger.debug("[rvs] Unbekannter Typ: %s", msg_type)
+
+    async def _send_to_rvs(self, message: dict) -> None:
+        """Sendet eine Nachricht an die App (via RVS)."""
+        if self.ws_rvs is None or not self.ws_rvs.open:
+            return
+
+        try:
+            await self.ws_rvs.send(json.dumps(message))
+        except Exception:
+            logger.exception("[rvs] Sendefehler")
+
+    # ── Log-Streaming an die App ─────────────────────────────
+
+    async def send_log_to_app(self, source: str, message: str, level: str = "info") -> None:
+        """Sendet einen Log-Eintrag an die App (erscheint im Log-Viewer)."""
+        await self._send_to_rvs({
+            "type": "log",
+            "payload": {
+                "source": source,
+                "message": message,
+                "level": level,
+            },
+            "timestamp": int(asyncio.get_event_loop().time() * 1000),
+        })
+
+    async def send_event_to_app(self, title: str, description: str) -> None:
+        """Sendet ein Event an die App (erscheint im Event-Feed)."""
+        await self._send_to_rvs({
+            "type": "event",
+            "payload": {
+                "title": title,
+                "description": description,
+            },
+            "timestamp": int(asyncio.get_event_loop().time() * 1000),
+        })
+
+    # ── Audio-Schleife (lokales Mikrofon) ────────────────────
 
     async def audio_loop(self) -> None:
-        """Haupt-Audio-Schleife: Wake-Word erkennen, aufnehmen, transkribieren.
-
-        Laeuft kontinuierlich und hoert auf das Wake-Word.
-        Bei Erkennung wird aufgenommen, transkribiert und an aria-core gesendet.
-        """
+        """Wake-Word erkennen, aufnehmen, transkribieren, an aria-core senden."""
         logger.info("Audio-Schleife gestartet — warte auf Wake-Word '%s'...", WakeWordDetector.WAKE_WORD)
 
-        # Audio-Stream fuer Wake-Word-Erkennung
         loop = asyncio.get_event_loop()
 
         while self.running:
             try:
-                # Einen Block Audio aufnehmen und auf Wake-Word pruefen
                 audio_chunk = sd.rec(
                     BLOCK_SIZE,
                     samplerate=SAMPLE_RATE,
@@ -517,22 +709,23 @@ class ARIABridge:
 
                 if detected:
                     logger.info("Wake-Word erkannt — starte Aufnahme")
+                    await self.send_event_to_app(
+                        "Wake-Word erkannt",
+                        "ARIA hoert zu...",
+                    )
 
-                    # Aufnahme im Thread-Pool (blockiert)
                     audio_data = await loop.run_in_executor(None, record_audio)
 
-                    # Transkription im Thread-Pool
                     text = await loop.run_in_executor(
                         None, self.stt_engine.transcribe, audio_data
                     )
 
                     if text.strip():
-                        # Modus-Wechsel lokal pruefen
                         new_mode = detect_mode_switch(text)
                         if new_mode is not None:
                             self.current_mode = new_mode
 
-                        await self.send_to_core(text)
+                        await self.send_to_core(text, source="bridge")
                     else:
                         logger.info("Keine Sprache erkannt — ignoriert")
 
@@ -543,13 +736,15 @@ class ARIABridge:
                 logger.exception("Fehler in der Audio-Schleife")
                 await asyncio.sleep(1)
 
+    # ── Run & Shutdown ───────────────────────────────────────
+
     async def run(self) -> None:
-        """Startet die Bridge mit allen Komponenten."""
+        """Startet die Bridge mit allen drei Verbindungen parallel."""
         self.running = True
 
-        # WebSocket-Verbindung und Audio-Schleife parallel ausfuehren
         tasks = [
             asyncio.create_task(self.connect_to_core()),
+            asyncio.create_task(self.connect_to_rvs()),
             asyncio.create_task(self.audio_loop()),
         ]