feat: Implement host system volume control and update player service to use it

app/services/volume.py

@@ -0,0 +1,251 @@
+"""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()