From 2593d31346b6560b2acbd291649b8056df278bfa Mon Sep 17 00:00:00 2001 From: ydy0615 Date: Sat, 27 Dec 2025 20:54:40 +0800 Subject: [PATCH] feat(ctf): add backend policy module with pathfinding logic MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .DS_Store | Bin 6148 -> 6148 bytes CTF/.DS_Store | Bin 0 -> 6148 bytes CTF/backend/mypolicy | 214 +++++++++++++++++++++++++ CTF/backend/{mypolicy.py => server.py} | 136 +++++++++------- 4 files changed, 291 insertions(+), 59 deletions(-) create mode 100644 CTF/.DS_Store create mode 100644 CTF/backend/mypolicy rename CTF/backend/{mypolicy.py => server.py} (62%) diff --git a/.DS_Store b/.DS_Store index 274367a4e84555db5fb2310dd8cefbc0d95eca63..eafd3ebea2def0d8c68088f460141ab26853cdcb 100644 GIT binary patch delta 37 tcmZoMXfc@J&&WJ6U^gT4WFE#FoM|b=$w~S7IgFccF-oy*X6N|J4*=#k3^o7& delta 32 ocmZoMXfc@J&&V_}VE1GL5thmPj8`{*XB1$Y*dV-_o#QV*0JKmF?EnA( diff --git a/CTF/.DS_Store b/CTF/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..f206f7d12e226929333ed34ae79e802b9aead495 GIT binary patch literal 6148 zcmeHKJ5EC}5S)cbL`svA(pTUHR+O9|7l7nJ10n^Ye--EAXqo*KM2~by0h*Q8W3P8? zd5X7h0odwucmS3F=5$AV`7kx#cc0izMT|)28P9mf@G^YtcZW&#?*ZrDVTUK|@%od0 zc=Iip6p#W^Knh3!DR5y0s=&?{7d}_VNdYPF_Z9H(L!&$P!Z9&E9UP(sAkG*L<2-r^ zV)Fp87mkUH&@8FMq*}EYmUPBj<@LfbG3l^sKCEuG>QF3h=lLztVLef!6p#X^3f$&$ z>Ggj@Khyu8lC+WnQs7@HV6*jhz2cLqwoV@BwYJe8>7MgVcjG)L9HJZ(qa1VL<@i35 bGOziZ`@L{X3_9aMC+cUwb&*MdYb$UBY7P}r literal 0 HcmV?d00001 diff --git a/CTF/backend/mypolicy b/CTF/backend/mypolicy new file mode 100644 index 0000000..1865b2c --- /dev/null +++ b/CTF/backend/mypolicy @@ -0,0 +1,214 @@ +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 + + # Determine Obstacles: Avoid opponents if we are in enemy territory + 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]} ") + 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()) \ No newline at end of file diff --git a/CTF/backend/mypolicy.py b/CTF/backend/server.py similarity index 62% rename from CTF/backend/mypolicy.py rename to CTF/backend/server.py index d33c5e1..f7d2d96 100644 --- a/CTF/backend/mypolicy.py +++ b/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 -player_to_flag_assignments = {} +update_threshold = 10 +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_with_flags = world.list_players(mine=False, inPrison=False, hasFlag=None) + 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 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): - continue + 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 # 最短距离，初始为无穷大 - prev = [None] * n # 前驱节点，用于路径回溯 + 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: @@ -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_y = next_idx // (self.width + 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_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"]} - - player_to_flag_assignments = { - name: pos for name, pos in player_to_flag_assignments.items() + flags_list = [] + for flags in opponents: + 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: - # Randomly assign one of the available enemy flags - f = random.choice(enemy_flags) - player_to_flag_assignments[p["name"]] = (f["posX"], f["posY"]) + if p["name"] not in active_player_names: + continue + 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: + 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,17 +179,14 @@ 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: - dest = player_to_flag_assignments[p["name"]] + elif p["name"] in player_to_flag_assign: + 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]) return player_moves @@ -181,10 +201,8 @@ async def main(): print(f"Usage: python3 {sys.argv[0]} ") 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: @@ -192,4 +210,4 @@ async def main(): sys.exit(1) if __name__ == "__main__": - asyncio.run(main()) + asyncio.run(main()) \ No newline at end of file