54 lines
2.3 KiB
Python
54 lines
2.3 KiB
Python
"""Observe actual deployed state/receiver; keep the test seat awake with motion.
|
|
|
|
Run on the box only while a controlled native-cursor session is active.
|
|
No clicks, protocol injection, or changes to the state producer are made.
|
|
"""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
duration = float(sys.argv[1])
|
|
output = Path(sys.argv[2])
|
|
pid = int(subprocess.check_output(['pgrep', '-x', 'moonlight']).strip())
|
|
proc = Path('/proc') / str(pid)
|
|
env = dict(x.split('=', 1) for x in (proc / 'environ').read_bytes().decode().split('\0') if '=' in x)
|
|
state = Path(env['VDI_NATIVE_CURSOR_STATE'])
|
|
start = time.monotonic()
|
|
last_motion = 0
|
|
result = dict(pid=pid, requested_seconds=duration, samples=0,
|
|
timestamp_after_pre_read_clock=0, timestamp_after_post_read_clock=0,
|
|
max_age_ms=0, survived=False)
|
|
try:
|
|
while time.monotonic() - start < duration:
|
|
if not proc.exists() or (proc / 'comm').read_text().strip() != 'moonlight':
|
|
raise RuntimeError('receiver exited')
|
|
before = time.monotonic_ns()
|
|
packet = state.read_bytes()
|
|
after = time.monotonic_ns()
|
|
stamp = struct.unpack_from('<Q', packet, 24)[0]
|
|
result['samples'] += 1
|
|
result['timestamp_after_pre_read_clock'] += stamp > before
|
|
result['timestamp_after_post_read_clock'] += stamp > after
|
|
result['max_age_ms'] = max(result['max_age_ms'], (after - stamp) / 1e6)
|
|
result['last_sequence'] = struct.unpack_from('<Q', packet, 16)[0]
|
|
if stamp > after or after - stamp > 2_000_000_000:
|
|
raise RuntimeError('invalid post-read timestamp')
|
|
now = time.monotonic()
|
|
if now - last_motion > 10:
|
|
px = 1800 + (int(now - start) // 10 % 2) * 50
|
|
motion_env = dict(os.environ, DISPLAY=':0', XAUTHORITY='/home/armbian/.Xauthority')
|
|
subprocess.run(['xdotool', 'mousemove', str(px), '1100'], env=motion_env, check=True)
|
|
last_motion = now
|
|
time.sleep(.004)
|
|
result['survived'] = True
|
|
except Exception as error:
|
|
result['error'] = str(error)
|
|
finally:
|
|
result['elapsed_seconds'] = round(time.monotonic() - start, 3)
|
|
output.write_text(json.dumps(result, indent=2) + '\n')
|
|
print(json.dumps(result), flush=True)
|