Sanad/dashboard/routes/audio_control.py

"""Audio control endpoints — mic mute, speaker mute, device profile selection."""

from __future__ import annotations

import asyncio
import os
import subprocess
import threading

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel

from Project.Sanad.core.logger import get_logger
from Project.Sanad.voice import audio_devices as ad

log = get_logger("audio_route")

router = APIRouter()

# ─────────────────────── G1 built-in speaker (DDS) ───────────────────────
#
# pactl set-sink-mute has NO effect on the G1 built-in speaker because
# sanad_voice.py streams PCM to it via the Unitree DDS AudioClient API,
# bypassing PulseAudio entirely. To actually silence the built-in speaker
# mid-playback we must call AudioClient.SetVolume(0) over DDS.
#
# This module keeps a lazily-initialized AudioClient + a cached volume so
# the dashboard can mute/unmute without waiting on DDS init for every click.

_g1_audio_client = None
_g1_audio_lock = threading.Lock()
_g1_current_volume: int = 100  # what's actually on the hardware right now
_g1_user_volume: int = 100     # the user's preferred "unmuted" level
_g1_init_error: str = ""


def _load_persisted_g1_volume() -> int:
    """Read the saved G1 volume from data/motions/config.json.

    Keys are `audio.g1_volume` (persistent target level 0-100). Returns
    100 if no value is stored — matches the default the Unitree SDK sets
    on the voice service.
    """
    try:
        from Project.Sanad.config import load_config
        cfg = load_config() or {}
        audio = cfg.get("audio") or {}
        vol = int(audio.get("g1_volume", 100))
        return max(0, min(100, vol))
    except Exception:
        return 100


def _save_persisted_g1_volume(level: int) -> None:
    """Persist the user's volume choice to config.json so it survives restart."""
    try:
        from Project.Sanad.config import load_config, save_config
        cfg = load_config() or {}
        audio = cfg.get("audio") if isinstance(cfg.get("audio"), dict) else {}
        audio["g1_volume"] = max(0, min(100, int(level)))
        cfg["audio"] = audio
        save_config(cfg)
    except Exception as exc:
        log.warning("could not persist g1_volume: %s", exc)


# Initialize user volume from the persisted value so the dashboard shows
# the correct level on first load even if no one has touched it yet.
_g1_user_volume = _load_persisted_g1_volume()
_g1_current_volume = _g1_user_volume


def _get_g1_audio_client():
    """Lazy-init AudioClient. Safe to call from multiple routes."""
    global _g1_audio_client, _g1_init_error
    if _g1_audio_client is not None:
        return _g1_audio_client
    try:
        from unitree_sdk2py.core.channel import ChannelFactoryInitialize
        from unitree_sdk2py.g1.audio.g1_audio_client import AudioClient
    except ImportError as exc:
        _g1_init_error = f"unitree_sdk2py not installed: {exc}"
        return None
    iface = os.environ.get("SANAD_DDS_INTERFACE", "eth0")
    # ChannelFactoryInitialize can only be called once per process. The
    # arm controller normally calls it first at startup — the second call
    # either no-ops or raises, so wrap it defensively.
    try:
        ChannelFactoryInitialize(0, iface)
    except Exception as exc:
        log.debug("ChannelFactoryInitialize already called or failed: %s", exc)
    try:
        client = AudioClient()
        client.SetTimeout(5.0)
        client.Init()
        _g1_audio_client = client
        log.info("G1 AudioClient initialized for dashboard mute control (iface=%s)", iface)
        return client
    except Exception as exc:
        _g1_init_error = f"AudioClient init failed: {exc}"
        log.warning("G1 AudioClient init failed: %s", exc)
        return None


