Add support for volume management on Linux

- Added support for volume increase, decrease, and set commands on Linux
- Updated volume.py to handle Linux-specific volume control using pulsectl
- Added pulsectl to requirements.txt

app/buttons/audio/volume.py

@@ -1,88 +1,118 @@
-from app.utils.platform import is_win
+from app.utils.platform import is_windows, is_linux
 
 import time
 import math
 import ctypes
-if is_win:
+if is_windows:
     import win32api
     import win32con
     from comtypes import CLSCTX_ALL
     from pycaw.pycaw import AudioUtilities, IAudioEndpointVolume, ISimpleAudioVolume
+else:
+    from pulsectl import Pulse
+
 from app.utils.logger import log
 
 
 
 def get_current_volume():
-    if not is_win:
-        log.error("This command is only available on Windows")
-        raise RuntimeError("This command is only available on Windows")
+    if is_windows:
+        devices = AudioUtilities.GetSpeakers()
+        interface = devices.Activate(IAudioEndpointVolume._iid_, CLSCTX_ALL, None)
+        volume = ctypes.cast(interface, ctypes.POINTER(IAudioEndpointVolume))
+        return volume.GetMasterVolumeLevelScalar()
+
+    elif is_linux:
+        with Pulse('volume') as pulse:
+            sink = pulse.sink_list()[0]
+            return round(sink.volume.value_flat, 2)
 
-    devices = AudioUtilities.GetSpeakers()
-    interface = devices.Activate(IAudioEndpointVolume._iid_, CLSCTX_ALL, None)
-    volume = ctypes.cast(interface, ctypes.POINTER(IAudioEndpointVolume))
-    return volume.GetMasterVolumeLevelScalar()
+    else:
+        raise NotImplementedError("This command is not implemented for this platform")
 
 
 def set_volume(target_volume):
-    if not is_win:
-        log.error("This command is only available on Windows")
-        raise RuntimeError("This command is only available on Windows")
-
-    current_volume = get_current_volume()
-    devices = AudioUtilities.GetSpeakers()
-    interface = devices.Activate(IAudioEndpointVolume._iid_, CLSCTX_ALL, None)
-    volume = ctypes.cast(interface, ctypes.POINTER(IAudioEndpointVolume))
-    while math.isclose(current_volume, target_volume, rel_tol=0.01) == False:
-        if current_volume > target_volume:
-            current_volume -= 0.01
-        else:
-            current_volume += 0.01
-        volume.SetMasterVolumeLevelScalar(current_volume, None)
-        time.sleep(0.01)
+    if is_windows:
+        current_volume = get_current_volume()
+        devices = AudioUtilities.GetSpeakers()
+        interface = devices.Activate(IAudioEndpointVolume._iid_, CLSCTX_ALL, None)
+        volume = ctypes.cast(interface, ctypes.POINTER(IAudioEndpointVolume))
+        while math.isclose(current_volume, target_volume, rel_tol=0.01) == False:
+            if current_volume > target_volume:
+                current_volume -= 0.01
+            else:
+                current_volume += 0.01
+            volume.SetMasterVolumeLevelScalar(current_volume, None)
+            time.sleep(0.01)
 
-    return current_volume
+        return current_volume
+
+    elif is_linux:
+        with Pulse('volume') as pulse:
+            sink = pulse.sink_list()[0]
+            volume = sink.volume
+            volume.value_flat = target_volume
+            pulse.volume_set(sink, volume)
+            return target_volume
+
+    else:
+        raise NotImplementedError("This command is not implemented for this platform.")
 
 
-def increase_volume(delta):
-    if not is_win:
-        log.error("This command is only available on Windows")
-        raise RuntimeError("This command is only available on Windows")
+def increase_volume(delta=1):
+    if is_windows:
+        win32api.keybd_event(win32con.VK_VOLUME_UP, 0)
+        win32api.keybd_event(win32con.VK_VOLUME_UP, 0, win32con.KEYEVENTF_KEYUP)
+        if delta == "":
+            return get_current_volume()
+        target_volume = get_current_volume() + (int(delta) / 100.0)
+        return set_volume(target_volume)
+
+    elif is_linux:
+        with Pulse('volume') as pulse:
+            sink = pulse.sink_list()[0]
+            target_volume = min(sink.volume.value_flat + (int(delta) / 100.0), 1.0)
+            volume = sink.volume
+            volume.value_flat = target_volume
+            pulse.volume_set(sink, volume)
+            return target_volume
 
-    win32api.keybd_event(win32con.VK_VOLUME_UP, 0)
-    win32api.keybd_event(win32con.VK_VOLUME_UP, 0, win32con.KEYEVENTF_KEYUP)
-    if delta == "":
-        return get_current_volume()
-    target_volume = get_current_volume() + (int(delta) / 100.0)
-    return set_volume(target_volume)
+    else:
+        raise NotImplementedError("This command is not implemented for this platform.")
 
 
-def decrease_volume(delta):
-    if not is_win:
-        log.error("This command is only available on Windows")
-        raise RuntimeError("This command is only available on Windows")
+def decrease_volume(delta=1):
+    if is_windows:
+        win32api.keybd_event(win32con.VK_VOLUME_DOWN, 0)
+        win32api.keybd_event(win32con.VK_VOLUME_DOWN, 0, win32con.KEYEVENTF_KEYUP)
+        if delta == "":
+            return get_current_volume()
+        target_volume = get_current_volume() - (int(delta) / 100.0)
+        return set_volume(target_volume)
+
+    elif is_linux:
+        with Pulse('volume') as pulse:
+            sink = pulse.sink_list()[0]
+            target_volume = max(sink.volume.value_flat - (int(delta) / 100.0), 0.0)
+            volume = sink.volume
+            volume.value_flat = target_volume
+            pulse.volume_set(sink, volume)
+            return target_volume
 
-    win32api.keybd_event(win32con.VK_VOLUME_DOWN, 0)
-    win32api.keybd_event(win32con.VK_VOLUME_DOWN, 0, win32con.KEYEVENTF_KEYUP)
-    if delta == "":
-        return get_current_volume()
-    target_volume = get_current_volume() - (int(delta) / 100.0)
-    return set_volume(target_volume)
+    else:
+        raise NotImplementedError("This command is not implemented for this platform.")
 
 
 def handle_command(message):
     if message.startswith("/volume +"):
-        delta = message.replace("/volume +", "")
-        if delta.replace(" ","") == "":
-            increase_volume("1")
-        else:
-            increase_volume(delta)
+        delta = message.replace("/volume +", "").strip()
+        delta = int(delta) if delta else 1
+        increase_volume(delta)
 
     elif message.startswith("/volume -"):
-        delta = message.replace("/volume -", "")
-        if delta.replace(" ","") == "":
-            decrease_volume("1")
-        else:
-            decrease_volume(delta)
+        delta = message.replace("/volume -", "").strip()
+        delta = int(delta) if delta else 1
+        decrease_volume(delta)
 
     elif message.startswith("/volume set"):
         target_volume = int(message.replace("/volume set ", "")) / 100.0

requirements.txt

@@ -8,7 +8,8 @@ notify_py==0.3.43
 keyboard==0.13.5
 PyAutoGUI==0.9.54
 pywin32; sys_platform == 'win32'
-pycaw==20240210
+pycaw==20240210; sys_platform == 'win32'
+pulsectl==24.11.0; sys_platform == 'linux'
 pillow==11.0.0
 PyAudio==0.2.14
 webcolors==24.11.1