Source

aichallenge-py / MyBot.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
#  MyBot.py
#  aichallenge.py
#
#  Created by Lars Yencken on 2011-11-02.
#  Copyright 2011 Lars Yencken. All rights reserved.
#

"""
Bot for Google's AI Challenge.
"""

import os
import sys
import heapq
from functools import partial
from itertools import product
import random
import math

import numpy as np
from scipy.signal import convolve

import settings
from ants import Ants
import mdp

class GameTurn(object):
    "Game state for an individual move."
    def __init__(self, ants, game):
        self.ants = ants
        self.game = game
        self.plans = set(ants.my_hills())

    def do_move_direction(self, loc, direction):
        "Move in a given direction, handling conflicting orders."
        new_loc = loc.dest(direction)
        if (self.ants.unoccupied(new_loc) and new_loc not in self.plans):
            self.ants.issue_order((loc, direction))
            self.plans.add(new_loc)
            return True
        else:
            return False

    def do_move_location(self, loc, dest):
        "Slow but exact pathfinding."
        p = self.find_path(loc, dest)
        if p:
            new_loc = p[0]
            d, = self.ants.direction(loc, new_loc)
            if self.do_move_direction(loc, d):
                for i, l in enumerate(p):
                    self.plans[i + 1].add(l)
                self.targets[dest] = loc
                return True

        return False
    
    def do_move_location_fast(self, loc, dest):
        "Fast and approximate pathfinding."
        directions = self.ants.direction(loc, dest)
        for direction in directions:
            if self.do_move_direction(loc, direction):
                self.targets[dest] = loc
                return True
        return False

    def find_path(self, loc, dest, max_depth=50):
        "A* search from the location to the destination."
        dest_d = partial(self.ants.distance, dest)
        passable = self.ants.passable
        unseen = self.game.unseen
        frontier = self.frontier
        def f(p):
            return dest_d(p[-1]) + len(p)

        def expand(p):
            l0 = p[-1]
            for direction in ('n', 's', 'e', 'w'):
                l = l0.dest(l0, direction)
                if l not in p and l not in self.plans[len(p)] and passable(l) \
                        and (l not in unseen or l in frontier):
                    yield p + (l,)

        p0 = loc,
        queue = [(f(p0), p0)]
        push = partial(heapq.heappush, queue)
        pop = partial(heapq.heappop, queue)

        visited = set()
        d = 0
        while queue:
            d += 1
            cost, p = pop()
            if p[-1] == dest:
                return p[1:]

            if d >= max_depth:
                # randomly choose between the best options so far
                push((cost, p))
                p = random.choice([v for v in queue if v[0] == cost])[1]
                return p[1:]

            for p_next in expand(p):
                l_next = p_next[-1]
                step = len(p)
                if l_next not in visited and l_next not in self.plans[step]:
                    visited.add(l_next)
                    push((f(p_next), p_next))

    def find_food(self):
        ant_dist = []
        distance = self.ants.distance
        for food_loc in self.ants.food():
            for ant_loc in list(self.needs_orders):
                dist = distance(ant_loc, food_loc)
                ant_dist.append((dist, ant_loc, food_loc))

        ant_dist.sort()
        for dist, ant_loc, food_loc in ant_dist:
            if food_loc not in self.targets \
                    and ant_loc not in self.targets.values():
                self.do_move_location(ant_loc, food_loc)

    def attack_hills(self):
        # attack enemy hills
        ant_dist = []
        for hill_loc in self.active_enemy_hills:
            for ant in list(self.needs_orders):
                dist = ant.distance(hill_loc)
                ant_dist.append((dist, ant))
        ant_dist.sort()
        for dist, ant_loc in ant_dist:
            if ant_loc in self.needs_orders:
                self.do_move_location(ant_loc, hill_loc)

    def explore(self):
        # explore unseen areas
        distance = self.ants.distance
        for ant_loc in list(self.needs_orders):
            if not self.have_time():
                break
            unseen_dist = []
            for unseen_loc in self.frontier:
                dist = distance(ant_loc, unseen_loc)
                unseen_dist.append((dist, unseen_loc))
            unseen_dist.sort()
            for dist, unseen_loc in unseen_dist:
                if self.do_move_location(ant_loc, unseen_loc):
                    break

    def print_frontier(self):
        self.game.print_map(self.frontier)

    def have_time(self):
        return self.ants.time_remaining() > 200

    def random_walk(self):
        "Do a random walk with all remaining ants."
        for loc in list(self.needs_orders):
            directions = ['n', 's', 'e', 'w']
            random.shuffle(directions)
            for d in directions:
                if self.do_move_direction(loc, d):
                    break

    def follow_scent(self):
        ants = self.ants
        directions = ['n', 's', 'e', 'w']
        value = self.game.value
        for loc in ants.my_ants():
            options = []
            for d in directions:
                dest = loc.dest(d)
                smell = value[dest]
                if dest not in self.plans and self.ants.passable(dest):
                    options.append((smell, d))
            options.sort(reverse=True)
            for smell, d in options:
                if self.do_move_direction(loc, d):
                    break

