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()
|