dtourolle
c41f99ea52
scenes.py aimed every press at window pixels, and the develop column had already moved under it: Compose now sits above Adjust, so the old exposure coordinate lands on a straighten slider. Every scene now names what it presses by its accessible label through the automation hook, places points on the photograph relative to the canvas, and opens its photographs by file name. Each starts from a known place and undoes what it did, so one can be recorded alone; the few that continue another's state name it, and running one runs that first into a scratch folder. Each scene also declares the pictures it makes and the sources they depend on. `record.sh --check` fails when the manual shows a picture no scene makes, or a scene makes one it does not show; it reads two files. `record.sh --changed` re-records the scenes whose sources, or own code, changed since the commit that last touched their pictures. record.sh builds with the automation feature, restores the library from DR_LIBRARY_SNAPSHOT, starts from a fresh profile and pins inference to the CPU; the launch screen is recorded from an empty profile of its own. Re-recorded with the ported scenes, and looked at frame by frame. What differs from the pictures they replace: - develop, presets, settings, local, compose, film, wb, light: the current develop column (Compose with Vertical and Horizontal above Adjust, the Label button), otherwise the same moments. - library pictures: the filter bar's colour-label chips; no collection left over from an earlier run in the sidebar; library-selection is the twelve alpine frames rather than eight of them and four New York ones. - library-rating rates two frames nobody had rated, so the stars are set and not cleared. - develop-zoom goes on past 1:1 with the wheel and ends on the file's pixels as hard-edged blocks. - repair covers a real mark on the road, with a size that fits it; film is shown on the Chinatown frame instead of the road. - panorama tries Perspective, Spherical and Cylindrical before filling. - launch, launch-folder and panorama-aligned came out byte-identical.
442 lines
14 KiB
Python
Executable File
442 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Drive the desktop build from outside, for the manual's screenshots.
|
||
|
||
drive.py launch [args...] start the app on the private X server
|
||
drive.py stop
|
||
drive.py shot OUT.png
|
||
drive.py click X Y [button]
|
||
drive.py move X Y
|
||
drive.py drag X1 Y1 X2 Y2 [steps]
|
||
drive.py type TEXT
|
||
drive.py key KEYSYM...
|
||
drive.py rec OUT.mp4 / cut start and stop a recording of the window
|
||
drive.py where the pointer, in window coordinates
|
||
|
||
By name — the accessible label a screen reader reads, answered by the app
|
||
itself when it is built with `--features automation` (see record.sh and
|
||
ui/dr-ui/src/automation.rs):
|
||
|
||
drive.py labels [TEXT] every labelled element on screen (containing TEXT)
|
||
drive.py ids [TEXT] every element with a markup id (containing TEXT)
|
||
drive.py locate NAME what NAME finds, as JSON
|
||
drive.py click-on NAME [button]
|
||
drive.py drag-on NAME DX DY [steps] from the element's centre
|
||
drive.py hold-on NAME SECONDS
|
||
drive.py wait-for NAME [timeout] until it is on screen
|
||
drive.py wait-gone NAME [timeout] until it is not
|
||
|
||
A NAME is an accessible label, exactly: `Exposure`, `Select`, `_MG_8393`.
|
||
Suffixes narrow it — `@ROLE` takes only elements of that role (`Button`,
|
||
`Slider`, `ListItem`, `Text`...), `#N` the N-th match in tree order from 0,
|
||
`#-1` the last (a sheet is drawn after what it covers). A leading `id:`
|
||
names an element by its id in the markup or its component type instead
|
||
(`id:canvas-image`, `id:move-area`, `id:Timeline`), for the few things a
|
||
scene aims at that are not controls. Only what Slint draws is answered,
|
||
and only where its centre is in the window: nothing hidden, nothing
|
||
scrolled out of its list. An element half under the edge of a scrolled
|
||
list still counts, so a scene that needs all of one scrolls it into view.
|
||
|
||
The input itself is still xdotool — a real pointer, pressed and held — so a
|
||
scene exercises exactly what a hand would. The hook only says where to aim.
|
||
|
||
Everything is in *window* pixels, at scale 1, with the window at the origin
|
||
of a 1920×1200 Xvfb on `DR_DISPLAY` (`:7`). The app gets its own XDG
|
||
profile under `DR_HOME` (`/var/tmp/dr-manual`), so nothing here touches the
|
||
library or settings of whoever is logged in; `DR_BIN` names the binary.
|
||
|
||
Two things that cost an afternoon, kept here so they cost nobody else one:
|
||
a Slint `TouchArea` wants a *held* press (`mousedown`, a beat, `mouseup`) —
|
||
xdotool's `click` is sometimes dropped; and under XWayland at scale 2 a
|
||
pointer warp lands at twice the coordinate asked for, which is why this runs
|
||
on Xvfb rather than the desktop.
|
||
"""
|
||
import json
|
||
import os
|
||
import re
|
||
import socket
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
|
||
HOME = os.environ.get('DR_HOME', '/var/tmp/dr-manual')
|
||
DISPLAY = os.environ.get('DR_DISPLAY', ':7')
|
||
BIN = os.environ.get('DR_BIN', 'target/release/darkroom-desktop')
|
||
SOCKET = os.environ.get('DR_AUTOMATION', f'{HOME}/automation.sock')
|
||
|
||
|
||
def app_env(xdg=None):
|
||
"""The app's environment: its own XDG tree (`DR_HOME/xdg` unless told
|
||
otherwise), X11 at scale 1, and the automation socket."""
|
||
xdg = xdg or f'{HOME}/xdg'
|
||
env = dict(
|
||
os.environ,
|
||
XDG_CONFIG_HOME=f'{xdg}/config',
|
||
XDG_DATA_HOME=f'{xdg}/data',
|
||
XDG_STATE_HOME=f'{xdg}/state',
|
||
RUST_LOG='info',
|
||
WINIT_X11_SCALE_FACTOR='1',
|
||
DISPLAY=DISPLAY,
|
||
DR_AUTOMATION=SOCKET,
|
||
)
|
||
env.pop('WAYLAND_DISPLAY', None)
|
||
return env
|
||
|
||
|
||
os.environ['DISPLAY'] = DISPLAY
|
||
|
||
|
||
def x(*args, check=True):
|
||
return subprocess.run(
|
||
['xdotool', *map(str, args)], capture_output=True, text=True, check=check
|
||
).stdout.strip()
|
||
|
||
|
||
def win():
|
||
return open(f'{HOME}/app.win').read().strip()
|
||
|
||
|
||
def geometry():
|
||
out = x('getwindowgeometry', win())
|
||
pos = out.split('Position: ')[1].split(' ')[0]
|
||
px, py = map(int, pos.split(','))
|
||
return px, py
|
||
|
||
|
||
def launch(args, xdg=None):
|
||
os.makedirs(HOME, exist_ok=True)
|
||
log = open(f'{HOME}/app.log', 'w')
|
||
p = subprocess.Popen([BIN, *args], env=app_env(xdg), stdout=log, stderr=subprocess.STDOUT)
|
||
open(f'{HOME}/app.pid', 'w').write(str(p.pid))
|
||
w = ''
|
||
for _ in range(120):
|
||
w = x('search', '--pid', p.pid, '--name', 'DarkRoom', check=False).split('\n')[-1]
|
||
if w:
|
||
break
|
||
time.sleep(0.5)
|
||
time.sleep(1.5)
|
||
W, H = os.environ.get('WIDTH', '1600'), os.environ.get('HEIGHT', '1100')
|
||
x('windowsize', '--sync', w, W, H)
|
||
time.sleep(0.5)
|
||
x('windowmove', '--sync', w, 0, 0)
|
||
x('windowfocus', '--sync', w, check=False)
|
||
open(f'{HOME}/app.win', 'w').write(w)
|
||
print(f'pid {p.pid} win {w}')
|
||
|
||
|
||
def stop():
|
||
"""SIGTERM to the app this profile launched, and wait for it to go.
|
||
Not `windowclose`: winit panics on that under Xvfb."""
|
||
try:
|
||
pid = int(open(f'{HOME}/app.pid').read())
|
||
os.kill(pid, 15)
|
||
except (OSError, ValueError) as e:
|
||
print(e)
|
||
return
|
||
for _ in range(100):
|
||
try:
|
||
os.kill(pid, 0)
|
||
time.sleep(0.1)
|
||
except OSError:
|
||
break
|
||
time.sleep(0.5)
|
||
|
||
|
||
def shot(out):
|
||
subprocess.run(['import', '-window', win(), out], check=True)
|
||
|
||
|
||
def move(px, py):
|
||
ox, oy = geometry()
|
||
x('mousemove', '--sync', ox + int(px), oy + int(py))
|
||
|
||
|
||
def click(px, py, button=1):
|
||
x('windowfocus', '--sync', win(), check=False)
|
||
move(px, py)
|
||
time.sleep(0.15)
|
||
x('mousedown', button)
|
||
time.sleep(0.12)
|
||
x('mouseup', button)
|
||
|
||
|
||
def drag(x1, y1, x2, y2, steps=20):
|
||
x('windowfocus', '--sync', win(), check=False)
|
||
move(x1, y1)
|
||
time.sleep(0.15)
|
||
x('mousedown', 1)
|
||
time.sleep(0.15)
|
||
for i in range(1, int(steps) + 1):
|
||
t = i / int(steps)
|
||
move(int(x1) + (int(x2) - int(x1)) * t, int(y1) + (int(y2) - int(y1)) * t)
|
||
time.sleep(0.03)
|
||
time.sleep(0.15)
|
||
x('mouseup', 1)
|
||
|
||
|
||
def drag_path(points, steps=20):
|
||
"""A drag through several points: out of the grid sideways first, then
|
||
to the row. A diagonal with much vertical in it is taken by the grid's
|
||
Flickable as a scroll before the DragArea can claim it."""
|
||
x('windowfocus', '--sync', win(), check=False)
|
||
(x1, y1), rest = points[0], points[1:]
|
||
move(x1, y1)
|
||
time.sleep(0.15)
|
||
x('mousedown', 1)
|
||
time.sleep(0.15)
|
||
for x2, y2 in rest:
|
||
for i in range(1, int(steps) + 1):
|
||
t = i / int(steps)
|
||
move(int(x1 + (x2 - x1) * t), int(y1 + (y2 - y1) * t))
|
||
time.sleep(0.03)
|
||
x1, y1 = x2, y2
|
||
time.sleep(0.4)
|
||
x('mouseup', 1)
|
||
|
||
|
||
# --- by name ----------------------------------------------------------------
|
||
class NotFound(Exception):
|
||
pass
|
||
|
||
|
||
def ask(request):
|
||
"""One request to the app's automation hook, and its JSON answer."""
|
||
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
|
||
s.settimeout(15)
|
||
s.connect(SOCKET)
|
||
s.sendall(request.encode() + b'\n')
|
||
buf = b''
|
||
while not buf.endswith(b'\n'):
|
||
chunk = s.recv(65536)
|
||
if not chunk:
|
||
break
|
||
buf += chunk
|
||
answer = json.loads(buf)
|
||
if isinstance(answer, dict) and 'error' in answer:
|
||
raise RuntimeError(f'{request}: {answer["error"]}')
|
||
return answer
|
||
|
||
|
||
def wait_ready(timeout=120):
|
||
"""Until the hook answers: the app is up and its event loop running."""
|
||
t0 = time.time()
|
||
while True:
|
||
try:
|
||
if ask('ping') == 'ok':
|
||
return
|
||
except OSError:
|
||
pass
|
||
if time.time() - t0 > timeout:
|
||
raise TimeoutError('the automation hook never answered; was the app built '
|
||
'with --features automation?')
|
||
time.sleep(0.5)
|
||
|
||
|
||
_window = None
|
||
|
||
|
||
def window_size():
|
||
global _window
|
||
if _window is None:
|
||
w = ask('window')
|
||
_window = (w['w'], w['h'])
|
||
return _window
|
||
|
||
|
||
def parse_name(name):
|
||
"""`LABEL`, `id:ID`, each with optional `@ROLE` and `#N`."""
|
||
m = re.match(r'^(.*?)(?:@(\w+))?(?:#(-?\d+))?$', name, re.S)
|
||
what, role, n = m.group(1), m.group(2), m.group(3)
|
||
return what, role, int(n) if n is not None else 0
|
||
|
||
|
||
def matches(name, within=None):
|
||
"""Every on-screen element `name` names, in tree order. `within` is an
|
||
optional (x0, y0, x1, y1) the element's centre must fall inside."""
|
||
what, role, _ = parse_name(name)
|
||
if what.startswith('id:'):
|
||
hits = ask(f'locate-id {what[3:]}')
|
||
else:
|
||
hits = ask(f'locate {what}')
|
||
W, H = window_size()
|
||
hits = [e for e in hits if e['w'] > 0 and e['h'] > 0 and e['opacity'] > 0
|
||
and 0 <= e['x'] + e['w'] / 2 < W and 0 <= e['y'] + e['h'] / 2 < H]
|
||
if role:
|
||
hits = [e for e in hits if (e['role'] or '').lower() == role.lower()]
|
||
if within:
|
||
x0, y0, x1, y1 = within
|
||
hits = [e for e in hits if x0 <= e['x'] + e['w'] / 2 < x1 and y0 <= e['y'] + e['h'] / 2 < y1]
|
||
return hits
|
||
|
||
|
||
def find(name, within=None):
|
||
"""The element `name` names, or NotFound."""
|
||
index = parse_name(name)[2]
|
||
hits = matches(name, within)
|
||
if not -len(hits) <= index < len(hits):
|
||
raise NotFound(f'{name!r}: {len(hits)} on screen')
|
||
return hits[index]
|
||
|
||
|
||
def rect(name, within=None):
|
||
"""(x0, y0, x1, y1) of `name`."""
|
||
e = find(name, within)
|
||
return e['x'], e['y'], e['x'] + e['w'], e['y'] + e['h']
|
||
|
||
|
||
def point(name, fx=0.5, fy=0.5, within=None):
|
||
"""A point inside `name`, as fractions of its width and height."""
|
||
e = find(name, within)
|
||
return int(e['x'] + e['w'] * float(fx)), int(e['y'] + e['h'] * float(fy))
|
||
|
||
|
||
def centre(name, within=None):
|
||
return point(name, within=within)
|
||
|
||
|
||
def present(name, within=None):
|
||
try:
|
||
find(name, within)
|
||
return True
|
||
except NotFound:
|
||
return False
|
||
|
||
|
||
def wait_for(name, timeout=30, within=None):
|
||
t0 = time.time()
|
||
while True:
|
||
try:
|
||
return find(name, within)
|
||
except (NotFound, OSError):
|
||
if time.time() - t0 > float(timeout):
|
||
raise
|
||
time.sleep(0.25)
|
||
|
||
|
||
def wait_gone(name, timeout=30):
|
||
t0 = time.time()
|
||
while time.time() - t0 < float(timeout):
|
||
if not present(name):
|
||
return
|
||
time.sleep(0.25)
|
||
raise TimeoutError(f'{name!r} still on screen after {timeout}s')
|
||
|
||
|
||
def click_on(name, button=1, within=None):
|
||
click(*centre(name, within), button)
|
||
|
||
|
||
def drag_on(name, dx, dy, steps=20, within=None, fx=0.5, fy=0.5):
|
||
"""Drag from a point in `name` (its centre by default) by (dx, dy)."""
|
||
cx, cy = point(name, fx, fy, within)
|
||
drag(cx, cy, cx + int(dx), cy + int(dy), steps)
|
||
|
||
|
||
def hold_on(name, seconds, within=None):
|
||
x('windowfocus', '--sync', win(), check=False)
|
||
move(*centre(name, within))
|
||
time.sleep(0.15)
|
||
x('mousedown', 1)
|
||
time.sleep(float(seconds))
|
||
x('mouseup', 1)
|
||
|
||
|
||
def photo(fx=0.5, fy=0.5):
|
||
"""A point on the photograph in develop, as fractions of the picture as
|
||
drawn — `contain`-fitted into the canvas, so not of the canvas itself."""
|
||
cx0, cy0, cx1, cy1 = rect('id:canvas-image')
|
||
img = ask('canvas')
|
||
cw, ch = cx1 - cx0, cy1 - cy0
|
||
if not img['w'] or not img['h']:
|
||
return int(cx0 + cw * fx), int(cy0 + ch * fy)
|
||
s = min(cw / img['w'], ch / img['h'])
|
||
w, h = img['w'] * s, img['h'] * s
|
||
return int(cx0 + (cw - w) / 2 + w * fx), int(cy0 + (ch - h) / 2 + h * fy)
|
||
|
||
|
||
def rec_start(out):
|
||
size = x('getwindowgeometry', win()).split('Geometry: ')[1].strip()
|
||
ox, oy = geometry()
|
||
p = subprocess.Popen([
|
||
'ffmpeg', '-hide_banner', '-loglevel', 'error', '-f', 'x11grab',
|
||
'-framerate', '15', '-video_size', size, '-i', f'{DISPLAY}+{ox},{oy}',
|
||
'-c:v', 'libx264', '-preset', 'ultrafast', '-qp', '0', '-y', out,
|
||
])
|
||
open(f'{HOME}/rec.pid', 'w').write(str(p.pid))
|
||
time.sleep(0.5)
|
||
|
||
|
||
def rec_stop():
|
||
"""Stop the recording `rec_start` began. The pid file goes with it, so
|
||
a second stop — or one after a scene failed — signals nothing."""
|
||
path = f'{HOME}/rec.pid'
|
||
pid = int(open(path).read())
|
||
os.remove(path)
|
||
os.kill(pid, 2)
|
||
for _ in range(50):
|
||
try:
|
||
os.kill(pid, 0)
|
||
time.sleep(0.1)
|
||
except OSError:
|
||
break
|
||
|
||
|
||
def where():
|
||
out = x('getmouselocation')
|
||
mx, my = [int(v.split(':')[1]) for v in out.split()[:2]]
|
||
ox, oy = geometry()
|
||
print(mx - ox, my - oy)
|
||
|
||
|
||
def print_elements(elements):
|
||
for e in elements:
|
||
name = e['label'] if e['label'] else f"id:{e['id']}" if e['id'] else f"({e['type']})"
|
||
print(f"{e['x']:5.0f} {e['y']:5.0f} {e['w']:5.0f}x{e['h']:<4.0f} "
|
||
f"{(e['role'] or ''):14} {name!r}"
|
||
+ (f" = {e['value']!r}" if e['value'] else ''))
|
||
|
||
|
||
if __name__ == '__main__':
|
||
cmd, *a = sys.argv[1:]
|
||
if cmd == 'launch':
|
||
launch(a)
|
||
elif cmd == 'stop':
|
||
stop()
|
||
elif cmd == 'shot':
|
||
shot(a[0])
|
||
elif cmd == 'click':
|
||
click(*a)
|
||
elif cmd == 'move':
|
||
move(*a)
|
||
elif cmd == 'drag':
|
||
drag(*a)
|
||
elif cmd == 'type':
|
||
x('windowfocus', '--sync', win(), check=False)
|
||
x('type', '--delay', '120', a[0])
|
||
elif cmd == 'key':
|
||
x('windowfocus', '--sync', win(), check=False)
|
||
x('key', '--delay', '80', *a)
|
||
elif cmd == 'rec':
|
||
rec_start(a[0])
|
||
elif cmd == 'cut':
|
||
rec_stop()
|
||
elif cmd == 'where':
|
||
where()
|
||
elif cmd == 'labels':
|
||
print_elements(ask('labels ' + (a[0] if a else '')))
|
||
elif cmd == 'ids':
|
||
print_elements(ask('ids ' + (a[0] if a else '')))
|
||
elif cmd == 'locate':
|
||
print(json.dumps(matches(a[0]), indent=1))
|
||
elif cmd == 'click-on':
|
||
click_on(*a)
|
||
elif cmd == 'drag-on':
|
||
drag_on(*a)
|
||
elif cmd == 'hold-on':
|
||
hold_on(*a)
|
||
elif cmd == 'wait-for':
|
||
print(wait_for(*a))
|
||
elif cmd == 'wait-gone':
|
||
wait_gone(*a)
|
||
else:
|
||
sys.exit(__doc__)
|