diff --git a/CTF/backend/mypolicy.py b/CTF/backend/mypolicy.py deleted file mode 100644 index 7439115..0000000 --- a/CTF/backend/mypolicy.py +++ /dev/null @@ -1,265 +0,0 @@ -import asyncio -from lib.game_engine import GameMap, run_game_server -import threading -import collections - -world = GameMap(show_gap_in_msec=1.0) -lock = threading.Lock() -last_updated_time = -1 -update_threshold = 1 -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): - global my_side_is_left - 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 - if my_side_is_left: - self.in_safe_zone = world.is_on_left((posx+1,posy)) == my_side_is_left - else: - self.in_safe_zone = world.is_on_left((posx-1,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) - if my_side_is_left: - if not world.is_on_left((x+1,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 - else: - if world.is_on_left((x-1,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, ): - self.edge[idx].append(n_idx) - else: - if self.grid[n_idx] not in (1, 2): - 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] - def closest(self, posx_start, posy_start, positions): - closest_pos = None - min_length = float('inf') - for pos in positions: - temp = self.length(posx_start, posy_start, pos[0], pos[1]) - if temp != -1 and temp < min_length: - min_length = temp - closest_pos = pos - return closest_pos - def closest_in_range(self, posx_start, posy_start, positions): - closest_pos = None - min_length = float('inf') - for pos in positions: - if my_side_is_left: - if pos[0] >= self.width // 2: - continue - else: - if pos[0] < self.width // 2: - continue - temp = self.length(posx_start, posy_start, pos[0], pos[1]) - if temp != -1 and temp < min_length: - min_length = temp - closest_pos = pos - return closest_pos,min_length -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) - enemy_players_flags = world.list_players(mine=False, inPrison=False, hasFlag=True) - enemy_players = world.list_players(mine=False, inPrison=False, hasFlag=False) - 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 = [] - regard_list = [] - pretend_list = [] - protect_list = [] - targets_list = [] - for flags in enemy_flags: - flags_list.append((flags["posX"],flags["posY"])) - for p in enemy_players_flags: - regard_list.append((p["posX"],p["posY"])) - for p in enemy_players: - pretend_list.append((p["posX"],p["posY"])) - for flags in my_flags: - protect_list.append((flags["posX"],flags["posY"])) - for t in my_targets: - targets_list.append((t[0],t[1])) - 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 - f = myMap.closest(p["posX"],p["posY"],flags_list) - if f is not None: - player_to_flag_assign[p["name"]] = (f[0],f[1]) - flags_list.remove((f[0],f[1])) - - player_moves = {} - - for p in my_players: - if p["hasFlag"]: - dest,temp = myMap.closest_in_range(p["posX"],p["posY"],my_targets) - f,lentime = myMap.closest_in_range(p["posX"],p["posY"],regard_list) - if (f is not None) and (lentime<=5): - dest = (f[0],f[1]) - elif p["name"] in player_to_flag_assign: - dest = player_to_flag_assign[p["name"]] - f,lentime = myMap.closest_in_range(p["posX"],p["posY"],regard_list) - if f is not None and lentime<=10: - dest = (f[0],f[1]) - else : - f,lentime = myMap.closest_in_range(p["posX"],p["posY"],pretend_list) - if f is not None and lentime<8: - dest = (f[0],f[1]) - else: - f,temp = myMap.closest_in_range(p["posX"],p["posY"],regard_list) - if f is not None: - dest = (f[0],f[1]) - else: - f = myMap.closest(p["posX"],p["posY"],pretend_list) - if f is not None: - dest = (f[0],f[1]) - else: - f= myMap.closest(p["posX"],p["posY"],protect_list) - if f is not None: - dest = (f[0],f[1]) - 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]} ") - 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/pick_flag_ai.py b/CTF/backend/pick_flag_ai.py index 1eb70d6..8f5327e 100644 --- a/CTF/backend/pick_flag_ai.py +++ b/CTF/backend/pick_flag_ai.py @@ -1,103 +1,266 @@ 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=1000.0) +world = GameMap(show_gap_in_msec=1.0) lock = threading.Lock() last_updated_time = -1 -update_threshold = 100 -player_to_flag_assignments = {} +update_threshold = 1 +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): + global my_side_is_left + 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 + if my_side_is_left: + self.in_safe_zone = world.is_on_left((posx+1,posy)) == my_side_is_left + else: + self.in_safe_zone = world.is_on_left((posx-1,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) + if my_side_is_left: + if not world.is_on_left((x+1,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 + else: + if world.is_on_left((x-1,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, ): + self.edge[idx].append(n_idx) + else: + if self.grid[n_idx] not in (1, 2): + 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] + def closest(self, posx_start, posy_start, positions): + closest_pos = None + min_length = float('inf') + for pos in positions: + temp = self.length(posx_start, posy_start, pos[0], pos[1]) + if temp != -1 and temp < min_length: + min_length = temp + closest_pos = pos + return closest_pos + def closest_in_range(self, posx_start, posy_start, positions): + closest_pos = None + min_length = float('inf') + for pos in positions: + if my_side_is_left: + if pos[0] >= self.width // 2: + continue + else: + if pos[0] < self.width // 2: + continue + temp = self.length(posx_start, posy_start, pos[0], pos[1]) + if temp != -1 and temp < min_length: + min_length = temp + closest_pos = pos + return closest_pos,min_length +myMap = Map() def start_game(req): - global player_to_flag_assignments + 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 - - # Render the map - # world.show(do_not_clear=False) + 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_players_flags = world.list_players(mine=False, inPrison=False, hasFlag=True) + enemy_players = world.list_players(mine=False, inPrison=False, hasFlag=False) 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)) - - # 2. Logic: Assign flags to players without flags - # We maintain the original logic of matching players to specific flag coordinates active_player_names = {p["name"] for p in my_players if not p["hasFlag"]} - - # Cleanup assignments for players captured or flags already taken - player_to_flag_assignments = { - name: pos for name, pos in player_to_flag_assignments.items() + flags_list = [] + regard_list = [] + pretend_list = [] + protect_list = [] + targets_list = [] + for flags in enemy_flags: + flags_list.append((flags["posX"],flags["posY"])) + for p in enemy_players_flags: + regard_list.append((p["posX"],p["posY"])) + for p in enemy_players: + pretend_list.append((p["posX"],p["posY"])) + for flags in my_flags: + protect_list.append((flags["posX"],flags["posY"])) + for t in my_targets: + targets_list.append((t[0],t[1])) + 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"]] in flags_list: + flags_list.remove(player_to_flag_assign[p["name"]]) + continue + f = myMap.closest(p["posX"],p["posY"],flags_list) + if f is not None: + player_to_flag_assign[p["name"]] = (f[0],f[1]) + flags_list.remove((f[0],f[1])) - # 3. Plan moves for each player player_moves = {} - my_side_is_left = world.is_on_left(my_targets[0]) 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_assignments: - dest = player_to_flag_assignments[p["name"]] + dest,temp = myMap.closest_in_range(p["posX"],p["posY"],my_targets) + f,lentime = myMap.closest_in_range(p["posX"],p["posY"],regard_list) + if f is not None and lentime<6: + dest = (f[0],f[1]) + elif p["name"] in player_to_flag_assign: + dest = player_to_flag_assign[p["name"]] + f,lentime = myMap.closest_in_range(p["posX"],p["posY"],regard_list) + if f is not None and lentime<12: + dest = (f[0],f[1]) + else : + f,lentime = myMap.closest_in_range(p["posX"],p["posY"],pretend_list) + if f is not None and lentime<6: + dest = (f[0],f[1]) 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) - - if len(path) > 1: - move = world.get_direction(curr_pos, path[1]) - player_moves[p["name"]] = move - + f,temp = myMap.closest_in_range(p["posX"],p["posY"],regard_list) + if f is not None: + dest = (f[0],f[1]) + else: + f = myMap.closest(p["posX"],p["posY"],pretend_list) + if f is not None: + dest = (f[0],f[1]) + else: + f= myMap.closest(p["posX"],p["posY"],protect_list) + if f is not None: + dest = (f[0],f[1]) + else: + continue + print(f"Player {p['name']} at ({p['posX']},{p['posY']}) moving towards ({dest[0]},{dest[1]})") + 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()) + asyncio.run(main()) \ No newline at end of file diff --git a/CTF/backend/server.py b/CTF/backend/server.py index 8f5327e..833d9b9 100644 --- a/CTF/backend/server.py +++ b/CTF/backend/server.py @@ -33,9 +33,9 @@ class Map: idx = self.convert_pos_to_index(x, y) self.grid[idx] = 1 if my_side_is_left: - self.in_safe_zone = world.is_on_left((posx+1,posy)) == my_side_is_left + self.in_safe_zone = world.is_on_left((min(posx+2,self.width),posy)) == my_side_is_left else: - self.in_safe_zone = world.is_on_left((posx-1,posy)) == my_side_is_left + self.in_safe_zone = world.is_on_left((max(posx-2,0),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) @@ -157,6 +157,27 @@ class Map: min_length = temp closest_pos = pos return closest_pos,min_length + def enemy_to_walls(self, enemy_posX,enemy_posY): + idx = self.convert_pos_to_index(enemy_posX, enemy_posY) + lengths = abs(enemy_posX-self.width//2) + return lengths + def choose_enemy(self, my_posX, my_posY,enemy_positions): + chosen_enemy = None + min_length = -1 + for pos in enemy_positions: + if my_side_is_left: + if pos[0] >= self.width // 2: + continue + else: + if pos[0] < self.width // 2: + continue + temp = self.enemy_to_walls(pos[0], pos[1]) + length = self.length(my_posX, my_posY, pos[0], pos[1]) + if length > temp*2+4 : continue + if temp < length: + length = temp + chosen_enemy = pos + return chosen_enemy,min_length myMap = Map() def start_game(req): global player_to_flag_assign,my_side_is_left @@ -208,24 +229,24 @@ def plan_next_actions(req): if f is not None: player_to_flag_assign[p["name"]] = (f[0],f[1]) flags_list.remove((f[0],f[1])) - player_moves = {} for p in my_players: if p["hasFlag"]: - dest,temp = myMap.closest_in_range(p["posX"],p["posY"],my_targets) + dest = myMap.closest(p["posX"],p["posY"],my_targets) f,lentime = myMap.closest_in_range(p["posX"],p["posY"],regard_list) - if f is not None and lentime<6: + if f is not None and lentime<4: dest = (f[0],f[1]) elif p["name"] in player_to_flag_assign: dest = player_to_flag_assign[p["name"]] f,lentime = myMap.closest_in_range(p["posX"],p["posY"],regard_list) - if f is not None and lentime<12: + if f is not None and lentime<14: dest = (f[0],f[1]) else : f,lentime = myMap.closest_in_range(p["posX"],p["posY"],pretend_list) - if f is not None and lentime<6: + if f is not None and lentime<8: dest = (f[0],f[1]) + pretend_list.remove((f[0],f[1])) else: f,temp = myMap.closest_in_range(p["posX"],p["posY"],regard_list) if f is not None: @@ -234,13 +255,14 @@ def plan_next_actions(req): f = myMap.closest(p["posX"],p["posY"],pretend_list) if f is not None: dest = (f[0],f[1]) + pretend_list.remove((f[0],f[1])) else: f= myMap.closest(p["posX"],p["posY"],protect_list) if f is not None: dest = (f[0],f[1]) + protect_list.remove((f[0],f[1])) else: continue - print(f"Player {p['name']} at ({p['posX']},{p['posY']}) moving towards ({dest[0]},{dest[1]})") player_moves[p["name"]] = myMap.guideance(p["posX"],p["posY"],dest[0],dest[1]) return player_moves