exapunks-hackmatch-bot/bot.py

#!/usr/bin/env python3
import os
import time
from collections import deque
from itertools import count
from Xlib import error
from detection import NOBLOCK
from interaction import get_exapunks_window, focus_window, screenshot_board, \
                        press_keys, listen_keys
from strategy import State


MAX_SPEED_ROWS = 3


def save_screenshot(win):
    board = screenshot_board(win).convert('RGB')
    os.makedirs('screens', exist_ok=True)
    for i in count(1):
        path = 'screens/board%d.png' % i
        if not os.path.exists(path):
            print('save screenshot in', path)
            board.save(path)
            break


if __name__ == '__main__':
    import sys
    verbose = '-q' not in sys.argv[1:]

    try:
        win = get_exapunks_window()
        focus_window(win)

        listen_keys({'s': lambda: save_screenshot(win)})

        buf = deque([], maxlen=3)

        def vprint(*args, **kwargs):
            if verbose:
                print(*args, **kwargs)

        def vprint_state(state):
            if verbose:
                state.print()

        while True:
            try:
                board = screenshot_board(win)

                start = time.time()
                state = State.detect(board)
                end = time.time()
                vprint('\033c', end='')
                vprint('parsed in', round((end - start) * 1000, 1), 'ms:')
                vprint_state(state)
                vprint()

                assert state.exa is not None

            except (TypeError, AssertionError):
                vprint('\rerror during parsing, wait for a bit...', end='')
                time.sleep(0.050)
                continue
            except error.BadMatch:
                vprint('\rEXAPUNKS window lost, wait for a bit...', end='')
                time.sleep(0.500)
                continue

            try:
                start = time.time()
                newstate = state.solve()
                end = time.time()
                vprint('thought for', round((end - start) * 1000, 1), 'ms')
            except AssertionError:
                print('error board 99:')
                state.print()
                board.convert('RGB').save('screens/board99.png')
                break

            if state.held == NOBLOCK and any(map(newstate.loops, buf)):
                vprint('\rloop detected, wait for a bit...', end='')
                time.sleep(0.03)
            elif newstate.moves:
                vprint('moves:', newstate.keys())
                vprint('score:', newstate.score)
                vprint()

                vprint('target after moves:')
                vprint_state(newstate)

                press_keys(win, newstate.keys())

                #keys_delay = len(newstate.moves) * 2 * KEY_DELAY
                #moves_delay = max(0, newstate.delay() - keys_delay)
                #vprint('wait for',  moves_delay, 'ms')
                #time.sleep(moves_delay / 1000)
                time.sleep(0.080)
            elif state.nrows - 2 <= MAX_SPEED_ROWS:
                vprint('no moves, speed up')
                press_keys(win, 'l')
                time.sleep(0.030)
            else:
                vprint('no moves')

            buf.append(newstate)
    except KeyboardInterrupt:
        print('interrupted, quitting')