Add squad based movement to formation

Formation as a hypothetical leader position that it
follows. Uses path reconstruction to get back on track
when unit has to go off the track because next position
is blocked. Still some TODOs are left.

Author: twitu

formation.py

@@ -6,30 +6,46 @@
 
 
 class Formation:
+    """
+    Formation represents a group of units that move together. The leader position
+    can be any unit and does not need to be centered on any particular unit. All
+    units move relative to the leader position which advances one step towards
+    destination on each turn.
+    """
 
-    def __init__(self, index, rel_pos, leader_index, units, is_valid_pos):
+    def __init__(self, index, rel_pos, leader_pos, units, is_valid_pos):
+        """
+        Formation object is unique to each unit.
+
+        Args:
+            index (int): index of unit in position list
+            rel_pos (List[Point]): relative position of other units to leader_pos
+            leader_pos (Point): leader position
+            units (List[Point]): unit positions
+            is_valid_pos (func): function to determine if postion is valid
+        """
         self.index = index
-        self.leader_index = leader_index
         self.units = units
-        self.leader = units[leader_index]
+        self.leader = leader_pos
         self.rel_pos = rel_pos
-        self.turn_queue = cycle(units)
         self.is_valid_pos = is_valid_pos
         self.path_finder = PathFinder(diagonal_cost(), diagonal_cost(), self.is_valid_pos)
-        self.leader_path = None
         self.moves = adjacent_octile()
         self.dest = None
 
-    def update_units(self, units):
+    def update_units(self, units, leader_pos):
         """
         Update formation unit positions. Decentralized formation requires
         each unit to observe its neighbours and perform updates.
 
+        Note: Should only be called when leader_path is not empty
+
         Args:
             units (List[Unit]): List of units
+            leader_pos (Point): Leader position
         """
         self.units = units
-        self.leader = units[self.leader_index]
+        self.leader = leader_pos
 
     def init_dest(self, dest):
         """
@@ -40,7 +56,7 @@ def init_dest(self, dest):
             dest (Point): destination point
         """
         self.dest = dest
-        self.leader_path = self.path_finder.find_path(self.moves, self.leader, dest)
+        return self.path_finder.find_path(self.moves, self.leader, dest)
 
     def find_path(self, src, dest):
         """
@@ -65,4 +81,7 @@ def predict_pos_from(self, pos):
         possible_pos = pos + self.rel_pos[self.index]
         store = self.path_finder.generic_a_star(self.moves, pos, possible_pos, 30)
         score_points = [(possible_pos.dist(pos), pos) for pos in store.keys()]
-        return min(score_points)
+        if score_points:
+            return min(score_points)[1]
+        else:
+            return None

game_map.py

@@ -9,12 +9,12 @@ def __init__(self, map_data):
     def is_valid_point(self, pos):
         """
         Checks static points and active points. Returns
-        true if pos is unoccupied in either
+        true if pos is unoccupied in both
 
         Args:
             pos (Point): position to check
         """
-        return self.static[pos.y][pos.x] and any(lambda x: x.cur_pos == pos for x in self.active)
+        return 0 <= pos.x < 100 and 0 <= pos.y < 100 and self.static[pos.y][pos.x] and all(lambda x: x.cur_pos != pos for x in self.active)
 
     def valid_next_pos(self, cur_unit):
         return [pos for pos in cur_unit.next_pos() if self.is_valid_point(pos)]

main.py

@@ -23,37 +23,27 @@ def is_valid_pos(cur_pos):
 if __name__ == "__main__":
     map_data = map_generator.generate_map(100, 100, seed=25, obstacle_density=0.35)
     game_map = Map(map_data)
-    start = Point(18, 65)
+    start = Point(2, 2)
     end = Point(25, 65)
-    cur_unit = SCOUT.copy(start)
-    cur_unit.set_game_state(game_map)
-    game_map.active.append(cur_unit)
-    cur_unit.find_path(end)
-    anim = animate_game_state(game_map)
+    rel_pos = [
+        Point(-1, 1),
+        Point(1, 1),
+        Point(0, 0),
+        Point(-1, -1),
+        Point(1, -1),
+    ]
+
+    units_pos = [start + pos for pos in rel_pos]
+    units = []
+    for i, unit_pos in enumerate(units_pos):
+        unit = FORMATION.copy(start + rel_pos[i])
+        unit.add_to_formation(Formation(i, rel_pos, start, units_pos, game_map.is_valid_point), i == 2)  # 2nd index unit is leader
+        unit.set_dest(end)
+        unit.game_map = game_map
+        units.append(unit)
+
+    game_map = Map(map_data)
+    game_map.active = units
+
+    anim = animate_game_state(game_map, frames=200)
     anim.save('run_game.gif', writer='imagemagick', fps=10)
-    # rel_pos = [
-    #     Point(-1, 1),
-    #     Point(1, 1),
-    #     Point(0, 0),
-    #     Point(-1, -1),
-    #     Point(1, -1),
-    # ]
-    # units_pos = [start + pos for pos in rel_pos]
-    # units = []
-    # for i, unit_pos in enumerate(units_pos):
-    #     unit = FORMATION.copy(start + rel_pos[i])
-    #     unit.add_to_formation(Formation(i, rel_pos, 2, units_pos, is_valid_pos), i == 2)  # 2nd index unit is leader
-    #     unit.set_dest(end)
-    #     units.append(unit)
-
-    # game_map = Map(map_data)
-    # game_map.active = units
-        
-    # for unit in cycle(units):
-    #     if unit.is_leader and unit.cur_pos == end:
-    #         break
-
-    #     units_pos = [unit.cur_pos for unit in units]
-    #     unit.formation.update_units(units_pos)
-    #     unit.cur_pos = unit.next_pos()
-    #     print(units)

mapworks/visualization.py

@@ -11,6 +11,7 @@
 def animate_game_state(game_map, frames=10, grid=True):
 
     fig = plt.figure()
+    fig.set_size_inches(12, 12, True)
     ax = plt.axes()
     active_scatter = None
     mock_scatter = None
@@ -43,7 +44,7 @@ def update_plot(i):
         active_scatter.set_offsets(units_pos)
         return active_scatter,
 
-    return animation.FuncAnimation(fig, update_plot, frames=frames, blit=False)
+    return animation.FuncAnimation(fig, update_plot, frames=frames, blit=True)
 
 
 def view_game_state(game_map, view_mock=False, grid=True):
@@ -57,6 +58,7 @@ def view_game_state(game_map, view_mock=False, grid=True):
     Args:
         game_map (Map object)
         view_mock (Boolean): mock objects are shown if true
+        grid (Boolean): True displays grid
     """
     # show active units
     unit_points = [game_unit.cur_pos for game_unit in game_map.active]
