ba-auto-daily/ba_auto/driver.py

"""Local PC/Steam/Proton control backend (xdotool/scrot wrappers)."""
import os
import subprocess
import time

import cv2

from ba_auto import config

PROBE_SHOT_PATH = os.path.join(config.SCRATCHPAD_DIR, "probe.png")


def run_command(args, **kwargs):
    kwargs.setdefault("env", config.ENV)
    kwargs.setdefault("check", True)
    return subprocess.run(args, **kwargs)


def focus_game():
    result = run_command(
        ["xdotool", "search", "--name", config.WINDOW_NAME],
        check=False, capture_output=True, text=True,
    )
    window_ids = result.stdout.split()
    if not window_ids:
        raise RuntimeError("Blue Archive window not found. Is the game running?")
    run_command(["xdotool", "windowactivate", window_ids[0]])
    # windowactivate alone isn't enough -- confirmed live (twice now, see
    # Handoff.md's original finding and plan.md's Bounty phase for a second
    # occurrence) that a stray anti-cheat XIGNCODE window can render
    # visibly on top of the game and intercept clicks at fixed positions
    # (notably the shared BACK_BUTTON coordinate) even while xdotool still
    # reports BlueArchive as the "active" window -- windowactivate changes
    # input focus, not stacking order, so it doesn't fix this by itself.
    # windowraise does. Cheap and harmless when no overlay is present.
    run_command(["xdotool", "windowraise", window_ids[0]])
    wait(0.5)


def click(x, y):
    # A combined "mousemove X Y click 1" invocation is unreliable here --
    # empirically (see plan.md's click-flakiness writeups) the game's input
    # handler sometimes misses clicks fired before it's processed the move.
    # Splitting into separate commands with a short pause between fixes it.
    run_command(["xdotool", "mousemove", str(x), str(y)])
    wait(0.2)
    run_command(["xdotool", "click", "1"])
    wait(0.5)


def move_mouse(x, y):
    run_command(["xdotool", "mousemove", str(x), str(y)])


def drag(start_x, start_y, end_x, end_y, duration=0.6, steps=12):
    """Click-and-drag gesture: mousedown at (start_x, start_y), move to
    (end_x, end_y) over `duration` seconds in `steps` increments, mouseup.

    Distinct from scroll()'s wheel-based gesture -- scroll()'s own comment
    already documents that a plain drag does NOT register as a list-scroll
    gesture in this Proton client, but that finding was specific to
    scrollable list widgets (mailbox/lesson/bounty/etc.); a room-view
    camera (e.g. cafe.py's horizontal pan) is a different UI surface, not a
    list, and needs an actual drag rather than a wheel click.
    """
    run_command(["xdotool", "mousemove", str(start_x), str(start_y)])
    wait(0.2)
    run_command(["xdotool", "mousedown", "1"])
    step_delay = duration / steps
    for i in range(1, steps + 1):
        x = start_x + (end_x - start_x) * i // steps
        y = start_y + (end_y - start_y) * i // steps
        run_command(["xdotool", "mousemove", str(x), str(y)])
        wait(step_delay)
    run_command(["xdotool", "mouseup", "1"])
    wait(0.3)


def scroll(x, y, direction, clicks=1):
    # xdotool button 4/5 = scroll wheel up/down. A drag (mousedown/move/mouseup)
    # does not register as a list-scroll gesture in this Proton client; the
    # wheel does.
    button = "4" if direction == "up" else "5"
    run_command(["xdotool", "mousemove", str(x), str(y)])
    wait(0.2)
    for _ in range(clicks):
        run_command(["xdotool", "click", button])
        wait(0.15)
    wait(0.3)


def keypress(key):
    run_command(["xdotool", "key", key])
    wait(0.5)


def screenshot(path):
    run_command(["scrot", "-a", "0,0,1920,1200", "-o", path])


def read_screenshot(path):
    """screenshot() + cv2.imread(), retrying the whole capture if the decode
    fails. scrot occasionally writes a truncated/corrupt PNG (observed live
    as a libpng "IDAT: invalid block type" decode error) while still exiting
    0, so a bad frame is a transient condition, not a hard failure -- same
    "retry, don't crash" contract as the rest of this project's screen-state
    checks. cv2.imread() returns None on a decode failure rather than
    raising, which used to propagate straight into a bare image[y, x] and
    crash the whole run (see plan.md/Handoff.md for the live incident this
    was found from, mid schedule-grid-close in lesson.py).

    Used for every screenshot-then-decode call site in this project
    (color_at/colors_at below, and detector.py's OCR/template-match reads)
    rather than each one pairing its own screenshot()+cv2.imread() and
    duplicating this retry -- see detector.py's OCR_SHOT_PATH/
    SPARKLE_SHOT_PATH call sites.
    """
    last_error = None
    for attempt in range(1, config.SCREENSHOT_DECODE_RETRIES + 1):
        screenshot(path)
        image = cv2.imread(path)
        if image is not None:
            return image
        last_error = f"cv2.imread returned None for {path} (attempt {attempt}/{config.SCREENSHOT_DECODE_RETRIES})"
        print(f"[driver] {last_error} -- retrying capture")
        wait(0.5)
    raise RuntimeError(f"[driver] giving up on screenshot capture: {last_error}")


def color_at(x, y):
    image = read_screenshot(PROBE_SHOT_PATH)
    b, g, r = image[y, x]
    return int(r), int(g), int(b)


def colors_at(points, image=None):
    """Sample multiple (x, y) points from a single screenshot, instead of one
    scrot capture per point -- added for navigation.is_header_bar_visible's
    multi-point header check, which otherwise called color_at() 8 times (8
    full-screen captures) for one logical check. Also more correct than
    looping color_at(): all points come from the exact same frame rather
    than points sampled sequentially across several hundred ms of separate
    captures, which could straddle a screen transition.

    `image` optionally supplies an already-decoded frame instead of taking a
    fresh screenshot -- for callers reading several independent things (e.g.
    cafe.py's rank-up header check + sparkle template match) off one shared
    capture rather than one apiece.

    Returns a list of (r, g, b) tuples in the same order as `points`.
    """
    img = image if image is not None else read_screenshot(PROBE_SHOT_PATH)
    colors = []
    for x, y in points:
        b, g, r = img[y, x]
        colors.append((int(r), int(g), int(b)))
    return colors


def wait(seconds):
    time.sleep(seconds)


def kill_game():
    # check=False -- pkill exits 1 with no output when nothing matches,
    # which is a normal outcome here (e.g. the process already died on its
    # own), not an error worth raising on.
    run_command(["pkill", "-f", config.GAME_PROCESS_NAME], check=False)


def launch_game():
    # The launch script (see config.GAME_LAUNCH_SCRIPT) blocks until the
    # game process exits, by design (it's meant to be run as a foreground
    # session launcher) -- Popen + start_new_session=True detaches it so
    # this process can keep polling for the window instead of hanging for
    # the whole play session.
    subprocess.Popen(
        [config.GAME_LAUNCH_SCRIPT],
        env=config.ENV,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        stdin=subprocess.DEVNULL,
        start_new_session=True,
    )


def is_game_running():
    result = run_command(
        ["pgrep", "-f", config.GAME_PROCESS_NAME],
        check=False, capture_output=True, text=True,
    )
    return bool(result.stdout.strip())


def window_exists():
    result = run_command(
        ["xdotool", "search", "--name", config.WINDOW_NAME],
        check=False, capture_output=True, text=True,
    )
    return bool(result.stdout.strip())