94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
"""Run on the box with drag-source.py visibly filling the Ubuntu video."""
|
|
import ctypes as C
|
|
import json
|
|
import os
|
|
import statistics
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
|
|
os.environ.update(DISPLAY=':0', XAUTHORITY='/home/armbian/.Xauthority')
|
|
x = C.CDLL('libX11.so.6')
|
|
t = C.CDLL('libXtst.so.6')
|
|
x.XOpenDisplay.restype = C.c_void_p
|
|
x.XDefaultRootWindow.argtypes = [C.c_void_p]
|
|
x.XDefaultRootWindow.restype = C.c_ulong
|
|
x.XSync.argtypes = [C.c_void_p, C.c_int]
|
|
x.XGetImage.argtypes = [C.c_void_p,C.c_ulong,C.c_int,C.c_int,C.c_uint,C.c_uint,C.c_ulong,C.c_int]
|
|
x.XGetImage.restype = C.c_void_p
|
|
x.XGetPixel.argtypes = [C.c_void_p,C.c_int,C.c_int]
|
|
x.XGetPixel.restype = C.c_ulong
|
|
x.XDestroyImage.argtypes = [C.c_void_p]
|
|
t.XTestFakeMotionEvent.argtypes = [C.c_void_p,C.c_int,C.c_int,C.c_int,C.c_ulong]
|
|
t.XTestFakeButtonEvent.argtypes = [C.c_void_p,C.c_uint,C.c_int,C.c_ulong]
|
|
d = x.XOpenDisplay(None)
|
|
assert d
|
|
root = x.XDefaultRootWindow(d)
|
|
|
|
def read():
|
|
image = x.XGetImage(d, root, 40, 1380, 1120, 1, 0xffffffff, 2)
|
|
assert image
|
|
try:
|
|
magic = sum(1 << i for i in range(8)
|
|
if (x.XGetPixel(image, 780+i*40, 0) & 0xffffff) > 0xc0c0c0)
|
|
assert magic == 0xa5, ('probe signature absent', magic)
|
|
pixels = [x.XGetPixel(image, i*40+20, 0) & 0xffffff for i in range(16)]
|
|
if any(0x404040 < v < 0xc0c0c0 for v in pixels):
|
|
return -1, False
|
|
seq = sum(1 << i for i,v in enumerate(pixels) if v > 0xc0c0c0)
|
|
held = (x.XGetPixel(image, 700, 0) & 0xffffff) > 0xc0c0c0
|
|
return seq, held
|
|
finally:
|
|
x.XDestroyImage(image)
|
|
|
|
pid = int(subprocess.check_output(['pgrep','-x','moonlight']).strip())
|
|
focus = int(subprocess.check_output(['xdotool','getwindowfocus','getwindowpid']).strip())
|
|
assert focus == pid
|
|
assert read() == (0, False), ('test window not visible/ready',read())
|
|
t.XTestFakeMotionEvent(d, -1, 800, 700, 0)
|
|
x.XSync(d, 0)
|
|
time.sleep(.5)
|
|
sent, frames = {}, []
|
|
start = time.monotonic()
|
|
last = -1
|
|
try:
|
|
t.XTestFakeButtonEvent(d, 1, 1, 0)
|
|
x.XSync(d, 0)
|
|
time.sleep(.8)
|
|
assert read()[1], 'held-button marker did not appear'
|
|
start = time.monotonic()
|
|
for seq in range(1, 301):
|
|
target = start + (seq-1)/60
|
|
time.sleep(max(0, target-time.monotonic()))
|
|
t.XTestFakeMotionEvent(d, -1, 800+4*seq, 700+seq, 0)
|
|
x.XSync(d, 0)
|
|
sent[seq] = time.monotonic()
|
|
seen, held = read()
|
|
now = time.monotonic()
|
|
if 0 <= seen <= 300 and seen != last and held:
|
|
frames.append(dict(seq=seen,t=now,age_ms=(now-sent[seen])*1000 if seen in sent else None))
|
|
last = seen
|
|
deadline = time.monotonic()+2
|
|
while last < 300 and time.monotonic()<deadline:
|
|
seen, held = read()
|
|
now = time.monotonic()
|
|
if 0 <= seen <= 300 and seen != last and held:
|
|
frames.append(dict(seq=seen,t=now,age_ms=(now-sent[seen])*1000 if seen in sent else None))
|
|
last = seen
|
|
time.sleep(.002)
|
|
finally:
|
|
t.XTestFakeButtonEvent(d, 1, 0, 0)
|
|
x.XSync(d, 0)
|
|
time.sleep(.8)
|
|
gaps = [(b['t']-a['t'])*1000 for a,b in zip(frames,frames[1:])]
|
|
ages = [a['age_ms'] for a in frames if a['age_ms'] is not None]
|
|
def stats(items):
|
|
items=sorted(items)
|
|
return dict(p50=statistics.median(items),p95=items[int(len(items)*.95)],maximum=max(items)) if items else None
|
|
result = dict(pid=pid,sent=300,elapsed=time.monotonic()-start,frames=frames,
|
|
fresh_video_updates_per_second=sum(f['t']<start+5 for f in frames)/5,
|
|
motion_to_decoded_marker_ms=stats(ages),frame_gap_ms=stats(gaps),
|
|
last_motion_seen=last,release_visible=not read()[1],
|
|
method='60Hz XTest held drag to first observed unique decoded marker; no USB/scanout')
|
|
print(json.dumps(result))
|