modified: CTF/backend/mypolicy

Introduce a new `CTF/backend/mypolicy` Python module that implements a comprehensive game policy for the capture‑the‑flag bot. The file defines a global world model, thread‑safe state handling, and a `Map` class that builds a grid representation of the game map, generates traversable edges, and provides BFS‑based `guideance` and `length` methods for pathfinding and distance calculation. This policy enables the bot to make informed movement decisions based on safe zones, walls, allies, and enemies. Additionally, macOS `.DS_Store` placeholder files were added to the repository.

CTF/backend/mypolicy

@@ -0,0 +1,213 @@
+import asyncio
+import random
+from lib.game_engine import GameMap, run_game_server
+import threading
+import collections
+
+# 1. Initialize the global world model
+world = GameMap(show_gap_in_msec=10.0)
+lock = threading.Lock()
+last_updated_time = -1
+update_threshold = 10
+player_to_flag_assign = {}
+my_side_is_left = None
+class Map:
+    def __init__(self):
+        self.width  = world.width
+        self.height = world.height
+        self.grid = [0] * (self.width * self.height)
+        self.edge = [[] for _ in range(self.width * self.height)]
+        self.in_safe_zone = None
+
+    def convert_pos_to_index(self, x, y):
+        return y * self.width + x
+
+    def update(self,posx,posy):
+        self.width  = world.width
+        self.height = world.height
+        self.edge = [[] for _ in range(self.width * self.height)]
+        self.grid = [0] * (self.width * self.height)
+        walls = world.walls
+        for wall in walls:
+            x, y = wall
+            idx = self.convert_pos_to_index(x, y)
+            self.grid[idx] = 1
+        self.in_safe_zone = world.is_on_left((posx,posy)) == my_side_is_left
+        enemy_players = world.list_players(mine=False, inPrison=False, hasFlag=None)
+        ally_players = world.list_players(mine=True, inPrison=False, hasFlag=None)
+        my_pos = self.convert_pos_to_index(posx, posy)
+        for ally in ally_players:
+            x, y = ally["posX"], ally["posY"]
+            idx = self.convert_pos_to_index(x, y)
+            self.grid[idx] = 3 
+        for enemy in enemy_players:
+            x, y = enemy["posX"], enemy["posY"]
+            idx = self.convert_pos_to_index(x, y)
+            self.grid[idx] = 2 
+            for dx, dy in [(0, 1), (0, -1), (1, 0), (-1, 0)]:
+                nx, ny = x + dx, y + dy
+                self.grid[self.convert_pos_to_index(nx,ny)] = 2 
+        for y in range(self.height):
+            for x in range(self.width):
+                idx = self.convert_pos_to_index(x, y)
+                if self.in_safe_zone:
+                    if (self.grid[idx] in (1,)) and idx != my_pos:
+                        continue
+                else:
+                    if (self.grid[idx] in (1, 2)) and idx != my_pos:
+                        continue
+                for dx, dy in [(0, 1), (0, -1), (1, 0), (-1, 0)]:
+                    nx, ny = x + dx, y + dy
+                    if 0 <= nx < self.width and 0 <= ny < self.height:
+                        n_idx = self.convert_pos_to_index(nx, ny)
+                        if self.in_safe_zone:
+                            if self.grid[n_idx] not in (1, 2):
+                                self.edge[idx].append(n_idx)
+                        else:
+                            if self.grid[n_idx] not in (1,):
+                                self.edge[idx].append(n_idx)
+    def guideance(self, posx_start, posy_start, posx_end, posy_end):
+        self.update(posx_start,posy_start)
+        src_idx = self.convert_pos_to_index(posx_start, posy_start)
+        dst_idx = self.convert_pos_to_index(posx_end,   posy_end)
+        n = self.width * self.height
+        dist = [float('inf')] * n
+        prev = [None] * n
+        dist[src_idx] = 0
+        queue = collections.deque([src_idx])
+        while queue:
+            u = queue.popleft()
+            if u == dst_idx:
+                break
+            for v in self.edge[u]:
+                if dist[v] == float('inf'):
+                    dist[v] = dist[u] + 1
+                    prev[v] = u
+                    queue.append(v)
+
+        # ---- 若不可达，返回空字符串 ----
+        if dist[dst_idx] == float('inf'):
+            return ""
+        # ---- 重建路径（逆序） ----
+        path = []
+        cur = dst_idx
+        while cur is not None:
+            path.append(cur)
+            cur = prev[cur]
+        path.reverse()   # 现在是 [src, ..., dst]
+        if len(path) < 2:
+            return ""
+        # ---- 计算第一步坐标并返回方向 ----
+        next_idx = path[1]
+        next_x = next_idx % (self.width)
+        next_y = next_idx // (self.height)
+        return world.get_direction((posx_start, posy_start), (next_x, next_y))
+    def length(self, posx_start, posy_start, posx_end, posy_end):
+        self.update(posx_start,posy_start)
+        src_idx = self.convert_pos_to_index(posx_start, posy_start)
+        dst_idx = self.convert_pos_to_index(posx_end,   posy_end)
+        n = self.width * self.height
+        dist = [float('inf')] * n          # 最短距离，初始为无穷大
+        prev = [None] * n                  # 前驱节点，用于路径回溯
+        dist[src_idx] = 0
+        queue = collections.deque([src_idx])
+        while queue:
+            u = queue.popleft()
+            if u == dst_idx:
+                break
+            for v in self.edge[u]:
+                if dist[v] == float('inf'):
+                    dist[v] = dist[u] + 1
+                    prev[v] = u
+                    queue.append(v)
+
+        # ---- 若不可达，返回空字符串 ----
+        if dist[dst_idx] == float('inf'):
+            return -1
+        return dist[dst_idx]
+
+myMap = Map()
+def start_game(req):
+    global player_to_flag_assign,my_side_is_left
+    world.init(req)
+    print("Start Game!!")
+    player_to_flag_assign = {}
+    print(f"Game Started! Side: {'Left' if world.is_on_left(list(world.my_team_target)[0]) else 'Right'}")
+    my_side_is_left = world.is_on_left(list(world.my_team_target)[0])
+
+def plan_next_actions(req):
+    if not world.update(req):
+        return
+    global player_to_flag_assign,myMap,my_side_is_left
+
+    my_players = world.list_players(mine=True, inPrison=False, hasFlag=None)
+    opponents = world.list_players(mine=False, inPrison=False, hasFlag=None)
+    enemy_flags = world.list_flags(mine=False, canPickup=True)
+    my_flags = world.list_flags(mine=True, canPickup=True)
+    my_targets = list(world.list_targets(mine=True))
+    active_player_names = {p["name"] for p in my_players if not p["hasFlag"]}
+    flags_list = []
+    for flags in my_flags:
+        flags_list.append((flags["posX"],flags["posY"]))
+    player_to_flag_assign = {
+        name: pos for name, pos in player_to_flag_assign.items()
+        if name in active_player_names
+    }
+    if enemy_flags:
+        for p in my_players:
+            if p["name"] not in active_player_names:
+                continue
+            if p["name"] in player_to_flag_assign and player_to_flag_assign[p["name"]] in flags_list:
+                flags_list.remove(player_to_flag_assign[p["name"]])
+                continue
+            closest_flag = None
+            min_length = float('inf')
+            for f in flags_list:
+                temp = myMap.length(p["posX"],p["posY"],f[0],f[1])
+                if temp != -1 and temp < min_length:
+                    min_length = temp
+                    closest_flag = f
+            f = closest_flag
+            if f is not None:
+                player_to_flag_assign[p["name"]] = (f[0],f[1])
+                print(f[0], f[1])
+                flags_list.remove((f[0],f[1]))
+
+    # 3. Plan moves for each player
+    player_moves = {}
+
+    for p in my_players:
+        curr_pos = (p["posX"], p["posY"])
+
+        # Determine Target: Either the assigned flag or the home target
+        if p["hasFlag"]:
+            dest = my_targets[0]
+        elif p["name"] in player_to_flag_assign:
+            dest = player_to_flag_assign[p["name"]]
+        else:
+            continue
+
+        player_moves[p["name"]] = myMap.guideance(p["posX"],p["posY"],dest[0],dest[1])
+
+    return player_moves
+
+def game_over(req):
+    print("Game Over!")
+    world.show(force=True)
+
+async def main():
+    import sys
+    if len(sys.argv) != 2:
+        print(f"Usage: python3 {sys.argv[0]} <port>")
+        print(f"Example: python3 {sys.argv[0]} 8080")
+        sys.exit(1)
+    port = int(sys.argv[1])
+    print(f"AI backend running on port {port} ...")
+    try:
+        await run_game_server(port, start_game, plan_next_actions, game_over)
+    except Exception as e:
+        print(f"Server Stopped: {e}")
+        sys.exit(1)
+
+if __name__ == "__main__":
+    asyncio.run(main())

