353 lines
16 KiB
Python
353 lines
16 KiB
Python
"""External Bot Agent with deliberate decision-making logic:
|
||
- Evaluates targets via Radar sensor.
|
||
- Intelligently navigates around impassable obstacles (mountains and forests).
|
||
- Decides whether to negotiate party formation or refuse & fight based on strength.
|
||
- Explicitly engages in 3-bout D20 battles when adjacent to an opposing party/refusing bot.
|
||
- Parses battle results, bout rolls, scores, and absorbed members.
|
||
- Configurable via CLI arguments or environment variables (URL, name, color, strength).
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
import argparse
|
||
import requests
|
||
from typing import Optional, Dict, Any
|
||
|
||
DEFAULT_SERVER_URL = "http://localhost:8000/api"
|
||
DEFAULT_BOT_NAME = "ExternalCyberBot"
|
||
DEFAULT_BOT_COLOR = "#10b981"
|
||
DEFAULT_BOT_STRENGTH = 4
|
||
|
||
|
||
def normalize_url(url: str) -> str:
|
||
"""Ensure the API URL ends with /api without trailing slashes."""
|
||
cleaned = url.rstrip("/")
|
||
if not cleaned.endswith("/api"):
|
||
cleaned = f"{cleaned}/api"
|
||
return cleaned
|
||
|
||
|
||
class SmartBotAgent:
|
||
def __init__(
|
||
self,
|
||
name: str = DEFAULT_BOT_NAME,
|
||
color: str = DEFAULT_BOT_COLOR,
|
||
strength: int = DEFAULT_BOT_STRENGTH,
|
||
server_url: str = DEFAULT_SERVER_URL,
|
||
):
|
||
self.name = name
|
||
self.color = color
|
||
self.strength = strength
|
||
self.base_url = normalize_url(server_url)
|
||
self.bot_id: Optional[str] = None
|
||
self.party_id: Optional[str] = None
|
||
self.is_leader: bool = False
|
||
|
||
def register(self):
|
||
"""Register the bot avatar on the 64x64 grid or reconnect if already present."""
|
||
try:
|
||
players = requests.get(f"{self.base_url}/players").json()
|
||
for p in players:
|
||
if p.get("name") == self.name:
|
||
self.bot_id = p["id"]
|
||
print(f"🔄 [RECONNECT] Reconnected to existing {self.name} (ID: {self.bot_id}, Str: {p.get('strength', self.strength)}) at ({p.get('x')}, {p.get('y')})")
|
||
return
|
||
except Exception:
|
||
pass
|
||
|
||
res = requests.post(
|
||
f"{self.base_url}/players",
|
||
json={"name": self.name, "color": self.color, "strength": self.strength},
|
||
)
|
||
if res.status_code == 400 and "already registered" in res.text:
|
||
players = requests.get(f"{self.base_url}/players").json()
|
||
for p in players:
|
||
if p.get("name") == self.name:
|
||
self.bot_id = p["id"]
|
||
print(f"🔄 [RECONNECT] Reconnected to existing {self.name} (ID: {self.bot_id}) at ({p.get('x')}, {p.get('y')})")
|
||
return
|
||
|
||
res.raise_for_status()
|
||
data = res.json()
|
||
self.bot_id = data["id"]
|
||
print(f"🚀 [REGISTER] Spawned {self.name} (ID: {self.bot_id}, Str: {self.strength}) at ({data['x']}, {data['y']})")
|
||
|
||
def refresh_status(self):
|
||
"""Update bot state (party membership, leader status, score)."""
|
||
res = requests.get(f"{self.base_url}/players/{self.bot_id}")
|
||
if res.status_code == 200:
|
||
data = res.json()
|
||
self.party_id = data.get("party_id")
|
||
self.is_leader = data.get("is_party_leader", False)
|
||
return data
|
||
return None
|
||
|
||
def decide_and_act(self):
|
||
"""Core AI decision loop executed when it is this bot's turn."""
|
||
my_info = self.refresh_status()
|
||
if not my_info:
|
||
return
|
||
|
||
print(f"\n🎮 --- Turn for {self.name} | Score: {my_info['score']} | Party: {self.party_id or 'Solo'} ---")
|
||
|
||
# 1. Consult Radar Sensor
|
||
radar_res = requests.get(f"{self.base_url}/players/{self.bot_id}/radar").json()
|
||
targets = radar_res.get("targets", [])
|
||
nearest = radar_res.get("nearest_target")
|
||
|
||
# 2. Check for immediate adjacent interaction (distance <= 1)
|
||
adjacent_target = None
|
||
for t in targets:
|
||
if t["distance"] <= 1 and not t["is_ally"]:
|
||
adjacent_target = t
|
||
break
|
||
|
||
if adjacent_target:
|
||
self._handle_adjacent_encounter(adjacent_target)
|
||
return
|
||
|
||
# 3. If no immediate adjacent enemy/recruit, move towards target
|
||
self._navigate_towards_goal(radar_res)
|
||
|
||
def _handle_adjacent_encounter(self, target: Dict[str, Any]):
|
||
"""Deliberate decision: Should we join, recruit, or fight?"""
|
||
target_name = target["name"]
|
||
target_str = target["strength"]
|
||
target_party = target.get("party_id")
|
||
|
||
print(f"🔍 [ADJACENT ENCOUNTER] Next to '{target_name}' (Str: {target_str}, Party: {target_party or 'None'})")
|
||
|
||
# SCENARIO A: I am a Solo Bot
|
||
if not self.party_id:
|
||
if not target_party:
|
||
# Both solo: agree to form a party! Stronger bot is leader.
|
||
if self.strength >= target_str:
|
||
print(f"🤝 [DECISION] Proposing party with {target_name}. I have >= strength, so I will lead!")
|
||
leader_id = self.bot_id
|
||
else:
|
||
print(f"🤝 [DECISION] Proposing party with {target_name}. They are stronger, so they will lead.")
|
||
leader_id = target["id"]
|
||
|
||
self._execute_party_formation([self.bot_id, target["id"]], leader_id)
|
||
else:
|
||
# Target belongs to a party: Check leader strength
|
||
# Leadership Rule: Desires equal or stronger leader. Refuses weaker leader!
|
||
# Fetch target's party leader
|
||
target_leader_str = target_str # default fallback
|
||
party_info = requests.get(f"{self.base_url}/parties/{target_party}").json()
|
||
if party_info:
|
||
leader_player = requests.get(f"{self.base_url}/players/{party_info['leader_id']}").json()
|
||
target_leader_str = leader_player.get("strength", 1)
|
||
|
||
if self.strength <= target_leader_str:
|
||
print(f"🤝 [DECISION] Party leader {party_info['leader_name']} has strength {target_leader_str} >= my {self.strength}. Willingly joining squad!")
|
||
# Step towards or engine auto-merges
|
||
self._step_or_attack(target)
|
||
else:
|
||
print(f"⚔️ [DECISION] Party leader has lower strength ({target_leader_str} < my {self.strength})! I REFUSE to join. Engaging in battle!")
|
||
self._initiate_battle(target["id"])
|
||
|
||
# SCENARIO B: I am in a Party
|
||
else:
|
||
if not self.is_leader:
|
||
print(f"🛡️ [PARTY MEMBER] Under command of party leader. Awaiting leader movement.")
|
||
requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
|
||
return
|
||
|
||
if not target_party:
|
||
# Party Leader vs Solo Bot:
|
||
if target_str <= self.strength:
|
||
print(f"🤝 [DECISION] Solo bot {target_name} is willing to join our squad under my leadership.")
|
||
self._step_or_attack(target)
|
||
else:
|
||
print(f"⚔️ [DECISION] Solo bot {target_name} refuses weaker leader! Squad is attacking!")
|
||
self._initiate_battle(target["id"])
|
||
else:
|
||
# Party vs Opposing Party: BATTLE!
|
||
print(f"⚔️ [DECISION] Hostile party detected: '{target.get('party_name')}'! Engaging in 3-Bout D20 battle!")
|
||
self._initiate_battle(target["id"])
|
||
|
||
def _execute_party_formation(self, member_ids, leader_id):
|
||
"""Form a party using the REST API."""
|
||
try:
|
||
res = requests.post(
|
||
f"{self.base_url}/parties",
|
||
json={"member_ids": member_ids, "leader_id": leader_id, "name": f"Squad_{self.name}"},
|
||
)
|
||
if res.status_code == 201:
|
||
party = res.json()
|
||
print(f"✅ [PARTY FORMED] Squad '{party['name']}' established! Leader: {party['leader_name']} | Str: {party['total_strength']}")
|
||
else:
|
||
# Fallback: pass turn if party creation rejected
|
||
requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
|
||
except Exception as e:
|
||
print(f"Party formation error: {e}")
|
||
requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
|
||
|
||
def _initiate_battle(self, opponent_id: str):
|
||
"""Explicitly call the 3-Bout D20 Battle endpoint."""
|
||
print(f"🎲 [BATTLE INITIATED] Clashing with opponent {opponent_id}...")
|
||
res = requests.post(
|
||
f"{self.base_url}/battles/fight",
|
||
json={"challenger_id": self.bot_id, "defender_id": opponent_id},
|
||
)
|
||
if res.status_code == 200:
|
||
battle = res.json()
|
||
print(f"\n--- ⚔️ 3-BOUT BATTLE RESOLUTION ---")
|
||
print(f"Bouts won: {battle['party1_name']} ({battle['party1_bouts_won']}) vs {battle['party2_name']} ({battle['party2_bouts_won']})")
|
||
for b in battle["bouts"]:
|
||
print(f" Bout #{b['bout_number']}: Roll {b['party1_roll']}×{b['party1_strength']} ({b['party1_score']}) vs Roll {b['party2_roll']}×{b['party2_strength']} ({b['party2_score']}) -> Winner: {b['winner_name']}")
|
||
|
||
print(f"🏆 Overall Winner: {battle['winner_party_name']} (Leader {battle['winner_leader_name']} receives +2 pts)")
|
||
print(f"💀 Defeated: {battle['defeated_party_name']} (Leader {battle['killed_leader_name']} -1 pt)")
|
||
if battle.get("absorbed_members"):
|
||
print(f"🧲 Absorbed {len(battle['absorbed_members'])} member(s) into {battle['winner_party_name']}")
|
||
else:
|
||
print(f"Battle failed ({res.status_code}): {res.text}")
|
||
|
||
def _get_best_move_towards(self, target_x: int, target_y: int, moves: Dict[str, Any]) -> Optional[str]:
|
||
"""Pick the available direction that minimizes Chebyshev distance to (target_x, target_y), strictly avoiding obstacles."""
|
||
valid_moves = {d: chk for d, chk in moves.items() if chk.get("available")}
|
||
if not valid_moves:
|
||
return None
|
||
|
||
def dist(chk: Dict[str, Any]) -> int:
|
||
return max(abs(chk["target_x"] - target_x), abs(chk["target_y"] - target_y))
|
||
|
||
return min(valid_moves.keys(), key=lambda d: dist(valid_moves[d]))
|
||
|
||
def _step_or_attack(self, target: Dict[str, Any]):
|
||
"""Move adjacent/towards the target while avoiding obstacles."""
|
||
moves_res = requests.get(f"{self.base_url}/players/{self.bot_id}/available-moves").json()
|
||
moves = moves_res.get("moves", {})
|
||
chosen = self._get_best_move_towards(target["x"], target["y"], moves)
|
||
|
||
if chosen:
|
||
res = requests.post(f"{self.base_url}/players/{self.bot_id}/move", json={"direction": chosen}).json()
|
||
if res.get("battle_triggered"):
|
||
print(f"⚔️ Move triggered battle! Winner: {res['battle_result']['winner_party_name']}")
|
||
elif res.get("party_formed_triggered"):
|
||
print(f"🤝 Move resulted in party alliance!")
|
||
else:
|
||
print("⚠️ No passable moves adjacent to target (terrain/border constraint). Passing turn.")
|
||
requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
|
||
|
||
def _navigate_towards_goal(self, radar_res: Dict[str, Any]):
|
||
"""Move towards the nearest target routing around obstacles."""
|
||
rec_dir = radar_res.get("recommended_direction")
|
||
nearest = radar_res.get("nearest_target")
|
||
moves_res = requests.get(f"{self.base_url}/players/{self.bot_id}/available-moves").json()
|
||
moves = moves_res.get("moves", {})
|
||
available = [d for d, chk in moves.items() if chk.get("available")]
|
||
|
||
if not available:
|
||
print("🚫 All adjacent paths blocked by borders or obstacle terrain (mountains/forests). Passing turn.")
|
||
requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
|
||
return
|
||
|
||
# 1. Prefer radar's obstacle-aware BFS pathfinder direction
|
||
if rec_dir and rec_dir in available:
|
||
chosen_dir = rec_dir
|
||
# 2. Otherwise pick the available direction that minimizes distance to the nearest target
|
||
elif nearest:
|
||
chosen_dir = self._get_best_move_towards(nearest["x"], nearest["y"], moves) or available[0]
|
||
# 3. Fallback to any valid open terrain cell
|
||
else:
|
||
chosen_dir = available[0]
|
||
|
||
print(f"🧭 Moving {chosen_dir} (Goal: {radar_res.get('bot_goal')}, Action: {radar_res.get('recommended_action')})")
|
||
res = requests.post(f"{self.base_url}/players/{self.bot_id}/move", json={"direction": chosen_dir}).json()
|
||
|
||
if res.get("battle_triggered"):
|
||
print(f"⚔️ Encounter battle! Winner: {res['battle_result']['winner_party_name']}")
|
||
elif res.get("party_formed_triggered"):
|
||
print(f"🤝 Formed or joined squad: {res.get('formed_party', {}).get('name')}")
|
||
|
||
def run(self):
|
||
"""Main game loop for external agent."""
|
||
self.register()
|
||
try:
|
||
while True:
|
||
turn_info = requests.get(f"{self.base_url}/turn").json()
|
||
if not turn_info.get("game_started", False):
|
||
# Check if bot was removed (e.g., board was reset)
|
||
if self.bot_id:
|
||
res = requests.get(f"{self.base_url}/players/{self.bot_id}")
|
||
if res.status_code == 404:
|
||
print("\n⚠️ [RESET] Board was regenerated. Rejoining lobby...")
|
||
self.register()
|
||
|
||
print("⏳ [LOBBY] Waiting for game to start via 'Start Game' in UI... ", end="\r", flush=True)
|
||
time.sleep(1.0)
|
||
continue
|
||
|
||
curr_player_id = turn_info.get("current_player_id")
|
||
|
||
if curr_player_id == self.bot_id:
|
||
self.decide_and_act()
|
||
|
||
# Check for game conclusion
|
||
conc = requests.get(f"{self.base_url}/game/conclusion").json()
|
||
if conc.get("concluded"):
|
||
print(f"\n🎉 [GAME CONCLUDED] All bots united under '{conc['winning_party_name']}'!")
|
||
break
|
||
else:
|
||
time.sleep(0.4)
|
||
|
||
except KeyboardInterrupt:
|
||
print(f"\nDisconnecting {self.name}...")
|
||
requests.delete(f"{self.base_url}/players/{self.bot_id}")
|
||
|
||
|
||
def main():
|
||
# Read defaults from environment variables if present
|
||
env_url = os.environ.get("BOT_SERVER_URL") or os.environ.get("SERVER_URL") or DEFAULT_SERVER_URL
|
||
env_name = os.environ.get("BOT_NAME", DEFAULT_BOT_NAME)
|
||
env_color = os.environ.get("BOT_COLOR", DEFAULT_BOT_COLOR)
|
||
env_strength = int(os.environ.get("BOT_STRENGTH", str(DEFAULT_BOT_STRENGTH)))
|
||
|
||
parser = argparse.ArgumentParser(
|
||
description="Autonomous External Bot Agent for botWebWars",
|
||
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
||
)
|
||
parser.add_argument(
|
||
"-u", "--url",
|
||
dest="server_url",
|
||
default=env_url,
|
||
help="Backend REST API base URL (env: BOT_SERVER_URL or SERVER_URL)",
|
||
)
|
||
parser.add_argument(
|
||
"-n", "--name",
|
||
dest="name",
|
||
default=env_name,
|
||
help="Display name for this bot (env: BOT_NAME)",
|
||
)
|
||
parser.add_argument(
|
||
"-c", "--color",
|
||
dest="color",
|
||
default=env_color,
|
||
help="Hex color code for the bot avatar, e.g. #10b981 (env: BOT_COLOR)",
|
||
)
|
||
parser.add_argument(
|
||
"-s", "--strength",
|
||
dest="strength",
|
||
type=int,
|
||
default=env_strength,
|
||
help="Strength attribute (1-10) for D20 battle multiplier (env: BOT_STRENGTH)",
|
||
)
|
||
|
||
args = parser.parse_args()
|
||
|
||
agent = SmartBotAgent(
|
||
name=args.name,
|
||
color=args.color,
|
||
strength=args.strength,
|
||
server_url=args.server_url,
|
||
)
|
||
agent.run()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|