# -*- coding: utf-8 -*- """ ALSA Volume Handler. Steuert die System-Lautstaerke ueber amixer fuer Playback (Lautsprecher) und Capture (Mikrofon). Reagiert auf volume_change und mute_toggle Events von HID-Geraeten oder anderen Quellen. Laeuft auf Client und Standalone — regelt die Hardware-Lautstaerke fuer alle Audio-Ein-/Ausgaben (TTS, Musik, Sounds, Wakeword-Erkennung). Quellen-Konzept: Aenderungen kennen eine "source" (z.B. "user", "duck", "hid"). Nur Aenderungen mit source="user" oder "hid" werden in client_config.json persistiert. Ducking (source="duck") aendert nur die ALSA-Lautstaerke, nicht die gespeicherten Werte. """ from __future__ import annotations import asyncio import shutil from typing import TYPE_CHECKING from trixy_core.utils.debug import pinfo, pdebug, perror if TYPE_CHECKING: from trixy_core.application import BaseApplication # Quellen die in die Config persistiert werden PERSIST_SOURCES = {"user", "hid"} class _MixerChannel: """Ein einzelner ALSA-Mixer-Kanal (Playback ODER Capture).""" def __init__( self, kind: str, control: str = "", capture: bool = False, ) -> None: self.kind = kind # "output" | "input" self.control = control self.capture = capture self.available = False self.muted = False async def detect(self) -> None: """Erkennt einen passenden Mixer-Control falls keiner gesetzt ist.""" if self.control: self.available = await _control_exists(self.control) return if self.capture: candidates = ["Capture", "Mic", "Microphone", "Front Mic", "Internal Mic"] else: candidates = ["Master", "PCM", "Speaker", "Headphone"] output = await _amixer("scontrols") for name in candidates: if f"'{name}'" in output: self.control = name self.available = True pdebug(f"ALSA {self.kind} Control erkannt: {name}") return # Fallback: ersten passenden Control nehmen for line in output.splitlines(): if "Simple mixer control" in line: start = line.index("'") + 1 end = line.index("'", start) name = line[start:end] self.control = name self.available = True pdebug(f"ALSA {self.kind} Control (Fallback): {name}") return async def get_volume(self) -> int: if not self.available: return 0 try: output = await _amixer("get", self.control) for line in output.splitlines(): if "%" in line: start = line.index("[") + 1 end = line.index("%", start) return int(line[start:end]) except Exception: pass return 0 async def set_volume(self, percent: int) -> None: if not self.available: return percent = max(0, min(100, percent)) await _amixer("set", self.control, f"{percent}%") pdebug(f"ALSA {self.kind} Volume: {percent}%") async def set_mute(self, muted: bool) -> None: if not self.available: return await _amixer("set", self.control, "mute" if muted else "unmute") self.muted = muted pdebug(f"ALSA {self.kind} {'MUTE' if muted else 'UNMUTE'}") async def read_mute(self) -> bool: if not self.available: return False output = await _amixer("get", self.control) self.muted = "[off]" in output return self.muted async def _amixer(*args: str) -> str: """Fuehrt einen amixer-Befehl aus.""" try: proc = await asyncio.create_subprocess_exec( "amixer", *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL, ) stdout, _ = await proc.communicate() return stdout.decode() except Exception as e: perror(f"amixer Fehler: {e}") return "" async def _control_exists(name: str) -> bool: output = await _amixer("scontrols") return f"'{name}'" in output class VolumeHandler: """ ALSA Volume Handler fuer Playback und Capture. Persistiert Aenderungen mit source in PERSIST_SOURCES in client_config.json. Ducking-Aenderungen (source="duck") werden NICHT persistiert. """ def __init__( self, application: "BaseApplication", output_control: str = "", input_control: str = "", config_name: str = "client", ) -> None: self._application = application self._output = _MixerChannel("output", output_control, capture=False) self._input = _MixerChannel("input", input_control, capture=True) self._available = False self._config_name = config_name @property def available(self) -> bool: return self._available @property def output(self) -> _MixerChannel: return self._output @property def input(self) -> _MixerChannel: return self._input async def start(self) -> None: """Startet den Volume Handler und wendet Config-Werte an.""" if not shutil.which("amixer"): pdebug("amixer nicht gefunden — Volume Handler deaktiviert") return await self._output.detect() await self._input.detect() if not (self._output.available or self._input.available): pdebug("Kein ALSA Mixer Control gefunden") return self._available = True # Config-Werte beim Start anwenden await self._apply_config() events = getattr(self._application, "events", None) if events: events.register("volume_change", self._on_volume_change) events.register("mute_toggle", self._on_mute_toggle) events.register("audio_volume_set", self._on_volume_set) events.register("audio_mute_set", self._on_mute_set) out_v = await self._output.get_volume() in_v = await self._input.get_volume() pinfo( f"Volume Handler gestartet — " f"Out: {self._output.control} ({out_v}%{', MUTE' if self._output.muted else ''}) " f"In: {self._input.control} ({in_v}%{', MUTE' if self._input.muted else ''})" ) async def stop(self) -> None: events = getattr(self._application, "events", None) if events: events.unregister("volume_change", self._on_volume_change) events.unregister("mute_toggle", self._on_mute_toggle) events.unregister("audio_volume_set", self._on_volume_set) events.unregister("audio_mute_set", self._on_mute_set) # === Config-Anbindung === def _audio_config(self): cfg = getattr(self._application, "_client_config", None) if cfg is None: cfg = getattr(self._application, "client_config", None) return getattr(cfg, "audio", None) if cfg else None async def _apply_config(self) -> None: """Wendet die in client_config.json gespeicherten Werte auf ALSA an.""" ac = self._audio_config() if ac is None: return if self._output.available: await self._output.set_volume(int(ac.output_volume)) await self._output.set_mute(bool(ac.output_muted)) if self._input.available: await self._input.set_volume(int(ac.input_volume)) await self._input.set_mute(bool(ac.input_muted)) def _persist(self, **changes) -> None: """Schreibt Aenderungen in client_config.json.""" ac = self._audio_config() if ac is None: return for key, value in changes.items(): if hasattr(ac, key): setattr(ac, key, value) cm = getattr(self._application, "config_manager", None) if cm is None: return try: cm.save(self._config_name) except Exception as e: perror(f"Volume-Persistenz fehlgeschlagen: {e}") # === Public API === async def set_output_volume(self, percent: int, source: str = "user") -> None: if not self._output.available: return await self._output.set_volume(percent) if source in PERSIST_SOURCES: self._persist(output_volume=max(0, min(100, percent))) await self._emit_changed("output", percent, source) async def set_input_volume(self, percent: int, source: str = "user") -> None: if not self._input.available: return await self._input.set_volume(percent) if source in PERSIST_SOURCES: self._persist(input_volume=max(0, min(100, percent))) await self._emit_changed("input", percent, source) async def set_output_mute(self, muted: bool, source: str = "user") -> None: if not self._output.available: return await self._output.set_mute(muted) if source in PERSIST_SOURCES: self._persist(output_muted=muted) async def set_input_mute(self, muted: bool, source: str = "user") -> None: if not self._input.available: return await self._input.set_mute(muted) if source in PERSIST_SOURCES: self._persist(input_muted=muted) async def restore_from_config(self, channel: str = "output") -> None: """Stellt die in der Config gespeicherte Lautstaerke wieder her. Wird nach Ducking aufgerufen — nutzt config-Werte statt temporaerer Variablen (solider). """ ac = self._audio_config() if ac is None: return if channel in ("output", "all") and self._output.available: await self._output.set_volume(int(ac.output_volume)) await self._output.set_mute(bool(ac.output_muted)) if channel in ("input", "all") and self._input.available: await self._input.set_volume(int(ac.input_volume)) await self._input.set_mute(bool(ac.input_muted)) async def get_status(self) -> dict: """Gibt aktuellen Status zurueck — fuer F6 Audio View.""" return { "output": { "available": self._output.available, "control": self._output.control, "volume": await self._output.get_volume() if self._output.available else 0, "muted": await self._output.read_mute() if self._output.available else False, }, "input": { "available": self._input.available, "control": self._input.control, "volume": await self._input.get_volume() if self._input.available else 0, "muted": await self._input.read_mute() if self._input.available else False, }, } # === Event-Handler === async def _on_volume_change(self, event_name: str, data: dict) -> None: """Handler fuer relative Aenderungen (HID Hotkeys).""" direction = data.get("direction", "") step = data.get("step", 0.05) target = data.get("target", "output") step_pct = max(1, int(step * 100)) channel = self._output if target == "output" else self._input if not channel.available: return current = await channel.get_volume() new_value = current + step_pct if direction == "up" else current - step_pct if target == "output": await self.set_output_volume(new_value, source="hid") else: await self.set_input_volume(new_value, source="hid") async def _on_mute_toggle(self, event_name: str, data: dict) -> None: mute_type = data.get("type", "speaker") if mute_type == "speaker": await self.set_output_mute(not self._output.muted, source="hid") elif mute_type in ("mic", "microphone"): await self.set_input_mute(not self._input.muted, source="hid") async def _on_volume_set(self, event_name: str, data: dict) -> None: """Handler fuer absolute Volume-Aenderungen (z.B. F6 Audio View, Ducking).""" target = data.get("target", "output") percent = int(data.get("volume", 0)) source = data.get("source", "user") if target == "output": await self.set_output_volume(percent, source=source) else: await self.set_input_volume(percent, source=source) async def _on_mute_set(self, event_name: str, data: dict) -> None: target = data.get("target", "output") muted = bool(data.get("muted", False)) source = data.get("source", "user") if target == "output": await self.set_output_mute(muted, source=source) else: await self.set_input_mute(muted, source=source) async def _emit_changed(self, target: str, percent: int, source: str) -> None: events = getattr(self._application, "events", None) if events: await events.emit("audio_level_changed", { "target": target, "volume": percent, "source": source, }) # === Legacy-Kompatibilitaet === @property def mixer_control(self) -> str: return self._output.control @property def is_muted(self) -> bool: return self._output.muted async def get_volume(self) -> int: return await self._output.get_volume() async def set_volume(self, percent: int) -> None: await self.set_output_volume(percent, source="user") async def volume_up(self, step_percent: int = 5) -> None: current = await self._output.get_volume() await self.set_output_volume(current + step_percent, source="hid") async def volume_down(self, step_percent: int = 5) -> None: current = await self._output.get_volume() await self.set_output_volume(current - step_percent, source="hid") async def mute(self) -> None: await self.set_output_mute(True, source="user") async def unmute(self) -> None: await self.set_output_mute(False, source="user") async def toggle_mute(self) -> None: await self.set_output_mute(not self._output.muted, source="user")