CTF/backend/server.py

@@ -4,13 +4,12 @@ from lib.game_engine import GameMap, run_game_server
 import threading
 import collections
 
-
 # 1. Initialize the global world model
 world = GameMap(show_gap_in_msec=10.0)
 lock = threading.Lock()
 last_updated_time = -1
-update_threshold = 100
+update_threshold = 10
-player_to_flag_assignments = {}
+player_to_flag_assign = {}
 my_side_is_left = None
 class Map:
     def __init__(self):
@@ -19,68 +18,63 @@ class Map:
         self.grid = [0] * (self.width * self.height)
         self.edge = [[] for _ in range(self.width * self.height)]
         self.in_safe_zone = None
+
     def convert_pos_to_index(self, x, y):
         return y * self.width + x
 
     def update(self,posx,posy):
+        self.width  = world.width
+        self.height = world.height
         self.edge = [[] for _ in range(self.width * self.height)]
         self.grid = [0] * (self.width * self.height)
         walls = world.walls
         for wall in walls:
-            x, y = wall["posX"], wall["posY"]
+            x, y = wall
             idx = self.convert_pos_to_index(x, y)
-            self.grid[idx] = 1  # Mark wall positions
+            self.grid[idx] = 1
-
         self.in_safe_zone = world.is_on_left((posx,posy)) == my_side_is_left
