chobit/services/tray/dev_trays.py

"""Dev-mode multi-tray: one indicator per service.

Spawns [Bridge:dev][Chat:dev][Char:dev][Speech:dev][Face:dev][Main] in a single
process sharing one Gtk.main() loop via lilith_tray.run_all().
"""

from __future__ import annotations

import json
import os
import signal
import subprocess
import sys
import time
from pathlib import Path
from typing import Callable

from lilith_tray import TrayApp, TrayConfig, TrayIcon, TrayMenuItem, run_all

import flight_recorder as fr

from chobit_tray import (
    PROJECT_ROOT,
    TRAY_ICONS_DIR,
    ChobitTray,
    _send_command,
    create_app,
)

SPEECH_UNIT = "chatterbox-tts"
LLM_URL = "http://127.0.0.1:8210"
REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379


def _icons() -> dict[str, TrayIcon]:
    return {
        "running": TrayIcon.from_file(TRAY_ICONS_DIR / "green-24.png"),
        "warning": TrayIcon.from_file(TRAY_ICONS_DIR / "yellow-24.png"),
        "stopped": TrayIcon.from_file(TRAY_ICONS_DIR / "red-24.png"),
    }


def _http_check(url: str) -> tuple[str, dict[str, str]]:
    """HTTP GET health check. Returns (icon_key, labels)."""
    import urllib.error
    import urllib.request

    try:
        with urllib.request.urlopen(f"{url}/health", timeout=1) as resp:
            body = resp.read().decode()
            try:
                data = json.loads(body)
                status = data.get("status", "ok")
            except json.JSONDecodeError:
                status = "ok"
            return "running", {"Status": status}
    except urllib.error.HTTPError as e:
        return "warning", {"Status": f"HTTP {e.code}"}
    except Exception:
        return "stopped", {"Status": "Unreachable"}


class DevServiceTray(TrayApp):
    """Lightweight dev tray backed by a check function."""

    def __init__(
        self,
        name: str,
        label: str,
        check_fn: Callable[[], tuple[str, dict[str, str]]],
        menu: list[TrayMenuItem] | None = None,
    ) -> None:
        self._check_fn = check_fn
        config = TrayConfig(
            name=name,
            icons=_icons(),
            initial_icon="stopped",
            menu=menu or [TrayMenuItem.quit(f"Quit {label}")],
            poll_interval=5,
            label=label,
        )
        super().__init__(config)

    def poll_status(self) -> str:
        icon_key, _ = self._check_fn()
        return icon_key

    def get_status_labels(self) -> dict[str, str]:
        _, labels = self._check_fn()
        return labels


# ── check functions ───────────────────────────────────────────────────────────



def _systemd_is_active(unit: str) -> bool:
    try:
        result = subprocess.run(
            ["systemctl", "--user", "is-active", unit],
            capture_output=True,
            text=True,
            timeout=2,
        )
        return result.stdout.strip() == "active"
    except Exception:
        return False


def _systemd_toggle(unit: str) -> tuple[str, bool, str]:
    """Start or stop a systemd unit. Returns (action, success, error_message)."""
    active = _systemd_is_active(unit)
    action = "stop" if active else "start"
    try:
        result = subprocess.run(
            ["systemctl", "--user", action, unit],
            capture_output=True,
            text=True,
            timeout=10,
        )
        if result.returncode == 0:
            return action, True, ""
        msg = (result.stderr.strip() or result.stdout.strip()).splitlines()[0] if (result.stderr or result.stdout) else "unknown error"
        return action, False, msg
    except Exception as e:
        return action, False, str(e)


def _check_char() -> tuple[str, dict[str, str]]:
    """Character config: reads personality_file from Godot AppState."""
    result = _send_command("state_get", section="companion")
    if result and not result.get("error"):
        pfile = result.get("personality_file", "—")
        name = Path(pfile).stem if pfile and pfile != "—" else "—"
        return "running", {"Persona": name}
    return "stopped", {"Godot": "Unreachable"}


def _check_chat() -> tuple[str, dict[str, str]]:
    return _http_check(LLM_URL)


def _check_bridge() -> tuple[str, dict[str, str]]:
    """Bridge: Redis ping."""
    try:
        import redis

        r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, socket_timeout=1)
        r.ping()
        r.close()
        return "running", {"Redis": "OK"}
    except Exception:
        return "stopped", {"Redis": "Unreachable"}


# ── specialised dev trays ─────────────────────────────────────────────────────

