exapunks-hackmatch-bot/strategy.py

import io
import time
from collections import deque
from contextlib import redirect_stdout
from copy import copy
from itertools import combinations
from detection import COLUMNS, NOBLOCK, detect_blocks, detect_exa, \
                      detect_held, print_board, is_basic, is_bomb


GRAB, DROP, SWAP, LEFT, RIGHT, SPEED = range(6)
MOVE_DELAYS = (
    # in milliseconds
    50,  # GRAB
    50,  # DROP
    50,  # SWAP
    30,  # LEFT
    30,  # RIGHT
    30,  # SPEED
)
MIN_BASIC_GROUP_SIZE = 4
MIN_BOMB_GROUP_SIZE = 2
COLSIZE_PRIO = 5
COLSIZE_PANIC = 8
BOMB_POINTS = 5


class State:
    def __init__(self, blocks, exa, held, colskip, busy, moves, blockgroups, groups):
        self.blocks = blocks
        self.exa = exa
        self.held = held
        self.colskip = colskip
        self.busy = busy
        self.moves = moves
        self.blockgroups = blockgroups
        self.groups = groups
        self.nrows = len(blocks) // COLUMNS

    @classmethod
    def detect(cls, board, pad=2):
        blocks = [NOBLOCK] * (COLUMNS * pad) + list(detect_blocks(board))
        exa = detect_exa(board)
        held = detect_held(board, exa)
        colskip = get_colskip(blocks)
        busy = get_busy(blocks, colskip)
        blockgroups, groups = get_groups(blocks)
        return cls(bytearray(blocks), exa, held, bytearray(colskip), busy, (),
                   bytearray(blockgroups), groups)

    def copy(self, deep):
        mcopy = copy if deep else lambda x: x
        return self.__class__(mcopy(self.blocks),
                              self.exa,
                              self.held,
                              mcopy(self.colskip),
                              self.busy,
                              self.moves,
                              mcopy(self.blockgroups),
                              mcopy(self.groups))

    def colbusy(self, col):
        return (self.busy >> col) & 1

    def colrows(self, col):
        return self.nrows - self.colskip[col]

    def maxrows(self):
        return max(map(self.colrows, range(COLUMNS)))

    def causes_panic(self):
        return self.max_colsize() >= COLSIZE_PANIC

    def max_colsize(self):
        return self.nrows - self.empty_rows()

    def empty_rows(self):
        for i, block in enumerate(self.blocks):
            if block != NOBLOCK:
                return i // COLUMNS
        return 0

    def holes(self):
        #return self.imbalance()

        start_row = self.empty_rows()
        score = 0
        for col in range(COLUMNS):
            for row in range(start_row, self.nrows):
                if self.blocks[row * COLUMNS + col] != NOBLOCK:
                    break
                score += row - start_row + 1
        return score

    def locked(self, i):
        size, block = self.groups[self.blockgroups[i]]
        if block == NOBLOCK:
            return False
        if is_basic(block):
            return size >= MIN_BASIC_GROUP_SIZE
        assert is_bomb(block)
        return size >= MIN_BOMB_GROUP_SIZE

    def move(self, *moves):
        deep = any(move in (GRAB, DROP, SWAP) for move in moves)
        s = self.copy(deep)
        s.moves += moves

        for move in moves:
            if move == LEFT:
                assert s.exa > 0
                s.exa -= 1

            elif move == RIGHT:
                assert s.exa < COLUMNS - 1
                s.exa += 1

            elif move == GRAB:
                assert not s.colbusy(s.exa)
                assert s.held == NOBLOCK
                row = s.colskip[s.exa]
                assert row < s.nrows
                i = row * COLUMNS + s.exa
                if not s.locked(i):
                    s.colskip[s.exa] += 1
                    s.held = s.remove(i)

            elif move == DROP:
                if s.held != NOBLOCK:
                    row = s.colskip[s.exa]
                    assert row > 0
                    i = (row - 1) * COLUMNS + s.exa
                    s.colskip[s.exa] -= 1
                    s.place(i, s.held)
                    s.held = NOBLOCK

            elif move == SWAP:
                assert not s.colbusy(s.exa)
                row = s.colskip[s.exa]
                i = row * COLUMNS + s.exa
                j = i + COLUMNS

                if j < len(s.blocks) and not s.locked(i) and not s.locked(j):
                    bi = s.blocks[i]
                    bj = s.blocks[j]
                    assert bi != NOBLOCK
                    assert bj != NOBLOCK
                    if bi != bj:
                        s.blocks[i] = NOBLOCK
                        s.blocks[j] = NOBLOCK
                        visited = set()
                        s.ungroup(i, visited)
                        s.ungroup(j, visited)
                        s.place(j, bi)
                        s.place(i, bj)

        return s

    def remove(self, i):
        block = self.blocks[i]
        assert block != NOBLOCK
        self.blocks[i] = NOBLOCK
        self.ungroup(i, set())
        return block

    def ungroup(self, i, visited):
        assert self.blocks[i] == NOBLOCK
        oldid = self.blockgroups[i]
        self.blockgroups[i] = 0
        self.groups[oldid] = 0, NOBLOCK

        for nb in neighbors(i, self.blocks):
            if self.blockgroups[nb] == oldid:
                self.create_group(nb, visited)

    def place(self, i, block):
        assert block != NOBLOCK
        assert self.blocks[i] == NOBLOCK
        self.blocks[i] = block
        self.create_group(i, set())

    def create_group(self, start, visited):
        def scan(i):
            if i not in visited:
                yield i
                visited.add(i)
                for nb in neighbors(i, self.blocks):
                    if self.blocks[nb] == block:
                        yield from scan(nb)

        block = self.blocks[start]
        group = tuple(scan(start))
        if group:
            newid = len(self.groups)
            self.groups.append((len(group), block))
            for j in group:
                self.blockgroups[j] = newid

    def fragmentation(self):
        """
        Minimize the sum of dist(i,j) between all blocks i,j of the same color.
        Magnify vertical distances to avoid column stacking.
        Add the vertical components of i and j to prioritize lower columns.
        """
        def dist(i, j):
            yi, xi = divmod(i, COLUMNS)
            yj, xj = divmod(j, COLUMNS)

            # for blocks in the same group, only count vertical distance so that
            # groups are spread out horizontally
            if self.blockgroups[i] == self.blockgroups[j]:
                return abs(yj - yi)

            return abs(xj - xi) + abs(yj - yi) * 2 - 1 + \
                   (self.nrows - yi) + (self.nrows - yj)

        colors = {}
        for i, block in enumerate(self.blocks):
            if block != NOBLOCK:
                colors.setdefault(block, []).append(i)

        return sum(dist(i, j) ** 2
                   for blocks in colors.values()
                   for i, j in combinations(blocks, 2))

    def points(self):
        points = 0

        for size, block in self.groups:
            if is_basic(block) and size >= MIN_BASIC_GROUP_SIZE:
                points += size
            elif is_bomb(block) and size >= MIN_BOMB_GROUP_SIZE:
                points += self.maxrows()

        return -points

    def imbalance(self):
        colsizes = tuple(self.nrows - skip
                         for col, skip in enumerate(self.colskip))
                         #if not self.colbusy(col))
        mean = sum(colsizes) / len(colsizes)
        return sum((size - mean) ** 2 for size in colsizes)

    def gen_moves(self):
        if self.held == NOBLOCK:
            yield self

            for src in self.gen_shift(True):
                yield from src.gen_stationary()

                for get in src.gen_get():
                    for dst in get.gen_shift(False):
                        yield from dst.gen_put()
        else:
            for dst in self.gen_shift(True):
                yield from dst.gen_put()

    def gen_shift(self, allow_noshift):
        if allow_noshift:
            yield self

        left = self
        for i in range(self.exa):
            left = left.move(LEFT)
            yield left

        right = self
        for i in range(COLUMNS - self.exa - 1):
            right = right.move(RIGHT)
            yield right

    def gen_stationary(self):
        # SWAP
        # GRAB, SWAP, DROP
        # GRAB, SWAP, DROP, SWAP
        # SWAP, GRAB, SWAP, DROP
        # SWAP, GRAB, SWAP, DROP, SWAP
        if not self.colbusy(self.exa):
            avail = self.colrows(self.exa)
            if avail >= 2:
                swap = self.move(SWAP)
                yield swap
                if avail >= 3:
                    grab = self.move(GRAB, SWAP, DROP)
                    yield grab
                    yield grab.move(SWAP)
                    swap = swap.move(GRAB, SWAP, DROP)
                    yield swap
                    yield swap.move(SWAP)

    def gen_get(self):
        # GRAB
        # SWAP, GRAB
        # GRAB, SWAP, DROP, SWAP, GRAB
        if not self.colbusy(self.exa):
            avail = self.colrows(self.exa)
            if avail >= 1:
                grab = self.move(GRAB)
                yield grab
                if avail >= 2:
                    yield self.move(SWAP, GRAB)
                    if avail >= 3:
                        yield grab.move(SWAP, DROP, SWAP, GRAB)

    def gen_put(self):
        # DROP
        # DROP, SWAP
        # DROP, SWAP, GRAB, SWAP, DROP
        if not self.colbusy(self.exa):
            avail = self.colrows(self.exa)
            drop = self.move(DROP)
            yield drop
            if avail >= 1:
                swap = drop.move(SWAP)
                yield swap
                if avail >= 2:
                    yield swap.move(GRAB, SWAP, DROP)

    def solve(self):
        assert self.exa is not None

        pool = deque(self.gen_moves())
        assert len(pool) > 0

        best_score = ()

        for key in self.score_keys():
            if len(pool) == 1:
                break

            for state in pool:
                state.score = key(state)

            best = min(state.score for state in pool)
            best_score += (best,)

            for i in range(len(pool)):
                state = pool.popleft()
                assert state.held == NOBLOCK
                if state.score == best:
                    pool.append(state)

        best = pool.popleft()
        best.score = best_score
        return best

    def score_keys(self):
        cls = self.__class__
        colsize = self.nrows - 2

        if colsize >= COLSIZE_PANIC:
            return cls.holes, cls.points, cls.nmoves, cls.fragmentation

        if colsize >= COLSIZE_PRIO:
            return cls.causes_panic, cls.points, cls.holes, \
                   cls.fragmentation, cls.nmoves

        return cls.points, cls.fragmentation, cls.holes, cls.nmoves

    def print_groupsizes(self):
        for start in range(len(self.blockgroups) - COLUMNS, -1, -COLUMNS):
            print(' '.join('%-2d' % self.groups[g][0]
                           for g in self.blockgroups[start:start + COLUMNS]))

    def print_groups(self):
        for start in range(len(self.blockgroups) - COLUMNS, -1, -COLUMNS):
            print(' '.join('%-2d' % g
                           for g in self.blockgroups[start:start + COLUMNS]))

    def print(self):
        print_board(self.blocks, self.exa, self.held)

    def tostring(self):
        stream = io.StringIO()
        with redirect_stdout(stream):
            self.print()
        return stream.getvalue()

    def has_same_exa(self, state):
        return self.exa == state.exa and self.held == state.held

    def nmoves(self):
        return len(self.moves)

    def delay(self):
        return moves_delay(self.moves)

    def keys(self):
        return moves_to_keys(self.moves)

    def loops(self, prev):
        return self.moves and \
               self.exa == prev.exa and \
               self.held == prev.held and \
               self.moves == prev.moves and \
               self.score == prev.score


