wumpus.py

# wumpus.py
# ---------
# Licensing Information:
# Please DO NOT DISTRIBUTE OR PUBLISH solutions to this project.
# You are free to use and extend these projects for EDUCATIONAL PURPOSES ONLY.
# The Hunt The Wumpus AI project was developed at University of Arizona
# by Clay Morrison (clayton@sista.arizona.edu), spring 2013.
# This project extends the python code provided by Peter Norvig as part of
# the Artificial Intelligence: A Modern Approach (AIMA) book example code;
# see http://aima.cs.berkeley.edu/code.html
# In particular, the following files come directly from the AIMA python
# code: ['agents.py', 'logic.py', 'search.py', 'utils.py']
# ('logic.py' has been modified by Clay Morrison in locations with the
# comment 'CTM')
# The file ['minisat.py'] implements a slim system call wrapper to the minisat
# (see http://minisat.se) SAT solver, and is directly based on the satispy
# python project, see https://github.com/netom/satispy .

from wumpus_agent import *
# from time import clock
import time

start_time = time.process_time()
# Code to measure
end_time = time.process_time()
elapsed_time = end_time - start_time
print(f"CPU time used: {elapsed_time} seconds")
import wumpus_environment


#-------------------------------------------------------------------------------
# Wumpus World Scenarios
#-------------------------------------------------------------------------------

class WumpusWorldScenario(object):
    """
    Construct a Wumpus World Scenario
    Objects that can be added to the environment:
        Wumpus()
        Pit()
        Gold()
        Wall()
        HybridWumpusAgent(heading)  # A propositional logic Wumpus World agent
        Explorer(program, heading)  # A non-logical Wumpus World agent (mostly for debugging)
    Provides methods to load layout from file
    Provides step and run methods to run the scenario
        with the provided agent's agent_program
    """

    def __init__(self, layout_file=None, agent=None, objects=None,
                 width=None, height=None, entrance=None, trace=True):
        """
        layout_file := (<string: layout_file_name>, <agent>)
        """
        if agent != None and not isinstance(agent, Explorer):
            raise Exception("agent must be type Explorer, got instance of class\n" \
                            + " {0}".format(agent.__class__))
        if layout_file:
            objects, width, height, entrance = self.load_layout(layout_file)

        self.width, self.height = width, height
        self.entrance = entrance
        self.agent = agent
        self.objects = objects
        self.trace = trace
        self.env = self.build_world(width, height, entrance, agent, objects)

    def build_world(self, width, height, entrance, agent, objects):
        """
        Create a WumpusEnvironment with dimensions width,height
        Set the environment entrance
        objects := [(<wumpus_environment_object>, <location: (<x>,<y>) >, ...]
        """
        env = WumpusEnvironment(width, height, entrance)
        if self.trace:
            agent = wumpus_environment.TraceAgent(agent)
        agent.register_environment(env)
        env.add_thing(agent, env.entrance)
        for (obj, loc) in objects:
            env.add_thing(obj, loc)
        return env

    def load_layout(self, layout_file):
        """
        Load text file specifying Wumpus Environment initial configuration
        Text file is N (rows) by M (columns) grid where each cell in a row
        consists of M comma-separated cells specs, where each cell contains
        either:
           '.' : space (really just a placeholder)
        or a one or more of (although typically just have one per cell):
           'W' : wumpus
           'P' : pit
           'G' : gold
           'A' : wumpus hunter agent (heading specified in agent object)
        """

        if layout_file.endswith('.lay'):
            layout = self.tryToLoad('layouts/' + layout_file)
            if not layout: layout = self.tryToLoad(layout_file)
        else:
            layout = self.tryToLoad('layouts/' + layout_file + '.lay')
            if not layout: layout = self.tryToLoad(layout_file + '.lay')

        if not layout:
            raise Exception("Could not find layout file: {0}".format(layout_file))

        print("Loaded layout '{0}'".format(layout_file))

        objects = []
        entrance = (1, 1)  # default entrance location

        ri = len(layout)
        largest_ci = 0
        for row in layout:
            ci = 0
            if row:
                ri -= 1
                row = row.split(',')
                for cell in row:
                    ci += 1
                    if ci > largest_ci: largest_ci = ci
                    for char in cell:
                        if char == 'W':
                            objects.append((Wumpus(), (ci, ri)))
                        elif char == 'P':
                            objects.append((Pit(), (ci, ri)))
                        elif char == 'G':
                            objects.append((Gold(), (ci, ri)))
                        elif char == 'A':
                            entrance = (ci, ri)

        return objects, largest_ci, len(layout) - ri, entrance

    def tryToLoad(self, fullname):
        if (not os.path.exists(fullname)): return None
        f = open(fullname)
        try:
            return [line.strip() for line in f]
        finally:
            f.close()

    def step(self):
        self.env.step()
        print()
        print("Current Wumpus Environment:")
        print(self.env.to_string())

    def run(self, steps=1000):
        print(self.env.to_string())
        for step in range(steps):
            if self.env.is_done():
                print("DONE.")
                slist = []
                if len(self.env.agents) > 0:
                    slist += ['Final Scores:']
                for agent in self.env.agents:
                    slist.append(' {0}={1}'.format(agent, agent.performance_measure))
                    if agent.verbose:
                        if hasattr(agent, 'number_of_clauses_over_epochs'):
                            print("number_of_clauses_over_epochs:" \
                                  + " {0}".format(agent.number_of_clauses_over_epochs))
                        if hasattr(agent, 'belief_loc_query_times'):
                            print("belief_loc_query_times:" \
                                  + " {0}".format(agent.belief_loc_query_times))
                print(''.join(slist))
                return
            self.step()

    def to_string(self):
        s = "Environment width={0}, height={1}\n".format(self.width, self.height)
        s += "Initial Position: {0}\n".format(self.entrance)
        s += "Actions: {0}\n".format(self.actions)
        return s

    def pprint(self):
        print(self.to_string())
        print(self.env.to_string())