class BridgeDevTray(DevServiceTray):
    """Dev tray for the bridge sidecar — can restart it in-place."""

    def __init__(self) -> None:
        menu = [
            TrayMenuItem.action("Restart Bridge", self._restart_bridge),
            TrayMenuItem.quit("Quit Bridge:dev"),
        ]
        super().__init__("Chobit-Bridge", "Bridge:dev", _check_bridge, menu)

    def _restart_bridge(self) -> None:
        fr.record("tray.restart_bridge", "Bridge sidecar restarted")
        bridge_script = PROJECT_ROOT / "services" / "bridge" / "chobit_bridge.py"
        pidfile = PROJECT_ROOT / ".bridge.pid"
        if pidfile.exists():
            try:
                old_pid = int(pidfile.read_text().strip())
                os.kill(old_pid, signal.SIGTERM)
                time.sleep(0.5)
            except (ProcessLookupError, ValueError, OSError):
                pass
        proc = subprocess.Popen(
            [sys.executable, str(bridge_script)],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        pidfile.write_text(str(proc.pid))


class SpeechDevTray(DevServiceTray):
    """Dev tray for the speech service — STT, TTS, and microphone controls."""

    def __init__(self) -> None:
        menu = [
            TrayMenuItem.action("Toggle Speech", self._toggle_speech),
            TrayMenuItem.action("Toggle Microphone", self._toggle_mic),
            TrayMenuItem.quit("Quit Speech:dev"),
        ]
        super().__init__("Chobit-Speech", "Speech:dev", self._check_speech_status, menu)

    def _check_speech_status(self) -> tuple[str, dict[str, str]]:
        """Speech systemd unit state + Godot mic state."""
        active = _systemd_is_active(SPEECH_UNIT)
        labels: dict[str, str] = {"Speech": "ON" if active else "OFF"}
        result = _send_command("status")
        labels["Mic"] = "ON" if (result and result.get("mic_enabled")) else "OFF"
        return ("running" if active else "stopped"), labels

    def _toggle_speech(self) -> None:
        action, ok, err = _systemd_toggle(SPEECH_UNIT)
        if ok:
            fr.record(f"tray.speech.{action}ed", f"Speech service {action}ped", {"unit": SPEECH_UNIT})
        else:
            fr.record("tray.speech.error", f"Failed to {action} speech service", {"unit": SPEECH_UNIT, "error": err})
        self.set_status_labels(self.get_status_labels())

    def _toggle_mic(self) -> None:
        fr.record("tray.toggle_mic", "Microphone toggled")
        _send_command("toggle_mic")
        self.set_status_labels(self.get_status_labels())


class FaceDevTray(DevServiceTray):
    """Dev tray for the vision sidecar — camera, face tracking, and gaze controls."""

    def __init__(self, main_tray: ChobitTray) -> None:
        self._main_tray = main_tray
        menu = [
            TrayMenuItem.action("Camera Settings", self._open_camera_settings),
            TrayMenuItem.separator(),
            TrayMenuItem.action("Toggle Face", self._toggle_face),
            TrayMenuItem.action("Toggle Gaze", self._toggle_gaze),
            TrayMenuItem.action("Toggle Halo", self._toggle_gaze_halo),
            TrayMenuItem.quit("Quit Face:dev"),
        ]
        super().__init__("Chobit-Face", "Face:dev", self._check_face_status, menu)

    def _check_face_status(self) -> tuple[str, dict[str, str]]:
        """Vision sidecar + Godot face/halo state."""
        labels: dict[str, str] = {}

        # Vision process liveness is the authoritative camera state.
        proc = self._main_tray._vision_process
        vision_running = proc is not None and proc.poll() is None

        # Redis: supplementary liveness / last-seen timestamp.
        redis_ok = False
        try:
            import redis

            r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, socket_timeout=1)
            r.ping()
            raw = r.get("chobit:face:last_seen")
            r.close()
            redis_ok = True
            if raw is not None:
                labels["Last seen"] = raw.decode()
        except Exception:
            labels["Vision"] = "Unreachable"

        labels["Camera"] = "ON" if vision_running else "OFF"

        # Godot status: face detection and halo.
        result = _send_command("status")
        if result:
            face_detected = result.get("face_detected", False)
            labels["Face"] = "Yes" if face_detected else "No"
            labels["Halo"] = "ON" if result.get("gaze_halo") else "OFF"
            if vision_running and redis_ok:
                icon = "running" if face_detected else "warning"
            else:
                icon = "stopped"
            return icon, labels

        labels["Face"] = "—"
        labels["Halo"] = "—"
        return ("warning" if redis_ok else "stopped"), labels

    def _open_camera_settings(self) -> None:
        fr.record("tray.open_camera_settings", "Camera settings panel opened")
        _send_command("open_settings_page", page="camera")

    def _toggle_face(self) -> None:
        fr.record("tray.toggle_face", "Vision sidecar toggled via Face:dev")
        self._main_tray.toggle_camera()
        self.set_status_labels(self.get_status_labels())

    def _toggle_gaze(self) -> None:
        fr.record("tray.toggle_gaze", "Gaze mode toggled via Face:dev")
        _send_command("toggle_gaze")
        self.set_status_labels(self.get_status_labels())

    def _toggle_gaze_halo(self) -> None:
        fr.record("tray.toggle_gaze_halo", "Gaze halo toggled")
        _send_command("toggle_gaze_halo")
        self.set_status_labels(self.get_status_labels())


# ── factory ───────────────────────────────────────────────────────────────────

def _reload_persona() -> None:
    fr.record("tray.reload_persona", "Persona reload requested")
    _send_command("reload_persona")


def main() -> None:
    main_tray = create_app(dev_mode=True)
    apps: list[TrayApp] = [
        BridgeDevTray(),
        DevServiceTray(
            "Chobit-Chat",
            "Chat:dev",
            _check_chat,
        ),
        DevServiceTray(
            "Chobit-Char",
            "Char:dev",
            _check_char,
            [
                TrayMenuItem.action("Reload Persona", _reload_persona),
                TrayMenuItem.quit("Quit Char:dev"),
            ],
        ),
        SpeechDevTray(),
        FaceDevTray(main_tray),
        main_tray,
    ]
    run_all(apps)


if __name__ == "__main__":
    main()