def _pactl(args: list[str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(["pactl", *args], check=True, text=True, capture_output=True)


def _get_muted(kind: str, name: str) -> bool:
    if not name:
        return False
    try:
        cmd = "get-source-mute" if kind == "source" else "get-sink-mute"
        r = _pactl([cmd, name])
        return (r.stdout or "").strip().lower().endswith("yes")
    except (FileNotFoundError, subprocess.CalledProcessError):
        return False


def _set_muted(kind: str, name: str, muted: bool) -> bool:
    if not name:
        return False
    cmd = "set-source-mute" if kind == "source" else "set-sink-mute"
    _pactl([cmd, name, "1" if muted else "0"])
    return _get_muted(kind, name)


def _current_sink_source() -> tuple[str, str]:
    cur = ad.current_selection()
    return cur.get("sink", ""), cur.get("source", "")


# ─────────────────────── status / mute ───────────────────────


@router.get("/status")
async def audio_status():
    """Return current device + mute state + G1 speaker volume.

    `speaker_muted` is the EFFECTIVE mute state — True if either the
    PulseAudio sink is muted OR the G1 built-in speaker volume is 0.
    `pulse_sink_muted` and `g1_speaker_muted` are the per-path states.
    `g1_current_volume` = what's live on the hardware.
    `g1_user_volume`    = the user's preferred unmuted level (what we
                          restore to when they un-mute).
    """
    def _do():
        sink, source = _current_sink_source()
        cur = ad.current_selection()
        pulse_muted = _get_muted("sink", sink)
        # Read cached state — avoid DDS GetVolume round-trips on every poll
        global _g1_current_volume, _g1_user_volume
        g1_muted = _g1_current_volume == 0
        return {
            "mic_muted": _get_muted("source", source),
            # Effective (OR of both paths) — the badge the user sees
            "speaker_muted": pulse_muted or g1_muted,
            # Per-path breakdown so the UI can distinguish
            "pulse_sink_muted": pulse_muted,
            "g1_speaker_muted": g1_muted,
            "g1_current_volume": _g1_current_volume,
            "g1_user_volume": _g1_user_volume,
            "g1_available": _g1_audio_client is not None or (_g1_init_error == ""),
            "sink": sink,
            "source": source,
            "current": cur,
            "pactl_available": ad.pactl_available(),
        }
    return await asyncio.to_thread(_do)


@router.post("/mic/mute")
async def toggle_mic(muted: bool | None = None):
    def _do():
        _, source = _current_sink_source()
        if not source:
            raise HTTPException(503, "No source device selected")
        target = muted if muted is not None else not _get_muted("source", source)
        try:
            actual = _set_muted("source", source, target)
        except (FileNotFoundError, subprocess.CalledProcessError) as exc:
            raise HTTPException(500, f"pactl failed: {exc}")
        return {"mic_muted": actual, "source": source}
    return await asyncio.to_thread(_do)


@router.post("/speaker/mute")
async def toggle_speaker(muted: bool | None = None):
    """Mute/unmute the SPEAKER — both the PulseAudio sink AND the G1
    built-in speaker, so the effect is audible regardless of which
    playback path is currently active (Anker PowerConf via PyAudio vs
    G1 built-in via Unitree DDS AudioClient).

    Each of the two paths is attempted independently; the response
    reports which one(s) succeeded. If either path is muted, the button
    shows as "Muted".
    """
    def _do():
        global _g1_current_volume, _g1_user_volume
        sink, _ = _current_sink_source()

        # Decide target state — if muted is None, toggle based on
        # whichever path is currently "not muted"
        if muted is None:
            pulse_cur = _get_muted("sink", sink) if sink else False
            g1_cur = _g1_current_volume == 0
            # Toggle: if anything is live, mute everything; else unmute all
            target = not (pulse_cur or g1_cur)
        else:
            target = bool(muted)

        result = {"speaker_muted": target, "pulse": None, "g1": None}

        # ── Path 1: PulseAudio sink (Anker PowerConf, USB, etc.) ──
        if sink:
            try:
                actual_pulse = _set_muted("sink", sink, target)
                result["pulse"] = {"ok": True, "muted": actual_pulse, "sink": sink}
            except (FileNotFoundError, subprocess.CalledProcessError) as exc:
                result["pulse"] = {"ok": False, "error": f"pactl failed: {exc}"}
        else:
            result["pulse"] = {"ok": False, "error": "no sink selected"}

        # ── Path 2: G1 built-in speaker via DDS AudioClient ──
        # Mute = SetVolume(0). Unmute = SetVolume(_g1_user_volume) so the
        # user's chosen level is restored (instead of always jumping back
        # to 100).
        client = _get_g1_audio_client()
        if client is None:
            result["g1"] = {"ok": False, "error": _g1_init_error or "AudioClient unavailable"}
        else:
            volume = 0 if target else _g1_user_volume
            try:
                with _g1_audio_lock:
                    code = client.SetVolume(volume)
                    _g1_current_volume = volume
                result["g1"] = {
                    "ok": True, "muted": volume == 0,
                    "volume": volume, "code": code,
                }
                log.info("G1 speaker volume set to %d (rc=%s)", volume, code)
            except Exception as exc:
                result["g1"] = {"ok": False, "error": f"SetVolume failed: {exc}"}

        # Final effective state — either path counts as muted
        pulse_muted = result["pulse"].get("muted", False) if result["pulse"] else False
        g1_muted = result["g1"].get("muted", False) if result["g1"] else False
        result["speaker_muted"] = bool(pulse_muted or g1_muted) if target else False
        result["sink"] = sink
        result["g1_current_volume"] = _g1_current_volume
        result["g1_user_volume"] = _g1_user_volume
        return result
    return await asyncio.to_thread(_do)


@router.post("/g1-speaker/mute")
async def toggle_g1_speaker_only(muted: bool | None = None):
    """Mute/unmute ONLY the G1 built-in speaker via DDS AudioClient.

    Useful for testing the DDS path in isolation — the normal
    /speaker/mute endpoint hits both PulseAudio and G1 at once.
    On unmute, restores the user's last chosen volume (not always 100).
    """
    def _do():
        global _g1_current_volume
        client = _get_g1_audio_client()
        if client is None:
            raise HTTPException(
                503,
                f"G1 AudioClient unavailable: {_g1_init_error or 'unknown'}",
            )
        if muted is None:
            target = _g1_current_volume > 0  # toggle
        else:
            target = bool(muted)
        volume = 0 if target else _g1_user_volume
        try:
            with _g1_audio_lock:
                code = client.SetVolume(volume)
                _g1_current_volume = volume
        except Exception as exc:
            raise HTTPException(500, f"SetVolume failed: {exc}")
        log.info("G1 speaker volume set to %d (rc=%s)", volume, code)
        return {
            "g1_muted": volume == 0,
            "volume": volume,
            "user_volume": _g1_user_volume,
            "return_code": code,
        }
    return await asyncio.to_thread(_do)


# ─────────────────────── G1 speaker volume (0-100) ───────────────────────


class G1VolumePayload(BaseModel):
    level: int  # 0..100


@router.get("/g1-speaker/volume")
async def get_g1_volume():
    """Return the current G1 speaker volume state.

    Response:
      {
        "available":      true,          # AudioClient available?
        "current_volume": 75,            # what's on hardware right now
        "user_volume":    75,            # user's preferred unmuted level
        "muted":          false,         # current_volume == 0
        "persisted":      75,            # value from config.json
      }
    """
    def _do():
        return {
            "available": _g1_audio_client is not None or (_g1_init_error == ""),
            "current_volume": _g1_current_volume,
            "user_volume": _g1_user_volume,
            "muted": _g1_current_volume == 0,
            "persisted": _load_persisted_g1_volume(),
            "init_error": _g1_init_error,
        }
    return await asyncio.to_thread(_do)


@router.post("/g1-speaker/volume")
async def set_g1_volume(payload: G1VolumePayload):
    """Set the G1 built-in speaker volume via DDS AudioClient.

    Body: `{"level": 0..100}`

    Effects:
      - Immediately applies to hardware via AudioClient.SetVolume(level).
      - Persists to data/motions/config.json under `audio.g1_volume` so
        it survives restarts.
      - If level > 0, updates _g1_user_volume (the "unmuted" restore
        target). level == 0 is a soft mute that preserves user_volume.
      - Takes effect on the live playback immediately — you can slide
        the volume down mid-speech and hear it get quieter.
    """
    def _do():
        global _g1_current_volume, _g1_user_volume
        level = int(payload.level)
        if not 0 <= level <= 100:
            raise HTTPException(400, "level must be 0..100")

        client = _get_g1_audio_client()
        if client is None:
            raise HTTPException(
                503,
                f"G1 AudioClient unavailable: {_g1_init_error or 'unknown'}",
            )
        try:
            with _g1_audio_lock:
                code = client.SetVolume(level)
                _g1_current_volume = level
                if level > 0:
                    # Only update the "preferred unmuted" level when the
                    # user is setting a non-zero volume. Setting 0 is a
                    # mute, which shouldn't overwrite their preference.
                    _g1_user_volume = level
        except Exception as exc:
            raise HTTPException(500, f"SetVolume failed: {exc}")

        # Persist the user's preferred level (not the current) so a
        # subsequent mute-then-restart restores to the preferred level
        _save_persisted_g1_volume(_g1_user_volume)
        log.info("G1 volume → %d (user_pref=%d, rc=%s)",
                 level, _g1_user_volume, code)
        return {
            "ok": True,
            "current_volume": level,
            "user_volume": _g1_user_volume,
            "muted": level == 0,
            "return_code": code,
            "persisted": True,
        }
    return await asyncio.to_thread(_do)


# ─────────────────────── device profiles ───────────────────────


@router.get("/devices")
async def list_devices():
    """Full device + profile listing for the dashboard picker."""
    return await asyncio.to_thread(ad.status)


@router.get("/profiles")
async def list_profiles():
    """Just the named profiles + which are currently plugged in."""
    def _do():
        from dataclasses import asdict
        detected = ad.detect_plugged_profiles() if ad.pactl_available() else []
        detected_ids = {d["profile"]["id"] for d in detected}
        return {
            "profiles": [
                {
                    **asdict(p),
                    "available": p.id in detected_ids,
                }
                for p in ad.PROFILES
            ],
            "detected_ids": list(detected_ids),
        }
    return await asyncio.to_thread(_do)


class ProfileSelect(BaseModel):
    profile_id: str


@router.post("/select-profile")
async def select_profile(payload: ProfileSelect):
    def _do():
        result = ad.select_profile(payload.profile_id)
        if not result.get("ok"):
            raise HTTPException(409, result.get("error") or "Could not select profile")
        # Best-effort: tell the audio_manager to refresh its cached state
        try:
            from Project.Sanad.main import audio_mgr
            if audio_mgr is not None and hasattr(audio_mgr, "refresh_devices"):
                audio_mgr.refresh_devices()
        except Exception:
            pass
        return result
    return await asyncio.to_thread(_do)


class ManualSelect(BaseModel):
    sink: str
    source: str


@router.post("/select-manual")
async def select_manual(payload: ManualSelect):
    def _do():
        if not payload.sink and not payload.source:
            raise HTTPException(400, "At least one of sink/source required")
        result = ad.select_manual(payload.sink, payload.source)
        if not result.get("ok"):
            raise HTTPException(500, str(result.get("errors") or "Selection failed"))
        try:
            from Project.Sanad.main import audio_mgr
            if audio_mgr is not None and hasattr(audio_mgr, "refresh_devices"):
                audio_mgr.refresh_devices()
        except Exception:
            pass
        return result
    return await asyncio.to_thread(_do)


@router.post("/refresh")
async def refresh_devices():
    """Re-scan plugged devices and re-resolve current selection."""
    return await asyncio.to_thread(ad.status)


@router.post("/apply")
async def apply_audio():
    """Re-scan all USB ports, resolve the best profile, and set pactl defaults.

    Use this after plugging/unplugging devices or switching USB ports.
    """
    def _do():
        result = ad.apply_current_selection()
        # Also refresh AudioManager so it picks up the new sink/source
        try:
            from Project.Sanad.main import audio_mgr
            if audio_mgr is not None:
                audio_mgr.refresh_devices()
        except Exception:
            pass
        return result
    return await asyncio.to_thread(_do)