"""Gestion audio système (ALSA/PulseAudio).""" import subprocess import os import tempfile import threading import time from typing import Optional import numpy as np from .logger import setup_logger class AudioOutput: """Gestionnaire de sortie audio système.""" def __init__(self): """Initialise le gestionnaire audio.""" self.logger = setup_logger(__name__) self.audio_system = self._detect_audio_system() self._lock = threading.Lock() def _detect_audio_system(self) -> str: """Détecte le système audio disponible.""" # Teste PulseAudio try: result = subprocess.run( ["pulseaudio", "--check"], capture_output=True, timeout=2 ) if result.returncode == 0: self.logger.info("PulseAudio détecté") return "pulse" except (subprocess.TimeoutExpired, FileNotFoundError): pass # Teste ALSA try: result = subprocess.run( ["aplay", "--version"], capture_output=True, timeout=2 ) if result.returncode == 0: self.logger.info("ALSA détecté") return "alsa" except (subprocess.TimeoutExpired, FileNotFoundError): pass self.logger.warning("Aucun système audio détecté") return "none" def play_wav_data(self, wav_data: bytes, blocking: bool = True) -> bool: """Joue des données WAV.""" if self.audio_system == "none": self.logger.warning("Pas de système audio - simulation lecture") return True with self._lock: return self._play_wav_internal(wav_data, blocking) def _play_wav_internal(self, wav_data: bytes, blocking: bool) -> bool: """Lecture WAV interne.""" with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp_file: try: tmp_file.write(wav_data) tmp_file.flush() if self.audio_system == "pulse": cmd = ["paplay", tmp_file.name] elif self.audio_system == "alsa": cmd = ["aplay", tmp_file.name] else: return False if blocking: result = subprocess.run(cmd, capture_output=True, timeout=30) success = result.returncode == 0 if not success: self.logger.error(f"Erreur lecture audio: {result.stderr}") return success else: # Lecture non-bloquante subprocess.Popen( cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) return True except subprocess.TimeoutExpired: self.logger.error("Timeout lecture audio") return False except Exception as e: self.logger.error(f"Erreur lecture WAV: {e}") return False finally: try: os.unlink(tmp_file.name) except OSError: pass def play_wav_file(self, file_path: str, blocking: bool = True) -> bool: """Joue un fichier WAV.""" if not os.path.exists(file_path): self.logger.error(f"Fichier WAV introuvable: {file_path}") return False try: with open(file_path, "rb") as f: wav_data = f.read() return self.play_wav_data(wav_data, blocking) except Exception as e: self.logger.error(f"Erreur lecture fichier WAV: {e}") return False def get_audio_devices(self) -> list: """Liste les périphériques audio disponibles.""" devices = [] if self.audio_system == "pulse": try: result = subprocess.run( ["pactl", "list", "short", "sinks"], capture_output=True, text=True, timeout=5, ) if result.returncode == 0: for line in result.stdout.strip().split("\n"): if line: parts = line.split("\t") if len(parts) >= 2: devices.append( { "id": parts[1], "name": parts[1], "system": "pulse", } ) except Exception as e: self.logger.warning(f"Erreur listage devices PulseAudio: {e}") elif self.audio_system == "alsa": try: result = subprocess.run( ["aplay", "-l"], capture_output=True, text=True, timeout=5 ) if result.returncode == 0: # Parse basique de la sortie aplay -l for line in result.stdout.split("\n"): if "card" in line and ":" in line: devices.append( { "id": line.strip(), "name": line.strip(), "system": "alsa", } ) except Exception as e: self.logger.warning(f"Erreur listage devices ALSA: {e}") return devices def set_volume(self, volume: float) -> bool: """Définit le volume système (0.0 à 1.0).""" volume = max(0.0, min(1.0, volume)) if self.audio_system == "pulse": try: # Volume PulseAudio en pourcentage volume_percent = int(volume * 100) result = subprocess.run( [ "pactl", "set-sink-volume", "@DEFAULT_SINK@", f"{volume_percent}%", ], capture_output=True, timeout=5, ) return result.returncode == 0 except Exception as e: self.logger.error(f"Erreur réglage volume PulseAudio: {e}") elif self.audio_system == "alsa": try: # Volume ALSA en pourcentage volume_percent = int(volume * 100) result = subprocess.run( ["amixer", "set", "Master", f"{volume_percent}%"], capture_output=True, timeout=5, ) return result.returncode == 0 except Exception as e: self.logger.error(f"Erreur réglage volume ALSA: {e}") return False def get_volume(self) -> Optional[float]: """Récupère le volume actuel.""" if self.audio_system == "pulse": try: result = subprocess.run( ["pactl", "get-sink-volume", "@DEFAULT_SINK@"], capture_output=True, text=True, timeout=5, ) if result.returncode == 0: # Parse basique pour extraire le pourcentage output = result.stdout if "%" in output: # Cherche le premier pourcentage for part in output.split(): if "%" in part: percent_str = part.replace("%", "") return float(percent_str) / 100.0 except Exception as e: self.logger.error(f"Erreur lecture volume PulseAudio: {e}") elif self.audio_system == "alsa": try: result = subprocess.run( ["amixer", "get", "Master"], capture_output=True, text=True, timeout=5, ) if result.returncode == 0: # Parse output amixer for line in result.stdout.split("\n"): if "[" in line and "%" in line: # Extrait pourcentage entre crochets start = line.find("[") + 1 end = line.find("%", start) if start > 0 and end > start: percent_str = line[start:end] return float(percent_str) / 100.0 except Exception as e: self.logger.error(f"Erreur lecture volume ALSA: {e}") return None