class MyBot(object):
    "Game state bot, persistent across turns."
    def __init__(self):
        pass
    
    def do_setup(self, ants):
        "Initialise state with game variables."
        self.rows = ants.rows
        self.cols = ants.cols
        self.seen = np.zeros((self.rows, self.cols), dtype=np.bool_)
        self.plannable = np.zeros((self.rows, self.cols), dtype=np.bool_)
        self.enemy_hills = set()
        self.destroyed_hills = set()
        self.turn_no = 0

    def do_turn(self, ants):
        "Give orders to all of our ants."
        self.turn_no += 1
        self.remember_hills(ants)
        self.remember_seen(ants)
        self.frontier = self.build_frontier()
        reward = self.get_reward_matrix(self.frontier, ants)
        self.value = mdp.value_iteration(reward, torus=True)

        if settings.DEBUG:
            self._dump_value_map()

        turn = GameTurn(ants, self)
        turn.follow_scent()

    def build_frontier(self):
        rows = self.rows
        cols = self.cols
        mask = np.array(
                [[0.0, -0.25, 0.0],
                [-0.25, 1.0, -0.25],
                [0.0, -0.25, 0.0]]
            )
        frontier = np.zeros((rows + 2, cols + 2))
        frontier[1:rows+1,1:cols+1] = self.seen
        frontier[rows+1,1:cols+1] = self.seen[0]
        frontier[0,1:cols+1] = self.seen[rows-1]
        frontier[1:rows+1,0] = self.seen[:, 0]
        frontier[1:rows+1,cols+1] = self.seen[:, cols-1]

        frontier = (convolve(frontier, mask) > 0)[2:rows+2,2:cols+2]

        if settings.DEBUG:
            self._dump_frontier(frontier)
                    
        return frontier

    def remember_seen(self, ants):
        visible = ants.visible
        seen = self.seen
        plannable = self.plannable
        for loc in filter(visible, product(xrange(ants.rows),
                xrange(ants.cols))):
            if loc not in self.seen:
                seen[loc] = True
                if ants.passable(loc):
                    plannable[loc] = True

    def remember_hills(self, ants):
        self.enemy_hills.update(h.loc for h in ants.enemy_hills())
        self.destroyed_hills.update(self.enemy_hills.intersection(
                ants.my_ants()))

    @property
    def active_enemy_hills(self):
        return self.enemy_hills.difference(self.destroyed_hills)

    def get_reward_matrix(self, frontier, ants):
        reward = np.ones((ants.rows, ants.cols), dtype=np.float32) * \
                settings.BASE_REWARD

        for loc in self.active_enemy_hills:
            reward[loc] = settings.ENEMY_HILL_REWARD

        for food in ants.food():
            reward[food] = settings.FOOD_REWARD

        for ant in ants.enemy_ants():
            reward[ant.loc] = settings.ENEMY_ANT_REWARD

        for y, x in zip(*np.nonzero(frontier)):
            reward[y, x] += settings.FRONTIER_REWARD 

        reward += (self.plannable == 0) * settings.BARRIER_REWARD

        return reward

    def print_map(self, m):
        render = lambda x: u'O' if x in m else u'•'
        for i in xrange(self.rows):
            l = u''.join(render((i, j)) for j in xrange(self.cols))
            print >> sys.stderr, l

    def _dump_value_map(self):
        import Image
        im = Image.new('L', (self.rows, self.cols))
        value = self.value
        value = 255 * (value > 0) * value / value.max()
        for i, j in product(xrange(self.rows), xrange(self.cols)):
            im.putpixel((i, j), int(math.ceil(value[i, j])))
        im.save(os.path.join(settings.LOG_DIR, '%.03d-value.png'
            % self.turn_no))

    def _dump_frontier(self, frontier):
        import Image
        im = Image.new('RGB', (self.rows, self.cols))
        seen = self.seen
        for i, j in product(xrange(self.rows), xrange(self.cols)):
            if frontier[i, j]:
                im.putpixel((i, j), (50, 50, 255))
            elif seen[i, j]:
                im.putpixel((i, j), (255, 255, 255))
        im.save(os.path.join(settings.LOG_DIR, '%.03d-frontier.png'
            % self.turn_no))

def dump_grayscale_image(m, filename):
    import Image
    rows, cols = m.shape
    im = Image.new('1', (cols, rows))
    for y, x in zip(*np.nonzero(m)):
        im.putpixel((x, y), 1)
    im.save(filename)
    
if __name__ == '__main__':
    args = sys.argv[1:]
    # psyco will speed up python a little, but is not needed
    try:
        import psyco
        psyco.full()
    except ImportError:
        pass
    
    try:
        if args:
            Ants.run(MyBot(), open(args[0]))
        else:
            Ants.run(MyBot())
    except KeyboardInterrupt:
        print('ctrl-c, leaving ...')
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.