feat(backend): add Map class with BFS pathfinding and faster updates
- Reduce `GameMap` update interval from 1000 ms to 10 ms for more responsive gameplay. - Introduce thread lock, update thresholds, and player‑to‑flag tracking variables for safer concurrent access. - Add `my_side_is_left` flag to determine team orientation. - Implement new `Map` class: * Builds a grid representation of the world, marking walls, enemy flag carriers, and allies. * Generates adjacency lists (`edge`) respecting safe‑zone rules. * Provides `guideance` method that runs a BFS (unit‑weight Dijkstra) to compute the shortest path and returns the first movement direction via `world.get_direction`. - Import `collections` for efficient deque usage. - Include compiled `.pyc` for the new module. These changes enable the policy to make quick, informed movement decisions based on real‑time map analysis.
This commit is contained in:
+111
-18
@@ -2,41 +2,139 @@ 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=10.0)
|
||||
lock = threading.Lock()
|
||||
last_updated_time = -1
|
||||
update_threshold = 100
|
||||
player_to_flag_assignments = {}
|
||||
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.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"]
|
||||
idx = self.convert_pos_to_index(x, y)
|
||||
self.grid[idx] = 1 # Mark wall positions
|
||||
|
||||
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)
|
||||
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
|
||||
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
|
||||
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:
|
||||
self.edge[idx].append(n_idx)
|
||||
else:
|
||||
if (self.grid[n_idx] not in (1, 3)) or n_idx == my_pos:
|
||||
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[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 + 1)
|
||||
next_y = next_idx // (self.width + 1)
|
||||
|
||||
return world.get_direction((posx_start, posy_start), (next_x, next_y))
|
||||
|
||||
myMap = Map()
|
||||
def start_game(req):
|
||||
global player_to_flag_assignments
|
||||
global player_to_flag_assignments,my_side_is_left
|
||||
world.init(req)
|
||||
print("Start Game!!")
|
||||
player_to_flag_assignments = {}
|
||||
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_assignments,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))
|
||||
|
||||
|
||||
# 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()
|
||||
if name in active_player_names
|
||||
@@ -51,7 +149,6 @@ def plan_next_actions(req):
|
||||
|
||||
# 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"])
|
||||
@@ -67,13 +164,10 @@ def plan_next_actions(req):
|
||||
# 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
|
||||
# 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
|
||||
|
||||
@@ -97,6 +191,5 @@ async def main():
|
||||
print(f"Server Stopped: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
|
||||
Reference in New Issue
Block a user