#-------------------------------------------------------------------------------

def world_scenario_hybrid_wumpus_agent_from_layout(layout_filename):
    """
    Create WumpusWorldScenario with an automated agent_program that will
        try to solve the Hunt The Wumpus game on its own.
    layout_filename := name of layout file to load
    """
    return WumpusWorldScenario(layout_file=layout_filename,
                               agent=HybridWumpusAgent('north', verbose=True),
                               trace=False)


#------------------------------------
# examples of constructing HybridWumpusAgent scenario
# specifying objects as list

def wscenario_4x4_HybridWumpusAgent():
    return WumpusWorldScenario(agent=HybridWumpusAgent('north', verbose=True),
                               objects=[(Wumpus(), (1, 3)),
                                        (Pit(), (3, 3)),
                                        (Pit(), (3, 1)),
                                        (Gold(), (2, 3))],
                               width=4, height=4, entrance=(1, 1),
                               trace=True)


#-------------------------------------------------------------------------------

def world_scenario_manual_with_kb_from_layout(layout_filename):
    """
    Create WumpusWorldScenario with a manual agent_program and Knowledge Base
        (see with_manual_kb_program)
    layout_filename := name of layout file to load
    """
    return WumpusWorldScenario(layout_file=layout_filename,
                               agent=with_manual_kb_program(HybridWumpusAgent('north',
                                                                              verbose=True)),
                               trace=False)


#------------------------------------
# examples of constructing manual wumpus agent with KB scenario
# specifying objects as list

def wscenario_4x4_manual_HybridWumpusAgent():
    return WumpusWorldScenario(agent=with_manual_kb_program(HybridWumpusAgent('north', verbose=True)),
                               objects=[(Wumpus(), (1, 3)),
                                        (Pit(), (3, 3)),
                                        (Pit(), (3, 1)),
                                        (Gold(), (2, 3))],
                               width=4, height=4, entrance=(1, 1),
                               trace=True)


#-------------------------------------------------------------------------------

def world_scenario_manual_from_layout(layout_filename):
    """
    Create WumpusWorldScenario with a manual agent_program (see with_manual_program)
    layout_filename := name of layout file to load
    """
    return WumpusWorldScenario(layout_file=layout_filename,
                               agent=with_manual_program(Explorer(heading='north', verbose=True)),
                               trace=False)


#------------------------------------
# examples of constructing manually-playable scenarios
# specifying objects as list or from layout file

