104 lines
3.8 KiB
Python
104 lines
3.8 KiB
Python
"""Run only with the controlled GTK acceptance window filling the stream."""
|
|
import ctypes as C
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import statistics
|
|
import struct
|
|
import subprocess
|
|
import time
|
|
|
|
os.environ.update(DISPLAY=':0', XAUTHORITY='/home/armbian/.Xauthority')
|
|
x = C.CDLL('libX11.so.6')
|
|
fix = C.CDLL('libXfixes.so.3')
|
|
test = C.CDLL('libXtst.so.6')
|
|
class Cursor(C.Structure):
|
|
_fields_ = [('x', C.c_short), ('y', C.c_short), ('w', C.c_ushort),
|
|
('h', C.c_ushort), ('hx', C.c_ushort), ('hy', C.c_ushort),
|
|
('serial', C.c_ulong), ('pixels', C.POINTER(C.c_ulong)),
|
|
('atom', C.c_ulong), ('name', C.c_char_p)]
|
|
x.XOpenDisplay.argtypes = [C.c_char_p]
|
|
x.XOpenDisplay.restype = C.c_void_p
|
|
x.XDefaultRootWindow.argtypes = [C.c_void_p]
|
|
x.XDefaultRootWindow.restype = C.c_ulong
|
|
x.XFree.argtypes = [C.c_void_p]
|
|
x.XSync.argtypes = [C.c_void_p, C.c_int]
|
|
x.XWarpPointer.argtypes = [C.c_void_p, C.c_ulong, C.c_ulong, C.c_int,
|
|
C.c_int, C.c_uint, C.c_uint, C.c_int, 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]
|
|
fix.XFixesGetCursorImage.argtypes = [C.c_void_p]
|
|
fix.XFixesGetCursorImage.restype = C.POINTER(Cursor)
|
|
test.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 native():
|
|
p = fix.XFixesGetCursorImage(d)
|
|
assert p
|
|
try:
|
|
c = p.contents
|
|
return dict(x=c.x, y=c.y, w=c.w, h=c.h, hx=c.hx, hy=c.hy,
|
|
pixels=[int(c.pixels[i]) & 0xffffffff for i in range(c.w*c.h)])
|
|
finally:
|
|
x.XFree(p)
|
|
|
|
|
|
def warp(px, py):
|
|
x.XWarpPointer(d, 0, root, 0, 0, 0, 0, px, py)
|
|
x.XSync(d, 0)
|
|
|
|
|
|
def decoded_sequence():
|
|
im = x.XGetImage(d, root, 40, 1370, 640, 1, 0xffffffff, 2)
|
|
assert im
|
|
try:
|
|
values = [x.XGetPixel(im, i*40+20, 0) & 0xffffff for i in range(16)]
|
|
if any(0x202020 <= v <= 0xe0e0e0 for v in values):
|
|
return -1
|
|
return sum((1 << i) for i, v in enumerate(values) if v > 0xe0e0e0)
|
|
finally:
|
|
x.XDestroyImage(im)
|
|
|
|
|
|
def flat_video_crop(px, py):
|
|
im = x.XGetImage(d, root, px-40, py-40, 100, 100, 0xffffffff, 2)
|
|
assert im
|
|
try:
|
|
pixels = [x.XGetPixel(im, a, b) & 0xffffff for b in range(100) for a in range(100)]
|
|
# No baked-in pointer: uniform acceptance-window background at this point.
|
|
ref = pixels[0]
|
|
bad = sum(any(abs(((v >> s) & 255) - ((ref >> s) & 255)) > 8
|
|
for s in (0, 8, 16)) for v in pixels)
|
|
return dict(reference=ref, nonuniform_pixels=bad)
|
|
finally:
|
|
x.XDestroyImage(im)
|
|
|
|
|
|
|
|
start_sequence=decoded_sequence()
|
|
results=[]
|
|
for i in range(16):
|
|
px,py=((300,800) if i%2==0 else (1500,900))
|
|
warp(px,py);time.sleep(.18)
|
|
start=time.monotonic_ns();test.XTestFakeButtonEvent(d,1,1,0);x.XSync(d,0)
|
|
time.sleep(.04);test.XTestFakeButtonEvent(d,1,0,0);x.XSync(d,0)
|
|
target=start_sequence+i+1
|
|
while decoded_sequence()!=target and time.monotonic_ns()-start<2000000000:time.sleep(.002)
|
|
seen=decoded_sequence();results.append(dict(seq=target,seen=seen,x=px,y=py));assert seen==target,results[-1]
|
|
time.sleep(.08)
|
|
# Two quick clicks before waiting for their combined numbered video update.
|
|
start=decoded_sequence()
|
|
for i in range(2):
|
|
test.XTestFakeButtonEvent(d,1,1,0);x.XSync(d,0);time.sleep(.04);test.XTestFakeButtonEvent(d,1,0,0);x.XSync(d,0);time.sleep(.07)
|
|
end=time.monotonic()+2
|
|
while decoded_sequence()!=start+2 and time.monotonic()<end:time.sleep(.002)
|
|
print(json.dumps(dict(clicks=results,double_click_start=start,double_click_end=decoded_sequence())))
|