fix: stop the ad mute stalling the poll loop for 200ms

CONTRIBUTING.md

@@ -39,6 +39,7 @@ manual verification scripts instead:
 ```bash
 python scripts/verify_smtc.py     # read-only; watch position extrapolation work
 python scripts/verify_hotkey.py   # press the hotkey, confirm it fires
+python scripts/verify_audio.py    # mutes and unmutes Spotify, and times it
 ```
 
 If you change timing behaviour, please add a case to `tests/test_logic.py` alongside

looptify/audio.py

@@ -2,26 +2,62 @@
 
 from __future__ import annotations
 
-from pycaw.pycaw import AudioUtilities, ISimpleAudioVolume
+import comtypes
+import psutil
+from pycaw.api.audiopolicy import IAudioSessionControl2, IAudioSessionManager2
+from pycaw.api.mmdeviceapi import IMMDeviceEnumerator
+from pycaw.constants import CLSID_MMDeviceEnumerator, EDataFlow, ERole
+from pycaw.pycaw import ISimpleAudioVolume
 
 _PROCESS_NAME = "spotify.exe"
 
 
+def _session_manager() -> IAudioSessionManager2:
+    """Get the default playback device's audio session manager.
+
+    pycaw's `AudioUtilities.GetAllSessions()` reaches this same object, but it
+    goes through `CreateDevice()`, which first reads the device's entire
+    property store — around 200 properties — to build a description nothing
+    here looks at. That cost ~200ms per call, and this runs on the asyncio
+    thread, so every ad boundary stalled the poll loop for longer than a poll
+    interval. Asking the endpoint directly costs ~5ms.
+    """
+    enumerator = comtypes.CoCreateInstance(
+        CLSID_MMDeviceEnumerator, IMMDeviceEnumerator, comtypes.CLSCTX_INPROC_SERVER
+    )
+    endpoint = enumerator.GetDefaultAudioEndpoint(
+        EDataFlow.eRender.value, ERole.eMultimedia.value
+    )
+    return endpoint.Activate(
+        IAudioSessionManager2._iid_, comtypes.CLSCTX_ALL, None
+    ).QueryInterface(IAudioSessionManager2)
+
+
 def set_spotify_muted(muted: bool) -> int:
     """Mute or unmute every Spotify session. Returns how many changed.
 
     0 is normal when Spotify is closed or has been idle.
     """
+    try:
+        sessions = _session_manager().GetSessionEnumerator()
+    except OSError:
+        # No playback device at all, so there is nothing to mute.
+        return 0
+
     changed = 0
-    for session in AudioUtilities.GetAllSessions():
-        process = session.Process
-        if process is None or process.name().lower() != _PROCESS_NAME:
+    for index in range(sessions.GetCount()):
+        control = sessions.GetSession(index)
+        if control is None:
             continue
         try:
-            volume = session._ctl.QueryInterface(ISimpleAudioVolume)
-            volume.SetMute(1 if muted else 0, None)
+            control2 = control.QueryInterface(IAudioSessionControl2)
+            pid = control2.GetProcessId()
+            # 0 is the system-sounds session, which owns no process.
+            if pid == 0 or psutil.Process(pid).name().lower() != _PROCESS_NAME:
+                continue
+            control2.QueryInterface(ISimpleAudioVolume).SetMute(1 if muted else 0, None)
             changed += 1
-        except OSError:
-            # A session can disappear between enumeration and use.
+        except (OSError, psutil.Error):
+            # A session, or its process, can vanish between enumeration and use.
             continue
     return changed

scripts/verify_audio.py

@@ -0,0 +1,85 @@
+"""Manual check: does muting still work, and is it fast enough to not stall the poll loop?
+
+Reads the mute state back through pycaw's own enumeration rather than through
+`audio.py`, so a broken implementation cannot vouch for itself.
+
+Spotify must be running. Always leaves it unmuted, even on failure.
+"""
+
+import statistics
+import sys
+import time
+from pathlib import Path
+
+# Python puts this script's own directory on sys.path, not the repo root.
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from pycaw.pycaw import AudioUtilities, ISimpleAudioVolume  # noqa: E402
+
+from looptify.audio import set_spotify_muted  # noqa: E402
+
+# set_spotify_muted runs on the asyncio thread, so it blocks the poll loop for
+# however long it takes. Anything near a poll interval (150ms) is too slow.
+BUDGET_MS = 50.0
+
+
+def read_mute_states() -> list[int]:
+    """Independent read-back: what does Windows say Spotify's sessions are set to?"""
+    states = []
+    for session in AudioUtilities.GetAllSessions():
+        process = session.Process
+        if process is None or process.name().lower() != "spotify.exe":
+            continue
+        states.append(session._ctl.QueryInterface(ISimpleAudioVolume).GetMute())
+    return states
+
+
+def main() -> int:
+    if not read_mute_states():
+        print("No Spotify audio sessions. Start Spotify and play something.")
+        return 2
+
+    failures = []
+    try:
+        changed = set_spotify_muted(True)
+        states = read_mute_states()
+        print(f"mute:   changed {changed} session(s), read back {states}")
+        if not states or not all(states):
+            failures.append("mute did not take effect")
+
+        changed = set_spotify_muted(False)
+        states = read_mute_states()
+        print(f"unmute: changed {changed} session(s), read back {states}")
+        if any(states):
+            failures.append("unmute did not take effect")
+
+        timings = []
+        for _ in range(10):
+            start = time.perf_counter()
+            set_spotify_muted(False)
+            timings.append((time.perf_counter() - start) * 1000)
+        timings.sort()
+        median = statistics.median(timings)
+        print(
+            f"timing: median {median:.1f}ms  min {timings[0]:.1f}ms  "
+            f"max {timings[-1]:.1f}ms  (budget {BUDGET_MS:.0f}ms)"
+        )
+        if median > BUDGET_MS:
+            failures.append(
+                f"too slow: {median:.1f}ms median blocks the poll loop for "
+                f"{median / 150:.1f} poll intervals"
+            )
+    finally:
+        set_spotify_muted(False)
+
+    if failures:
+        print("\nFAIL")
+        for failure in failures:
+            print(f"  - {failure}")
+        return 1
+    print("\nPASS")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())