def wscenario_4x4_manual():
    return WumpusWorldScenario(agent=with_manual_program(Explorer(heading='north', verbose=True)),
                               objects=[(Wumpus(), (1, 3)),
                                        (Pit(), (3, 3)),
                                        (Pit(), (3, 1)),
                                        (Gold(), (2, 3))],
                               width=4, height=4, entrance=(1, 1),
                               trace=False)


def wscenario_4x4_manual_book_from_layout():
    return WumpusWorldScenario(layout_file="wumpus_4x4_book",
                               agent=with_manual_program(Explorer(heading='north', verbose=True)),
                               trace=False)


def wscenario_4x4_manual_layout2_from_layout():
    return WumpusWorldScenario(layout_file="wumpus_4x4_2",
                               agent=with_manual_program(Explorer(heading='north', verbose=True)),
                               trace=False)


#-------------------------------------------------------------------------------
# Manual agent program
#-------------------------------------------------------------------------------

def with_manual_program(agent):
    """
    Take <agent> and replaces its agent_program with manual_program.
    manual_program waits for keyboard input and executes command.
    This uses a closure.  Three cheers for closures !!!
    (if you don't know what a closure is, read this:
       http://en.wikipedia.org/wiki/Closure_(computer_science) )
    """

    helping = ['?', 'help']
    stopping = ['quit', 'stop', 'exit']
    actions = ['TurnRight', 'TurnLeft', 'Forward', 'Grab', 'Climb', 'Shoot', 'Wait']

    def show_commands():
        print("   The following are valid Hunt The Wumpus action:")
        print("     {0}".format(', '.join(map(lambda a: '\'{0}\''.format(a), actions))))
        print("   Enter {0} to get this command info" \
              .format(' or '.join(map(lambda a: '\'{0}\''.format(a), helping))))
        print("   Enter {0} to stop playing" \
              .format(' or '.join(map(lambda a: '\'{0}\''.format(a), stopping))))
        print("   Enter 'env' to display current wumpus environment")

    def manual_program(percept):
        print("[{0}] You perceive: {1}".format(agent.time, agent.pretty_percept_vector(percept)))
        action = None
        while not action:
            val = input("Enter Action ('?' for list of commands): ")
            val = val.strip()
            if val in helping:
                print()
                show_commands()
                print()
            elif val in stopping:
                action = 'Stop'
            elif val == 'env':
                print()
                print("Current wumpus environment:")
                print(agent.env.to_string())
            elif val in actions:
                action = val
            else:
                print("'{0}' is an invalid command;".format(val) + " try again (enter '?' for list of commands)")
        agent.time += 1
        return action

    agent.program = manual_program
    return agent


#-------------------------------------------------------------------------------
# Manual agent program with Knowledge Base
#-------------------------------------------------------------------------------

