feat(ai): add Map class for pathfinding and speed up world updates
- Remove unused `random` import and introduce `collections` for BFS queue. - Decrease `GameMap` update interval from 1000 ms to 1 ms and lower `update_threshold` from 100 to 1 for near‑real‑time world state. - Add thread‑safe lock, timestamp tracking, and rename `player_to_flag_assignments` to `player_to_flag_assign`. - Introduce `my_side_is_left` flag to differentiate safe‑zone logic. - Implement new `Map` class: * Stores grid, edges, and safe‑zone status. * Updates grid with walls, allies, enemies, and safe‑zone handling. * Computes adjacency edges respecting obstacles. * Provides `guideance` (BFS) to obtain the first movement direction towards a target. - Overall refactor improves AI flag‑picking speed and decision accuracy.
This commit is contained in:
@@ -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]} <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())
|
|
||||||
+215
-52
@@ -1,103 +1,266 @@
|
|||||||
import asyncio
|
import asyncio
|
||||||
import random
|
|
||||||
from lib.game_engine import GameMap, run_game_server
|
from lib.game_engine import GameMap, run_game_server
|
||||||
import threading
|
import threading
|
||||||
|
import collections
|
||||||
|
|
||||||
|
world = GameMap(show_gap_in_msec=1.0)
|
||||||
# 1. Initialize the global world model
|
|
||||||
world = GameMap(show_gap_in_msec=1000.0)
|
|
||||||
lock = threading.Lock()
|
lock = threading.Lock()
|
||||||
last_updated_time = -1
|
last_updated_time = -1
|
||||||
update_threshold = 100
|
update_threshold = 1
|
||||||
player_to_flag_assignments = {}
|
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):
|
def start_game(req):
|
||||||
global player_to_flag_assignments
|
global player_to_flag_assign,my_side_is_left
|
||||||
world.init(req)
|
world.init(req)
|
||||||
print("Start Game!!")
|
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'}")
|
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):
|
def plan_next_actions(req):
|
||||||
if not world.update(req):
|
if not world.update(req):
|
||||||
return
|
return
|
||||||
|
global player_to_flag_assign,myMap,my_side_is_left
|
||||||
global player_to_flag_assignments
|
|
||||||
|
|
||||||
# Render the map
|
|
||||||
# world.show(do_not_clear=False)
|
|
||||||
|
|
||||||
my_players = world.list_players(mine=True, inPrison=False, hasFlag=None)
|
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)
|
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_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"]}
|
active_player_names = {p["name"] for p in my_players if not p["hasFlag"]}
|
||||||
|
flags_list = []
|
||||||
# Cleanup assignments for players captured or flags already taken
|
regard_list = []
|
||||||
player_to_flag_assignments = {
|
pretend_list = []
|
||||||
name: pos for name, pos in player_to_flag_assignments.items()
|
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 name in active_player_names
|
||||||
}
|
}
|
||||||
|
|
||||||
if enemy_flags:
|
if enemy_flags:
|
||||||
for p in my_players:
|
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"]] in flags_list:
|
||||||
player_to_flag_assignments[p["name"]] = (f["posX"], f["posY"])
|
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 = {}
|
player_moves = {}
|
||||||
my_side_is_left = world.is_on_left(my_targets[0])
|
|
||||||
|
|
||||||
for p in my_players:
|
for p in my_players:
|
||||||
curr_pos = (p["posX"], p["posY"])
|
|
||||||
|
|
||||||
# Determine Target: Either the assigned flag or the home target
|
|
||||||
if p["hasFlag"]:
|
if p["hasFlag"]:
|
||||||
dest = my_targets[0]
|
dest,temp = myMap.closest_in_range(p["posX"],p["posY"],my_targets)
|
||||||
elif p["name"] in player_to_flag_assignments:
|
f,lentime = myMap.closest_in_range(p["posX"],p["posY"],regard_list)
|
||||||
dest = player_to_flag_assignments[p["name"]]
|
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:
|
else:
|
||||||
continue
|
f,temp = myMap.closest_in_range(p["posX"],p["posY"],regard_list)
|
||||||
|
if f is not None:
|
||||||
# Determine Obstacles: Avoid opponents if we are in enemy territory
|
dest = (f[0],f[1])
|
||||||
is_safe = world.is_on_left(curr_pos) == my_side_is_left
|
else:
|
||||||
blockers = [] if is_safe else [(o["posX"], o["posY"]) for o in opponents]
|
f = myMap.closest(p["posX"],p["posY"],pretend_list)
|
||||||
|
if f is not None:
|
||||||
# Calculate Path
|
dest = (f[0],f[1])
|
||||||
path = world.route_to(curr_pos, dest, extra_obstacles=blockers)
|
else:
|
||||||
|
f= myMap.closest(p["posX"],p["posY"],protect_list)
|
||||||
if len(path) > 1:
|
if f is not None:
|
||||||
move = world.get_direction(curr_pos, path[1])
|
dest = (f[0],f[1])
|
||||||
player_moves[p["name"]] = move
|
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
|
return player_moves
|
||||||
|
|
||||||
def game_over(req):
|
def game_over(req):
|
||||||
print("Game Over!")
|
print("Game Over!")
|
||||||
world.show(force=True)
|
world.show(force=True)
|
||||||
|
|
||||||
|
|
||||||
async def main():
|
async def main():
|
||||||
import sys
|
import sys
|
||||||
if len(sys.argv) != 2:
|
if len(sys.argv) != 2:
|
||||||
print(f"Usage: python3 {sys.argv[0]} <port>")
|
print(f"Usage: python3 {sys.argv[0]} <port>")
|
||||||
print(f"Example: python3 {sys.argv[0]} 8080")
|
print(f"Example: python3 {sys.argv[0]} 8080")
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
|
|
||||||
port = int(sys.argv[1])
|
port = int(sys.argv[1])
|
||||||
print(f"AI backend running on port {port} ...")
|
print(f"AI backend running on port {port} ...")
|
||||||
|
|
||||||
try:
|
try:
|
||||||
await run_game_server(port, start_game, plan_next_actions, game_over)
|
await run_game_server(port, start_game, plan_next_actions, game_over)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
print(f"Server Stopped: {e}")
|
print(f"Server Stopped: {e}")
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
asyncio.run(main())
|
asyncio.run(main())
|
||||||
+30
-8
@@ -33,9 +33,9 @@ class Map:
|
|||||||
idx = self.convert_pos_to_index(x, y)
|
idx = self.convert_pos_to_index(x, y)
|
||||||
self.grid[idx] = 1
|
self.grid[idx] = 1
|
||||||
if my_side_is_left:
|
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:
|
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)
|
enemy_players = world.list_players(mine=False, inPrison=False, hasFlag=None)
|
||||||
ally_players = world.list_players(mine=True, inPrison=False, hasFlag=None)
|
ally_players = world.list_players(mine=True, inPrison=False, hasFlag=None)
|
||||||
my_pos = self.convert_pos_to_index(posx, posy)
|
my_pos = self.convert_pos_to_index(posx, posy)
|
||||||
@@ -157,6 +157,27 @@ class Map:
|
|||||||
min_length = temp
|
min_length = temp
|
||||||
closest_pos = pos
|
closest_pos = pos
|
||||||
return closest_pos,min_length
|
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()
|
myMap = Map()
|
||||||
def start_game(req):
|
def start_game(req):
|
||||||
global player_to_flag_assign,my_side_is_left
|
global player_to_flag_assign,my_side_is_left
|
||||||
@@ -208,24 +229,24 @@ def plan_next_actions(req):
|
|||||||
if f is not None:
|
if f is not None:
|
||||||
player_to_flag_assign[p["name"]] = (f[0],f[1])
|
player_to_flag_assign[p["name"]] = (f[0],f[1])
|
||||||
flags_list.remove((f[0],f[1]))
|
flags_list.remove((f[0],f[1]))
|
||||||
|
|
||||||
player_moves = {}
|
player_moves = {}
|
||||||
|
|
||||||
for p in my_players:
|
for p in my_players:
|
||||||
if p["hasFlag"]:
|
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)
|
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])
|
dest = (f[0],f[1])
|
||||||
elif p["name"] in player_to_flag_assign:
|
elif p["name"] in player_to_flag_assign:
|
||||||
dest = player_to_flag_assign[p["name"]]
|
dest = player_to_flag_assign[p["name"]]
|
||||||
f,lentime = myMap.closest_in_range(p["posX"],p["posY"],regard_list)
|
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])
|
dest = (f[0],f[1])
|
||||||
else :
|
else :
|
||||||
f,lentime = myMap.closest_in_range(p["posX"],p["posY"],pretend_list)
|
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])
|
dest = (f[0],f[1])
|
||||||
|
pretend_list.remove((f[0],f[1]))
|
||||||
else:
|
else:
|
||||||
f,temp = myMap.closest_in_range(p["posX"],p["posY"],regard_list)
|
f,temp = myMap.closest_in_range(p["posX"],p["posY"],regard_list)
|
||||||
if f is not None:
|
if f is not None:
|
||||||
@@ -234,13 +255,14 @@ def plan_next_actions(req):
|
|||||||
f = myMap.closest(p["posX"],p["posY"],pretend_list)
|
f = myMap.closest(p["posX"],p["posY"],pretend_list)
|
||||||
if f is not None:
|
if f is not None:
|
||||||
dest = (f[0],f[1])
|
dest = (f[0],f[1])
|
||||||
|
pretend_list.remove((f[0],f[1]))
|
||||||
else:
|
else:
|
||||||
f= myMap.closest(p["posX"],p["posY"],protect_list)
|
f= myMap.closest(p["posX"],p["posY"],protect_list)
|
||||||
if f is not None:
|
if f is not None:
|
||||||
dest = (f[0],f[1])
|
dest = (f[0],f[1])
|
||||||
|
protect_list.remove((f[0],f[1]))
|
||||||
else:
|
else:
|
||||||
continue
|
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])
|
player_moves[p["name"]] = myMap.guideance(p["posX"],p["posY"],dest[0],dest[1])
|
||||||
return player_moves
|
return player_moves
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user