-
+        enemy_players = world.list_players(mine=False, inPrison=False, hasFlag=None)
-        enemy_players_with_flags = world.list_players(mine=False, inPrison=False, hasFlag=None)
         ally_players = world.list_players(mine=True, inPrison=False, hasFlag=None)
         my_pos = self.convert_pos_to_index(posx, posy)
-        for enemy in enemy_players_with_flags:
-            x, y = enemy["posX"], enemy["posY"]
-            idx = self.convert_pos_to_index(x, y)
-            self.grid[idx] = 2  # Mark enemy players with flags as obstacles
         for ally in ally_players:
             x, y = ally["posX"], ally["posY"]
             idx = self.convert_pos_to_index(x, y)
-            self.grid[idx] = 3  # Mark ally players as free space
+            self.grid[idx] = 3 
+        for enemy in enemy_players:
+            x, y = enemy["posX"], enemy["posY"]
+            idx = self.convert_pos_to_index(x, y)
+            self.grid[idx] = 2 
+            for dx, dy in [(0, 1), (0, -1), (1, 0), (-1, 0)]:
+                nx, ny = x + dx, y + dy
+                self.grid[self.convert_pos_to_index(nx,ny)] = 2 
         for y in range(self.height):
             for x in range(self.width):
                 idx = self.convert_pos_to_index(x, y)