def with_manual_kb_program(agent):
    """
    Take <agent> and replaces its agent_program with manual_kb_program.
    Assumes the <agent> is a HybridWumpusAgent.
    (TODO: separate out logical agent from HybridWumpusAgent)
    Agent program that waits for keyboard input and executes command.
    Also provides interface for doing KB queries.
    Closures rock!
    """

    helping = ['?', 'help']
    stopping = ['quit', 'stop', 'exit']
    actions = ['TurnRight', 'TurnLeft', 'Forward', 'Grab', 'Climb', 'Shoot', 'Wait']
    queries = [('qp', 'Query a single proposition;\n' \
                + '           E.g. \'qp B1_1\' or \'qp OK1_1_3\', \'qp HeadingWest4\''),
               ('qpl', 'Query a-temporal location-based proposition at all x,y locations;\n' \
                + '           E.g., \'qpl P\' runs all queries of P<x>_<y>'),
               ('qplt', 'Query temporal and location-based propositions at all x,y locations;\n' \
                + '           E.g., \'qplt OK 4\' runs all queries of the OK<x>_<y>_4'),
               ('q!', 'Run ALL queries for optionally specified time (default is current time);\n' \
                + '           (can be time consuming!)')]

    def show_commands():
        print("Available Commands:")
        print("   The following are valid Hunt The Wumpus actions:")
        print("     {0}".format(', '.join(map(lambda a: '\'{0}\''.format(a), actions))))
        print("   Enter {0} to get this command info" \
              .format(' or '.join(map(lambda a: '\'{0}\''.format(a), helping))))
        print("   Enter {0} to stop playing" \
              .format(' or '.join(map(lambda a: '\'{0}\''.format(a), stopping))))
        print("   Enter 'env' to display current wumpus environment")
        print("   Enter 'kbsat' to check if the agent's KB is satisfiable")
        print("      If the KB is NOT satisfiable, then there's a contradiction that needs fixing.")
        print("      NOTE: A satisfiable KB does not mean there aren't other problems.")
        print("   Enter 'save-axioms' to save all of the KB axioms to 'kb-axioms.txt'")
        print("      This will overwrite any existing 'kb-axioms.txt'")
        print("   Enter 'save-clauses' to save all of the KB clauses to text file 'kb-clauses.txt'")
        print("      This will overwrite any existing 'kb-clauses.txt'")
        print("   Enter 'props' to list all of the proposition bases")
        print("   Queries:")
        for query, desc in queries:
            print("      {0} : {1}".format(query, desc))

    def show_propositions():
        print("Proposition Bases:")
        print("   Atemporal location-based propositions (include x,y index: P<x>_<y>)")
        print("     '" + '\', \''.join(proposition_bases_atemporal_location) + '\'')
        print("   Perceptual propositions (include time index: P<t>)")
        print("     '" + '\', \''.join(proposition_bases_perceptual_fluents) + '\'')
        print("   Location fluent propositions (include x,y and time index: P<x>_<y>_<t>)")
        print("     '" + '\', \''.join(proposition_bases_location_fluents) + '\'')
        print("   State fluent propositions (include time index: P<t>)")
        print("     '" + '\', \''.join(proposition_bases_state_fluents[:4]) + '\',')
        print("     '" + '\', \''.join(proposition_bases_state_fluents[4:]) + '\'')
        print("   Action propositions (include time index: P<t>)")
        print("     '" + '\', \''.join(proposition_bases_actions) + '\'')

    def write_list_to_text_file(filename, list):
        outfile = file(filename, 'w')
        for item in list:
            outfile.write('{0}\n'.format(item))
        outfile.close()

    def check_kb_status():
        """
        Tests whether the agent KB is satisfiable.
        If not, that means the KB contains a contradiction that needs fixing.
        However, being satisfiable does not mean the KB is correct.
        """
        result = minisat(agent.kb.clauses)
        if result:
            print("Agent KB is satisfiable")
        else:
            print("Agent KB is NOT satisfiable!!  There is contradiction that needs fixing!")

    def simple_query(proposition):
        """
        Executes a simple query to the agent KB for specified proposition.
        """
        result = agent.kb.ask(expr(proposition))
        if result == None:
            print("{0}: Unknown!".format(proposition))
        else:
            print("{0}: {1}".format(proposition, result))

    def location_based_query(proposition_base):
        """
        Executes queries for the specified type of proposition, for
        each x,y location.
        proposition_base := as all of the propositions include in their
        name 1 or more indexes (for time and/or x,y location), the
        proposition_base is the simple string representing the base
        of the proposition witout the indexes, which are added in
        code, below.
        time := the time index of the propositions being queried
        """
        display_env = WumpusEnvironment(agent.width, agent.height)
        start_time = clock()
        print("Running queries for: {0}<x>_<y>".format(proposition_base))
        for x in range(1, agent.width + 1):
            for y in range(1, agent.height + 1):
                query = expr('{0}{1}_{2}'.format(proposition_base, x, y))
                result = agent.kb.ask(query)
                if result == None:
                    display_env.add_thing(Proposition(query, '?'), (x, y))
                else:
                    display_env.add_thing(Proposition(query, result), (x, y))
        end_time = clock()
        print("          >>> time elapsed while making queries:" \
              + " {0}".format(end_time - start_time))
        print(display_env.to_string(agent.time,
                                    title="All {0}<x>_<y> queries".format(proposition_base)))

    def location_time_based_query(proposition_base, time):
        """
        Executes queries for the specified type of proposition, for
        each x,y location, at the specified time.
        proposition_base := as all of the propositions include in their
        name 1 or more indexes (for time and/or x,y location), the
        proposition_base is the simple string representing the base
        of the proposition witout the indexes, which are added in
        code, below.
        time := the time index of the propositions being queried
        """
        display_env = WumpusEnvironment(agent.width, agent.height)
        start_time = clock()
        print("Running queries for: {0}<x>_<y>_{1}".format(proposition_base, time))
        for x in range(1, agent.width + 1):
            for y in range(1, agent.height + 1):
                query = expr('{0}{1}_{2}_{3}'.format(proposition_base, x, y, time))
                result = agent.kb.ask(query)
                if result == None:
                    display_env.add_thing(Proposition(query, '?'), (x, y))
                else:
                    display_env.add_thing(Proposition(query, result), (x, y))
        end_time = clock()
        print("          >>> time elapsed while making queries:" \
              + " {0}".format(end_time - start_time))
        print(display_env.to_string(agent.time,
                                    title="All {0}<x>_<y>_{1} queries".format(proposition_base,
                                                                              time)))

    def run_all_queries(time):
        check_kb_status()
        for p in proposition_bases_perceptual_fluents:
            simple_query(p + '{0}'.format(time))
        for p in proposition_bases_atemporal_location:
            location_based_query(p)
        for p in proposition_bases_location_fluents:
            location_time_based_query(p, time)
        for p in proposition_bases_state_fluents:
            simple_query(p + '{0}'.format(time))
        # remove the quotes below and add quotes to the following if-statement
        # in order to query all actions from time 0 to now
        '''
        print "Querying actions from time 0 to {0}".format(time)
        for p in propositions_actions:
            for t in range(time+1):
                simple_query(p + '{0}'.format(t))
        '''
        if time - 1 > 0:
            print("Actions from previous time: {0}".format(time - 1))
            for p in proposition_bases_actions:
                simple_query(p + '{0}'.format(time - 1))

        print("FINISHED running all queries for time {0}".format(time))

    def manual_kb_program(percept):

        print("------------------------------------------------------------------")
        print("At time {0}".format(agent.time))
        # update current location and heading based on current KB knowledge state
        print("     HWA.infer_and_set_belief_location()")
        agent.infer_and_set_belief_location()
        print("     HWA.infer_and_set_belief_heading()")
        agent.infer_and_set_belief_heading()

        percept_sentence = agent.make_percept_sentence(percept)
        print("     HWA.agent_program(): kb.tell(percept_sentence):")
        print("         {0}".format(percept_sentence))
        agent.kb.tell(percept_sentence)  # update the agent's KB based on percepts

        clauses_before = len(agent.kb.clauses)
        print("     HWA.agent_program(): Prepare to add temporal axioms")
        print("         Number of clauses in KB before: {0}".format(clauses_before))
        agent.add_temporal_axioms()
        clauses_after = len(agent.kb.clauses)
        print("         Number of clauses in KB after: {0}".format(clauses_after))
        print("         Total clauses added to KB: {0}".format(clauses_after - clauses_before))
        agent.number_of_clauses_over_epochs.append(len(agent.kb.clauses))

        action = None
        while not action:
            print("[{0}] You perceive: {1}".format(agent.time,
                                                   agent.pretty_percept_vector(percept)))
            val = raw_input("Enter Action ('?' for list of commands): ")
            val = val.strip()
            if val in helping:
                print()
                show_commands()
                print()
            elif val in stopping:
                action = 'Stop'
            elif val in actions:
                action = val
            elif val == 'env':
                print()
                print("Current wumpus environment:")
                print(agent.env.to_string())
            elif val == 'props':
                print()
                show_propositions()
                print()
            elif val == 'kbsat':
                check_kb_status()
                print()
            elif val == 'save-axioms':
                write_list_to_text_file('kb-axioms.txt', agent.kb.axioms)
                print("   Saved to 'kb-axioms.txt'")
                print()
            elif val == 'save-clauses':
                write_list_to_text_file('kb-clauses.txt', agent.kb.clauses)
                print("   Saved to 'kb-clauses.txt'")
                print()
            else:
                q = val.split(' ')
                if len(q) == 2 and q[0] == 'qp':
                    simple_query(q[1])
                    print()
                elif len(q) == 2 and q[0] == 'qpl':
                    location_based_query(q[1])
                    print()
                elif len(q) == 3 and q[0] == 'qplt':
                    location_time_based_query(q[1], q[2])
                    print()
                elif q[0] == 'q!':
                    if len(q) == 2:
                        t = int(q[1])
                        run_all_queries(t)
                    else:
                        run_all_queries(agent.time)
                    print()
                else:
                    print("'{0}' is an invalid command;".format(val) \
                          + " try again (enter '?' for list of commands)")
                    print()

        # update KB with selected action
        agent.kb.tell(add_time_stamp(action, agent.time))

        agent.time += 1

        return action

    agent.program = manual_kb_program
    return agent


#-------------------------------------------------------------------------------
# Test MiniSat connection
#-------------------------------------------------------------------------------

def run_minisat_test():
    """
    Test connection to MiniSat
    """
    import logic

    queries = [("(P | ~P)", True),  # SAT
               ("(P & ~P)", False),  # UNSAT
               ("(P | R) <=> (~(Q | R) & (R >> ~(S <=> T)))", True)  # SAT
               ]

    print("Running simple MiniSat test:")
    t = 1
    failed = []
    for query, expected_result in queries:
        print("-----------------------------------------------------")
        print("Test {0}".format(t))
        print("  Query:      '{0}'".format(query))
        query = logic.conjuncts(logic.to_cnf(logic.expr(query)))
        result = minisat(query, None, variable=None, value=True, verbose=False)
        print("  Query CNF:  {0}".format(query))
        print("  Result:     {0}   (Expected: {1})".format(result.success, expected_result))
        if result.success != expected_result:
            print("    FAILURE: unexpected result.")
            failed.append(t)
        if result.success:
            print("  Variable Assignment: {0}".format(result.varmap))
        t += 1
    print("-----------------------------------------------------")
    if not failed:
        print("Successfully passed {0} tests.".format(len(queries)))
    else:
        print("Passed {0} test(s).".format(len(queries) - len(failed)))
        print("The following tests failed: {0}".format(failure))
    print("DONE.")


