"""Volume service for host system volume control.""" import platform from app.core.logging import get_logger logger = get_logger(__name__) # Constants MIN_VOLUME = 0 MAX_VOLUME = 100 class VolumeService: """Service for controlling host system volume.""" def __init__(self) -> None: """Initialize volume service.""" self._system = platform.system().lower() self._pycaw_available = False self._pulsectl_available = False # Try to import Windows volume control if self._system == "windows": try: from comtypes import ( # noqa: PLC0415 CLSCTX_ALL, # type: ignore[import-untyped] ) from pycaw.pycaw import ( # type: ignore[import-untyped] # noqa: PLC0415 AudioUtilities, IAudioEndpointVolume, ) self._AudioUtilities = AudioUtilities self._IAudioEndpointVolume = IAudioEndpointVolume self._CLSCTX_ALL = CLSCTX_ALL self._pycaw_available = True logger.info("Windows volume control (pycaw) initialized") except ImportError as e: logger.warning("pycaw not available: %s", e) # Try to import Linux volume control elif self._system == "linux": try: import pulsectl # type: ignore[import-untyped] # noqa: PLC0415 self._pulsectl = pulsectl self._pulsectl_available = True logger.info("Linux volume control (pulsectl) initialized") except ImportError as e: logger.warning("pulsectl not available: %s", e) else: logger.warning("Volume control not supported on %s", self._system) def get_volume(self) -> int | None: """Get the current system volume as a percentage (0-100). Returns: Volume level as percentage, or None if not available """ try: if self._system == "windows" and self._pycaw_available: return self._get_windows_volume() if self._system == "linux" and self._pulsectl_available: return self._get_linux_volume() except Exception: logger.exception("Failed to get volume") return None else: logger.warning("Volume control not available for this system") return None def set_volume(self, volume: int) -> bool: """Set the system volume to a percentage (0-100). Args: volume: Volume level as percentage (0-100) Returns: True if successful, False otherwise """ if not (MIN_VOLUME <= volume <= MAX_VOLUME): logger.error( "Volume must be between %s and %s, got %s", MIN_VOLUME, MAX_VOLUME, volume, ) return False try: if self._system == "windows" and self._pycaw_available: return self._set_windows_volume(volume) if self._system == "linux" and self._pulsectl_available: return self._set_linux_volume(volume) except Exception: logger.exception("Failed to set volume") return False else: logger.warning("Volume control not available for this system") return False def is_muted(self) -> bool | None: """Check if the system is muted. Returns: True if muted, False if not muted, None if not available """ try: if self._system == "windows" and self._pycaw_available: return self._get_windows_mute_status() if self._system == "linux" and self._pulsectl_available: return self._get_linux_mute_status() except Exception: logger.exception("Failed to get mute status") return None else: logger.warning("Mute status not available for this system") return None def set_mute(self, *, muted: bool) -> bool: """Set the system mute status. Args: muted: True to mute, False to unmute Returns: True if successful, False otherwise """ try: if self._system == "windows" and self._pycaw_available: return self._set_windows_mute(muted=muted) if self._system == "linux" and self._pulsectl_available: return self._set_linux_mute(muted=muted) except Exception: logger.exception("Failed to set mute status") return False else: logger.warning("Mute control not available for this system") return False def _get_windows_volume(self) -> int: """Get Windows volume using pycaw.""" devices = self._AudioUtilities.GetSpeakers() interface = devices.Activate( self._IAudioEndpointVolume._iid_, self._CLSCTX_ALL, None, ) volume = interface.QueryInterface(self._IAudioEndpointVolume) current_volume = volume.GetMasterVolume() # Convert from scalar (0.0-1.0) to percentage (0-100) return int(current_volume * MAX_VOLUME) def _set_windows_volume(self, volume_percent: int) -> bool: """Set Windows volume using pycaw.""" devices = self._AudioUtilities.GetSpeakers() interface = devices.Activate( self._IAudioEndpointVolume._iid_, self._CLSCTX_ALL, None, ) volume = interface.QueryInterface(self._IAudioEndpointVolume) # Convert from percentage (0-100) to scalar (0.0-1.0) volume_scalar = volume_percent / MAX_VOLUME volume.SetMasterVolume(volume_scalar, None) logger.info("Windows volume set to %s%%", volume_percent) return True def _get_windows_mute_status(self) -> bool: """Get Windows mute status using pycaw.""" devices = self._AudioUtilities.GetSpeakers() interface = devices.Activate( self._IAudioEndpointVolume._iid_, self._CLSCTX_ALL, None, ) volume = interface.QueryInterface(self._IAudioEndpointVolume) return bool(volume.GetMute()) def _set_windows_mute(self, *, muted: bool) -> bool: """Set Windows mute status using pycaw.""" devices = self._AudioUtilities.GetSpeakers() interface = devices.Activate( self._IAudioEndpointVolume._iid_, self._CLSCTX_ALL, None, ) volume = interface.QueryInterface(self._IAudioEndpointVolume) volume.SetMute(muted, None) logger.info("Windows mute set to %s", muted) return True def _get_linux_volume(self) -> int: """Get Linux volume using pulsectl.""" with self._pulsectl.Pulse("volume-service") as pulse: # Get the default sink (output device) default_sink = pulse.get_sink_by_name(pulse.server_info().default_sink_name) if default_sink is None: logger.error("No default audio sink found") return MIN_VOLUME # Get volume as percentage (PulseAudio uses 0.0-1.0, we convert to 0-100) volume = default_sink.volume avg_volume = sum(volume.values) / len(volume.values) return int(avg_volume * MAX_VOLUME) def _set_linux_volume(self, volume_percent: int) -> bool: """Set Linux volume using pulsectl.""" with self._pulsectl.Pulse("volume-service") as pulse: # Get the default sink (output device) default_sink = pulse.get_sink_by_name(pulse.server_info().default_sink_name) if default_sink is None: logger.error("No default audio sink found") return False # Convert percentage to PulseAudio volume (0.0-1.0) volume_scalar = volume_percent / MAX_VOLUME # Set volume for all channels pulse.volume_set_all_chans(default_sink, volume_scalar) logger.info("Linux volume set to %s%%", volume_percent) return True def _get_linux_mute_status(self) -> bool: """Get Linux mute status using pulsectl.""" with self._pulsectl.Pulse("volume-service") as pulse: # Get the default sink (output device) default_sink = pulse.get_sink_by_name(pulse.server_info().default_sink_name) if default_sink is None: logger.error("No default audio sink found") return False return bool(default_sink.mute) def _set_linux_mute(self, *, muted: bool) -> bool: """Set Linux mute status using pulsectl.""" with self._pulsectl.Pulse("volume-service") as pulse: # Get the default sink (output device) default_sink = pulse.get_sink_by_name(pulse.server_info().default_sink_name) if default_sink is None: logger.error("No default audio sink found") return False # Set mute status pulse.mute(default_sink, muted) logger.info("Linux mute set to %s", muted) return True # Global volume service instance volume_service = VolumeService()