-                if self.grid[idx] in (1, 2):
+                if self.in_safe_zone:
+                    if (self.grid[idx] in (1,)) and idx != my_pos:
+                        continue
+                else:
+                    if (self.grid[idx] in (1, 2)) and idx != my_pos:
                         continue
                 for dx, dy in [(0, 1), (0, -1), (1, 0), (-1, 0)]:
                     nx, ny = x + dx, y + dy
                     if 0 <= nx < self.width and 0 <= ny < self.height:
                         n_idx = self.convert_pos_to_index(nx, ny)
                         if self.in_safe_zone:
-                            if (self.grid[n_idx] not in (1, 2, 3)) or n_idx == my_pos:
+                            if self.grid[n_idx] not in (1, 2):
                                 self.edge[idx].append(n_idx)
                         else:
-                            if (self.grid[n_idx] not in (1, 3)) or n_idx == my_pos:
+                            if self.grid[n_idx] not in (1,):
                                 self.edge[idx].append(n_idx)
     def guideance(self, posx_start, posy_start, posx_end, posy_end):
-        """
-        Compute the shortest path from the start position to the end position
-        using BFS (equivalent to Dijkstra with unit edge weight) on the pre‑built
-        adjacency list `self.edge`. Returns the first move direction via
-        `world.get_direction`. If no path exists or inputs are invalid, returns an
-        empty string.
-        """
         self.update(posx_start,posy_start)
-        # ---- 参数合法性检查 ----
-        if not (0 <= posx_start < self.width and 0 <= posy_start < self.height):
-            return ""
-        if not (0 <= posx_end < self.width and 0 <= posy_end < self.height):
-            return ""
         src_idx = self.convert_pos_to_index(posx_start, posy_start)
         dst_idx = self.convert_pos_to_index(posx_end,   posy_end)
         n = self.width * self.height
-        dist = [float('inf')] * n          # 最短距离，初始为无穷大
+        dist = [float('inf')] * n
-        prev = [None] * n                  # 前驱节点，用于路径回溯
+        prev = [None] * n
         dist[src_idx] = 0
         queue = collections.deque([src_idx])
-
         while queue:
             u = queue.popleft()
             if u == dst_idx:
@@ -94,7 +88,6 @@ class Map:
         # ---- 若不可达，返回空字符串 ----
         if dist[dst_idx] == float('inf'):
             return ""
-
         # ---- 重建路径（逆序） ----
         path = []
         cur = dst_idx
@@ -102,50 +95,80 @@ class Map:
             path.append(cur)
             cur = prev[cur]
         path.reverse()   # 现在是 [src, ..., dst]
-
         if len(path) < 2:
             return ""
-
         # ---- 计算第一步坐标并返回方向 ----
         next_idx = path[1]
-        next_x = next_idx % (self.width + 1)
+        next_x = next_idx % (self.width)
-        next_y = next_idx // (self.width + 1)
+        next_y = next_idx // (self.height)
-
         return world.get_direction((posx_start, posy_start), (next_x, next_y))
+    def length(self, posx_start, posy_start, posx_end, posy_end):
+        self.update(posx_start,posy_start)
+        src_idx = self.convert_pos_to_index(posx_start, posy_start)
+        dst_idx = self.convert_pos_to_index(posx_end,   posy_end)
+        n = self.width * self.height
+        dist = [float('inf')] * n          # 最短距离，初始为无穷大
+        prev = [None] * n                  # 前驱节点，用于路径回溯
+        dist[src_idx] = 0
+        queue = collections.deque([src_idx])
+        while queue:
+            u = queue.popleft()
+            if u == dst_idx:
+                break
+            for v in self.edge[u]:
+                if dist[v] == float('inf'):
+                    dist[v] = dist[u] + 1
+                    prev[v] = u
+                    queue.append(v)
+
+        # ---- 若不可达，返回空字符串 ----
+        if dist[dst_idx] == float('inf'):
+            return -1
+        return dist[dst_idx]
 
 myMap = Map()
 def start_game(req):
