feat: Linux bulk importer for the Allflame Voyage Solver
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>
This commit is contained in:
Claude Opus 5
Executable
+299
@@ -0,0 +1,299 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user