coupled roborally game logic with web server

app.py

@@ -2,29 +2,39 @@ from flask import Flask, render_template, request, session, make_response
 import random
 import socket
 import time
+import roborally
 
 app = Flask(__name__)
 app.secret_key = b'RoboRallyRolling'
 
-random.seed(0)
 
-moves = ['forward', 'forward x2', 'forward x3', 'backward', 'turn left', 'turn right', 'turn around']
+probabilities = [0.21428571428571427, 0.14285714285714285, 0.07142857142857142, 0.07142857142857142, 0.21428571428571427, 0.21428571428571427, 0.07142857142857142]
+
+deck = roborally.CardDeck()
 
 class Game:
     def __init__(self):
         self.action_stack = {}
         self.processing_done = False    # indicates whether all commands in the current round have been processed
 
-        self.comm_socket = socket.socket()  # socket for communicating with the program controlling the robots
+        self.board = roborally.Board()
 
+        self.comm_socket = socket.socket()  # socket for communicating with the program controlling the robots
+        try:
+            self.comm_socket.connect(('192.168.1.222', 1337))
+        except socket.error:
+            print("could not connect to robot control socket!")
 
     def ready(self):
         # have all players chosen an action?
         return len(self.action_stack.keys()) == Player.player_counter
 
     def register_actions(self, player_id, actions):
-        if not player_id in self.action_stack.keys():
+        # store the selected actions for the given player
+        if not player_id in self.action_stack.keys():   # make sure that the actions have not been registered before
+            print("registered actions: ", [str(a) for a in actions])
             self.action_stack[player_id] = actions
+            deck.return_cards(actions)  # put cards back into the deck
             self.processing_done = False
             return True
         else:
@@ -32,15 +42,34 @@ class Game:
             return False
 
     def process_actions(self):
-        # send commands to the robots in the order of priority
+        l = []
-        for i in range(5):
+        for p in self.action_stack.keys():
-            current_actions = []
+            l.append(zip([p] * len(self.action_stack[p]), self.action_stack[p]))
-            for p in self.action_stack.keys():
+        chosen_cards = list(zip(*l))
-                current_actions += [(p, self.action_stack[p][i])]
-            print("actions in step {}: {}".format(i, ["robot {} action {}".format(c[0], c[1]) for c in current_actions]))
-            time.sleep(1)
 
-            #self.comm_socket.send()
+        # apply the chosen commands to the board which generates a list of movement commands to send to the control program
+        cmd_list = self.board.apply_actions(chosen_cards)
+
+        if False:
+            # send movements to the program
+            for c in current_actions:
+                if c[0] == 0:
+                    print("{}, {}\n".format(c[1].action, 11))
+                    self.comm_socket.sendall("{}, {}\n".format(c[1].action, 11).encode())
+                elif c[0] == 1:
+                    print("{}, {}\n".format(c[1].action, 14))
+                    self.comm_socket.sendall("{}, {}\n".format(c[1].action, 14).encode())
+                data = self.comm_socket.recv(32)
+
+                if data == b'OK\n':
+                    print("an error occured while processing the commands")
+                    self.processing_done = True
+                    self.action_stack = {}
+                    return
+
+        time.sleep(0.5)
+
+        #self.comm_socket.send()
 
 
         # clear the action stack for the next round
@@ -48,21 +77,6 @@ class Game:
 
         self.processing_done = True
 
-class Card:
-    card_counter = 0
-    def __init__(self):
-        self.number = Card.card_counter
-        Card.card_counter += 1
-        self.action = random.choice(moves)
-        self.priority = random.randint(0, 100)
-
-    def __str__(self):
-        return "Card No. " + str(self.number) + " " + self.action + " " + str(self.priority)
-
-card_deck = {}
-for i in range(0,20):
-    card_deck[i] = Card()
-
 class Player:
     MAX_PLAYERS = 3
     player_counter = 0
@@ -73,20 +87,17 @@ class Player:
 
             self.max_cards = 9
 
-            self.player_hand = random.sample(list(card_deck.values()), self.max_cards)
+            self.player_hand = deck.draw_cards(self.max_cards)
             print("current hand: ", [str(c) for c in self.player_hand])
 
             self.action_count = 5
-
-
             self.action_chosen = False
 
         else:
             print("max players reached!")
 
-
     def draw_new_cards(self):