-    global player_to_flag_assignments,my_side_is_left
+    global player_to_flag_assign,my_side_is_left
     world.init(req)
     print("Start Game!!")
-    player_to_flag_assignments = {}
+    player_to_flag_assign = {}
     print(f"Game Started! Side: {'Left' if world.is_on_left(list(world.my_team_target)[0]) else 'Right'}")
     my_side_is_left = world.is_on_left(list(world.my_team_target)[0])
 
 def plan_next_actions(req):
     if not world.update(req):
         return
-    global player_to_flag_assignments,myMap,my_side_is_left
+    global player_to_flag_assign,myMap,my_side_is_left
 
     my_players = world.list_players(mine=True, inPrison=False, hasFlag=None)
     opponents = world.list_players(mine=False, inPrison=False, hasFlag=None)
     enemy_flags = world.list_flags(mine=False, canPickup=True)
+
     my_targets = list(world.list_targets(mine=True))
 
-
     active_player_names = {p["name"] for p in my_players if not p["hasFlag"]}
-
+    flags_list = []
-    player_to_flag_assignments = {
+    for flags in opponents:
-        name: pos for name, pos in player_to_flag_assignments.items()
+        flags_list.append((flags["posX"],flags["posY"]))
+    player_to_flag_assign = {
+        name: pos for name, pos in player_to_flag_assign.items()
         if name in active_player_names
     }
-
     if enemy_flags:
         for p in my_players:
-            if not p["hasFlag"] and p["name"] not in player_to_flag_assignments:
+            if p["name"] not in active_player_names:
-                # Randomly assign one of the available enemy flags
+                continue
-                f = random.choice(enemy_flags)
+            if p["name"] in player_to_flag_assign and player_to_flag_assign[p["name"]] != None and player_to_flag_assign[p["name"]] in flags_list:
-                player_to_flag_assignments[p["name"]] = (f["posX"], f["posY"])
+                continue
+            closest_flag = None
+            min_length = float('inf')
+            for f in enemy_flags:
+                temp = myMap.length(p["posX"],p["posY"],f["posX"],f["posY"])
+                if temp != -1 and temp < min_length:
+                    min_length = temp
+                    closest_flag = f
+            f = closest_flag if closest_flag is not None else random.choice(enemy_flags)
+            player_to_flag_assign[p["name"]] = (f["posX"], f["posY"])
 
     # 3. Plan moves for each player
     player_moves = {}
@@ -156,16 +179,13 @@ def plan_next_actions(req):
         # Determine Target: Either the assigned flag or the home target
         if p["hasFlag"]:
             dest = my_targets[0]
-        elif p["name"] in player_to_flag_assignments:
+        elif p["name"] in player_to_flag_assign:
-            dest = player_to_flag_assignments[p["name"]]
+            dest = player_to_flag_assign[p["name"]]
         else:
             continue
 
         # Determine Obstacles: Avoid opponents if we are in enemy territory
         is_safe = world.is_on_left(curr_pos) == my_side_is_left
-        blockers = [] if is_safe else [(o["posX"], o["posY"]) for o in opponents]
-        # Calculate Path
-        # path = world.route_to(curr_pos, dest, extra_obstacles=blockers)
 
         player_moves[p["name"]] = myMap.guideance(p["posX"],p["posY"],dest[0],dest[1])
 
@@ -181,10 +201,8 @@ async def main():
         print(f"Usage: python3 {sys.argv[0]} <port>")
         print(f"Example: python3 {sys.argv[0]} 8080")
         sys.exit(1)
-
     port = int(sys.argv[1])
     print(f"AI backend running on port {port} ...")
-
     try:
         await run_game_server(port, start_game, plan_next_actions, game_over)
     except Exception as e: