diff --git a/agents_4e.py b/agents_4e.py new file mode 100644 index 000000000..debd9441e --- /dev/null +++ b/agents_4e.py @@ -0,0 +1,1044 @@ +"""Implement Agents and Environments (Chapters 1-2). + +The class hierarchies are as follows: + +Thing ## A physical object that can exist in an environment + Agent + Wumpus + Dirt + Wall + ... + +Environment ## An environment holds objects, runs simulations + XYEnvironment + VacuumEnvironment + WumpusEnvironment + +An agent program is a callable instance, taking percepts and choosing actions + SimpleReflexAgentProgram + ... + +EnvGUI ## A window with a graphical representation of the Environment + +EnvToolbar ## contains buttons for controlling EnvGUI + +EnvCanvas ## Canvas to display the environment of an EnvGUI + +""" + +# TO DO: +# Implement grabbing correctly. +# When an object is grabbed, does it still have a location? +# What if it is released? +# What if the grabbed or the grabber is deleted? +# What if the grabber moves? +# +# Speed control in GUI does not have any effect -- fix it. + +from utils import distance_squared, turn_heading +from statistics import mean +from ipythonblocks import BlockGrid +from IPython.display import HTML, display +from time import sleep + +import random +import copy +import collections + + +# ______________________________________________________________________________ + + +class Thing: + """This represents any physical object that can appear in an Environment. + You subclass Thing to get the things you want. Each thing can have a + .__name__ slot (used for output only).""" + + def __repr__(self): + return '<{}>'.format(getattr(self, '__name__', self.__class__.__name__)) + + def is_alive(self): + """Things that are 'alive' should return true.""" + return hasattr(self, 'alive') and self.alive + + def show_state(self): + """Display the agent's internal state. Subclasses should override.""" + print("I don't know how to show_state.") + + def display(self, canvas, x, y, width, height): + """Display an image of this Thing on the canvas.""" + # Do we need this? + pass + + +class Agent(Thing): + """An Agent is a subclass of Thing with one required slot, + .program, which should hold a function that takes one argument, the + percept, and returns an action. (What counts as a percept or action + will depend on the specific environment in which the agent exists.) + Note that 'program' is a slot, not a method. If it were a method, + then the program could 'cheat' and look at aspects of the agent. + It's not supposed to do that: the program can only look at the + percepts. An agent program that needs a model of the world (and of + the agent itself) will have to build and maintain its own model. + There is an optional slot, .performance, which is a number giving + the performance measure of the agent in its environment.""" + + def __init__(self, program=None): + self.alive = True + self.bump = False + self.holding = [] + self.performance = 0 + if program is None or not isinstance(program, collections.Callable): + print("Can't find a valid program for {}, falling back to default.".format( + self.__class__.__name__)) + + def program(percept): + return eval(input('Percept={}; action? '.format(percept))) + + self.program = program + + def can_grab(self, thing): + """Return True if this agent can grab this thing. + Override for appropriate subclasses of Agent and Thing.""" + return False + + +def TraceAgent(agent): + """Wrap the agent's program to print its input and output. This will let + you see what the agent is doing in the environment.""" + old_program = agent.program + + def new_program(percept): + action = old_program(percept) + print('{} perceives {} and does {}'.format(agent, percept, action)) + return action + agent.program = new_program + return agent + +# ______________________________________________________________________________ + + +def TableDrivenAgentProgram(table): + """This agent selects an action based on the percept sequence. + It is practical only for tiny domains. + To customize it, provide as table a dictionary of all + {percept_sequence:action} pairs. [Figure 2.7]""" + percepts = [] + + def program(percept): + percepts.append(percept) + action = table.get(tuple(percepts)) + return action + return program + + +def RandomAgentProgram(actions): + """An agent that chooses an action at random, ignoring all percepts. + >>> list = ['Right', 'Left', 'Suck', 'NoOp'] + >>> program = RandomAgentProgram(list) + >>> agent = Agent(program) + >>> environment = TrivialVacuumEnvironment() + >>> environment.add_thing(agent) + >>> environment.run() + >>> environment.status == {(1, 0): 'Clean' , (0, 0): 'Clean'} + True + """ + return lambda percept: random.choice(actions) + +# ______________________________________________________________________________ + + +def SimpleReflexAgentProgram(rules, interpret_input): + """This agent takes action based solely on the percept. [Figure 2.10]""" + def program(percept): + state = interpret_input(percept) + rule = rule_match(state, rules) + action = rule.action + return action + return program + + +def ModelBasedReflexAgentProgram(rules, update_state, trainsition_model, sensor_model): + """This agent takes action based on the percept and state. [Figure 2.12]""" + def program(percept): + program.state = update_state(program.state, program.action, percept, trainsition_model, sensor_model) + rule = rule_match(program.state, rules) + action = rule.action + return action + program.state = program.action = None + return program + + +def rule_match(state, rules): + """Find the first rule that matches state.""" + for rule in rules: + if rule.matches(state): + return rule + +# ______________________________________________________________________________ + + +loc_A, loc_B = (0, 0), (1, 0) # The two locations for the Vacuum world + + +def RandomVacuumAgent(): + """Randomly choose one of the actions from the vacuum environment. + >>> agent = RandomVacuumAgent() + >>> environment = TrivialVacuumEnvironment() + >>> environment.add_thing(agent) + >>> environment.run() + >>> environment.status == {(1,0):'Clean' , (0,0) : 'Clean'} + True + """ + return Agent(RandomAgentProgram(['Right', 'Left', 'Suck', 'NoOp'])) + + +def TableDrivenVacuumAgent(): + """[Figure 2.3]""" + table = {((loc_A, 'Clean'),): 'Right', + ((loc_A, 'Dirty'),): 'Suck', + ((loc_B, 'Clean'),): 'Left', + ((loc_B, 'Dirty'),): 'Suck', + ((loc_A, 'Dirty'), (loc_A, 'Clean')): 'Right', + ((loc_A, 'Clean'), (loc_B, 'Dirty')): 'Suck', + ((loc_B, 'Clean'), (loc_A, 'Dirty')): 'Suck', + ((loc_B, 'Dirty'), (loc_B, 'Clean')): 'Left', + ((loc_A, 'Dirty'), (loc_A, 'Clean'), (loc_B, 'Dirty')): 'Suck', + ((loc_B, 'Dirty'), (loc_B, 'Clean'), (loc_A, 'Dirty')): 'Suck' + } + return Agent(TableDrivenAgentProgram(table)) + + +def ReflexVacuumAgent(): + """A reflex agent for the two-state vacuum environment. [Figure 2.8] + >>> agent = ReflexVacuumAgent() + >>> environment = TrivialVacuumEnvironment() + >>> environment.add_thing(agent) + >>> environment.run() + >>> environment.status == {(1,0):'Clean' , (0,0) : 'Clean'} + True + """ + def program(percept): + location, status = percept + if status == 'Dirty': + return 'Suck' + elif location == loc_A: + return 'Right' + elif location == loc_B: + return 'Left' + return Agent(program) + + +def ModelBasedVacuumAgent(): + """An agent that keeps track of what locations are clean or dirty. + >>> agent = ModelBasedVacuumAgent() + >>> environment = TrivialVacuumEnvironment() + >>> environment.add_thing(agent) + >>> environment.run() + >>> environment.status == {(1,0):'Clean' , (0,0) : 'Clean'} + True + """ + model = {loc_A: None, loc_B: None} + + def program(percept): + """Same as ReflexVacuumAgent, except if everything is clean, do NoOp.""" + location, status = percept + model[location] = status # Update the model here + if model[loc_A] == model[loc_B] == 'Clean': + return 'NoOp' + elif status == 'Dirty': + return 'Suck' + elif location == loc_A: + return 'Right' + elif location == loc_B: + return 'Left' + return Agent(program) + +# ______________________________________________________________________________ + + +class Environment: + """Abstract class representing an Environment. 'Real' Environment classes + inherit from this. Your Environment will typically need to implement: + percept: Define the percept that an agent sees. + execute_action: Define the effects of executing an action. + Also update the agent.performance slot. + The environment keeps a list of .things and .agents (which is a subset + of .things). Each agent has a .performance slot, initialized to 0. + Each thing has a .location slot, even though some environments may not + need this.""" + + def __init__(self): + self.things = [] + self.agents = [] + + def thing_classes(self): + return [] # List of classes that can go into environment + + def percept(self, agent): + """Return the percept that the agent sees at this point. (Implement this.)""" + raise NotImplementedError + + def execute_action(self, agent, action): + """Change the world to reflect this action. (Implement this.)""" + raise NotImplementedError + + def default_location(self, thing): + """Default location to place a new thing with unspecified location.""" + return None + + def exogenous_change(self): + """If there is spontaneous change in the world, override this.""" + pass + + def is_done(self): + """By default, we're done when we can't find a live agent.""" + return not any(agent.is_alive() for agent in self.agents) + + def step(self): + """Run the environment for one time step. If the + actions and exogenous changes are independent, this method will + do. If there are interactions between them, you'll need to + override this method.""" + if not self.is_done(): + actions = [] + for agent in self.agents: + if agent.alive: + actions.append(agent.program(self.percept(agent))) + else: + actions.append("") + for (agent, action) in zip(self.agents, actions): + self.execute_action(agent, action) + self.exogenous_change() + + def run(self, steps=1000): + """Run the Environment for given number of time steps.""" + for step in range(steps): + if self.is_done(): + return + self.step() + + def list_things_at(self, location, tclass=Thing): + """Return all things exactly at a given location.""" + return [thing for thing in self.things + if thing.location == location and isinstance(thing, tclass)] + + def some_things_at(self, location, tclass=Thing): + """Return true if at least one of the things at location + is an instance of class tclass (or a subclass).""" + return self.list_things_at(location, tclass) != [] + + def add_thing(self, thing, location=None): + """Add a thing to the environment, setting its location. For + convenience, if thing is an agent program we make a new agent + for it. (Shouldn't need to override this.)""" + if not isinstance(thing, Thing): + thing = Agent(thing) + if thing in self.things: + print("Can't add the same thing twice") + else: + thing.location = location if location is not None else self.default_location(thing) + self.things.append(thing) + if isinstance(thing, Agent): + thing.performance = 0 + self.agents.append(thing) + + def delete_thing(self, thing): + """Remove a thing from the environment.""" + try: + self.things.remove(thing) + except ValueError as e: + print(e) + print(" in Environment delete_thing") + print(" Thing to be removed: {} at {}".format(thing, thing.location)) + print(" from list: {}".format([(thing, thing.location) for thing in self.things])) + if thing in self.agents: + self.agents.remove(thing) + + +class Direction: + """A direction class for agents that want to move in a 2D plane + Usage: + d = Direction("down") + To change directions: + d = d + "right" or d = d + Direction.R #Both do the same thing + Note that the argument to __add__ must be a string and not a Direction object. + Also, it (the argument) can only be right or left.""" + + R = "right" + L = "left" + U = "up" + D = "down" + + def __init__(self, direction): + self.direction = direction + + def __add__(self, heading): + """ + >>> d = Direction('right') + >>> l1 = d.__add__(Direction.L) + >>> l2 = d.__add__(Direction.R) + >>> l1.direction + 'up' + >>> l2.direction + 'down' + >>> d = Direction('down') + >>> l1 = d.__add__('right') + >>> l2 = d.__add__('left') + >>> l1.direction == Direction.L + True + >>> l2.direction == Direction.R + True + """ + if self.direction == self.R: + return{ + self.R: Direction(self.D), + self.L: Direction(self.U), + }.get(heading, None) + elif self.direction == self.L: + return{ + self.R: Direction(self.U), + self.L: Direction(self.D), + }.get(heading, None) + elif self.direction == self.U: + return{ + self.R: Direction(self.R), + self.L: Direction(self.L), + }.get(heading, None) + elif self.direction == self.D: + return{ + self.R: Direction(self.L), + self.L: Direction(self.R), + }.get(heading, None) + + def move_forward(self, from_location): + """ + >>> d = Direction('up') + >>> l1 = d.move_forward((0, 0)) + >>> l1 + (0, -1) + >>> d = Direction(Direction.R) + >>> l1 = d.move_forward((0, 0)) + >>> l1 + (1, 0) + """ + x, y = from_location + if self.direction == self.R: + return (x + 1, y) + elif self.direction == self.L: + return (x - 1, y) + elif self.direction == self.U: + return (x, y - 1) + elif self.direction == self.D: + return (x, y + 1) + + +class XYEnvironment(Environment): + """This class is for environments on a 2D plane, with locations + labelled by (x, y) points, either discrete or continuous. + + Agents perceive things within a radius. Each agent in the + environment has a .location slot which should be a location such + as (0, 1), and a .holding slot, which should be a list of things + that are held.""" + + def __init__(self, width=10, height=10): + super().__init__() + + self.width = width + self.height = height + self.observers = [] + # Sets iteration start and end (no walls). + self.x_start, self.y_start = (0, 0) + self.x_end, self.y_end = (self.width, self.height) + + perceptible_distance = 1 + + def things_near(self, location, radius=None): + """Return all things within radius of location.""" + if radius is None: + radius = self.perceptible_distance + radius2 = radius * radius + return [(thing, radius2 - distance_squared(location, thing.location)) + for thing in self.things if distance_squared( + location, thing.location) <= radius2] + + def percept(self, agent): + """By default, agent perceives things within a default radius.""" + return self.things_near(agent.location) + + def execute_action(self, agent, action): + agent.bump = False + if action == 'TurnRight': + agent.direction += Direction.R + elif action == 'TurnLeft': + agent.direction += Direction.L + elif action == 'Forward': + agent.bump = self.move_to(agent, agent.direction.move_forward(agent.location)) +# elif action == 'Grab': +# things = [thing for thing in self.list_things_at(agent.location) +# if agent.can_grab(thing)] +# if things: +# agent.holding.append(things[0]) + elif action == 'Release': + if agent.holding: + agent.holding.pop() + + def default_location(self, thing): + return (random.choice(self.width), random.choice(self.height)) + + def move_to(self, thing, destination): + """Move a thing to a new location. Returns True on success or False if there is an Obstacle. + If thing is holding anything, they move with him.""" + thing.bump = self.some_things_at(destination, Obstacle) + if not thing.bump: + thing.location = destination + for o in self.observers: + o.thing_moved(thing) + for t in thing.holding: + self.delete_thing(t) + self.add_thing(t, destination) + t.location = destination + return thing.bump + + def add_thing(self, thing, location=(1, 1), exclude_duplicate_class_items=False): + """Add things to the world. If (exclude_duplicate_class_items) then the item won't be + added if the location has at least one item of the same class.""" + if (self.is_inbounds(location)): + if (exclude_duplicate_class_items and + any(isinstance(t, thing.__class__) for t in self.list_things_at(location))): + return + super().add_thing(thing, location) + + def is_inbounds(self, location): + """Checks to make sure that the location is inbounds (within walls if we have walls)""" + x, y = location + return not (x < self.x_start or x >= self.x_end or y < self.y_start or y >= self.y_end) + + def random_location_inbounds(self, exclude=None): + """Returns a random location that is inbounds (within walls if we have walls)""" + location = (random.randint(self.x_start, self.x_end), + random.randint(self.y_start, self.y_end)) + if exclude is not None: + while(location == exclude): + location = (random.randint(self.x_start, self.x_end), + random.randint(self.y_start, self.y_end)) + return location + + def delete_thing(self, thing): + """Deletes thing, and everything it is holding (if thing is an agent)""" + if isinstance(thing, Agent): + for obj in thing.holding: + super().delete_thing(obj) + for obs in self.observers: + obs.thing_deleted(obj) + + super().delete_thing(thing) + for obs in self.observers: + obs.thing_deleted(thing) + + def add_walls(self): + """Put walls around the entire perimeter of the grid.""" + for x in range(self.width): + self.add_thing(Wall(), (x, 0)) + self.add_thing(Wall(), (x, self.height - 1)) + for y in range(1, self.height-1): + self.add_thing(Wall(), (0, y)) + self.add_thing(Wall(), (self.width - 1, y)) + + # Updates iteration start and end (with walls). + self.x_start, self.y_start = (1, 1) + self.x_end, self.y_end = (self.width - 1, self.height - 1) + + def add_observer(self, observer): + """Adds an observer to the list of observers. + An observer is typically an EnvGUI. + + Each observer is notified of changes in move_to and add_thing, + by calling the observer's methods thing_moved(thing) + and thing_added(thing, loc).""" + self.observers.append(observer) + + def turn_heading(self, heading, inc): + """Return the heading to the left (inc=+1) or right (inc=-1) of heading.""" + return turn_heading(heading, inc) + + +class Obstacle(Thing): + """Something that can cause a bump, preventing an agent from + moving into the same square it's in.""" + pass + + +class Wall(Obstacle): + pass + +# ______________________________________________________________________________ + + +class GraphicEnvironment(XYEnvironment): + def __init__(self, width=10, height=10, boundary=True, color={}, display=False): + """Define all the usual XYEnvironment characteristics, + but initialise a BlockGrid for GUI too.""" + super().__init__(width, height) + self.grid = BlockGrid(width, height, fill=(200, 200, 200)) + if display: + self.grid.show() + self.visible = True + else: + self.visible = False + self.bounded = boundary + self.colors = color + + def get_world(self): + """Returns all the items in the world in a format + understandable by the ipythonblocks BlockGrid.""" + result = [] + x_start, y_start = (0, 0) + x_end, y_end = self.width, self.height + for x in range(x_start, x_end): + row = [] + for y in range(y_start, y_end): + row.append(self.list_things_at([x, y])) + result.append(row) + return result + + """ + def run(self, steps=1000, delay=1): + "" "Run the Environment for given number of time steps, + but update the GUI too." "" + for step in range(steps): + sleep(delay) + if self.visible: + self.reveal() + if self.is_done(): + if self.visible: + self.reveal() + return + self.step() + if self.visible: + self.reveal() + """ + + def run(self, steps=1000, delay=1): + """Run the Environment for given number of time steps, + but update the GUI too.""" + for step in range(steps): + self.update(delay) + if self.is_done(): + break + self.step() + self.update(delay) + + def update(self, delay=1): + sleep(delay) + if self.visible: + self.conceal() + self.reveal() + else: + self.reveal() + + def reveal(self): + """Display the BlockGrid for this world - the last thing to be added + at a location defines the location color.""" + self.draw_world() + self.grid.show() + self.visible = True + + def draw_world(self): + self.grid[:] = (200, 200, 200) + world = self.get_world() + for x in range(0, len(world)): + for y in range(0, len(world[x])): + if len(world[x][y]): + self.grid[y, x] = self.colors[world[x][y][-1].__class__.__name__] + + def conceal(self): + """Hide the BlockGrid for this world""" + self.visible = False + display(HTML('')) + + +# ______________________________________________________________________________ +# Continuous environment + +class ContinuousWorld(Environment): + """Model for Continuous World""" + + def __init__(self, width=10, height=10): + super().__init__() + self.width = width + self.height = height + + def add_obstacle(self, coordinates): + self.things.append(PolygonObstacle(coordinates)) + + +class PolygonObstacle(Obstacle): + + def __init__(self, coordinates): + """Coordinates is a list of tuples.""" + super().__init__() + self.coordinates = coordinates + +# ______________________________________________________________________________ +# Vacuum environment + + +class Dirt(Thing): + pass + + +class VacuumEnvironment(XYEnvironment): + + """The environment of [Ex. 2.12]. Agent perceives dirty or clean, + and bump (into obstacle) or not; 2D discrete world of unknown size; + performance measure is 100 for each dirt cleaned, and -1 for + each turn taken.""" + + def __init__(self, width=10, height=10): + super().__init__(width, height) + self.add_walls() + + def thing_classes(self): + return [Wall, Dirt, ReflexVacuumAgent, RandomVacuumAgent, + TableDrivenVacuumAgent, ModelBasedVacuumAgent] + + def percept(self, agent): + """The percept is a tuple of ('Dirty' or 'Clean', 'Bump' or 'None'). + Unlike the TrivialVacuumEnvironment, location is NOT perceived.""" + status = ('Dirty' if self.some_things_at( + agent.location, Dirt) else 'Clean') + bump = ('Bump' if agent.bump else'None') + return (status, bump) + + def execute_action(self, agent, action): + agent.bump = False + if action == 'Suck': + dirt_list = self.list_things_at(agent.location, Dirt) + if dirt_list != []: + dirt = dirt_list[0] + agent.performance += 100 + self.delete_thing(dirt) + else: + super().execute_action(agent, action) + + if action != 'NoOp': + agent.performance -= 1 + + +class TrivialVacuumEnvironment(Environment): + + """This environment has two locations, A and B. Each can be Dirty + or Clean. The agent perceives its location and the location's + status. This serves as an example of how to implement a simple + Environment.""" + + def __init__(self): + super().__init__() + self.status = {loc_A: random.choice(['Clean', 'Dirty']), + loc_B: random.choice(['Clean', 'Dirty'])} + + def thing_classes(self): + return [Wall, Dirt, ReflexVacuumAgent, RandomVacuumAgent, + TableDrivenVacuumAgent, ModelBasedVacuumAgent] + + def percept(self, agent): + """Returns the agent's location, and the location status (Dirty/Clean).""" + return (agent.location, self.status[agent.location]) + + def execute_action(self, agent, action): + """Change agent's location and/or location's status; track performance. + Score 10 for each dirt cleaned; -1 for each move.""" + if action == 'Right': + agent.location = loc_B + agent.performance -= 1 + elif action == 'Left': + agent.location = loc_A + agent.performance -= 1 + elif action == 'Suck': + if self.status[agent.location] == 'Dirty': + agent.performance += 10 + self.status[agent.location] = 'Clean' + + def default_location(self, thing): + """Agents start in either location at random.""" + return random.choice([loc_A, loc_B]) + +# ______________________________________________________________________________ +# The Wumpus World + + +class Gold(Thing): + + def __eq__(self, rhs): + """All Gold are equal""" + return rhs.__class__ == Gold + pass + + +class Bump(Thing): + pass + + +class Glitter(Thing): + pass + + +class Pit(Thing): + pass + + +class Breeze(Thing): + pass + + +class Arrow(Thing): + pass + + +class Scream(Thing): + pass + + +class Wumpus(Agent): + screamed = False + pass + + +class Stench(Thing): + pass + + +class Explorer(Agent): + holding = [] + has_arrow = True + killed_by = "" + direction = Direction("right") + + def can_grab(self, thing): + """Explorer can only grab gold""" + return thing.__class__ == Gold + + +class WumpusEnvironment(XYEnvironment): + pit_probability = 0.2 # Probability to spawn a pit in a location. (From Chapter 7.2) + # Room should be 4x4 grid of rooms. The extra 2 for walls + + def __init__(self, agent_program, width=6, height=6): + super().__init__(width, height) + self.init_world(agent_program) + + def init_world(self, program): + """Spawn items in the world based on probabilities from the book""" + + "WALLS" + self.add_walls() + + "PITS" + for x in range(self.x_start, self.x_end): + for y in range(self.y_start, self.y_end): + if random.random() < self.pit_probability: + self.add_thing(Pit(), (x, y), True) + self.add_thing(Breeze(), (x - 1, y), True) + self.add_thing(Breeze(), (x, y - 1), True) + self.add_thing(Breeze(), (x + 1, y), True) + self.add_thing(Breeze(), (x, y + 1), True) + + "WUMPUS" + w_x, w_y = self.random_location_inbounds(exclude=(1, 1)) + self.add_thing(Wumpus(lambda x: ""), (w_x, w_y), True) + self.add_thing(Stench(), (w_x - 1, w_y), True) + self.add_thing(Stench(), (w_x + 1, w_y), True) + self.add_thing(Stench(), (w_x, w_y - 1), True) + self.add_thing(Stench(), (w_x, w_y + 1), True) + + "GOLD" + self.add_thing(Gold(), self.random_location_inbounds(exclude=(1, 1)), True) + + "AGENT" + self.add_thing(Explorer(program), (1, 1), True) + + def get_world(self, show_walls=True): + """Return the items in the world""" + result = [] + x_start, y_start = (0, 0) if show_walls else (1, 1) + + if show_walls: + x_end, y_end = self.width, self.height + else: + x_end, y_end = self.width - 1, self.height - 1 + + for x in range(x_start, x_end): + row = [] + for y in range(y_start, y_end): + row.append(self.list_things_at((x, y))) + result.append(row) + return result + + def percepts_from(self, agent, location, tclass=Thing): + """Return percepts from a given location, + and replaces some items with percepts from chapter 7.""" + thing_percepts = { + Gold: Glitter(), + Wall: Bump(), + Wumpus: Stench(), + Pit: Breeze()} + + """Agents don't need to get their percepts""" + thing_percepts[agent.__class__] = None + + """Gold only glitters in its cell""" + if location != agent.location: + thing_percepts[Gold] = None + + result = [thing_percepts.get(thing.__class__, thing) for thing in self.things + if thing.location == location and isinstance(thing, tclass)] + return result if len(result) else [None] + + def percept(self, agent): + """Return things in adjacent (not diagonal) cells of the agent. + Result format: [Left, Right, Up, Down, Center / Current location]""" + x, y = agent.location + result = [] + result.append(self.percepts_from(agent, (x - 1, y))) + result.append(self.percepts_from(agent, (x + 1, y))) + result.append(self.percepts_from(agent, (x, y - 1))) + result.append(self.percepts_from(agent, (x, y + 1))) + result.append(self.percepts_from(agent, (x, y))) + + """The wumpus gives out a loud scream once it's killed.""" + wumpus = [thing for thing in self.things if isinstance(thing, Wumpus)] + if len(wumpus) and not wumpus[0].alive and not wumpus[0].screamed: + result[-1].append(Scream()) + wumpus[0].screamed = True + + return result + + def execute_action(self, agent, action): + """Modify the state of the environment based on the agent's actions. + Performance score taken directly out of the book.""" + + if isinstance(agent, Explorer) and self.in_danger(agent): + return + + agent.bump = False + if action == 'TurnRight': + agent.direction += Direction.R + agent.performance -= 1 + elif action == 'TurnLeft': + agent.direction += Direction.L + agent.performance -= 1 + elif action == 'Forward': + agent.bump = self.move_to(agent, agent.direction.move_forward(agent.location)) + agent.performance -= 1 + elif action == 'Grab': + things = [thing for thing in self.list_things_at(agent.location) + if agent.can_grab(thing)] + if len(things): + print("Grabbing", things[0].__class__.__name__) + if len(things): + agent.holding.append(things[0]) + agent.performance -= 1 + elif action == 'Climb': + if agent.location == (1, 1): # Agent can only climb out of (1,1) + agent.performance += 1000 if Gold() in agent.holding else 0 + self.delete_thing(agent) + elif action == 'Shoot': + """The arrow travels straight down the path the agent is facing""" + if agent.has_arrow: + arrow_travel = agent.direction.move_forward(agent.location) + while(self.is_inbounds(arrow_travel)): + wumpus = [thing for thing in self.list_things_at(arrow_travel) + if isinstance(thing, Wumpus)] + if len(wumpus): + wumpus[0].alive = False + break + arrow_travel = agent.direction.move_forward(agent.location) + agent.has_arrow = False + + def in_danger(self, agent): + """Check if Explorer is in danger (Pit or Wumpus), if he is, kill him""" + for thing in self.list_things_at(agent.location): + if isinstance(thing, Pit) or (isinstance(thing, Wumpus) and thing.alive): + agent.alive = False + agent.performance -= 1000 + agent.killed_by = thing.__class__.__name__ + return True + return False + + def is_done(self): + """The game is over when the Explorer is killed + or if he climbs out of the cave only at (1,1).""" + explorer = [agent for agent in self.agents if isinstance(agent, Explorer)] + if len(explorer): + if explorer[0].alive: + return False + else: + print("Death by {} [-1000].".format(explorer[0].killed_by)) + else: + print("Explorer climbed out {}." + .format( + "with Gold [+1000]!" if Gold() not in self.things else "without Gold [+0]")) + return True + + + # TODO: Arrow needs to be implemented +# ______________________________________________________________________________ + + +def compare_agents(EnvFactory, AgentFactories, n=10, steps=1000): + """See how well each of several agents do in n instances of an environment. + Pass in a factory (constructor) for environments, and several for agents. + Create n instances of the environment, and run each agent in copies of + each one for steps. Return a list of (agent, average-score) tuples. + >>> environment = TrivialVacuumEnvironment + >>> agents = [ModelBasedVacuumAgent, ReflexVacuumAgent] + >>> result = compare_agents(environment, agents) + >>> performance_ModelBasedVacummAgent = result[0][1] + >>> performance_ReflexVacummAgent = result[1][1] + >>> performance_ReflexVacummAgent <= performance_ModelBasedVacummAgent + True + """ + envs = [EnvFactory() for i in range(n)] + return [(A, test_agent(A, steps, copy.deepcopy(envs))) + for A in AgentFactories] + + +def test_agent(AgentFactory, steps, envs): + """Return the mean score of running an agent in each of the envs, for steps + >>> def constant_prog(percept): + ... return percept + ... + >>> agent = Agent(constant_prog) + >>> result = agent.program(5) + >>> result == 5 + True + """ + def score(env): + agent = AgentFactory() + env.add_thing(agent) + env.run(steps) + return agent.performance + return mean(map(score, envs)) + +# _________________________________________________________________________ + + +__doc__ += """ +>>> a = ReflexVacuumAgent() +>>> a.program((loc_A, 'Clean')) +'Right' +>>> a.program((loc_B, 'Clean')) +'Left' +>>> a.program((loc_A, 'Dirty')) +'Suck' +>>> a.program((loc_A, 'Dirty')) +'Suck' + +>>> e = TrivialVacuumEnvironment() +>>> e.add_thing(ModelBasedVacuumAgent()) +>>> e.run(5) + +""" diff --git a/games4e.ipynb b/games4e.ipynb index 380466662..5b619f7ed 100644 --- a/games4e.ipynb +++ b/games4e.ipynb @@ -1659,7 +1659,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.7.0" + "version": "3.7.2" } }, "nbformat": 4, diff --git a/games4e.py b/games4e.py new file mode 100644 index 000000000..f32259175 --- /dev/null +++ b/games4e.py @@ -0,0 +1,630 @@ +"""Games, or Adversarial Search (Chapter 5)""" + +from collections import namedtuple +import random +import itertools +import copy +from utils import argmax, vector_add, MCT_Node, ucb + +infinity = float('inf') +GameState = namedtuple('GameState', 'to_move, utility, board, moves') +StochasticGameState = namedtuple('StochasticGameState', 'to_move, utility, board, moves, chance') + +# ______________________________________________________________________________ +# Minimax Search + + +def minimax_decision(state, game): + """Given a state in a game, calculate the best move by searching + forward all the way to the terminal states. [Figure 5.3]""" + + player = game.to_move(state) + + def max_value(state): + if game.terminal_test(state): + return game.utility(state, player) + v = -infinity + for a in game.actions(state): + v = max(v, min_value(game.result(state, a))) + return v + + def min_value(state): + if game.terminal_test(state): + return game.utility(state, player) + v = infinity + for a in game.actions(state): + v = min(v, max_value(game.result(state, a))) + return v + + # Body of minimax_decision: + return argmax(game.actions(state), + key=lambda a: min_value(game.result(state, a))) + +# ______________________________________________________________________________ + + +def expectiminimax(state, game): + """Return the best move for a player after dice are thrown. The game tree + includes chance nodes along with min and max nodes. [Figure 5.11]""" + player = game.to_move(state) + + def max_value(state): + v = -infinity + for a in game.actions(state): + v = max(v, chance_node(state, a)) + return v + + def min_value(state): + v = infinity + for a in game.actions(state): + v = min(v, chance_node(state, a)) + return v + + def chance_node(state, action): + res_state = game.result(state, action) + if game.terminal_test(res_state): + return game.utility(res_state, player) + sum_chances = 0 + num_chances = len(game.chances(res_state)) + for chance in game.chances(res_state): + res_state = game.outcome(res_state, chance) + util = 0 + if res_state.to_move == player: + util = max_value(res_state) + else: + util = min_value(res_state) + sum_chances += util * game.probability(chance) + return sum_chances / num_chances + + # Body of expectiminimax: + return argmax(game.actions(state), + key=lambda a: chance_node(state, a), default=None) + + +def alphabeta_search(state, game): + """Search game to determine best action; use alpha-beta pruning. + As in [Figure 5.7], this version searches all the way to the leaves.""" + + player = game.to_move(state) + + # Functions used by alphabeta + def max_value(state, alpha, beta): + if game.terminal_test(state): + return game.utility(state, player) + v = -infinity + for a in game.actions(state): + v = max(v, min_value(game.result(state, a), alpha, beta)) + if v >= beta: + return v + alpha = max(alpha, v) + return v + + def min_value(state, alpha, beta): + if game.terminal_test(state): + return game.utility(state, player) + v = infinity + for a in game.actions(state): + v = min(v, max_value(game.result(state, a), alpha, beta)) + if v <= alpha: + return v + beta = min(beta, v) + return v + + # Body of alphabeta_search: + best_score = -infinity + beta = infinity + best_action = None + for a in game.actions(state): + v = min_value(game.result(state, a), best_score, beta) + if v > best_score: + best_score = v + best_action = a + return best_action + + +def alphabeta_cutoff_search(state, game, d=4, cutoff_test=None, eval_fn=None): + """Search game to determine best action; use alpha-beta pruning. + This version cuts off search and uses an evaluation function.""" + + player = game.to_move(state) + + # Functions used by alphabeta + def max_value(state, alpha, beta, depth): + if cutoff_test(state, depth): + return eval_fn(state) + v = -infinity + for a in game.actions(state): + v = max(v, min_value(game.result(state, a), + alpha, beta, depth + 1)) + if v >= beta: + return v + alpha = max(alpha, v) + return v + + def min_value(state, alpha, beta, depth): + if cutoff_test(state, depth): + return eval_fn(state) + v = infinity + for a in game.actions(state): + v = min(v, max_value(game.result(state, a), + alpha, beta, depth + 1)) + if v <= alpha: + return v + beta = min(beta, v) + return v + + # Body of alphabeta_cutoff_search starts here: + # The default test cuts off at depth d or at a terminal state + cutoff_test = (cutoff_test or + (lambda state, depth: depth > d or + game.terminal_test(state))) + eval_fn = eval_fn or (lambda state: game.utility(state, player)) + best_score = -infinity + beta = infinity + best_action = None + for a in game.actions(state): + v = min_value(game.result(state, a), best_score, beta, 1) + if v > best_score: + best_score = v + best_action = a + return best_action + + +# ______________________________________________________________________________ +# Monte Carlo Tree Search + + +def monte_carlo_tree_search(state, game, N=1000): + def select(n): + """select a leaf node in the tree""" + if n.children: + return select(max(n.children.keys(), key=ucb)) + else: + return n + + def expand(n): + """expand the leaf node by adding all its children states""" + if not n.children and not game.terminal_test(n.state): + n.children = {MCT_Node(state=game.result(n.state, action), parent=n): action for action in + game.actions(n.state)} + return select(n) + + def simulate(game, state): + """simulate the utility of current state by random picking a step""" + player = game.to_move(state) + while not game.terminal_test(state): + action = random.choice(list(game.actions(state))) + state = game.result(state, action) + v = game.utility(state, player) + return -v + + def backprop(n, utility): + """passing the utility back to all parent nodes""" + if utility > 0: + n.U += utility + # if utility == 0: + # n.U += 0.5 + n.N += 1 + if n.parent: + backprop(n.parent, -utility) + + root = MCT_Node(state=state) + + while N > 0: + leaf = select(root) + child = expand(leaf) + result = simulate(game, child.state) + backprop(child, result) + N -= 1 + max_state = max(root.children, key=lambda p: p.N) + + return root.children.get(max_state) + +# ______________________________________________________________________________ +# Players for Games + + +def query_player(game, state): + """Make a move by querying standard input.""" + print("current state:") + game.display(state) + print("available moves: {}".format(game.actions(state))) + print("") + move = None + if game.actions(state): + move_string = input('Your move? ') + try: + move = eval(move_string) + except NameError: + move = move_string + else: + print('no legal moves: passing turn to next player') + return move + + +def random_player(game, state): + """A player that chooses a legal move at random.""" + return random.choice(game.actions(state)) if game.actions(state) else None + + +def alphabeta_player(game, state): + return alphabeta_search(state, game) + + +def expectiminimax_player(game, state): + return expectiminimax(state, game) + + +def mcts_player(game, state): + return monte_carlo_tree_search(state, game) + + +# ______________________________________________________________________________ +# Some Sample Games + + +class Game: + """A game is similar to a problem, but it has a utility for each + state and a terminal test instead of a path cost and a goal + test. To create a game, subclass this class and implement actions, + result, utility, and terminal_test. You may override display and + successors or you can inherit their default methods. You will also + need to set the .initial attribute to the initial state; this can + be done in the constructor.""" + + def actions(self, state): + """Return a list of the allowable moves at this point.""" + raise NotImplementedError + + def result(self, state, move): + """Return the state that results from making a move from a state.""" + raise NotImplementedError + + def utility(self, state, player): + """Return the value of this final state to player.""" + raise NotImplementedError + + def terminal_test(self, state): + """Return True if this is a final state for the game.""" + return not self.actions(state) + + def to_move(self, state): + """Return the player whose move it is in this state.""" + return state.to_move + + def display(self, state): + """Print or otherwise display the state.""" + print(state) + + def __repr__(self): + return '<{}>'.format(self.__class__.__name__) + + def play_game(self, *players): + """Play an n-person, move-alternating game.""" + state = self.initial + while True: + for player in players: + move = player(self, state) + state = self.result(state, move) + if self.terminal_test(state): + self.display(state) + return self.utility(state, self.to_move(self.initial)) + +class StochasticGame(Game): + """A stochastic game includes uncertain events which influence + the moves of players at each state. To create a stochastic game, subclass + this class and implement chances and outcome along with the other + unimplemented game class methods.""" + + def chances(self, state): + """Return a list of all possible uncertain events at a state.""" + raise NotImplementedError + + def outcome(self, state, chance): + """Return the state which is the outcome of a chance trial.""" + raise NotImplementedError + + def probability(self, chance): + """Return the probability of occurence of a chance.""" + raise NotImplementedError + + def play_game(self, *players): + """Play an n-person, move-alternating stochastic game.""" + state = self.initial + while True: + for player in players: + chance = random.choice(self.chances(state)) + state = self.outcome(state, chance) + move = player(self, state) + state = self.result(state, move) + if self.terminal_test(state): + self.display(state) + return self.utility(state, self.to_move(self.initial)) + +class Fig52Game(Game): + """The game represented in [Figure 5.2]. Serves as a simple test case.""" + + succs = dict(A=dict(a1='B', a2='C', a3='D'), + B=dict(b1='B1', b2='B2', b3='B3'), + C=dict(c1='C1', c2='C2', c3='C3'), + D=dict(d1='D1', d2='D2', d3='D3')) + utils = dict(B1=3, B2=12, B3=8, C1=2, C2=4, C3=6, D1=14, D2=5, D3=2) + initial = 'A' + + def actions(self, state): + return list(self.succs.get(state, {}).keys()) + + def result(self, state, move): + return self.succs[state][move] + + def utility(self, state, player): + if player == 'MAX': + return self.utils[state] + else: + return -self.utils[state] + + def terminal_test(self, state): + return state not in ('A', 'B', 'C', 'D') + + def to_move(self, state): + return 'MIN' if state in 'BCD' else 'MAX' + + +class Fig52Extended(Game): + """Similar to Fig52Game but bigger. Useful for visualisation""" + + succs = {i:dict(l=i*3+1, m=i*3+2, r=i*3+3) for i in range(13)} + utils = dict() + + def actions(self, state): + return sorted(list(self.succs.get(state, {}).keys())) + + def result(self, state, move): + return self.succs[state][move] + + def utility(self, state, player): + if player == 'MAX': + return self.utils[state] + else: + return -self.utils[state] + + def terminal_test(self, state): + return state not in range(13) + + def to_move(self, state): + return 'MIN' if state in {1, 2, 3} else 'MAX' + +class TicTacToe(Game): + """Play TicTacToe on an h x v board, with Max (first player) playing 'X'. + A state has the player to move, a cached utility, a list of moves in + the form of a list of (x, y) positions, and a board, in the form of + a dict of {(x, y): Player} entries, where Player is 'X' or 'O'.""" + + def __init__(self, h=3, v=3, k=3): + self.h = h + self.v = v + self.k = k + moves = [(x, y) for x in range(1, h + 1) + for y in range(1, v + 1)] + self.initial = GameState(to_move='X', utility=0, board={}, moves=moves) + + def actions(self, state): + """Legal moves are any square not yet taken.""" + return state.moves + + def result(self, state, move): + if move not in state.moves: + return state # Illegal move has no effect + board = state.board.copy() + board[move] = state.to_move + moves = list(state.moves) + moves.remove(move) + return GameState(to_move=('O' if state.to_move == 'X' else 'X'), + utility=self.compute_utility(board, move, state.to_move), + board=board, moves=moves) + + def utility(self, state, player): + """Return the value to player; 1 for win, -1 for loss, 0 otherwise.""" + return state.utility if player == 'X' else -state.utility + + def terminal_test(self, state): + """A state is terminal if it is won or there are no empty squares.""" + return state.utility != 0 or len(state.moves) == 0 + + def display(self, state): + board = state.board + for x in range(1, self.h + 1): + for y in range(1, self.v + 1): + print(board.get((x, y), '.'), end=' ') + print() + + def compute_utility(self, board, move, player): + """If 'X' wins with this move, return 1; if 'O' wins return -1; else return 0.""" + if (self.k_in_row(board, move, player, (0, 1)) or + self.k_in_row(board, move, player, (1, 0)) or + self.k_in_row(board, move, player, (1, -1)) or + self.k_in_row(board, move, player, (1, 1))): + return +1 if player == 'X' else -1 + else: + return 0 + + def k_in_row(self, board, move, player, delta_x_y): + """Return true if there is a line through move on board for player.""" + (delta_x, delta_y) = delta_x_y + x, y = move + n = 0 # n is number of moves in row + while board.get((x, y)) == player: + n += 1 + x, y = x + delta_x, y + delta_y + x, y = move + while board.get((x, y)) == player: + n += 1 + x, y = x - delta_x, y - delta_y + n -= 1 # Because we counted move itself twice + return n >= self.k + + +class ConnectFour(TicTacToe): + """A TicTacToe-like game in which you can only make a move on the bottom + row, or in a square directly above an occupied square. Traditionally + played on a 7x6 board and requiring 4 in a row.""" + + def __init__(self, h=7, v=6, k=4): + TicTacToe.__init__(self, h, v, k) + + def actions(self, state): + return [(x, y) for (x, y) in state.moves + if y == 1 or (x, y - 1) in state.board] + + +class Backgammon(StochasticGame): + """A two player game where the goal of each player is to move all the + checkers off the board. The moves for each state are determined by + rolling a pair of dice.""" + + def __init__(self): + """Initial state of the game""" + point = {'W' : 0, 'B' : 0} + board = [point.copy() for index in range(24)] + board[0]['B'] = board[23]['W'] = 2 + board[5]['W'] = board[18]['B'] = 5 + board[7]['W'] = board[16]['B'] = 3 + board[11]['B'] = board[12]['W'] = 5 + self.allow_bear_off = {'W' : False, 'B' : False} + self.direction = {'W' : -1, 'B' : 1} + self.initial = StochasticGameState(to_move='W', + utility=0, + board=board, + moves=self.get_all_moves(board, 'W'), chance=None) + + def actions(self, state): + """Return a list of legal moves for a state.""" + player = state.to_move + moves = state.moves + if len(moves) == 1 and len(moves[0]) == 1: + return moves + legal_moves = [] + for move in moves: + board = copy.deepcopy(state.board) + if self.is_legal_move(board, move, state.chance, player): + legal_moves.append(move) + return legal_moves + + def result(self, state, move): + board = copy.deepcopy(state.board) + player = state.to_move + self.move_checker(board, move[0], state.chance[0], player) + if len(move) == 2: + self.move_checker(board, move[1], state.chance[1], player) + to_move = ('W' if player == 'B' else 'B') + return StochasticGameState(to_move=to_move, + utility=self.compute_utility(board, move, player), + board=board, + moves=self.get_all_moves(board, to_move), chance=None) + + def utility(self, state, player): + """Return the value to player; 1 for win, -1 for loss, 0 otherwise.""" + return state.utility if player == 'W' else -state.utility + + def terminal_test(self, state): + """A state is terminal if one player wins.""" + return state.utility != 0 + + def get_all_moves(self, board, player): + """All possible moves for a player i.e. all possible ways of + choosing two checkers of a player from the board for a move + at a given state.""" + all_points = board + taken_points = [index for index, point in enumerate(all_points) + if point[player] > 0] + if self.checkers_at_home(board, player) == 1: + return [(taken_points[0], )] + moves = list(itertools.permutations(taken_points, 2)) + moves = moves + [(index, index) for index, point in enumerate(all_points) + if point[player] >= 2] + return moves + + def display(self, state): + """Display state of the game.""" + board = state.board + player = state.to_move + print("current state : ") + for index, point in enumerate(board): + print("point : ", index, " W : ", point['W'], " B : ", point['B']) + print("to play : ", player) + + def compute_utility(self, board, move, player): + """If 'W' wins with this move, return 1; if 'B' wins return -1; else return 0.""" + util = {'W' : 1, 'B' : -1} + for idx in range(0, 24): + if board[idx][player] > 0: + return 0 + return util[player] + + def checkers_at_home(self, board, player): + """Return the no. of checkers at home for a player.""" + sum_range = range(0, 7) if player == 'W' else range(17, 24) + count = 0 + for idx in sum_range: + count = count + board[idx][player] + return count + + def is_legal_move(self, board, start, steps, player): + """Move is a tuple which contains starting points of checkers to be + moved during a player's turn. An on-board move is legal if both the destinations + are open. A bear-off move is the one where a checker is moved off-board. + It is legal only after a player has moved all his checkers to his home.""" + dest1, dest2 = vector_add(start, steps) + dest_range = range(0, 24) + move1_legal = move2_legal = False + if dest1 in dest_range: + if self.is_point_open(player, board[dest1]): + self.move_checker(board, start[0], steps[0], player) + move1_legal = True + else: + if self.allow_bear_off[player]: + self.move_checker(board, start[0], steps[0], player) + move1_legal = True + if not move1_legal: + return False + if dest2 in dest_range: + if self.is_point_open(player, board[dest2]): + move2_legal = True + else: + if self.allow_bear_off[player]: + move2_legal = True + return move1_legal and move2_legal + + def move_checker(self, board, start, steps, player): + """Move a checker from starting point by a given number of steps""" + dest = start + steps + dest_range = range(0, 24) + board[start][player] -= 1 + if dest in dest_range: + board[dest][player] += 1 + if self.checkers_at_home(board, player) == 15: + self.allow_bear_off[player] = True + + def is_point_open(self, player, point): + """A point is open for a player if the no. of opponent's + checkers already present on it is 0 or 1. A player can + move a checker to a point only if it is open.""" + opponent = 'B' if player == 'W' else 'W' + return point[opponent] <= 1 + + def chances(self, state): + """Return a list of all possible dice rolls at a state.""" + dice_rolls = list(itertools.combinations_with_replacement([1, 2, 3, 4, 5, 6], 2)) + return dice_rolls + + def outcome(self, state, chance): + """Return the state which is the outcome of a dice roll.""" + dice = tuple(map((self.direction[state.to_move]).__mul__, chance)) + return StochasticGameState(to_move=state.to_move, + utility=state.utility, + board=state.board, + moves=state.moves, chance=dice) + + def probability(self, chance): + """Return the probability of occurence of a dice roll.""" + return 1/36 if chance[0] == chance[1] else 1/18 diff --git a/tests/test_agents_4e.py b/tests/test_agents_4e.py new file mode 100644 index 000000000..ca082887e --- /dev/null +++ b/tests/test_agents_4e.py @@ -0,0 +1,374 @@ +import random +from agents_4e import Direction +from agents_4e import Agent +from agents_4e import ReflexVacuumAgent, ModelBasedVacuumAgent, TrivialVacuumEnvironment, compare_agents,\ + RandomVacuumAgent, TableDrivenVacuumAgent, TableDrivenAgentProgram, RandomAgentProgram, \ + SimpleReflexAgentProgram, ModelBasedReflexAgentProgram, rule_match +from agents_4e import Wall, Gold, Explorer, Thing, Bump, Glitter, WumpusEnvironment, Pit, \ + VacuumEnvironment, Dirt + + +random.seed("aima-python") + + +def test_move_forward(): + d = Direction("up") + l1 = d.move_forward((0, 0)) + assert l1 == (0, -1) + + d = Direction(Direction.R) + l1 = d.move_forward((0, 0)) + assert l1 == (1, 0) + + d = Direction(Direction.D) + l1 = d.move_forward((0, 0)) + assert l1 == (0, 1) + + d = Direction("left") + l1 = d.move_forward((0, 0)) + assert l1 == (-1, 0) + + l2 = d.move_forward((1, 0)) + assert l2 == (0, 0) + + +def test_add(): + d = Direction(Direction.U) + l1 = d + "right" + l2 = d + "left" + assert l1.direction == Direction.R + assert l2.direction == Direction.L + + d = Direction("right") + l1 = d.__add__(Direction.L) + l2 = d.__add__(Direction.R) + assert l1.direction == "up" + assert l2.direction == "down" + + d = Direction("down") + l1 = d.__add__("right") + l2 = d.__add__("left") + assert l1.direction == Direction.L + assert l2.direction == Direction.R + + d = Direction(Direction.L) + l1 = d + Direction.R + l2 = d + Direction.L + assert l1.direction == Direction.U + assert l2.direction == Direction.D + + +def test_RandomAgentProgram() : + #create a list of all the actions a vacuum cleaner can perform + list = ['Right', 'Left', 'Suck', 'NoOp'] + # create a program and then an object of the RandomAgentProgram + program = RandomAgentProgram(list) + + agent = Agent(program) + # create an object of TrivialVacuumEnvironment + environment = TrivialVacuumEnvironment() + # add agent to the environment + environment.add_thing(agent) + # run the environment + environment.run() + # check final status of the environment + assert environment.status == {(1, 0): 'Clean' , (0, 0): 'Clean'} + + +def test_RandomVacuumAgent() : + # create an object of the RandomVacuumAgent + agent = RandomVacuumAgent() + # create an object of TrivialVacuumEnvironment + environment = TrivialVacuumEnvironment() + # add agent to the environment + environment.add_thing(agent) + # run the environment + environment.run() + # check final status of the environment + assert environment.status == {(1,0):'Clean' , (0,0) : 'Clean'} + + +def test_TableDrivenAgent(): + loc_A, loc_B = (0, 0), (1, 0) + # table defining all the possible states of the agent + table = {((loc_A, 'Clean'),): 'Right', + ((loc_A, 'Dirty'),): 'Suck', + ((loc_B, 'Clean'),): 'Left', + ((loc_B, 'Dirty'),): 'Suck', + ((loc_A, 'Dirty'), (loc_A, 'Clean')): 'Right', + ((loc_A, 'Clean'), (loc_B, 'Dirty')): 'Suck', + ((loc_B, 'Clean'), (loc_A, 'Dirty')): 'Suck', + ((loc_B, 'Dirty'), (loc_B, 'Clean')): 'Left', + ((loc_A, 'Dirty'), (loc_A, 'Clean'), (loc_B, 'Dirty')): 'Suck', + ((loc_B, 'Dirty'), (loc_B, 'Clean'), (loc_A, 'Dirty')): 'Suck' + } + + # create an program and then an object of the TableDrivenAgent + program = TableDrivenAgentProgram(table) + agent = Agent(program) + # create an object of TrivialVacuumEnvironment + environment = TrivialVacuumEnvironment() + # initializing some environment status + environment.status = {loc_A:'Dirty', loc_B:'Dirty'} + # add agent to the environment + environment.add_thing(agent) + + # run the environment by single step everytime to check how environment evolves using TableDrivenAgentProgram + environment.run(steps = 1) + assert environment.status == {(1,0): 'Clean', (0,0): 'Dirty'} + + environment.run(steps = 1) + assert environment.status == {(1,0): 'Clean', (0,0): 'Dirty'} + + environment.run(steps = 1) + assert environment.status == {(1,0): 'Clean', (0,0): 'Clean'} + + +def test_ReflexVacuumAgent() : + # create an object of the ReflexVacuumAgent + agent = ReflexVacuumAgent() + # create an object of TrivialVacuumEnvironment + environment = TrivialVacuumEnvironment() + # add agent to the environment + environment.add_thing(agent) + # run the environment + environment.run() + # check final status of the environment + assert environment.status == {(1,0):'Clean' , (0,0) : 'Clean'} + + +def test_SimpleReflexAgentProgram(): + class Rule: + + def __init__(self, state, action): + self.__state = state + self.action = action + + def matches(self, state): + return self.__state == state + + loc_A = (0, 0) + loc_B = (1, 0) + + # create rules for a two state Vacuum Environment + rules = [Rule((loc_A, "Dirty"), "Suck"), Rule((loc_A, "Clean"), "Right"), + Rule((loc_B, "Dirty"), "Suck"), Rule((loc_B, "Clean"), "Left")] + + def interpret_input(state): + return state + + # create a program and then an object of the SimpleReflexAgentProgram + program = SimpleReflexAgentProgram(rules, interpret_input) + agent = Agent(program) + # create an object of TrivialVacuumEnvironment + environment = TrivialVacuumEnvironment() + # add agent to the environment + environment.add_thing(agent) + # run the environment + environment.run() + # check final status of the environment + assert environment.status == {(1,0):'Clean' , (0,0) : 'Clean'} + + +def test_ModelBasedReflexAgentProgram(): + class Rule: + + def __init__(self, state, action): + self.__state = state + self.action = action + + def matches(self, state): + return self.__state == state + + loc_A = (0, 0) + loc_B = (1, 0) + + # create rules for a two-state vacuum environment + rules = [Rule((loc_A, "Dirty"), "Suck"), Rule((loc_A, "Clean"), "Right"), + Rule((loc_B, "Dirty"), "Suck"), Rule((loc_B, "Clean"), "Left")] + + def update_state(state, action, percept, transition_model, sensor_model): + return percept + + # create a program and then an object of the ModelBasedReflexAgentProgram class + program = ModelBasedReflexAgentProgram(rules, update_state, None, None) + agent = Agent(program) + # create an object of TrivialVacuumEnvironment + environment = TrivialVacuumEnvironment() + # add agent to the environment + environment.add_thing(agent) + # run the environment + environment.run() + # check final status of the environment + assert environment.status == {(1, 0): 'Clean', (0, 0): 'Clean'} + + +def test_ModelBasedVacuumAgent() : + # create an object of the ModelBasedVacuumAgent + agent = ModelBasedVacuumAgent() + # create an object of TrivialVacuumEnvironment + environment = TrivialVacuumEnvironment() + # add agent to the environment + environment.add_thing(agent) + # run the environment + environment.run() + # check final status of the environment + assert environment.status == {(1,0):'Clean' , (0,0) : 'Clean'} + + +def test_TableDrivenVacuumAgent() : + # create an object of the TableDrivenVacuumAgent + agent = TableDrivenVacuumAgent() + # create an object of the TrivialVacuumEnvironment + environment = TrivialVacuumEnvironment() + # add agent to the environment + environment.add_thing(agent) + # run the environment + environment.run() + # check final status of the environment + assert environment.status == {(1, 0):'Clean', (0, 0):'Clean'} + + +def test_compare_agents() : + environment = TrivialVacuumEnvironment + agents = [ModelBasedVacuumAgent, ReflexVacuumAgent] + + result = compare_agents(environment, agents) + performance_ModelBasedVacummAgent = result[0][1] + performance_ReflexVacummAgent = result[1][1] + + # The performance of ModelBasedVacuumAgent will be at least as good as that of + # ReflexVacuumAgent, since ModelBasedVacuumAgent can identify when it has + # reached the terminal state (both locations being clean) and will perform + # NoOp leading to 0 performance change, whereas ReflexVacuumAgent cannot + # identify the terminal state and thus will keep moving, leading to worse + # performance compared to ModelBasedVacuumAgent. + assert performance_ReflexVacummAgent <= performance_ModelBasedVacummAgent + + +def test_TableDrivenAgentProgram(): + table = {(('foo', 1),): 'action1', + (('foo', 2),): 'action2', + (('bar', 1),): 'action3', + (('bar', 2),): 'action1', + (('foo', 1), ('foo', 1),): 'action2', + (('foo', 1), ('foo', 2),): 'action3', + } + agent_program = TableDrivenAgentProgram(table) + assert agent_program(('foo', 1)) == 'action1' + assert agent_program(('foo', 2)) == 'action3' + assert agent_program(('invalid percept',)) == None + + +def test_Agent(): + def constant_prog(percept): + return percept + agent = Agent(constant_prog) + result = agent.program(5) + assert result == 5 + +def test_VacuumEnvironment(): + # Initialize Vacuum Environment + v = VacuumEnvironment(6,6) + #Get an agent + agent = ModelBasedVacuumAgent() + agent.direction = Direction(Direction.R) + v.add_thing(agent) + v.add_thing(Dirt(), location=(2,1)) + + # Check if things are added properly + assert len([x for x in v.things if isinstance(x, Wall)]) == 20 + assert len([x for x in v.things if isinstance(x, Dirt)]) == 1 + + #Let the action begin! + assert v.percept(agent) == ("Clean", "None") + v.execute_action(agent, "Forward") + assert v.percept(agent) == ("Dirty", "None") + v.execute_action(agent, "TurnLeft") + v.execute_action(agent, "Forward") + assert v.percept(agent) == ("Dirty", "Bump") + v.execute_action(agent, "Suck") + assert v.percept(agent) == ("Clean", "None") + old_performance = agent.performance + v.execute_action(agent, "NoOp") + assert old_performance == agent.performance + +def test_WumpusEnvironment(): + def constant_prog(percept): + return percept + # Initialize Wumpus Environment + w = WumpusEnvironment(constant_prog) + + #Check if things are added properly + assert len([x for x in w.things if isinstance(x, Wall)]) == 20 + assert any(map(lambda x: isinstance(x, Gold), w.things)) + assert any(map(lambda x: isinstance(x, Explorer), w.things)) + assert not any(map(lambda x: not isinstance(x,Thing), w.things)) + + #Check that gold and wumpus are not present on (1,1) + assert not any(map(lambda x: isinstance(x, Gold) or isinstance(x,WumpusEnvironment), + w.list_things_at((1, 1)))) + + #Check if w.get_world() segments objects correctly + assert len(w.get_world()) == 6 + for row in w.get_world(): + assert len(row) == 6 + + #Start the game! + agent = [x for x in w.things if isinstance(x, Explorer)][0] + gold = [x for x in w.things if isinstance(x, Gold)][0] + pit = [x for x in w.things if isinstance(x, Pit)][0] + + assert w.is_done()==False + + #Check Walls + agent.location = (1, 2) + percepts = w.percept(agent) + assert len(percepts) == 5 + assert any(map(lambda x: isinstance(x,Bump), percepts[0])) + + #Check Gold + agent.location = gold.location + percepts = w.percept(agent) + assert any(map(lambda x: isinstance(x,Glitter), percepts[4])) + agent.location = (gold.location[0], gold.location[1]+1) + percepts = w.percept(agent) + assert not any(map(lambda x: isinstance(x,Glitter), percepts[4])) + + #Check agent death + agent.location = pit.location + assert w.in_danger(agent) == True + assert agent.alive == False + assert agent.killed_by == Pit.__name__ + assert agent.performance == -1000 + + assert w.is_done()==True + +def test_WumpusEnvironmentActions(): + def constant_prog(percept): + return percept + # Initialize Wumpus Environment + w = WumpusEnvironment(constant_prog) + + agent = [x for x in w.things if isinstance(x, Explorer)][0] + gold = [x for x in w.things if isinstance(x, Gold)][0] + pit = [x for x in w.things if isinstance(x, Pit)][0] + + agent.location = (1, 1) + assert agent.direction.direction == "right" + w.execute_action(agent, 'TurnRight') + assert agent.direction.direction == "down" + w.execute_action(agent, 'TurnLeft') + assert agent.direction.direction == "right" + w.execute_action(agent, 'Forward') + assert agent.location == (2, 1) + + agent.location = gold.location + w.execute_action(agent, 'Grab') + assert agent.holding == [gold] + + agent.location = (1, 1) + w.execute_action(agent, 'Climb') + assert not any(map(lambda x: isinstance(x, Explorer), w.things)) + + assert w.is_done()==True \ No newline at end of file diff --git a/tests/test_games_4e.py b/tests/test_games_4e.py new file mode 100644 index 000000000..1cfb78763 --- /dev/null +++ b/tests/test_games_4e.py @@ -0,0 +1,88 @@ +from games4e import * + +# Creating the game instances +f52 = Fig52Game() +ttt = TicTacToe() +con4 = ConnectFour() + + +def gen_state(to_move='X', x_positions=[], o_positions=[], h=3, v=3, k=3): + """Given whose turn it is to move, the positions of X's on the board, the + positions of O's on the board, and, (optionally) number of rows, columns + and how many consecutive X's or O's required to win, return the corresponding + game state""" + + moves = set([(x, y) for x in range(1, h + 1) for y in range(1, v + 1)]) \ + - set(x_positions) - set(o_positions) + moves = list(moves) + board = {} + for pos in x_positions: + board[pos] = 'X' + for pos in o_positions: + board[pos] = 'O' + return GameState(to_move=to_move, utility=0, board=board, moves=moves) + + +def test_minimax_decision(): + assert minimax_decision('A', f52) == 'a1' + assert minimax_decision('B', f52) == 'b1' + assert minimax_decision('C', f52) == 'c1' + assert minimax_decision('D', f52) == 'd3' + + +def test_alphabeta_search(): + assert alphabeta_search('A', f52) == 'a1' + assert alphabeta_search('B', f52) == 'b1' + assert alphabeta_search('C', f52) == 'c1' + assert alphabeta_search('D', f52) == 'd3' + + state = gen_state(to_move='X', x_positions=[(1, 1), (3, 3)], + o_positions=[(1, 2), (3, 2)]) + assert alphabeta_search(state, ttt) == (2, 2) + + state = gen_state(to_move='O', x_positions=[(1, 1), (3, 1), (3, 3)], + o_positions=[(1, 2), (3, 2)]) + assert alphabeta_search(state, ttt) == (2, 2) + + state = gen_state(to_move='O', x_positions=[(1, 1)], + o_positions=[]) + assert alphabeta_search(state, ttt) == (2, 2) + + state = gen_state(to_move='X', x_positions=[(1, 1), (3, 1)], + o_positions=[(2, 2), (3, 1)]) + assert alphabeta_search(state, ttt) == (1, 3) + + +def test_monte_carlo_tree_search(): + state = gen_state(to_move='X', x_positions=[(1, 1), (3, 3)], + o_positions=[(1, 2), (3, 2)]) + assert monte_carlo_tree_search(state, ttt) == (2, 2) + + state = gen_state(to_move='O', x_positions=[(1, 1), (3, 1), (3, 3)], + o_positions=[(1, 2), (3, 2)]) + assert monte_carlo_tree_search(state, ttt) == (2, 2) + + state = gen_state(to_move='O', x_positions=[(1, 1)], + o_positions=[]) + assert monte_carlo_tree_search(state, ttt) == (2, 2) + + state = gen_state(to_move='X', x_positions=[(1, 1), (3, 1)], + o_positions=[(2, 2), (3, 1)]) + assert monte_carlo_tree_search(state, ttt) == (1, 3) + + # should never lose to a random or alphabeta player in a ttt game + assert ttt.play_game(mcts_player, random_player) >= 0 + assert ttt.play_game(mcts_player, alphabeta_player) >= 0 + + # should never lose to a random player in a connect four game + assert con4.play_game(mcts_player, random_player) >= 0 + + +def test_random_tests(): + assert Fig52Game().play_game(alphabeta_player, alphabeta_player) == 3 + + # The player 'X' (one who plays first) in TicTacToe never loses: + assert ttt.play_game(alphabeta_player, alphabeta_player) >= 0 + + # The player 'X' (one who plays first) in TicTacToe never loses: + assert ttt.play_game(alphabeta_player, random_player) >= 0 diff --git a/utils.py b/utils.py index c2644b787..45dd03636 100644 --- a/utils.py +++ b/utils.py @@ -794,6 +794,21 @@ def __delitem__(self, key): heapq.heapify(self.heap) +# ______________________________________________________________________________ +# Monte Carlo tree node and ucb function +class MCT_Node: + """Node in the Monte Carlo search tree, keeps track of the children states""" + def __init__(self, parent=None, state=None, U=0, N=0): + self.__dict__.update(parent=parent, state=state, U=U, N=N) + self.children = {} + self.actions = None + + +def ucb(n, C=1.4): + return (float('inf') if n.N == 0 else + n.U / n.N + C * math.sqrt(math.log(n.parent.N)/n.N)) + + # ______________________________________________________________________________ # Useful Shorthands