feat(backend): integrate dual AI strategies with conditional switching and A* pathfinding

- Add imports for server0 and server1 modules
- Modify start_game, plan_next_actions, and game_over to invoke both strategies
- Implement dynamic strategy selection based on player prison counts
- Add A* pathfinding with danger weights for attacking players, including safe/dangerous position calculations and opponent avoidance
- Update traditional pathfinding for defenders and flag carriers in enemy territory
- Include binary updates to .DS_Store files (likely system artifacts)
This commit is contained in:
2025-12-28 13:39:06 +08:00
parent 6cae6a2ad8
commit fc9f98aa3d
5 changed files with 1471 additions and 13 deletions

BIN
.DS_Store vendored

Binary file not shown.

BIN
CTF/.DS_Store vendored

Binary file not shown.

View File

@@ -2,6 +2,8 @@ import asyncio
from lib.game_engine import GameMap, run_game_server
import threading
import collections
import server0
import server1
world = GameMap(show_gap_in_msec=1.0)
lock = threading.Lock()
@@ -179,15 +181,23 @@ class Map:
chosen_enemy = pos
return chosen_enemy,min_length
myMap = Map()
rounds = 0
def start_game(req):
global player_to_flag_assign,my_side_is_left
server1.start_game(req)
server0.start_game(req)
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])
global rounds
rounds = 0
def plan_next_actions(req):
global rounds
rounds+=1
# if rounds < 50:
# return server0.plan_next_actions(req)
# else:
# return server1.plan_next_actions(req)
# return server0.plan_next_actions(req)
if not world.update(req):
return
global player_to_flag_assign,myMap,my_side_is_left
@@ -198,6 +208,17 @@ def plan_next_actions(req):
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))
my_in_prison = len(world.list_players(mine=True, inPrison=True, hasFlag=None))
op_in_prison = len(world.list_players(False,True,None))
if my_in_prison <= 0 and op_in_prison <= 0:
return server0.plan_next_actions(req)
elif my_in_prison > op_in_prison:
return server1.plan_next_actions(req)
else:
return server0.plan_next_actions(req)
active_player_names = {p["name"] for p in my_players if not p["hasFlag"]}
flags_list = []
regard_list = []
@@ -242,7 +263,6 @@ def plan_next_actions(req):
f,lentime = myMap.closest_in_range(p["posX"],p["posY"],regard_list)
if f is not None and lentime<14:
dest = (f[0],f[1])
regard_list.remove((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:
@@ -252,7 +272,6 @@ def plan_next_actions(req):
f,temp = myMap.closest_in_range(p["posX"],p["posY"],regard_list)
if f is not None:
dest = (f[0],f[1])
regard_list.remove((f[0],f[1]))
else:
f = myMap.closest(p["posX"],p["posY"],pretend_list)
if f is not None:
@@ -265,15 +284,14 @@ def plan_next_actions(req):
protect_list.remove((f[0],f[1]))
else:
continue
if dest is not None:
player_moves[p["name"]] = myMap.guideance(p["posX"],p["posY"],dest[0],dest[1])
else:
player_moves[p["name"]] = myMap.guideance(p["posX"],p["posY"],p["posX"],p["posY"])
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)
server1.game_over(req)
server0.game_over(req)
# print("Game Over!")
# world.show(force=True)
async def main():
import sys

1147
CTF/backend/server0.py Normal file

File diff suppressed because it is too large Load Diff

293
CTF/backend/server1.py Normal file
View File

@@ -0,0 +1,293 @@
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((min(posx+2,self.width),posy)) == my_side_is_left
else:
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)
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
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
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 = 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<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<14:
dest = (f[0],f[1])
regard_list.remove((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])
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:
dest = (f[0],f[1])
regard_list.remove((f[0],f[1]))
else:
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
if dest is not None:
player_moves[p["name"]] = myMap.guideance(p["posX"],p["posY"],dest[0],dest[1])
else:
player_moves[p["name"]] = myMap.guideance(p["posX"],p["posY"],p["posX"],p["posY"])
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())