125 lines
3.8 KiB
Python
125 lines
3.8 KiB
Python
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
|