Add cross-platform system volume control with linear amplification mapping

- New mediahive/volume_control.py module controls master volume on Windows (IAudioEndpointVolume via ctypes), macOS (osascript), and Linux/PipeWire (wpctl). - UI slider x in [0, 1.5] maps to linear amp via: amp = 0.02*x if x < 0.1 amp = (exp(6*x)-1)/402.42879349 otherwise - PipeWire supports >100% amplification natively (up to x=1.5). - Windows and macOS clamp at x=1.0 (100%) since their system mixers do not support amplification above unity. - Gamepad D-pad up/down now controls system volume in 0.01 steps. - JsApi exposes set_volume(), get_volume(), and volume_max() for the frontend to query platform limits and control volume directly. - No external dependencies: uses ctypes on Windows, subprocess on Linux/macOS. No PulseAudio fallback. - Add ruff ignore rules for ctypes COM boilerplate patterns.
2026-05-29 18:21:18 +00:00
parent cac3348ec1
commit 359ca0c5b4
3 changed files with 508 additions and 4 deletions
@@ -0,0 +1,475 @@
+"""Cross-platform master volume control with linear amplification mapping.
+
+The UI slider produces a value ``x`` in the range ``[0, 1.5]``.  This module
+converts ``x`` to a linear amplitude coefficient ``amp`` using:
+
+    amp = 0.02 * x                 if x < 0.1
+    amp = (exp(6 * x) - 1) / 402.42879349   otherwise
+
+``amp`` is then applied to the system mixer in a platform-specific way:
+
+* **Linux / PipeWire** – ``wpctl`` accepts linear values > 1.0 natively, so
+  amplification up to ~150 % (and beyond) works without clipping.
+* **Windows** – ``IAudioEndpointVolume`` uses an audio-tapered scalar and does
+  not support > 100 %.  ``amp`` is clamped to 1.0 and mapped to the scalar.
+* **macOS** – ``osascript`` sets a 0-100 slider and does not support > 100 %.
+  ``amp`` is clamped to 1.0 and mapped to that range.
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+import subprocess
+import sys
+import threading
+from collections.abc import Callable
+
+logger = logging.getLogger("mediahive.volume")
+
+# ---------------------------------------------------------------------------
+# Linear amplitude formula
+# ---------------------------------------------------------------------------
+
+_AMP_DENOMINATOR = 402.42879349
+
+
+def slider_to_amp(x: float) -> float:
+    """Convert UI slider position ``x`` (0.0 .. 1.5) to linear amplitude.
+
+    Returns a value in the range ``[0, ~20]`` where ``1.0`` means unity gain.
+    """
+    if x <= 0.0:
+        return 0.0
+    if x < 0.1:
+        return 0.02 * x
+    return (math.exp(6.0 * x) - 1.0) / _AMP_DENOMINATOR
+
+
+def amp_to_slider(amp: float) -> float:
+    """Inverse of :func:`slider_to_amp` for the ``x >= 0.1`` branch.
+
+    Returns a slider position in ``[0.1, 1.5]`` (or slightly above when the
+    platform reports amplification > 1.0).
+    """
+    if amp <= 0.0:
+        return 0.0
+    # For very small values we fall back to the linear branch
+    if amp < 0.002042892590415348:  # value at x == 0.1
+        return amp / 0.02
+    return math.log(amp * _AMP_DENOMINATOR + 1.0) / 6.0
+
+
+# ---------------------------------------------------------------------------
+# Platform backends
+# ---------------------------------------------------------------------------
+
+
+def _set_volume_windows(amp: float) -> None:
+    """Set master volume on Windows via ``IAudioEndpointVolume`` (ctypes)."""
+    import ctypes
+    import uuid
+    from ctypes import (
+        POINTER,
+        Structure,
+        byref,
+        c_float,
+        c_uint32,
+        c_void_p,
+        cast,
+        wintypes,
+    )
+
+    class GUID(Structure):
+        _fields_ = [
+            ("Data1", wintypes.DWORD),
+            ("Data2", wintypes.WORD),
+            ("Data3", wintypes.WORD),
+            ("Data4", wintypes.BYTE * 8),
+        ]
+
+    def _uuid_to_guid(u: uuid.UUID) -> GUID:
+        g = GUID()
+        g.Data1 = u.time_low
+        g.Data2 = u.time_mid
+        g.Data3 = u.time_hi_version
+        g.Data4[0] = u.clock_seq_hi_variant & 0xFF
+        g.Data4[1] = u.clock_seq_low & 0xFF
+        for i in range(6):
+            g.Data4[2 + i] = (u.node >> (40 - i * 8)) & 0xFF
+        return g
+
+    CLSID_MMDeviceEnumerator = _uuid_to_guid(
+        uuid.UUID("{BCDE0395-E52F-467C-8E3D-C4579291692E}")
+    )
+    IID_IMMDeviceEnumerator = _uuid_to_guid(
+        uuid.UUID("{A95664D2-9614-4F35-A746-DE8DB63617E6}")
+    )
+    IID_IAudioEndpointVolume = _uuid_to_guid(
+        uuid.UUID("{5CDF2C82-841E-4546-9722-0CF74078229A}")
+    )
+    IID_IMMDevice = _uuid_to_guid(uuid.UUID("{D666063F-1587-4E43-81F1-B948E807363F}"))
+
+    CLSCTX_ALL = 23
+
+    ole32 = ctypes.windll.ole32
+    ole32.CoInitializeEx(None, 0)
+
+    CoCreateInstance = ole32.CoCreateInstance
+    CoCreateInstance.argtypes = [
+        POINTER(GUID),
+        c_void_p,
+        wintypes.DWORD,
+        POINTER(GUID),
+        POINTER(c_void_p),
+    ]
+    CoCreateInstance.restype = wintypes.HRESULT
+
+    enumerator_ptr = c_void_p()
+    hr = CoCreateInstance(
+        byref(CLSID_MMDeviceEnumerator),
+        None,
+        CLSCTX_ALL,
+        byref(IID_IMMDeviceEnumerator),
+        byref(enumerator_ptr),
+    )
+    if hr != 0:
+        raise OSError(f"CoCreateInstance failed: {hr:#x}")
+
+    try:
+        vtable_ptr = cast(enumerator_ptr, POINTER(c_void_p)).contents
+        vtable = cast(vtable_ptr, POINTER(c_void_p * 6))
+        GetDefaultAudioEndpoint = ctypes.WINFUNCTYPE(
+            wintypes.HRESULT, c_void_p, c_uint32, c_uint32, POINTER(c_void_p)
+        )(vtable[0][4])
+
+        device_ptr = c_void_p()
+        hr = GetDefaultAudioEndpoint(enumerator_ptr, 0, 0, byref(device_ptr))
+        if hr != 0:
+            raise OSError(f"GetDefaultAudioEndpoint failed: {hr:#x}")
+
+        try:
+            device_vtable_ptr = cast(device_ptr, POINTER(c_void_p)).contents
+            device_vtable = cast(device_vtable_ptr, POINTER(c_void_p * 7))
+            Activate = ctypes.WINFUNCTYPE(
+                wintypes.HRESULT,
+                c_void_p,
+                POINTER(GUID),
+                wintypes.DWORD,
+                c_void_p,
+                POINTER(c_void_p),
+            )(device_vtable[0][3])
+
+            volume_ptr = c_void_p()
+            hr = Activate(
+                device_ptr,
+                byref(IID_IAudioEndpointVolume),
+                CLSCTX_ALL,
+                None,
+                byref(volume_ptr),
+            )
+            if hr != 0:
+                raise OSError(f"Activate(IAudioEndpointVolume) failed: {hr:#x}")
+
+            try:
+                vol_vtable_ptr = cast(volume_ptr, POINTER(c_void_p)).contents
+                vol_vtable = cast(vol_vtable_ptr, POINTER(c_void_p * 22))
+
+                GetVolumeRange = ctypes.WINFUNCTYPE(
+                    wintypes.HRESULT,
+                    c_void_p,
+                    POINTER(c_float),
+                    POINTER(c_float),
+                    POINTER(c_float),
+                )(vol_vtable[0][20])
+
+                min_db = c_float()
+                max_db = c_float()
+                step_db = c_float()
+                hr = GetVolumeRange(
+                    volume_ptr, byref(min_db), byref(max_db), byref(step_db)
+                )
+                if hr != 0:
+                    raise OSError(f"GetVolumeRange failed: {hr:#x}")
+
+                SetMasterVolumeLevel = ctypes.WINFUNCTYPE(
+                    wintypes.HRESULT, c_void_p, c_float, POINTER(GUID)
+                )(vol_vtable[0][6])
+                SetMasterVolumeLevelScalar = ctypes.WINFUNCTYPE(
+                    wintypes.HRESULT, c_void_p, c_float, POINTER(GUID)
+                )(vol_vtable[0][7])
+
+                # Windows does not natively support >100% system volume.
+                # Clamp to unity gain and map to the audio-tapered scalar.
+                # The exponent 0.573 was empirically derived from measurements
+                # on a reference Windows system.
+                clamped = min(amp, 1.0)
+                scalar = clamped**0.573
+                SetMasterVolumeLevelScalar(volume_ptr, scalar, None)
+            finally:
+                Release = ctypes.WINFUNCTYPE(wintypes.ULONG, c_void_p)(vol_vtable[0][2])
+                Release(volume_ptr)
+        finally:
+            Release = ctypes.WINFUNCTYPE(wintypes.ULONG, c_void_p)(device_vtable[0][2])
+            Release(device_ptr)
+    finally:
+        Release = ctypes.WINFUNCTYPE(wintypes.ULONG, c_void_p)(vtable[0][2])
+        Release(enumerator_ptr)
+
+
+def _get_volume_windows() -> float:
+    """Return current linear amplitude on Windows."""
+    import ctypes
+    import uuid
+    from ctypes import (
+        POINTER,
+        Structure,
+        byref,
+        c_float,
+        c_uint32,
+        c_void_p,
+        cast,
+        wintypes,
+    )
+
+    class GUID(Structure):
+        _fields_ = [
+            ("Data1", wintypes.DWORD),
+            ("Data2", wintypes.WORD),
+            ("Data3", wintypes.WORD),
+            ("Data4", wintypes.BYTE * 8),
+        ]
+
+    def _uuid_to_guid(u: uuid.UUID) -> GUID:
+        g = GUID()
+        g.Data1 = u.time_low
+        g.Data2 = u.time_mid
+        g.Data3 = u.time_hi_version
+        g.Data4[0] = u.clock_seq_hi_variant & 0xFF
+        g.Data4[1] = u.clock_seq_low & 0xFF
+        for i in range(6):
+            g.Data4[2 + i] = (u.node >> (40 - i * 8)) & 0xFF
+        return g
+
+    CLSID_MMDeviceEnumerator = _uuid_to_guid(
+        uuid.UUID("{BCDE0395-E52F-467C-8E3D-C4579291692E}")
+    )
+    IID_IMMDeviceEnumerator = _uuid_to_guid(
+        uuid.UUID("{A95664D2-9614-4F35-A746-DE8DB63617E6}")
+    )
+    IID_IAudioEndpointVolume = _uuid_to_guid(
+        uuid.UUID("{5CDF2C82-841E-4546-9722-0CF74078229A}")
+    )
+    IID_IMMDevice = _uuid_to_guid(uuid.UUID("{D666063F-1587-4E43-81F1-B948E807363F}"))
+
+    CLSCTX_ALL = 23
+
+    ole32 = ctypes.windll.ole32
+    ole32.CoInitializeEx(None, 0)
+
+    CoCreateInstance = ole32.CoCreateInstance
+    CoCreateInstance.argtypes = [
+        POINTER(GUID),
+        c_void_p,
+        wintypes.DWORD,
+        POINTER(GUID),
+        POINTER(c_void_p),
+    ]
+    CoCreateInstance.restype = wintypes.HRESULT
+
+    enumerator_ptr = c_void_p()
+    hr = CoCreateInstance(
+        byref(CLSID_MMDeviceEnumerator),
+        None,
+        CLSCTX_ALL,
+        byref(IID_IMMDeviceEnumerator),
+        byref(enumerator_ptr),
+    )
+    if hr != 0:
+        raise OSError(f"CoCreateInstance failed: {hr:#x}")
+
+    try:
+        vtable_ptr = cast(enumerator_ptr, POINTER(c_void_p)).contents
+        vtable = cast(vtable_ptr, POINTER(c_void_p * 6))
+        GetDefaultAudioEndpoint = ctypes.WINFUNCTYPE(
+            wintypes.HRESULT, c_void_p, c_uint32, c_uint32, POINTER(c_void_p)
+        )(vtable[0][4])
+
+        device_ptr = c_void_p()
+        hr = GetDefaultAudioEndpoint(enumerator_ptr, 0, 0, byref(device_ptr))
+        if hr != 0:
+            raise OSError(f"GetDefaultAudioEndpoint failed: {hr:#x}")
+
+        try:
+            device_vtable_ptr = cast(device_ptr, POINTER(c_void_p)).contents
+            device_vtable = cast(device_vtable_ptr, POINTER(c_void_p * 7))
+            Activate = ctypes.WINFUNCTYPE(
+                wintypes.HRESULT,
+                c_void_p,
+                POINTER(GUID),
+                wintypes.DWORD,
+                c_void_p,
+                POINTER(c_void_p),
+            )(device_vtable[0][3])
+
+            volume_ptr = c_void_p()
+            hr = Activate(
+                device_ptr,
+                byref(IID_IAudioEndpointVolume),
+                CLSCTX_ALL,
+                None,
+                byref(volume_ptr),
+            )
+            if hr != 0:
+                raise OSError(f"Activate failed: {hr:#x}")
+
+            try:
+                vol_vtable_ptr = cast(volume_ptr, POINTER(c_void_p)).contents
+                vol_vtable = cast(vol_vtable_ptr, POINTER(c_void_p * 22))
+
+                GetMasterVolumeLevel = ctypes.WINFUNCTYPE(
+                    wintypes.HRESULT, c_void_p, POINTER(c_float)
+                )(vol_vtable[0][8])
+                GetMasterVolumeLevelScalar = ctypes.WINFUNCTYPE(
+                    wintypes.HRESULT, c_void_p, POINTER(c_float)
+                )(vol_vtable[0][9])
+
+                db = c_float()
+                hr = GetMasterVolumeLevel(volume_ptr, byref(db))
+                if hr == 0 and db.value > -90.0:
+                    return 10.0 ** (db.value / 20.0)
+
+                scalar = c_float()
+                hr = GetMasterVolumeLevelScalar(volume_ptr, byref(scalar))
+                if hr != 0:
+                    raise OSError(f"GetMasterVolumeLevelScalar failed: {hr:#x}")
+                # Invert the audio-tapered mapping
+                return scalar.value ** (1.0 / 0.573)
+            finally:
+                Release = ctypes.WINFUNCTYPE(wintypes.ULONG, c_void_p)(vol_vtable[0][2])
+                Release(volume_ptr)
+        finally:
+            Release = ctypes.WINFUNCTYPE(wintypes.ULONG, c_void_p)(device_vtable[0][2])
+            Release(device_ptr)
+    finally:
+        Release = ctypes.WINFUNCTYPE(wintypes.ULONG, c_void_p)(vtable[0][2])
+        Release(enumerator_ptr)
+
+
+def _set_volume_macos(amp: float) -> None:
+    """Set master volume on macOS via AppleScript."""
+    # macOS does not natively support >100% system volume.
+    clamped = min(amp, 1.0)
+    level = int(clamped * 100)
+    subprocess.run(
+        ["osascript", "-e", f"set volume output volume {level}"],
+        check=False,
+        capture_output=True,
+    )
+
+
+def _get_volume_macos() -> float:
+    """Return current linear amplitude on macOS."""
+    result = subprocess.run(
+        ["osascript", "-e", "output volume of (get volume settings)"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    try:
+        return int(result.stdout.strip()) / 100.0
+    except ValueError, AttributeError:
+        return 1.0
+
+
+def _set_volume_linux(amp: float) -> None:
+    """Set master volume on Linux via PipeWire (``wpctl``)."""
+    # wpctl is the native PipeWire CLI and accepts linear values > 1.0
+    # directly.  We do not fall back to PulseAudio tools.
+    subprocess.run(
+        ["wpctl", "set-volume", "@DEFAULT_AUDIO_SINK@", str(amp)],
+        check=True,
+        capture_output=True,
+    )
+
+
+def _get_volume_linux() -> float:
+    """Return current linear amplitude on Linux via PipeWire (``wpctl``)."""
+    result = subprocess.run(
+        ["wpctl", "get-volume", "@DEFAULT_AUDIO_SINK@"],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    # Output: "Volume: 0.40" or "Volume: 0.50 [MUTED]"
+    line = result.stdout.strip()
+    parts = line.split()
+    return float(parts[1])
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+_setter: Callable[[float], None] | None = None
+_getter: Callable[[], float] | None = None
+_lock = threading.Lock()
+
+
+def _init_backend() -> None:
+    """Lazily select the platform backend."""
+    global _setter, _getter
+    with _lock:
+        if _setter is not None:
+            return
+        if sys.platform == "win32":
+            _setter = _set_volume_windows
+            _getter = _get_volume_windows
+        elif sys.platform == "darwin":
+            _setter = _set_volume_macos
+            _getter = _get_volume_macos
+        else:
+            _setter = _set_volume_linux
+            _getter = _get_volume_linux
+
+
+def set_volume(x: float) -> None:
+    """Set master volume from slider position ``x`` (0.0 .. 1.5).
+
+    The value is converted to a linear amplitude coefficient and applied to
+    the system mixer using the platform-native API.
+    """
+    _init_backend()
+    amp = slider_to_amp(x)
+    assert _setter is not None
+    try:
+        _setter(amp)
+    except Exception:
+        logger.exception("Failed to set volume (x=%.3f, amp=%.6f)", x, amp)
+
+
+def get_volume() -> float:
+    """Return the current slider position (0.0 .. 1.5) by reading the OS mixer.
+
+    On platforms that do not support amplification above 100 % the returned
+    value will never exceed ``1.0``.
+    """
+    _init_backend()
+    assert _getter is not None
+    try:
+        amp = _getter()
+    except Exception:
+        logger.exception("Failed to get volume")
+        return 1.0
+    return amp_to_slider(amp)
+
+
+def volume_max() -> float:
+    """Return the maximum slider position supported on this platform.
+
+    * Linux / PipeWire – ``1.5`` (amplification above 100 % is supported).
+    * Windows / macOS  – ``1.0`` (the OS mixer does not amplify above unity).
+    """
+    if sys.platform in ("win32", "darwin"):
+        return 1.0
+    return 1.5
@@ -27,6 +27,7 @@ import uvicorn
 import webview

 from mediahive.config import load_config, save_config
+from mediahive.volume_control import get_volume, set_volume, volume_max

 logger = logging.getLogger("mediahive.winmain")

@@ -52,6 +53,9 @@ MPC_BE_SEEK_BEGIN_COMMAND = 1085
 MPC_BE_RESUME_APPLY_THRESHOLD_MS = 15000
 MPC_BE_RESUME_CLEAR_MARGIN_MS = 15000
 MPC_BE_PLAYBACK_STATE_FLUSH_SECONDS = 1.0
+VOLUME_MIN = 0.0
+VOLUME_MAX = 1.5
+VOLUME_STEP = 0.01


 class _XINPUT_GAMEPAD(ctypes.Structure):
@@ -91,8 +95,6 @@ _XINPUT_BUTTONS = {
 }

 _MPC_BE_COMMANDS = {
-    0x0001: 907,
-    0x0002: 908,
    0x1000: 889,
    0x2000: 816,
    0x8000: 909,
@@ -104,8 +106,6 @@ _MPC_BE_SEEK_MASK_TO_COMMANDS = {
 }

 _MPC_BE_REPEATABLE_MASKS = {
-    0x0001,
-    0x0002,
    *_MPC_BE_SEEK_MASK_TO_COMMANDS,
 }

@@ -520,6 +520,12 @@ def _start_gamepad_remote(
                    if is_connected != last_connected[slot]:
                        last_connected[slot] = is_connected

+                    # Volume control via D-pad up / down (system master volume)
+                    if current_mask & 0x0001 and not last_pressed_masks[slot] & 0x0001:
+                        set_volume(min(volume_max(), get_volume() + VOLUME_STEP))
+                    if current_mask & 0x0002 and not last_pressed_masks[slot] & 0x0002:
+                        set_volume(max(VOLUME_MIN, get_volume() - VOLUME_STEP))
+
                    for mask, command_id in _MPC_BE_COMMANDS.items():
                        is_pressed = bool(current_mask & mask)
                        was_pressed = bool(last_pressed_masks[slot] & mask)
@@ -646,6 +652,21 @@ class JsApi:
        result = self._window.create_file_dialog(webview.FOLDER_DIALOG)
        return result[0] if result else None

+    def set_volume(self, x: float) -> None:
+        """Set system master volume from slider position ``x`` (0.0 .. 1.5)."""
+        # Clamp to the platform's maximum so the slider never exceeds what
+        # the OS can actually apply (1.0 on Windows/macOS, 1.5 on Linux).
+        clamped = max(VOLUME_MIN, min(volume_max(), float(x)))
+        set_volume(clamped)
+
+    def get_volume(self) -> float:
+        """Return current volume slider position (0.0 .. 1.5)."""
+        return get_volume()
+
+    def volume_max(self) -> float:
+        """Return the maximum volume slider position for this platform."""
+        return volume_max()
+

 def _prepend_meipass_to_path() -> None:
    """When frozen, ensure bundled binaries (ffmpeg) are found first on PATH."""
@@ -107,4 +107,12 @@ ignore = [
    "D415",
    "D400",
    "ANN401",
+    # Allow en-dash in docstrings (used for list formatting)
+    "RUF002",
+    # Allow ctypes COM variable names (CLSID_*, IID_*, etc.)
+    "N806",
+    # Allow inline comments that describe output formats
+    "ERA001",
+    # Allow unused local variables in ctypes COM boilerplate
+    "F841",
 ]