355 lines
16 KiB
Python
355 lines
16 KiB
Python
"""Autonomous Heuristic Troll Agent for botWebWars:
|
||
- Registers as character_type='troll' and piece_type='troll'.
|
||
- Trolls do not band together or form alliances.
|
||
- Trolls do not battle each other.
|
||
- Trolls only hunt and battle players and player squads.
|
||
- Trolls do not gain strength from victory, but earn +2 victory points (score).
|
||
- Trolls take damage (1-3 HP) upon defeat and die if HP reaches 0.
|
||
- Trolls cannot duel Gary the Wizard.
|
||
- Trolls can take a rest turn to SLEEP (POST /api/players/{id}/sleep), recovering +0.1 HP.
|
||
- Configurable via CLI arguments or environment variables.
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
import argparse
|
||
import requests
|
||
from typing import Optional, Dict, Any, List
|
||
|
||
DEFAULT_SERVER_URL = os.getenv("TROLL_SERVER_URL", os.getenv("BOT_SERVER_URL", "http://localhost:8000/api"))
|
||
DEFAULT_TROLL_NAME = os.getenv("TROLL_NAME", "GorgonTroll")
|
||
DEFAULT_TROLL_COLOR = os.getenv("TROLL_COLOR", "#16a34a")
|
||
DEFAULT_TROLL_STRENGTH = float(os.getenv("TROLL_STRENGTH", "3.0"))
|
||
DEFAULT_TROLL_HEALTH = float(os.getenv("TROLL_HEALTH", "10.0"))
|
||
DEFAULT_SLEEP_THRESHOLD = float(os.getenv("TROLL_SLEEP_THRESHOLD", "6.0"))
|
||
|
||
|
||
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 SmartTrollAgent:
|
||
def __init__(
|
||
self,
|
||
name: str = DEFAULT_TROLL_NAME,
|
||
color: str = DEFAULT_TROLL_COLOR,
|
||
strength: float = DEFAULT_TROLL_STRENGTH,
|
||
health: float = DEFAULT_TROLL_HEALTH,
|
||
server_url: str = DEFAULT_SERVER_URL,
|
||
sleep_threshold: float = DEFAULT_SLEEP_THRESHOLD,
|
||
loop_delay: float = 1.0,
|
||
):
|
||
self.name = name
|
||
self.color = color
|
||
self.strength = strength
|
||
self.health = health
|
||
self.server_url = server_url
|
||
self.base_url = normalize_url(server_url)
|
||
self.sleep_threshold = sleep_threshold
|
||
self.loop_delay = loop_delay
|
||
self.bot_id: Optional[str] = None
|
||
|
||
def register(self):
|
||
"""Register the troll 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 Troll {self.name} "
|
||
f"(ID: {self.bot_id}, Str: {p.get('strength', self.strength)}, "
|
||
f"HP: {p.get('health', self.health)}) at ({p.get('x')}, {p.get('y')})"
|
||
)
|
||
return
|
||
except Exception:
|
||
pass
|
||
|
||
payload = {
|
||
"name": self.name,
|
||
"color": self.color,
|
||
"strength": self.strength,
|
||
"health": self.health,
|
||
"piece_type": "troll",
|
||
"character_type": "troll",
|
||
}
|
||
|
||
res = requests.post(f"{self.base_url}/players", json=payload)
|
||
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 Troll {self.name} "
|
||
f"(ID: {self.bot_id}, HP: {p.get('health', self.health)}) at ({p.get('x')}, {p.get('y')})"
|
||
)
|
||
return
|
||
|
||
res.raise_for_status()
|
||
data = res.json()
|
||
self.bot_id = data["id"]
|
||
print(
|
||
f"👹 [REGISTER] Spawned Troll {self.name} (ID: {self.bot_id}, Str: {self.strength}, "
|
||
f"HP: {data.get('health', self.health)}) at ({data['x']}, {data['y']})"
|
||
)
|
||
|
||
def refresh_status(self) -> Optional[Dict[str, Any]]:
|
||
"""Update troll status (health, score, position, alive status)."""
|
||
try:
|
||
res = requests.get(f"{self.base_url}/players/{self.bot_id}")
|
||
if res.status_code == 200:
|
||
return res.json()
|
||
except Exception as e:
|
||
print(f"Status refresh error: {e}")
|
||
return None
|
||
|
||
def sleep_and_heal(self) -> bool:
|
||
"""Execute sleep turn to recover +0.1 HP."""
|
||
try:
|
||
res = requests.post(f"{self.base_url}/players/{self.bot_id}/sleep")
|
||
if res.status_code == 200:
|
||
data = res.json()
|
||
print(
|
||
f"💤 [SLEEP] {self.name} curled up and slept for 1 turn. "
|
||
f"(+{data.get('health_gained', 0.1)} HP -> ❤️ {data.get('new_health')} HP)"
|
||
)
|
||
return True
|
||
else:
|
||
print(f"Sleep failed ({res.status_code}): {res.text}")
|
||
requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
|
||
return False
|
||
except Exception as e:
|
||
print(f"Error sleeping: {e}")
|
||
requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
|
||
return False
|
||
|
||
def pass_turn(self):
|
||
"""Pass turn to next queued entity."""
|
||
try:
|
||
requests.post(f"{self.base_url}/players/{self.bot_id}/pass")
|
||
print(f"⏩ [PASS] {self.name} passed turn.")
|
||
except Exception as e:
|
||
print(f"Error passing turn: {e}")
|
||
|
||
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)."""
|
||
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))
|
||
|
||
# Prioritize moves without penalty, then minimum distance
|
||
return min(valid_moves.keys(), key=lambda d: (valid_moves[d].get("strength_penalty", 0.0) > 0, dist(valid_moves[d])))
|
||
|
||
def _find_adjacent_player(self, my_x: int, my_y: int) -> Optional[Dict[str, Any]]:
|
||
"""Find an adjacent living player or party to attack (ignoring other trolls)."""
|
||
try:
|
||
players: List[Dict[str, Any]] = requests.get(f"{self.base_url}/players").json()
|
||
for p in players:
|
||
if p.get("id") == self.bot_id:
|
||
continue
|
||
# Trolls ignore other trolls
|
||
if p.get("character_type") == "troll":
|
||
continue
|
||
# Ignore dead players
|
||
if p.get("is_alive") is False or p.get("health", 10) <= 0:
|
||
continue
|
||
chebyshev = max(abs(p["x"] - my_x), abs(p["y"] - my_y))
|
||
if chebyshev <= 1:
|
||
return p
|
||
except Exception as e:
|
||
print(f"Error scanning adjacent players: {e}")
|
||
return None
|
||
|
||
def attack_player(self, defender: Dict[str, Any]):
|
||
"""Initiate mandatory 3-bout D20 battle against adjacent player or party."""
|
||
target_label = f"Squad '{defender.get('party_id')}'" if defender.get("party_id") else f"Player {defender.get('name')}"
|
||
print(f"⚔️ [BATTLE CLASH] Troll {self.name} engages {target_label} in 3-bout D20 combat!")
|
||
try:
|
||
res = requests.post(
|
||
f"{self.base_url}/battles/fight",
|
||
json={"challenger_id": self.bot_id, "defender_id": defender["id"]},
|
||
)
|
||
if res.status_code == 200:
|
||
b = res.json()
|
||
won = b.get("winner_leader_name") == self.name or b.get("winner_party_name", "").startswith(self.name)
|
||
outcome = "VICTORY (+2 Victory Points!)" if won else "DEFEAT (Health Damage sustained)"
|
||
print(f"⚔️ [RESULT] {outcome}: {b.get('winner_party_name')} defeated {b.get('defeated_party_name')}")
|
||
for bout in b.get("bouts", []):
|
||
print(
|
||
f" Bout #{bout['bout_number']}: "
|
||
f"{bout['party1_name']} D20({bout['party1_roll']})×Str({bout['party1_strength']})={bout['party1_score']:.1f} vs "
|
||
f"{bout['party2_name']} D20({bout['party2_roll']})×Str({bout['party2_strength']})={bout['party2_score']:.1f} "
|
||
f"-> Winner: {bout['winner']}"
|
||
)
|
||
else:
|
||
print(f"Battle initiation failed ({res.status_code}): {res.text}")
|
||
self.pass_turn()
|
||
except Exception as e:
|
||
print(f"Error executing battle: {e}")
|
||
self.pass_turn()
|
||
|
||
def decide_and_act(self):
|
||
"""Evaluate troll radar, health condition, and surroundings to execute best action."""
|
||
my_status = self.refresh_status()
|
||
if not my_status:
|
||
self.pass_turn()
|
||
return
|
||
|
||
my_x = my_status.get("x", 0)
|
||
my_y = my_status.get("y", 0)
|
||
my_hp = float(my_status.get("health", 10.0))
|
||
|
||
# Check for immediate adjacent player to attack
|
||
adjacent_player = self._find_adjacent_player(my_x, my_y)
|
||
if adjacent_player:
|
||
self.attack_player(adjacent_player)
|
||
return
|
||
|
||
# Check radar sensor
|
||
try:
|
||
radar_res = requests.get(f"{self.base_url}/players/{self.bot_id}/radar").json()
|
||
except Exception as e:
|
||
print(f"Radar failure: {e}")
|
||
radar_res = {}
|
||
|
||
recommended_action = radar_res.get("recommended_action")
|
||
|
||
# Sleep condition:
|
||
# 1. Radar recommends sleep (e.g. damaged and no immediate contact)
|
||
# 2. Or current health is below configured sleep threshold
|
||
if (recommended_action == "sleep" or my_hp < self.sleep_threshold) and my_hp < float(my_status.get("max_health", 10.0)):
|
||
print(f"🩸 Low health alert (HP: {my_hp:.1f} < threshold: {self.sleep_threshold:.1f}). Taking sleep turn to heal...")
|
||
self.sleep_and_heal()
|
||
return
|
||
|
||
# Navigation: Hunt closest player
|
||
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:
|
||
# If completely cornered/trapped, rest and sleep instead of passing
|
||
if my_hp < float(my_status.get("max_health", 10.0)):
|
||
print("🚫 All adjacent paths blocked by terrain. Sleeping to regenerate health...")
|
||
self.sleep_and_heal()
|
||
else:
|
||
print("🚫 All adjacent paths blocked by terrain. Passing turn.")
|
||
self.pass_turn()
|
||
return
|
||
|
||
# 1. Prefer radar BFS route direction
|
||
if rec_dir and rec_dir in available:
|
||
chosen_dir = rec_dir
|
||
# 2. Move towards nearest player target
|
||
elif nearest:
|
||
chosen_dir = self._get_best_move_towards(nearest["x"], nearest["y"], moves) or available[0]
|
||
# 3. Fallback open move
|
||
else:
|
||
chosen_dir = available[0]
|
||
|
||
target_info = f"closest player '{nearest.get('name')}' at dist {nearest.get('distance')}" if nearest else "open frontier"
|
||
print(f"👹 [HUNT] Moving {chosen_dir} towards {target_info} (HP: {my_hp:.1f}, Str: {self.strength:.1f})")
|
||
|
||
try:
|
||
res = requests.post(
|
||
f"{self.base_url}/players/{self.bot_id}/move",
|
||
json={"direction": chosen_dir},
|
||
).json()
|
||
|
||
if res.get("battle_triggered") and res.get("battle_result"):
|
||
b = res["battle_result"]
|
||
print(f"⚔️ Move clash! Winner: {b.get('winner_party_name')} (Defeated: {b.get('defeated_party_name')})")
|
||
except Exception as e:
|
||
print(f"Error during move: {e}")
|
||
self.pass_turn()
|
||
|
||
def run(self):
|
||
"""Main game loop for autonomous troll agent."""
|
||
self.register()
|
||
try:
|
||
while True:
|
||
# Check life status
|
||
my_status = self.refresh_status()
|
||
if my_status and (my_status.get("is_alive") is False or my_status.get("health", 10) <= 0):
|
||
print(f"\n🪦 [FALLEN TROLL] {self.name} has fallen (0 HP)! Gravestone marked on board.")
|
||
print(f"Final Achieved Score: {my_status.get('score', 0)} pts preserved on scoreboard. Spectating...")
|
||
while True:
|
||
try:
|
||
conc = requests.get(f"{self.base_url}/game/conclusion").json()
|
||
if conc.get("concluded"):
|
||
print(f"\n🎉 [GAME CONCLUDED] Game ended! Winner: '{conc.get('winning_party_name')}'")
|
||
return
|
||
except Exception:
|
||
pass
|
||
time.sleep(2.0)
|
||
|
||
turn_info = requests.get(f"{self.base_url}/turn").json()
|
||
if not turn_info.get("game_started", False):
|
||
# Check if bot was removed by board 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:
|
||
print(f"\n⚡ [MY TURN] Troll {self.name}'s turn (Round {turn_info.get('round_number')}, Turn {turn_info.get('turn_number')})")
|
||
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] Arena concluded! Winner: '{conc.get('winning_party_name')}'!")
|
||
break
|
||
|
||
time.sleep(self.loop_delay)
|
||
|
||
except KeyboardInterrupt:
|
||
print(f"\n👋 Troll agent {self.name} disconnecting gracefully.")
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Autonomous Heuristic Troll Agent for botWebWars")
|
||
parser.add_argument("-u", "--url", default=DEFAULT_SERVER_URL, help="botWebWars API base URL")
|
||
parser.add_argument("-n", "--name", default=DEFAULT_TROLL_NAME, help="Troll name")
|
||
parser.add_argument("-c", "--color", default=DEFAULT_TROLL_COLOR, help="Troll color hex code")
|
||
parser.add_argument("-s", "--strength", type=float, default=DEFAULT_TROLL_STRENGTH, help="Starting strength (1-10)")
|
||
parser.add_argument("-H", "--health", type=float, default=DEFAULT_TROLL_HEALTH, help="Starting health points (default: 10.0)")
|
||
parser.add_argument("--sleep-threshold", type=float, default=DEFAULT_SLEEP_THRESHOLD, help="HP threshold below which troll sleeps to heal (default: 6.0)")
|
||
parser.add_argument("--loop-delay", type=float, default=1.0, help="Polling interval in seconds (default: 1.0)")
|
||
|
||
args = parser.parse_args()
|
||
|
||
agent = SmartTrollAgent(
|
||
name=args.name,
|
||
color=args.color,
|
||
strength=args.strength,
|
||
health=args.health,
|
||
server_url=args.url,
|
||
sleep_threshold=args.sleep_threshold,
|
||
loop_delay=args.loop_delay,
|
||
)
|
||
agent.run()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|