diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 000000000..2398f62e3 --- /dev/null +++ b/.coveragerc @@ -0,0 +1,3 @@ +[report] +omit = + tests/* \ No newline at end of file diff --git a/.travis.yml b/.travis.yml index cc770609a..18019d2ff 100644 --- a/.travis.yml +++ b/.travis.yml @@ -15,10 +15,11 @@ install: - pip install networkx - pip install ipywidgets - pip install Pillow + - pip install pytest-cov - pip install ipythonblocks script: - - py.test + - py.test --cov=./ - python -m doctest -v *.py after_success: diff --git a/README.md b/README.md index 5efe0fd60..11ea2e62e 100644 --- a/README.md +++ b/README.md @@ -172,7 +172,7 @@ Here is a table of the implemented data structures, the figure, name of the impl # Acknowledgements -Many thanks for contributions over the years. I got bug reports, corrected code, and other support from Darius Bacon, Phil Ruggera, Peng Shao, Amit Patil, Ted Nienstedt, Jim Martin, Ben Catanzariti, and others. Now that the project is on GitHub, you can see the [contributors](https://github.com/aimacode/aima-python/graphs/contributors) who are doing a great job of actively improving the project. Many thanks to all contributors, especially @darius, @SnShine, @reachtarunhere, @MrDupin, @Chipe1, @ad71 and @MariannaSpyrakou. +Many thanks for contributions over the years. I got bug reports, corrected code, and other support from Darius Bacon, Phil Ruggera, Peng Shao, Amit Patil, Ted Nienstedt, Jim Martin, Ben Catanzariti, and others. Now that the project is on GitHub, you can see the [contributors](https://github.com/aimacode/aima-python/graphs/contributors) who are doing a great job of actively improving the project. Many thanks to all contributors, especially @darius, @SnShine, @reachtarunhere, @antmarakis, @Chipe1, @ad71 and @MariannaSpyrakou. [agents]:../master/agents.py 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/index.ipynb b/index.ipynb index 2ae5742bb..f9da121f2 100644 --- a/index.ipynb +++ b/index.ipynb @@ -18,7 +18,7 @@ "\n", "3. [**Search**](./search.ipynb)\n", "\n", - "4. [**Search - 4th edition**](./search-4e.ipynb)\n", + "4. [**Search - 4th edition**](./search4e.ipynb)\n", "\n", "4. [**Games**](./games.ipynb)\n", "\n", diff --git a/learning.py b/learning.py index 3759d6c76..7fd000950 100644 --- a/learning.py +++ b/learning.py @@ -751,7 +751,7 @@ def BackPropagationLearner(dataset, net, learning_rate, epochs, activation=sigmo # Error for the MSE cost function err = [t_val[i] - o_nodes[i].value for i in range(o_units)] - # The activation function used is relu or sigmoid function + # Calculate delta at output if node.activation == sigmoid: delta[-1] = [sigmoid_derivative(o_nodes[i].value) * err[i] for i in range(o_units)] elif node.activation == relu: 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