#!/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()