Adding Project Files
This commit is contained in:
360
set_volume.py
Normal file
360
set_volume.py
Normal file
@@ -0,0 +1,360 @@
|
||||
import json
|
||||
import sys
|
||||
import os
|
||||
import traceback
|
||||
|
||||
# --- Added: support a simple --list mode BEFORE the usage check ---
|
||||
if len(sys.argv) >= 2 and sys.argv[1] == "--list":
|
||||
# print running audio session application/process names (one per line) and exit
|
||||
if os.name == "nt":
|
||||
try:
|
||||
from comtypes import CoInitialize
|
||||
from pycaw.pycaw import AudioUtilities
|
||||
except Exception as e:
|
||||
sys.stderr.write("pycaw not available: " + str(e) + "\n")
|
||||
sys.exit(2)
|
||||
try:
|
||||
CoInitialize()
|
||||
sessions = AudioUtilities.GetAllSessions()
|
||||
for session in sessions:
|
||||
try:
|
||||
if session.Process and session.Process.name():
|
||||
name = session.Process.name().lower().replace(".exe", "")
|
||||
print(name, flush=True)
|
||||
except Exception:
|
||||
continue
|
||||
sys.exit(0)
|
||||
except Exception as e:
|
||||
sys.stderr.write("Windows list failed: " + str(e) + "\n")
|
||||
sys.stderr.write(traceback.format_exc())
|
||||
sys.exit(3)
|
||||
elif os.name == "posix":
|
||||
try:
|
||||
import pulsectl
|
||||
except Exception as e:
|
||||
sys.stderr.write("pulsectl not available: " + str(e) + "\n")
|
||||
sys.exit(2)
|
||||
try:
|
||||
with pulsectl.Pulse('dreckshub-list') as pulse:
|
||||
for sink_input in pulse.sink_input_list():
|
||||
app_name = (sink_input.proplist.get('application.name') or "").lower()
|
||||
if app_name:
|
||||
print(app_name, flush=True)
|
||||
sys.exit(0)
|
||||
except Exception as e:
|
||||
sys.stderr.write("Pulse list failed: " + str(e) + "\n")
|
||||
sys.stderr.write(traceback.format_exc())
|
||||
sys.exit(3)
|
||||
else:
|
||||
sys.stderr.write("Unsupported platform for --list: " + os.name + "\n")
|
||||
sys.exit(2)
|
||||
|
||||
def fail(msg):
|
||||
sys.stderr.write(msg + "\n")
|
||||
sys.stderr.flush()
|
||||
sys.exit(1)
|
||||
|
||||
def info(msg):
|
||||
sys.stdout.write(msg + "\n")
|
||||
sys.stdout.flush()
|
||||
|
||||
# --- Added: small helper functions to allow a persistent stdin server mode ---
|
||||
def list_sessions():
|
||||
"""Return a list of running audio session names (lowercased) for the current platform."""
|
||||
names = []
|
||||
try:
|
||||
if os.name == "nt":
|
||||
try:
|
||||
from comtypes import CoInitialize
|
||||
from pycaw.pycaw import AudioUtilities
|
||||
except Exception as e:
|
||||
raise RuntimeError("pycaw not available: " + str(e))
|
||||
CoInitialize()
|
||||
sessions = AudioUtilities.GetAllSessions()
|
||||
for session in sessions:
|
||||
try:
|
||||
if session.Process and session.Process.name():
|
||||
name = session.Process.name().lower().replace(".exe", "")
|
||||
names.append(name)
|
||||
except Exception:
|
||||
continue
|
||||
elif os.name == "posix":
|
||||
try:
|
||||
import pulsectl
|
||||
except Exception as e:
|
||||
raise RuntimeError("pulsectl not available: " + str(e))
|
||||
with pulsectl.Pulse('dreckshub-list') as pulse:
|
||||
for sink_input in pulse.sink_input_list():
|
||||
app_name = (sink_input.proplist.get('application.name') or "").lower()
|
||||
if app_name:
|
||||
names.append(app_name)
|
||||
else:
|
||||
raise RuntimeError("Unsupported platform for list: " + os.name)
|
||||
except Exception as e:
|
||||
# propagate exception for caller to handle
|
||||
raise
|
||||
return names
|
||||
|
||||
def set_volume_for(proc_raw, value):
|
||||
"""Set volume for matching sessions (proc_raw may be process name or substring).
|
||||
Returns dict: { matched: n, matched_names: [...] } or raises on fatal errors.
|
||||
"""
|
||||
proc = (proc_raw or "").lower().replace(".exe", "")
|
||||
value = max(0.0, min(1.0, float(value)))
|
||||
matched_names = []
|
||||
found = 0
|
||||
|
||||
if os.name == "nt":
|
||||
try:
|
||||
from comtypes import CoInitialize
|
||||
from pycaw.pycaw import AudioUtilities
|
||||
except Exception as e:
|
||||
raise RuntimeError("pycaw not available: " + str(e))
|
||||
|
||||
CoInitialize()
|
||||
sessions = AudioUtilities.GetAllSessions()
|
||||
# First pass: exact/process-name contains target
|
||||
for session in sessions:
|
||||
pname = None
|
||||
try:
|
||||
if session.Process:
|
||||
pname = session.Process.name()
|
||||
except Exception:
|
||||
pname = None
|
||||
pname_l = (pname or "").lower()
|
||||
name_clean = pname_l.replace(".exe", "")
|
||||
try:
|
||||
if name_clean == proc or proc in name_clean:
|
||||
try:
|
||||
session.SimpleAudioVolume.SetMasterVolume(value, None)
|
||||
found += 1
|
||||
matched_names.append(pname or "<unknown>")
|
||||
except Exception as ex:
|
||||
# continue trying others, but record nothing here
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
# Second pass: substring matches
|
||||
if found == 0:
|
||||
for session in sessions:
|
||||
try:
|
||||
if not session.Process:
|
||||
continue
|
||||
pname = session.Process.name()
|
||||
pname_l = pname.lower()
|
||||
if proc in pname_l and pname not in matched_names:
|
||||
try:
|
||||
session.SimpleAudioVolume.SetMasterVolume(value, None)
|
||||
found += 1
|
||||
matched_names.append(pname)
|
||||
except Exception:
|
||||
pass
|
||||
except Exception:
|
||||
continue
|
||||
return {"matched": found, "matched_names": matched_names}
|
||||
|
||||
elif os.name == "posix":
|
||||
try:
|
||||
import pulsectl
|
||||
except Exception as e:
|
||||
raise RuntimeError("pulsectl not available: " + str(e))
|
||||
with pulsectl.Pulse('dreckshub-set-volume') as pulse:
|
||||
sink_inputs = pulse.sink_input_list()
|
||||
for sink_input in sink_inputs:
|
||||
app_name = (sink_input.proplist.get('application.name') or "").lower()
|
||||
try:
|
||||
if proc == app_name or proc in app_name:
|
||||
pulse.volume_set_all_chans(sink_input, value)
|
||||
found += 1
|
||||
matched_names.append(app_name)
|
||||
except Exception:
|
||||
continue
|
||||
return {"matched": found, "matched_names": matched_names}
|
||||
else:
|
||||
raise RuntimeError("Unsupported platform: " + os.name)
|
||||
|
||||
# --- NEW: server mode via stdin / stdout (JSON lines) ---
|
||||
if len(sys.argv) >= 2 and sys.argv[1] == "--server-stdin":
|
||||
# Run a simple REPL: read JSON per-line, respond with JSON per-line.
|
||||
# Commands: {"cmd":"list"} -> {"ok":true,"list":[...]}
|
||||
# {"cmd":"set","process":"name","value":0.5} -> {"ok":true,"matched":1}
|
||||
sys.stdout.write("") # ensure stdout available
|
||||
sys.stdout.flush()
|
||||
for raw in sys.stdin:
|
||||
line = raw.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
obj = json.loads(line)
|
||||
except Exception as e:
|
||||
sys.stdout.write(json.dumps({"ok": False, "error": "invalid-json", "detail": str(e)}) + "\n")
|
||||
sys.stdout.flush()
|
||||
continue
|
||||
|
||||
cmd = obj.get("cmd")
|
||||
try:
|
||||
if cmd == "list":
|
||||
try:
|
||||
items = list_sessions()
|
||||
sys.stdout.write(json.dumps({"ok": True, "list": items}) + "\n")
|
||||
except Exception as e:
|
||||
sys.stdout.write(json.dumps({"ok": False, "error": "list-failed", "detail": str(e)}) + "\n")
|
||||
sys.stdout.flush()
|
||||
elif cmd == "set":
|
||||
proc = obj.get("process") or ""
|
||||
val = obj.get("value", 0)
|
||||
try:
|
||||
res = set_volume_for(proc, val)
|
||||
sys.stdout.write(json.dumps({"ok": True, "matched": res.get("matched", 0), "matched_names": res.get("matched_names", [])}) + "\n")
|
||||
except Exception as e:
|
||||
sys.stdout.write(json.dumps({"ok": False, "error": "set-failed", "detail": str(e)}) + "\n")
|
||||
sys.stdout.flush()
|
||||
else:
|
||||
sys.stdout.write(json.dumps({"ok": False, "error": "unknown-cmd", "detail": cmd}) + "\n")
|
||||
sys.stdout.flush()
|
||||
except Exception as e:
|
||||
sys.stdout.write(json.dumps({"ok": False, "error": "internal", "detail": str(e)}) + "\n")
|
||||
sys.stdout.flush()
|
||||
# stdin closed -> exit
|
||||
sys.exit(0)
|
||||
|
||||
if len(sys.argv) < 3:
|
||||
fail("Usage: set_volume.py <process_name> <value_float_0_1> [optional_tag]")
|
||||
|
||||
proc_raw = sys.argv[1]
|
||||
proc = proc_raw.lower()
|
||||
try:
|
||||
value = float(sys.argv[2])
|
||||
except Exception as e:
|
||||
fail("Invalid value: " + str(e))
|
||||
|
||||
value = max(0.0, min(1.0, value))
|
||||
|
||||
info(f"set_volume.py start -> proc='{proc_raw}' normalized='{value}' platform='{os.name}'")
|
||||
|
||||
if os.name == "nt":
|
||||
try:
|
||||
from comtypes import CoInitialize
|
||||
from pycaw.pycaw import AudioUtilities
|
||||
except Exception as e:
|
||||
fail("pycaw not available: " + str(e))
|
||||
|
||||
try:
|
||||
CoInitialize()
|
||||
sessions = AudioUtilities.GetAllSessions()
|
||||
info(f"Found {len(sessions)} audio session(s)")
|
||||
found = 0
|
||||
tried = 0
|
||||
matched_names = []
|
||||
|
||||
target = proc.replace(".exe", "").lower()
|
||||
|
||||
# First pass: exact/process-name contains target
|
||||
for session in sessions:
|
||||
tried += 1
|
||||
pname = None
|
||||
try:
|
||||
if session.Process:
|
||||
pname = session.Process.name()
|
||||
except Exception:
|
||||
pname = None
|
||||
pname_l = (pname or "").lower()
|
||||
name_clean = pname_l.replace(".exe", "")
|
||||
info(f"Session[{tried}] -> process='{pname}' cleaned='{name_clean}'")
|
||||
try:
|
||||
if name_clean == target or target in name_clean:
|
||||
info(f" -> MATCH candidate: '{pname}' ; setting volume to {value}")
|
||||
try:
|
||||
session.SimpleAudioVolume.SetMasterVolume(value, None)
|
||||
info(f" -> OK: set for '{pname}'")
|
||||
found += 1
|
||||
matched_names.append(pname or "<unknown>")
|
||||
except Exception as ex:
|
||||
sys.stderr.write(f" -> ERROR setting volume for '{pname}': {ex}\n")
|
||||
sys.stderr.flush()
|
||||
except Exception as e:
|
||||
sys.stderr.write(f" -> ERROR inspecting session '{pname}': {e}\n")
|
||||
sys.stderr.flush()
|
||||
|
||||
# Second pass: try partial substring matches if nothing found
|
||||
if found == 0:
|
||||
info("No exact/primary matches found; trying substring matches against full process names")
|
||||
for session in sessions:
|
||||
try:
|
||||
if not session.Process:
|
||||
continue
|
||||
pname = session.Process.name()
|
||||
pname_l = pname.lower()
|
||||
if proc in pname_l and pname not in matched_names:
|
||||
info(f" -> substring MATCH candidate: '{pname}' ; setting volume to {value}")
|
||||
try:
|
||||
session.SimpleAudioVolume.SetMasterVolume(value, None)
|
||||
info(f" -> OK: set for '{pname}'")
|
||||
found += 1
|
||||
matched_names.append(pname)
|
||||
except Exception as ex:
|
||||
sys.stderr.write(f" -> ERROR setting volume for '{pname}': {ex}\n")
|
||||
sys.stderr.flush()
|
||||
except Exception as e:
|
||||
sys.stderr.write(f" -> ERROR during substring pass: {e}\n")
|
||||
sys.stderr.flush()
|
||||
|
||||
info(f"Completed Windows set-volume: matched {found} session(s).")
|
||||
if found == 0:
|
||||
info("Available session process names (for debugging):")
|
||||
for session in sessions:
|
||||
try:
|
||||
if session.Process and session.Process.name():
|
||||
info(" - " + session.Process.name())
|
||||
continue
|
||||
except Exception:
|
||||
pass
|
||||
# fallback unknown entry
|
||||
info(" - <unknown>")
|
||||
sys.exit(0)
|
||||
except Exception as e:
|
||||
sys.stderr.write("Windows volume change failed: " + str(e) + "\n")
|
||||
sys.stderr.write(traceback.format_exc())
|
||||
sys.stderr.flush()
|
||||
sys.exit(2)
|
||||
|
||||
elif os.name == "posix":
|
||||
try:
|
||||
import pulsectl
|
||||
except Exception as e:
|
||||
fail("pulsectl not available: " + str(e))
|
||||
|
||||
try:
|
||||
found = 0
|
||||
with pulsectl.Pulse('dreckshub-set-volume') as pulse:
|
||||
sink_inputs = pulse.sink_input_list()
|
||||
info(f"Found {len(sink_inputs)} sink_input(s)")
|
||||
for idx, sink_input in enumerate(sink_inputs, start=1):
|
||||
app_name = (sink_input.proplist.get('application.name') or "").lower()
|
||||
pid = sink_input.proplist.get('application.process.id') or ""
|
||||
info(f"SinkInput[{idx}] -> app_name='{app_name}' pid='{pid}' index={sink_input.index}")
|
||||
try:
|
||||
if proc == app_name or proc in app_name:
|
||||
info(f" -> MATCH candidate: '{app_name}' ; setting volume to {value}")
|
||||
pulse.volume_set_all_chans(sink_input, value)
|
||||
info(f" -> OK: set for sink_input index={sink_input.index}")
|
||||
found += 1
|
||||
except Exception as ex:
|
||||
sys.stderr.write(f" -> ERROR setting volume for sink_input index={sink_input.index}: {ex}\n")
|
||||
sys.stderr.flush()
|
||||
|
||||
if found == 0:
|
||||
info("No matching sink_input found. Listing available application names for debug:")
|
||||
for idx, sink_input in enumerate(sink_inputs, start=1):
|
||||
app_name = (sink_input.proplist.get('application.name') or "")
|
||||
info(f" - [{idx}] '{app_name}' (index={sink_input.index})")
|
||||
info(f"Completed PulseAudio set-volume: matched {found} sink_input(s).")
|
||||
sys.exit(0)
|
||||
except Exception as e:
|
||||
sys.stderr.write("PulseAudio volume change failed: " + str(e) + "\n")
|
||||
sys.stderr.write(traceback.format_exc())
|
||||
sys.stderr.flush()
|
||||
sys.exit(3)
|
||||
else:
|
||||
fail("Unsupported platform: " + os.name)
|
||||
|
||||
Reference in New Issue
Block a user