85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
"""Read only the target game's Windows audio-session state and volume meter.
|
|
|
|
Does not record audio, access microphones, or change device/volume settings.
|
|
"""
|
|
import argparse
|
|
import ctypes as c
|
|
import json
|
|
import time
|
|
import uuid
|
|
|
|
p=argparse.ArgumentParser()
|
|
p.add_argument("pid",type=int)
|
|
a=p.parse_args()
|
|
ole=c.OleDLL("ole32")
|
|
P=c.c_void_p
|
|
I=c.c_int32
|
|
U=c.c_uint32
|
|
F=c.c_float
|
|
def guid(s): return (c.c_ubyte*16).from_buffer_copy(uuid.UUID(s).bytes_le)
|
|
def call(obj,index,types=(),args=()):
|
|
vt=c.cast(obj,c.POINTER(c.POINTER(P))).contents
|
|
return c.WINFUNCTYPE(I,P,*types)(vt[index])(obj,*args)
|
|
def check(hr):
|
|
if hr<0: raise OSError(f"COM HRESULT 0x{hr&0xffffffff:08x}")
|
|
def qi(obj,iid):
|
|
out=P()
|
|
check(call(obj,0,(P,c.POINTER(P)),(guid(iid),c.byref(out))))
|
|
return out
|
|
refs=[]
|
|
def keep(obj):
|
|
refs.append(obj)
|
|
return obj
|
|
ole.CoInitializeEx.argtypes=[P,U]
|
|
ole.CoCreateInstance.argtypes=[P,P,U,P,c.POINTER(P)]
|
|
ole.CoTaskMemFree.argtypes=[P]
|
|
check(ole.CoInitializeEx(None,0))
|
|
try:
|
|
enum=P()
|
|
check(ole.CoCreateInstance(guid("BCDE0395-E52F-467C-8E3D-C4579291692E"),None,23,guid("A95664D2-9614-4F35-A746-DE8DB63617E6"),c.byref(enum)))
|
|
keep(enum)
|
|
devices=P()
|
|
check(call(enum,3,(I,U,c.POINTER(P)),(0,1,c.byref(devices))))
|
|
keep(devices)
|
|
count=U()
|
|
check(call(devices,3,(c.POINTER(U),),(c.byref(count),)))
|
|
found=0
|
|
for i in range(count.value):
|
|
dev=P()
|
|
check(call(devices,4,(U,c.POINTER(P)),(i,c.byref(dev))))
|
|
keep(dev)
|
|
manager=P()
|
|
check(call(dev,3,(P,U,P,c.POINTER(P)),(guid("77AA99A0-1BD6-484F-8BC7-2C654C9A9B6F"),23,None,c.byref(manager))))
|
|
keep(manager)
|
|
sessions=P()
|
|
check(call(manager,5,(c.POINTER(P),),(c.byref(sessions),)))
|
|
keep(sessions)
|
|
sc=I()
|
|
check(call(sessions,3,(c.POINTER(I),),(c.byref(sc),)))
|
|
for j in range(sc.value):
|
|
session=P()
|
|
check(call(sessions,4,(I,c.POINTER(P)),(j,c.byref(session))))
|
|
keep(session)
|
|
sc2=keep(qi(session,"BFB7FF88-7239-4FC9-8FA2-07C950BE9C6D"))
|
|
pid=U()
|
|
check(call(sc2,14,(c.POINTER(U),),(c.byref(pid),)))
|
|
if pid.value!=a.pid: continue
|
|
found+=1
|
|
meter=keep(qi(session,"C02216F6-8C67-4B5B-9D00-D008E73E0064"))
|
|
volume=keep(qi(session,"87CE5498-68D6-44E5-9215-6DA47EF883D8"))
|
|
state,muted=I(),I()
|
|
vol,peak=F(),F()
|
|
check(call(session,3,(c.POINTER(I),),(c.byref(state),)))
|
|
check(call(volume,4,(c.POINTER(F),),(c.byref(vol),)))
|
|
check(call(volume,6,(c.POINTER(I),),(c.byref(muted),)))
|
|
maximum=0.0
|
|
for _ in range(12):
|
|
check(call(meter,3,(c.POINTER(F),),(c.byref(peak),)))
|
|
maximum=max(maximum,peak.value)
|
|
time.sleep(0.25)
|
|
print(json.dumps({"pid":a.pid,"active":state.value==1,"muted":bool(muted.value),"volume":vol.value,"peak_over_3_seconds":maximum}))
|
|
if not found: print(json.dumps({"pid":a.pid,"audio_sessions_found":0}))
|
|
finally:
|
|
for ref in reversed(refs): call(ref,2)
|
|
ole.CoUninitialize()
|