@@ -88,7 +90,6 @@ def view_game_state(game_map, view_mock=False, grid=True):
     plt.connect('button_press_event', mouse_move)
     plt.grid(grid)
     plt.show()
-    return
 
 
 def view_map(map_data, stores=None, paths=None, points=None, grid=True):

path_finding.py

@@ -203,6 +203,10 @@ def best_potential_step(self, game_map, cur_unit):
         If path and range is given, finds the best potential score point
         within the range of the next step prescribed by path.
 
+        Note: function does not compare does not calculate score of current
+        position. This may cause thrashing as unit might try to move to points
+        that are not an improvement on the current position. TODO
+
         Args:
             cur_pos Point: current unit
             path List(Point): prescribed path from source to destination
@@ -216,24 +220,24 @@ def best_potential_step(self, game_map, cur_unit):
         best_state = min(potential_values)
         return best_state[1]
 
-    def reconstruct_path(self, cur_pos, path):
+    def reconstruct_path(self, moves, cur_pos, path):
         """
         TODO
         Reconstructs a new path from current position,
-        that joins the given path at some point
+        that joins the given path at some point.
+
+        Note: consider intersecting intersecting further along the
+        path based on point sensitivity TODO
 
         Args:
+            moves: set of moves possible on each turn
             cur_pos: current position of unit
             path: previous path from which unit has deviated
 
         Return:
             New path
         """
-        # check if path is None and return None
-        # Find position of intersection with path,
-        # simple case is head of path
-        # complex case can be to analyze sensitivity of path points
-        # and set intersection at most sensitive point
-        # calculate path from cur_pos to intersection
-        # return concatenated path
-        return None
+        dest = path.popleft()
+        correction = self.find_path(moves, cur_pos, dest)
+        path.extendleft(correction.reverse())
+        return path

unit.py

@@ -1,5 +1,6 @@
 from collections import deque
 
+from mock_object import PathObject
 from path_finding import PathFinder
 from movement_cost import diagonal_cost, linear_cost
 from moves import adjacent_linear, bc19_9_radius, adjacent_octile
@@ -84,46 +85,54 @@ def __init__(self, *args):
         self.leader_path = None
         self.path = None
         self.game_map = None
+        self.at_objective = False # Temporary measure should be replace by concept of unit state
 
     def copy(self, cur_pos):
         if not cur_pos:
             cur_pos = self.cur_pos
-        return FormationUnit(cur_pos, self.poten_func, self.next_moves, self.move_cost_func, self.sight_range)
+        return FormationUnit(cur_pos, self.name, self.poten_func, self.next_moves, self.move_cost_func, self.sight_range)
 
     def add_to_formation(self, formation, is_leader):
         self.is_leader = is_leader
         self.formation = formation
 
     def set_dest(self, dest):
-        self.formation.init_dest(dest)
-        if self.is_leader:
-            self.path = deque(self.formation.leader_path)
-        else:
-            self.leader_path = deque(self.formation.leader_path)
+        self.leader_path = deque(self.formation.init_dest(dest))
 
     def find_path(self):
         if len(self.leader_path) > FormationUnit.MAX_PREDICT:
             future_leader_pos = self.leader_path[FormationUnit.MAX_PREDICT]
         else:
             future_leader_pos = self.leader_path[-1]
 
-        _, short_dest = self.formation.predict_pos_from(future_leader_pos)
-        self.path = deque(self.formation.find_path(self.cur_pos, short_dest))
+        short_dest = self.formation.predict_pos_from(future_leader_pos)
+        if not short_dest:
+            self.path = deque([self.cur_pos])
+        else:
+            self.path = deque(self.formation.find_path(self.cur_pos, short_dest))
 
     def update_pos(self):
+        if self.at_objective:
+            return self.cur_pos
+
         if not self.path:
             self.find_path()
 
+        self.update_formation()
         self.leader_path.popleft()
+        if not self.leader_path:  # reached objective when leader path is empty
+            self.at_objective = True
         next_pos = self.path.popleft()
-        if self.game_map.is_valid_pos(next_pos):
+        if self.game_map.is_valid_point(next_pos):
             self.cur_pos = next_pos
         else:
-            self.cur_pos = next_pos
+            vis_path = [pos for pos in self.path if self.can_see_point(pos)]
+            self.game_map.mock = [PathObject(vis_path)]
+            self.cur_pos = self.path_finder.best_potential_step(self.game_map, self)
 
     def update_formation(self):
         leader_pos = self.leader_path[0]
-        units = self.game_map.active
+        units = self.game_map.active  # temporary measure as all active units are part of formation
         self.formation.update_units(units, leader_pos)
 
     def __str__(self):