#-------------------------------------------------------------------------------
# Command-line interface
#-------------------------------------------------------------------------------

def default(str):
    return str + ' [Default: %default]'


def readCommand(argv):
    """
    Processes the command used to run wumpus.py from the command line.
    """
    from optparse import OptionParser
    usageStr = """
    USAGE:     python wumpus.py <options>
    EXAMPLES:  (1) python wumpus.py
                   - starts simple manual Hunt The Wumpus game
               (2) python wumpus.py -k OR python wumpus.py --kb
                   - starts simple manual Hunt The Wumpus game with
                   knowledge base and interactive queries possible
    """
    parser = OptionParser(usageStr)

    parser.add_option('-k', '--kb', action='store_true', dest='kb', default=False,
                      help=default("Instantiate a queriable knowledge base"))
    parser.add_option('-y', '--hybrid', action='store_true', dest='hybrid', default=False,
                      help=default("Run hybrid wumpus agent" \
                                   + " (takes precedence over -k option)"))
    parser.add_option('-l', '--layout', dest='layout', default=None,
                      help=default("Load layout file"))

    parser.add_option('-t', '--test', action='store_true', dest='test_minisat',
                      default=False,
                      help=default("Test connection to command-line MiniSat"))

    options, otherjunk = parser.parse_args(argv)

    if len(otherjunk) != 0:
        raise Exception("Command line input not understood: " + str(otherjunk))

    return options


def run_command(options):
    if options.test_minisat:
        run_minisat_test()
        return
    if options.hybrid:
        if options.layout:
            s = world_scenario_hybrid_wumpus_agent_from_layout(options.layout)
        else:
            s = wscenario_4x4_HybridWumpusAgent()
    elif options.kb:
        if options.layout:
            s = world_scenario_manual_with_kb_from_layout(options.layout)
        else:
            s = wscenario_4x4_manual_HybridWumpusAgent()
    else:
        if options.layout:
            s = world_scenario_manual_from_layout(options.layout)
        else:
            s = wscenario_4x4_manual()
    s.run()


#-------------------------------------------------------------------------------

if __name__ == '__main__':
    """
    The main funciton called when wumpus_test.py is run from the command line:
    > python wumpus_test.py <options>
    """
    options = readCommand(sys.argv[1:])
    run_command(options)