33 lines
2.2 KiB
Python
33 lines
2.2 KiB
Python
import ctypes,ctypes.util,subprocess,os,time,json,statistics
|
|
os.environ.update(DISPLAY=':0',XAUTHORITY='/home/armbian/.Xauthority')
|
|
x=ctypes.CDLL(ctypes.util.find_library('X11'));x.XOpenDisplay.argtypes=[ctypes.c_char_p];x.XOpenDisplay.restype=ctypes.c_void_p
|
|
x.XDefaultRootWindow.argtypes=[ctypes.c_void_p];x.XDefaultRootWindow.restype=ctypes.c_ulong
|
|
x.XGetImage.argtypes=[ctypes.c_void_p,ctypes.c_ulong,ctypes.c_int,ctypes.c_int,ctypes.c_uint,ctypes.c_uint,ctypes.c_ulong,ctypes.c_int];x.XGetImage.restype=ctypes.c_void_p
|
|
x.XGetPixel.argtypes=[ctypes.c_void_p,ctypes.c_int,ctypes.c_int];x.XGetPixel.restype=ctypes.c_ulong
|
|
x.XDestroyImage.argtypes=[ctypes.c_void_p];x.XCloseDisplay.argtypes=[ctypes.c_void_p]
|
|
d=x.XOpenDisplay(b':0');assert d;root=x.XDefaultRootWindow(d)
|
|
prefix='/home/armbian/vdi-build/rkmpp-experiment/build/moonlight/app/moonlight stream 192.168.1.71 '
|
|
items=[l.split(None,1) for l in subprocess.check_output(['ps','-eo','pid,args'],text=True).splitlines()[1:] if len(l.split(None,1))==2 and l.split(None,1)[1].startswith(prefix)]
|
|
assert len(items)==1
|
|
ids=subprocess.check_output(['xdotool','search','--onlyvisible','--pid',items[0][0],'--class','Moonlight'],text=True).split();assert ids
|
|
subprocess.run(['xdotool','windowactivate','--sync',ids[-1]],check=True)
|
|
def pixel():
|
|
im=x.XGetImage(d,root,2250,130,1,1,0xffffffff,2);assert im
|
|
v=x.XGetPixel(im,0,0);x.XDestroyImage(im)
|
|
return v&0xffffff
|
|
samples=[];errors=[]
|
|
for i in range(24):
|
|
v=pixel();assert v<0x202020 or v>0xe0e0e0,hex(v)
|
|
start=time.monotonic_ns()
|
|
subprocess.run(['xdotool','key','--clearmodifiers','--delay','0','F9'],check=True,stdout=subprocess.DEVNULL)
|
|
while time.monotonic_ns()-start<2000000000:
|
|
cur=pixel()
|
|
if (v<0x202020 and cur>0xe0e0e0) or (v>0xe0e0e0 and cur<0x202020):
|
|
samples.append((time.monotonic_ns()-start)/1e6);break
|
|
time.sleep(.002)
|
|
else:errors.append(i)
|
|
time.sleep(.12)
|
|
x.XCloseDisplay(d)
|
|
vals=sorted(samples)
|
|
print(json.dumps({'method':'synthetic F9 injection to decoded video marker in X11 root; includes xdotool startup and polling, excludes physical USB and panel scanout','samples_ms':samples,'timeouts':errors,'p50_ms':statistics.median(vals) if vals else None,'p95_ms':vals[min(len(vals)-1,int(len(vals)*.95))] if vals else None}))
|