-        self.player_hand += random.sample(list(card_deck.values()), self.max_cards - len(self.player_hand))
+        self.player_hand += deck.draw_cards(self.max_cards - len(self.player_hand))
 
 players = {}
 game = Game()
@@ -148,8 +159,10 @@ def hello_world():
             i1 = int(request.form.get('drag'))  # number of first card
             i2 = int(request.form.get('drop'))  # number of second card
 
-            card1 = card_deck[i1]   # get card by number
+            card1 = deck.deck[i1]   # get card by number
-            card2 = card_deck[i2]
+            card2 = deck.deck[i2]
+
+            print("swapping {} and {}".format(card1, card2))
 
             j1 = player_hand.index(card1)   # get index of card in the hand
             j2 = player_hand.index(card2)

roborally.py

@@ -0,0 +1,438 @@
+import random
+import sys
+random.seed(0)
+
+class Card:
+    possible_moves = ['forward', 'forward x2', 'forward x3', 'backward', 'turn left', 'turn right', 'turn around']
+    card_counter = 0
+
+    def __init__(self):
+        self.number = Card.card_counter
+        Card.card_counter += 1
+        self.action = random.choice(Card.possible_moves)
+        self.priority = random.randint(0, 100)
+
+    def __str__(self):
+        return "Card No. " + str(self.number) + " " + self.action + " " + str(self.priority)
+
+    def __repr__(self):
+        return self.action + " (" + str(self.priority) + ")"
+
+
+class CardDeck:
+    def __init__(self, n=84):
+        self.deck = {}
+        # generate cards
+        for i in range(0, n):
+            self.deck[i] = Card()
+        self.dealt = set()
+        self.discard_pile = set()
+
+    def draw_cards(self, n=1):
+        available = set(self.deck.keys()).difference(self.dealt)
+        # print("{} cards are available".format(len(available)))
+
+        if len(available) < n:
+            drawn = list(available)  # give out remaining cards
+            # print("drawing remaining {} cards".format(len(drawn)))
+            self.dealt = self.dealt.union(drawn)
+
+            # put the cards from the discard pile back into the game
+            self.dealt = self.dealt - self.discard_pile
+            self.discard_pile = set()  # reset the discard pile
+
+            # draw rest of cards
+            available = set(self.deck.keys()).difference(self.dealt)
+            # print("drawing another {} cards".format(n - len(drawn)))
+            drawn += random.sample(available, n - len(drawn))
+        else:
+            drawn = random.sample(available, n)
+        # print("cards drawn: {}".format(drawn))
+
+        self.dealt = self.dealt.union(drawn)
+
+        return [self.deck[i] for i in drawn]
+
+    def return_cards(self, cards):
+        self.discard_pile = self.discard_pile.union(set([c.number for c in cards]))
+        pass
+
+
+class Robot:
+    # dictionary mapping the current orientation and a turn command to the resulting orientation
+    resulting_orientation = {
+        '^': {'turn left': '<', 'turn right': '>', 'turn around': 'v'},
+        '>': {'turn left': '^', 'turn right': 'v', 'turn around': '<'},
+        'v': {'turn left': '>', 'turn right': '<', 'turn around': '^'},
+        '<': {'turn left': 'v', 'turn right': '^', 'turn around': '>'},
+    }
+
+    # dictionary mapping the current orientation and the target orientation to the necessary turn command
+    necessary_turn = {
+            '^': {'>': 'turn right', 'v': 'turn around', '<': 'turn left'},
+            '>': {'v': 'turn right', '<': 'turn around', '^': 'turn left'},
+            'v': {'<': 'turn right', '^': 'turn around', '>': 'turn left'},
+            '<': {'^': 'turn right', '>': 'turn around', 'v': 'turn left'},
+    }
+
+    # dictionary mapping an orientation to its opposite
+    opposites = {'^': 'v', '>': '<', 'v': '^', '<': '>'}
+
+    def __init__(self, x, y, orientation, id, board):
+        self.x = x
+        self.y = y
+        self.orientation = orientation
+        self.id = id
+
+        self.board = board
+
+        # mark the tile on the board as occupied
+        self.board[(x,y)].occupant = self
+
+
+    def get_accessed_tiles(self, count, forward=True):
+        # create a list of all tiles the robot would enter if it drives <count> steps forward
+        tiles = []
+        current_tile = self.board[(self.x, self.y)]
+        for i in range(1, count + 1):
+            if forward:
+                current_tile = self.board.get(current_tile.get_neighbor_coordinates(self.orientation))
+            else:
+                current_tile = self.board.get(current_tile.get_neighbor_coordinates(Robot.opposites[self.orientation]))
+            if current_tile is None:
+                return tiles
+            else:
+                tiles.append(current_tile)
+        return tiles
+
+    def is_pushable(self, direction):
+        # check if the robot can be pushed in the given direction
+        # this is the case if there is a non-blocking tile next to the robot or if there is another robot that is pushable
+        robot_tile = self.board[(self.x, self.y)]
+        neighbor_tile = self.board.get(robot_tile.get_neighbor_coordinates(direction))
+        if neighbor_tile is None:   # neighbor tile could not be found -> robot would be pushed out of the board
+            return False
+        else:
+            if neighbor_tile.is_empty():
+                return True
+            elif neighbor_tile.modifier == '#':   # if there's a wall on the neighbor tile the robot cannot be pushed there
+                return False
+            else:
+                # this means there's another robot on the neighbor tile -> check if it can be pushed away
+                return neighbor_tile.occupant.is_pushable(direction)
+
+    def has_opposite_orientation(self, direction):
+        opposites = [('^', 'v'), ('>', '<'), ('v', '^'), ('<', '>')]
+        return (self.orientation, direction) in opposites
+
+    def get_turn_direction(self, target_orienation):
+        return Robot.necessary_turn[self.orientation][target_orienation]
+
+    def get_opposite_orientation(self):
+        return Robot.opposites[self.orientation]
+
+    def turn(self, type):
+        # change the orientation of the robot
+        self.orientation = Robot.resulting_orientation[self.orientation][type]
+
+        return "{}, {}".format(self.id, type)
+
+    def move(self, type):
+        # move the robot forward or backward
+        # this involves
+        tile = self.board[(self.x, self.y)]
+        if type == 'forward':
+            target_tile = self.board[tile.get_neighbor_coordinates(self.orientation)]
+
+            if target_tile.occupant is not None:
+                print("error: target tile is not empty")
+                sys.exit(1)
+
+            tile.occupant = None    # delete the robot from the current tile
+            target_tile.occupant = self     # place the robot in the next tile
+            self.x = target_tile.x
+            self.y = target_tile.y
+
+            # return the move for sending to the controller
+            return "{}, forward".format(self.id)
+        elif type == 'backward':
+            opposite_orientation = self.get_opposite_orientation()
+            target_tile = self.board[tile.get_neighbor_coordinates(opposite_orientation)]
+
+            if target_tile.occupant is not None:
+                print("error: target tile is not empty")
+                sys.exit(1)
+
+            tile.occupant = None  # delete the robot from the current tile
+            target_tile.occupant = self  # place the robot in the next tile
+            self.x = target_tile.x
+            self.y = target_tile.y
+
+            # return the move for sending to the controller
+            return "{}, backward".format(self.id)
+        else:
+            print("error: invalid move")
+            sys.exit(1)
+
+    def nop(self):
+        # do nothing command
+        return "{}, nop".format(self.id)
+
+    def __str__(self):
+        return str(self.id)
+
+
+class Tile:
+    # possible modifiers:
+    # conveyors: <, >, ^, v
+    # repair station: r
+    # flag: f<number>
+    def __init__(self, x, y, modifier=None):
+        self.modifier = modifier
+        self.occupant = None
+        self.x = x
+        self.y = y
+
+    def get_neighbor_coordinates(self, direction):
+        # get the coordinates of the neighboring tile in the given direction
+        if direction == '^':
+            return (self.x, self.y - 1)
+        elif direction == '>':
+            return (self.x + 1, self.y)
+        elif direction == 'v':
+            return (self.x, self.y + 1)
+        elif direction == '<':
+            return (self.x - 1, self.y)
+        else:
+            print("error: unknown direction")
+            sys.exit(1)
+
+    def is_empty(self):
+        # check if the tile is non-empty and does not contain a wall
+        return self.occupant is None and self.modifier != '#'
+
+    def __str__(self):
+        if self.is_empty():
+            if self.modifier is None:
+                return ' '
+            else:
+                return self.modifier
+        else:
+            if self.occupant is None:
+                return self.modifier
+            else:
+                return str(self.occupant)
+
+    def __repr__(self):
+        return "({}, {})  occ: {} mod: {}".format(self.x, self.y, self.occupant, self.modifier)
+
+
+class Board:
+    x_dims = 12  # number of tiles in x direction
+    y_dims = 6  # number of tiles in y direction
+
+    def __init__(self):
+        self.board = {}
+        for x in range(Board.x_dims + 2):
+            for y in range(Board.y_dims + 2):
+                if (x == 0) or (x == Board.x_dims + 1) or (y == 0) or (y == Board.y_dims + 1):
+                    # place walls around the board
+                    self.board[(x, y)] = Tile(x, y, '#')
+                elif x == 1 and (y >= 1) and (y < 4):
+                    self.board[(x, y)] = Tile(x, y, 'v')
+                elif y == 4:
+                    self.board[(x, y)] = Tile(x, y, '>')
+                else:
+                    self.board[(x,y)] = Tile(x,y)
+
+        self.robots = {}
+        self.robots[0] = Robot(3, 1, '>', 0, self.board)
+        self.robots[1] = Robot(2, 1, 'v', 1, self.board)
+
+    def handle_push(self, direction, pushed_robot, forward=True, pushing_robot=None):
+        cmd_list = []
+        # push robot out of the way
+        if pushed_robot.orientation == direction:
+            if forward:
+                # the pushed robot can just drive forward
+                cmd_list += self.handle_single_action('forward', pushed_robot)
+            else:
+                # the pushed robot can just drive backward
+                cmd_list += self.handle_single_action('backward', pushed_robot)
+        elif pushed_robot.has_opposite_orientation(direction):
+            if forward:
+                # the pushed robot can drive backward
+                cmd_list += self.handle_single_action('backward', pushed_robot)
+            else:
+                # the pushed robot drives forward
+                cmd_list += self.handle_single_action('forward', pushed_robot)
+        else:
+            # we first have to turn the pushed robot s.t. it faces in the same orientation as the
+            # pushing robot
+            turn_direction = pushed_robot.get_turn_direction(direction)
+            cmd_list += self.handle_single_action(turn_direction, pushed_robot)
+
+            if forward:
+                # then the pushed robot drives one step forward
+                cmd_list += self.handle_single_action('forward', pushed_robot)
+            else:
+                # if its pushed backward it instead drives on step backward
+                cmd_list += self.handle_single_action('backward', pushed_robot)
+
+            # afterwards we turn the robot back to the original orientation
+            if turn_direction == 'turn left':
+                turn_back_direction = 'turn right'
+            elif turn_direction == 'turn right':
+                turn_back_direction = 'turn left'
+            else:
+                print("error: invalid turn direction")
+                sys.exit(1)
+            cmd_list += self.handle_single_action(turn_back_direction, pushed_robot)
+
+        if pushing_robot is not None:
+            # now the tile should be empty so the pushing robot can move into the tile
+            if forward:
+                cmd_list.append(pushing_robot.move('forward'))
+            else:
+                cmd_list.append(pushing_robot.move('backward'))
+        return cmd_list
+
+    def handle_single_action(self, action, robot):
+        cmd_list = []
+        if 'forward' in action:  # driving forward
+            if "x2" in action:
+                move_count = 2
+            elif "x3" in action:
+                move_count = 3
+            else:
+                move_count = 1
+            accessed_tiles = robot.get_accessed_tiles(move_count)
+
+            for tile in accessed_tiles:
+                if tile is None:
+                    # this case should not happen
+                    print("error: unknown state occured")
+                    sys.exit(1)
+                elif tile.is_empty():
+                    # if the tile is empty we can just move there
+                    cmd_list.append(robot.move('forward'))
+                elif tile.modifier == '#':    # robot hits a wall -> stop the robot
+                    cmd_list.append(robot.nop())
+                    return cmd_list
+                elif any([(tile.x, tile.y) == (r.x, r.y) for r in
+                          self.robots.values()]):  # robots hits a tile occupied by another robot
+                    pushed_robot = next(filter(lambda r: (tile.x, tile.y) == (r.x, r.y), self.robots.values()))
+                    if pushed_robot.is_pushable(robot.orientation):  # check if robot is pushable in the given direction
+                        cmd_list += self.handle_push(direction=robot.orientation, pushed_robot=pushed_robot, forward=True, pushing_robot=robot)
+                    else:
+                        cmd_list.append(robot.nop())
+                        return cmd_list
+                else:
+                    # this case should not happen
+                    print("error: unknown state occured")
+                    sys.exit(1)
+        elif action == 'backward':
+            # basically do the same as with forward
+            accessed_tiles = robot.get_accessed_tiles(1, forward=False)
+
+            for tile in accessed_tiles:
+                if tile is None:
+                    # this case should not happen
+                    print("error: unknown state occured")
+                    sys.exit(1)
+                elif tile.is_empty():
+                    # if the tile is empty we can just move there
+                    cmd_list.append(robot.move('backward'))
+                elif tile.modifier == '#':  # robot hits a wall -> stop the robot
+                    cmd_list.append(robot.nop())
+                    return cmd_list
+                elif any([(tile.x, tile.y) == (r.x, r.y) for r in
+                          self.robots.values()]):  # robots hits a tile occupied by another robot
+                    pushed_robot = next(filter(lambda r: (tile.x, tile.y) == (r.x, r.y), self.robots.values()))
+                    if pushed_robot.is_pushable(Robot.opposites[robot.orientation]):  # check if robot is pushable in the given direction
+                        cmd_list += self.handle_push(direction=robot.orientation, pushed_robot=pushed_robot, forward=False, pushing_robot=robot)
+                    else:
+                        cmd_list.append(robot.nop())
+                        return cmd_list
+                else:
+                    # this case should not happen
+                    print("error: unknown state occured")
+                    sys.exit(1)
+        else:   # this means we have a turn action
+            cmd_list.append(robot.turn(action))
+
+        return cmd_list
+
+    def handle_board_element(self, robot):
+        cmd_list = []
+        tile = self.board[(robot.x, robot.y)]
+        if tile.modifier in ['^', '>', 'v', '<']:
+            # board element pushes the robot to next tile
+            if robot.is_pushable(tile.modifier):
+                cmd_list += self.handle_push(direction=tile.modifier, pushed_robot=robot, forward=True)
+            else:
+                cmd_list.append(robot.nop())
+        return cmd_list
+
+    def apply_actions(self, cards):
+        cmd_list = []
+        # apply the actions to the board and generate a list of movement commands
+
+        for i, phase in enumerate(cards): # process register phases
+            print("processing phase {}".format(i+1))
+            # sort actions by priority
+            sorted_actions = sorted(phase, key=lambda a: a[1].priority)
+
+            for a in sorted_actions:
+                robot_id = a[0]
+                robot = self.robots[robot_id]
+                action = a[1].action
+
+                print("robot {} action {}".format(robot, action))
+
+                cmd_list += self.handle_single_action(action, robot)
+
+            # apply the actions caused by board elements at the end of the phase
+            for robot_id in self.robots:
+                robot = self.robots[robot_id]
+                cmd_list += self.handle_board_element(robot)
+
+            print(cmd_list)
+            print(self)
+            pass
+        return cmd_list
+
+    def __str__(self):
+        #output = '#' * (Board.x_dims + 2) + '\n'
+        output = ''
+        for y in range(Board.y_dims+2):
+            for x in range(Board.x_dims+2):
+                if any((r.x, r.y) == (x,y) for r in self.robots.values()):
+                    r = list(filter(lambda r: (r.x,r.y) == (x,y), self.robots.values()))[0]
+                    output += str(r.id)
+                else:
+                    output += str(self.board[(x, y)])
+            output += '\n'
+        #output += '#' * (Board.x_dims + 2)
+        return output
+
+if __name__ == "__main__":
+    n = 5
+
+    deck = CardDeck()
+
+    player_1_cards = random.sample(list(filter(lambda c: 'backward' in c.action, deck.deck.values())), n)
+    player_2_cards = random.sample(list(filter(lambda c: 'turn around' in c.action, deck.deck.values())), n)
+    #player_1_cards = deck.draw_cards(40)
+    #player_2_cards = deck.draw_cards(40)
+
+    cards_1 = [(0, c) for c in player_1_cards]
+    cards_2 = [(1, c) for c in player_2_cards]
+
+    chosen_cards = list(zip(cards_1, cards_2))
+
+    b = Board()
+    print(b)
+
+    b.apply_actions(chosen_cards)
+