templisandClaude Opus 5
e65f9db3ae
Mirrors the solver's Windows-only voyage-import.ahk on KDE Wayland: - poe-clip-bridge.py: clipboard watcher that keeps only PoE item text, serves it on loopback and can trigger a sweep from the browser - voyage-sweep.py: hovers every cell of the in-game chart grid, Ctrl+C's it and hands over the whole batch in one go - userscript: feeds the batch into the solver page without focusing it Includes notes on the four Wayland pointer pitfalls this ran into. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
300 lines
11 KiB
Python
Executable File
300 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Bulk-import PoE Charts into the Allflame Voyage Solver (Linux/KDE Wayland).
|
||
|
||
Linux counterpart of the site's voyage-import.ahk: hovers every cell of the
|
||
in-game chart grid, Ctrl+C's it, collects the item texts and puts the whole
|
||
batch into the clipboard, where poe-clip-bridge + the Violentmonkey userscript
|
||
pick it up and paste it into the solver.
|
||
|
||
voyage-sweep.py calibrate # capture the grid corners
|
||
voyage-sweep.py run # do the sweep
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import pathlib
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
CONFIG = pathlib.Path.home() / ".config" / "voyage-sweep.json"
|
||
DEFAULTS = {
|
||
"cols": 6,
|
||
"rows": 10,
|
||
"tl": None, # [x, y] centre of the top-left cell
|
||
"br": None, # [x, y] centre of the bottom-right cell
|
||
"hover_delay": 0.14, # let the tooltip build before Ctrl+C
|
||
"clip_timeout": 0.6, # how long to wait for clipboard content
|
||
"poe_window": "Path of Exile",
|
||
}
|
||
CHART_RE = re.compile(r"^[ \t]*Item Class\s*[:：]", re.IGNORECASE | re.MULTILINE)
|
||
MAX_STEP = 150 # larger relative jumps get distorted by pointer acceleration
|
||
ENV = {**os.environ, "YDOTOOL_SOCKET": f"/run/user/{os.getuid()}/.ydotool_socket"}
|
||
|
||
|
||
def run(cmd, **kw):
|
||
return subprocess.run(cmd, env=ENV, capture_output=True, text=True, **kw)
|
||
|
||
|
||
def load_config():
|
||
cfg = dict(DEFAULTS)
|
||
if CONFIG.exists():
|
||
cfg.update(json.loads(CONFIG.read_text()))
|
||
return cfg
|
||
|
||
|
||
def save_config(cfg):
|
||
CONFIG.write_text(json.dumps(cfg, indent=2) + "\n")
|
||
|
||
|
||
# ---------- pointer ----------
|
||
|
||
def cursor():
|
||
"""Pointer position. Reliable while the pointer is over an XWayland window."""
|
||
out = run(["xdotool", "getmouselocation", "--shell"]).stdout
|
||
pos = dict(l.split("=", 1) for l in out.strip().splitlines() if "=" in l)
|
||
return int(pos["X"]), int(pos["Y"])
|
||
|
||
|
||
def cursor_kwin():
|
||
"""Pointer position straight from KWin. Slow (~0.3s) but always correct,
|
||
also while the pointer sits on a Wayland-native window where X sees nothing."""
|
||
js = pathlib.Path(f"/tmp/voyage-probe-{os.getpid()}.js")
|
||
js.write_text('print("VOYAGEPROBE " + workspace.cursorPos.x + ","'
|
||
' + workspace.cursorPos.y);\n')
|
||
name = f"voyageprobe{os.getpid()}"
|
||
since = time.strftime("%Y-%m-%d %H:%M:%S")
|
||
sid = run(["qdbus6", "org.kde.KWin", "/Scripting",
|
||
"org.kde.kwin.Scripting.loadScript", str(js), name]).stdout.strip()
|
||
run(["qdbus6", "org.kde.KWin", f"/Scripting/Script{sid}",
|
||
"org.kde.kwin.Script.run"])
|
||
time.sleep(0.25)
|
||
log = run(["journalctl", "_COMM=kwin_wayland", "--since", since,
|
||
"--no-pager", "-o", "cat"]).stdout
|
||
run(["qdbus6", "org.kde.KWin", "/Scripting",
|
||
"org.kde.kwin.Scripting.unloadScript", name])
|
||
js.unlink(missing_ok=True)
|
||
hits = re.findall(r"VOYAGEPROBE (\d+),(\d+)", log)
|
||
if not hits:
|
||
raise RuntimeError("KWin cursor probe returned nothing")
|
||
return int(hits[-1][0]), int(hits[-1][1])
|
||
|
||
|
||
# Injected motion is not applied 1:1 - this display multiplies it (measured 2.0).
|
||
# The factor is learned from the actual movement instead of being hardcoded.
|
||
_gain = 1.0
|
||
|
||
|
||
def move_to(x, y, tries=40, coarse=False, tol=1):
|
||
"""Closed loop: step, re-read, correct, and learn the motion gain."""
|
||
global _gain
|
||
read = cursor_kwin if coarse else cursor
|
||
for _ in range(tries):
|
||
cx, cy = read()
|
||
dx, dy = x - cx, y - cy
|
||
if abs(dx) <= tol and abs(dy) <= tol:
|
||
return True
|
||
sx = max(-MAX_STEP, min(MAX_STEP, round(dx / _gain)))
|
||
sy = max(-MAX_STEP, min(MAX_STEP, round(dy / _gain)))
|
||
run(["ydotool", "mousemove", "-x", str(sx), "-y", str(sy)])
|
||
time.sleep(0.03)
|
||
nx, ny = read()
|
||
for sent, moved in ((sx, nx - cx), (sy, ny - cy)):
|
||
if abs(sent) >= 20 and abs(moved) >= 1:
|
||
_gain = max(0.2, min(8.0, 0.7 * _gain + 0.3 * (moved / sent)))
|
||
break
|
||
return False
|
||
|
||
|
||
def ctrl_c():
|
||
run(["ydotool", "key", "29:1", "46:1", "46:0", "29:0"]) # LEFTCTRL + C
|
||
|
||
|
||
# ---------- clipboard ----------
|
||
|
||
def clip_arm(token):
|
||
"""Put a unique marker in the clipboard.
|
||
|
||
'wl-copy --clear' is asynchronous: a poll right after it still reads the
|
||
previous owner's data, which made every cell report the first chart again.
|
||
Waiting for the marker to disappear is race-free.
|
||
"""
|
||
subprocess.run(["wl-copy"], input=token, text=True, env=ENV)
|
||
|
||
|
||
def clip_read():
|
||
r = run(["wl-paste", "--no-newline", "--type", "text/plain"])
|
||
return r.stdout if r.returncode == 0 else ""
|
||
|
||
|
||
def clip_wait(timeout, token):
|
||
"""Wait until the game replaced our marker with the copied item text."""
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
text = clip_read().strip()
|
||
if text and text != token:
|
||
return text
|
||
time.sleep(0.03)
|
||
return ""
|
||
|
||
|
||
# ---------- grid ----------
|
||
|
||
def poe_geometry(cfg):
|
||
"""Window origin+size of the PoE window (an XWayland client, so X knows it)."""
|
||
wid = run(["xdotool", "search", "--name", cfg["poe_window"]]).stdout.split()
|
||
if not wid:
|
||
sys.exit(f"No window named '{cfg['poe_window']}' found - is PoE running?")
|
||
out = run(["xdotool", "getwindowgeometry", "--shell", wid[-1]]).stdout
|
||
g = dict(l.split("=", 1) for l in out.strip().splitlines() if "=" in l)
|
||
return int(g["X"]), int(g["Y"]), int(g["WIDTH"]), int(g["HEIGHT"])
|
||
|
||
|
||
def cell_pos(cfg, row, col, origin):
|
||
"""Grid coordinates are stored relative to the window, so moving the
|
||
PoE window does not invalidate the calibration."""
|
||
tlx, tly = cfg["tl"]
|
||
brx, bry = cfg["br"]
|
||
dx = (brx - tlx) / (cfg["cols"] - 1) if cfg["cols"] > 1 else 0
|
||
dy = (bry - tly) / (cfg["rows"] - 1) if cfg["rows"] > 1 else 0
|
||
return round(origin[0] + tlx + col * dx), round(origin[1] + tly + row * dy)
|
||
|
||
|
||
def bridge(action):
|
||
"""Pause/resume the clipboard bridge over its local HTTP API.
|
||
|
||
Not 'systemctl stop': when the sweep is started by the bridge itself
|
||
(Firefox bookmark), stopping the unit would kill this process too.
|
||
"""
|
||
token_file = pathlib.Path.home() / ".config" / "poe-clip-bridge" / "token"
|
||
if not token_file.exists():
|
||
return
|
||
token = urllib.parse.quote(token_file.read_text().strip())
|
||
try:
|
||
urllib.request.urlopen(
|
||
f"http://127.0.0.1:8477/{action}?token={token}", timeout=2).read()
|
||
except OSError:
|
||
pass # bridge not running - sweep still works, just no auto-import
|
||
|
||
|
||
def countdown(msg, seconds=5):
|
||
print(msg)
|
||
for i in range(seconds, 0, -1):
|
||
print(f" {i} ...", end="\r", flush=True)
|
||
time.sleep(1)
|
||
print(" " * 20, end="\r")
|
||
|
||
|
||
# ---------- commands ----------
|
||
|
||
def cmd_calibrate(cfg, args):
|
||
cfg["cols"], cfg["rows"] = args.cols, args.rows
|
||
ox, oy = poe_geometry(cfg)[:2]
|
||
print(f"Grid: {cfg['cols']} columns x {cfg['rows']} rows. Window at {ox},{oy}.")
|
||
countdown("\nPut the mouse on the CENTRE of the TOP-LEFT cell.")
|
||
cx, cy = cursor()
|
||
cfg["tl"] = [cx - ox, cy - oy]
|
||
print(f" top-left = {cfg['tl']} (window-relative)")
|
||
countdown("\nNow the CENTRE of the BOTTOM-RIGHT cell (even if empty).")
|
||
cx, cy = cursor()
|
||
cfg["br"] = [cx - ox, cy - oy]
|
||
print(f" bottom-right = {cfg['br']} (window-relative)")
|
||
if cfg["tl"] == cfg["br"]:
|
||
sys.exit("Both corners are identical - nothing saved.")
|
||
save_config(cfg)
|
||
print(f"\nSaved to {CONFIG}. Cell spacing: "
|
||
f"{(cfg['br'][0] - cfg['tl'][0]) / max(1, cfg['cols'] - 1):.1f} x "
|
||
f"{(cfg['br'][1] - cfg['tl'][1]) / max(1, cfg['rows'] - 1):.1f} px")
|
||
|
||
|
||
def preflight():
|
||
sock = pathlib.Path(ENV["YDOTOOL_SOCKET"])
|
||
if not sock.exists():
|
||
sys.exit("ydotoold is not running (no %s).\nStart it with: "
|
||
"systemctl --user start ydotool.service" % sock)
|
||
|
||
|
||
def cmd_run(cfg, args):
|
||
if not cfg["tl"] or not cfg["br"]:
|
||
sys.exit("Not calibrated yet - run: voyage-sweep.py calibrate")
|
||
preflight()
|
||
rows = args.rows or cfg["rows"]
|
||
|
||
ox, oy, ww, wh = poe_geometry(cfg)
|
||
subprocess.run(["kdotool", "search", "--name", cfg["poe_window"],
|
||
"windowactivate"], capture_output=True)
|
||
countdown("\nSweeping - keep hands off mouse and keyboard.", args.delay)
|
||
|
||
# The fast X11 pointer read only works while the pointer is over the game,
|
||
# so get it in there first using KWin's own (slower) cursor position.
|
||
if not move_to(ox + ww // 2, oy + wh // 2, coarse=True, tol=25):
|
||
sys.exit("Could not move the pointer into the PoE window.")
|
||
|
||
bridge("pause") # otherwise every single copy gets imported separately
|
||
charts, skipped, first, identical = [], 0, None, True
|
||
try:
|
||
for row in range(rows):
|
||
for col in range(cfg["cols"]):
|
||
x, y = cell_pos(cfg, row, col, (ox, oy))
|
||
token = f"voyage-sweep-{row}-{col}-{time.monotonic_ns()}"
|
||
clip_arm(token)
|
||
if not move_to(x, y):
|
||
print(f" cell {row + 1}/{col + 1}: pointer did not reach {x},{y}")
|
||
skipped += 1
|
||
continue
|
||
time.sleep(cfg["hover_delay"])
|
||
ctrl_c()
|
||
text = clip_wait(cfg["clip_timeout"], token)
|
||
if not text or not CHART_RE.search(text):
|
||
skipped += 1
|
||
continue
|
||
charts.append(text)
|
||
if first is None:
|
||
first = text
|
||
elif text != first:
|
||
identical = False
|
||
print(f" charts {len(charts)} skipped {skipped}", end="\r", flush=True)
|
||
except KeyboardInterrupt:
|
||
print("\nAborted.")
|
||
finally:
|
||
bridge("resume")
|
||
|
||
print(" " * 40, end="\r")
|
||
if len(charts) >= 5 and identical:
|
||
sys.exit("\nEvery cell copied the SAME chart - calibration is off "
|
||
"(or the PoE window moved). Re-run: voyage-sweep.py calibrate")
|
||
if not charts:
|
||
sys.exit(f"\nNothing copied ({skipped} empty/non-chart cells).")
|
||
|
||
blob = "\n".join(charts)
|
||
subprocess.run(["wl-copy"], input=blob, text=True)
|
||
print(f"\n{len(charts)} charts copied, {skipped} cells empty/non-chart.")
|
||
print("Batch is in the clipboard - the userscript imports it into the solver.")
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description=__doc__,
|
||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||
|
||
cal = sub.add_parser("calibrate", help="capture the grid corners")
|
||
cal.add_argument("--cols", type=int, default=DEFAULTS["cols"])
|
||
cal.add_argument("--rows", type=int, default=DEFAULTS["rows"])
|
||
|
||
sweep = sub.add_parser("run", help="sweep the grid and import")
|
||
sweep.add_argument("--delay", type=int, default=5,
|
||
help="seconds before the sweep starts (default 5)")
|
||
sweep.add_argument("--rows", type=int, help="sweep only the first N rows")
|
||
|
||
args = ap.parse_args()
|
||
cfg = load_config()
|
||
{"calibrate": cmd_calibrate, "run": cmd_run}[args.cmd](cfg, args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|