def get_colskip(blocks):
    def colskip(col):
        for row, block in enumerate(blocks[col::COLUMNS]):
            if block != NOBLOCK:
                return row
        return len(blocks) // COLUMNS

    return list(map(colskip, range(COLUMNS)))


def get_busy(blocks, colskip):
    mask = 0
    for col, skip in enumerate(colskip):
        start = (skip + 1) * COLUMNS + col
        colbusy = NOBLOCK in blocks[start::COLUMNS]
        mask |= colbusy << col
    return mask


def scan_group(blocks, i, block, visited):
    yield i
    visited.add(i)
    for nb in neighbors(i, blocks):
        if blocks[nb] == block and nb not in visited:
            yield from scan_group(blocks, nb, block, visited)


def get_groups(blocks):
    blockgroups = [0] * len(blocks)
    groups = [(0, NOBLOCK)]
    visited = set()

    for i, block in enumerate(blocks):
        if block != NOBLOCK and i not in visited:
            groupid = len(groups)
            group = tuple(scan_group(blocks, i, block, visited))
            groups.append((len(group), block))
            for j in group:
                blockgroups[j] = groupid

    return blockgroups, groups


def neighbors(i, blocks):
    y, x = divmod(i, COLUMNS)
    if x > 0:
        yield i - 1
    if x < COLUMNS - 1:
        yield i + 1
    if y > 0:
        yield i - COLUMNS
    if y < len(blocks) // COLUMNS - 1:
        yield i + COLUMNS


def move_to_key(move):
    return 'jjkadl'[move]


def moves_to_keys(moves):
    return ''.join(move_to_key(move) for move in moves)


def moves_delay(moves):
    return sum(MOVE_DELAYS[m] for m in moves)


if __name__ == '__main__':
    import sys
    from PIL import Image

    board = Image.open('screens/board%d.png' % int(sys.argv[1])).convert('HSV')
    state = State.detect(board)

    print('parsed:')
    state.print()
    print()

    print('groups:')
    state.print_groups()
    print()

    print('groupsizes:')
    state.print_groupsizes()
    print()

    start = time.time()
    newstate = state.solve()
    end = time.time()
    print('best move:', newstate.keys())
    print('score:', newstate.score)
    print('elapsed:', round((end - start) * 1000, 1), 'ms')
    print()

    print('target after move:')
    newstate.print()
    #print()

    #print('generated moves:')
    #for state in state.gen_moves():
    #    print(state.keys())