volume-knob

Python/volume-knob/config.yaml

@@ -0,0 +1,17 @@
+modes:
+  - name: "Master"
+    process: "master"
+  - name: "Media"
+    process:
+      - "chrome.exe"
+      - "firefox.exe"
+      - "vlc.exe"
+      - "Spotify.exe"
+  - name: "Other"
+    process: "other"
+
+mqtt:
+  broker: "192.168.178.6"
+  port: 9001
+  topic: "zigbee2mqtt/Desk knob"
+  transport: "websockets"

Python/volume-knob/main.py

@@ -0,0 +1,124 @@
+import paho.mqtt.client as mqtt
+import json
+from pycaw.pycaw import AudioUtilities, ISimpleAudioVolume, IAudioEndpointVolume
+import yaml
+from comtypes import CLSCTX_ALL
+import time
+import threading
+
+with open("config.yaml") as f:
+    try:
+        config = yaml.safe_load(f)
+    except yaml.YAMLError as exc:
+        print(exc)
+
+stopped = False
+
+modes = config.get("modes")
+current_mode_index = 0
+# MQTT broker details
+broker_address = config.get("mqtt").get("broker")
+broker_port = config.get("mqtt").get("port", 1883)
+topic = config.get("mqtt").get("topic")
+
+
+def used_process_names():
+    result = []
+    for mode in modes:
+        if mode.get("process"):
+            if isinstance(mode.get("process"), str):
+                result.append(mode.get("process"))
+            else:
+                result.extend(mode.get("process"))
+    return result
+
+
+def delta_volume_master(mode, delta):
+    devices = AudioUtilities.GetSpeakers()
+    interface = devices.Activate(IAudioEndpointVolume._iid_, CLSCTX_ALL, None)
+    volume = interface.QueryInterface(IAudioEndpointVolume)
+    min_volume, max_volume, step = volume.GetVolumeRange()
+    current_volume = volume.GetMasterVolumeLevel()
+    new_volume = max(min_volume, min(max_volume, current_volume + 5 * delta))
+    volume.SetMasterVolumeLevel(new_volume, None)
+
+
+def delta_volume_process(mode, delta):
+    is_other = mode.get("process") == "other"
+    process_name = mode.get("process")
+    if is_other:
+        process_names = [
+            session.Process.name()
+            for session in AudioUtilities.GetAllSessions()
+            if session.Process.name() not in used_process_names()
+        ]
+    elif isinstance(process_name, str):
+        process_names = [process_name]
+    else:
+        process_names = process_name
+    sessions = AudioUtilities.GetAllSessions()
+    for session in sessions:
+        volume = session._ctl.QueryInterface(ISimpleAudioVolume)
+        if session.Process and session.Process.name() in process_names:
+            min_volume = 0
+            max_volume = 1
+            step = 0.2
+            old_volume = volume.GetMasterVolume()
+            new_volume = max(min_volume, min(max_volume, old_volume + step * delta))
+            volume.SetMasterVolume(new_volume, None)
+
+
+# Function to adjust volume based on delta
+def delta_volume(delta):
+    current_mode = modes[current_mode_index]
+    if current_mode == "master":
+        delta_volume_master(current_mode, delta)
+    else:
+        delta_volume_process(current_mode, delta)
+
+
+previous_action = None
+
+
+# Callback function to handle incoming messages
+def on_message(client, userdata, message):
+    global previous_action, current_mode_index
+    payload = json.loads(message.payload.decode("utf-8"))
+    action = payload.get("action")
+    if action == "rotate_left":
+        # delta_volume(1)  # Decrease volume when knob is rotated left
+        previous_action = action
+    elif action == "rotate_right":
+        # delta_volume(-1)  # Increase volume when knob is rotated right
+        previous_action = action
+    elif action == "single":
+        previous_action = None
+        print("Switching mode to", modes[(current_mode_index + 1) % len(modes)])
+        current_mode_index = (current_mode_index + 1) % len(modes)
+
+
+def listen_for_messages():
+    global previous_action
+    while not stopped:
+        time.sleep(1)
+        if previous_action:
+            delta_volume(1 if previous_action == "rotate_left" else -1)
+
+
+transport = config.get("mqtt").get("transport", "tcp")
+# Create MQTT client
+client = mqtt.Client(transport=transport)
+client.on_message = on_message
+
+# Connect to broker and subscribe to topic
+client.connect(broker_address, broker_port, 60)
+client.subscribe(topic)
+
+x = threading.Thread(target=listen_for_messages)
+x.start()
+
+
+try:
+    client.loop_forever()
+except KeyboardInterrupt:
+    stopped = True

Python/volume-knob/requirements.txt

@@ -0,0 +1,3 @@
+paho-mqtt
+pycaw
+comtypes