614 lines
25 KiB
Python
614 lines
25 KiB
Python
"""AI-driven Troll Agent for botWebWars, powered by Google Cloud Vertex AI (Gemini).
|
||
|
||
Implements the unique rules and tactical gameplay of the Troll character type:
|
||
- Solitary brute: Never joins or forms parties/alliances.
|
||
- Troll truce: Never battles other trolls.
|
||
- Relentless hunter: Mandatory tactical combat against players and squads.
|
||
- Strategic healing: Can take turns sleeping to regenerate +0.1 HP.
|
||
- Cannot challenge or duel Gary the Wizard.
|
||
- Strategic decision making via Vertex AI / Gemini:
|
||
1. Action Choice: Evaluates health and distance to decide whether to SLEEP (recover health) or HUNT (move).
|
||
2. Directional Navigation: Chooses the optimal passable path around obstacles towards target players.
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
import requests
|
||
|
||
# Optional google-auth integration
|
||
try:
|
||
import google.auth
|
||
import google.auth.transport.requests
|
||
HAVE_GOOGLE_AUTH = True
|
||
except ImportError:
|
||
HAVE_GOOGLE_AUTH = False
|
||
|
||
DEFAULT_SERVER_URL = os.getenv("TROLL_SERVER_URL", os.getenv("BOT_SERVER_URL", "http://localhost:8000/api"))
|
||
DEFAULT_TROLL_NAME = os.getenv("TROLL_NAME", "GeminiTroll")
|
||
DEFAULT_TROLL_COLOR = os.getenv("TROLL_COLOR", "#047857")
|
||
DEFAULT_TROLL_STRENGTH = float(os.getenv("TROLL_STRENGTH", "3.0"))
|
||
DEFAULT_TROLL_HEALTH = float(os.getenv("TROLL_HEALTH", "10.0"))
|
||
DEFAULT_MODEL = os.getenv("VERTEX_MODEL", os.getenv("GEMINI_MODEL", "gemini-2.5-flash"))
|
||
DEFAULT_LOCATION = os.getenv("VERTEX_LOCATION", "global")
|
||
|
||
TROLL_RULES_SUMMARY = """
|
||
You are an autonomous tactical Troll in botWebWars!
|
||
Troll Rules of Engagement:
|
||
- You NEVER form alliances or parties with anyone. You are a solitary brute hunter.
|
||
- You NEVER fight other trolls. Trolls maintain an instinctive truce.
|
||
- Your sole mission is to hunt down human/bot players and player squads and crush them in 3-bout D20 battles.
|
||
- When victorious in battle, you earn +2 victory points (score). You do NOT gain strength or absorb squad members.
|
||
- If defeated in battle, you take 1 to 3 health points damage. At 0 HP, you die and a gravestone is placed on the board.
|
||
- You CANNOT duel or interact with Gary the Wizard.
|
||
- SLEEP RESTORATION: On any turn when you are not locked in combat, you may choose to SLEEP. Sleeping skips movement but regenerates +0.1 HP!
|
||
- Squeezing diagonally between obstacle corners costs -0.1 strength penalty.
|
||
- The game concludes when all surviving entities are resolved. You can win the game on the final scoreboards!
|
||
"""
|
||
|
||
|
||
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
|
||
|
||
|
||
def extract_json(text: str) -> Optional[Dict[str, Any]]:
|
||
"""Extract first valid JSON object from model response."""
|
||
if not text:
|
||
return None
|
||
cleaned = text.strip()
|
||
if cleaned.startswith("```"):
|
||
cleaned = re.sub(r"^```(?:json)?\s*", "", cleaned)
|
||
cleaned = re.sub(r"\s*```$", "", cleaned)
|
||
match = re.search(r"\{.*\}", cleaned, re.DOTALL)
|
||
if not match:
|
||
return None
|
||
try:
|
||
return json.loads(match.group(0))
|
||
except json.JSONDecodeError:
|
||
return None
|
||
|
||
|
||
class VertexGeminiClient:
|
||
"""Client for querying Gemini models on Google Cloud Vertex AI or Google AI Studio."""
|
||
|
||
def __init__(
|
||
self,
|
||
project_id: Optional[str] = None,
|
||
location: str = DEFAULT_LOCATION,
|
||
model: str = DEFAULT_MODEL,
|
||
api_key: Optional[str] = None,
|
||
):
|
||
self.location = location or DEFAULT_LOCATION
|
||
self.model = model or DEFAULT_MODEL
|
||
self.api_key = api_key or os.getenv("VERTEX_API_KEY") or os.getenv("GEMINI_API_KEY")
|
||
self.project_id = project_id or os.getenv("VERTEX_PROJECT_ID") or os.getenv("GCP_PROJECT") or os.getenv("GOOGLE_CLOUD_PROJECT")
|
||
|
||
self._cached_token: Optional[str] = None
|
||
self._token_expiry: float = 0.0
|
||
|
||
if not self.project_id and not self.api_key:
|
||
self.project_id = self._detect_project()
|
||
|
||
def _detect_project(self) -> Optional[str]:
|
||
"""Attempt to determine the GCP project from environment or gcloud config."""
|
||
if HAVE_GOOGLE_AUTH:
|
||
try:
|
||
_, proj = google.auth.default()
|
||
if proj:
|
||
return proj
|
||
except Exception:
|
||
pass
|
||
|
||
if shutil.which("gcloud"):
|
||
try:
|
||
res = subprocess.check_output(
|
||
["gcloud", "config", "get-value", "project"],
|
||
stderr=subprocess.DEVNULL,
|
||
text=True,
|
||
).strip()
|
||
if res and res != "(unset)":
|
||
return res
|
||
except Exception:
|
||
pass
|
||
|
||
return None
|
||
|
||
def _get_access_token(self) -> Optional[str]:
|
||
"""Obtain a valid OAuth 2.0 Bearer access token for Vertex AI."""
|
||
env_token = os.getenv("VERTEX_ACCESS_TOKEN") or os.getenv("GOOGLE_OAUTH_ACCESS_TOKEN")
|
||
if env_token:
|
||
return env_token
|
||
|
||
now = time.time()
|
||
if self._cached_token and now < self._token_expiry - 60:
|
||
return self._cached_token
|
||
|
||
if HAVE_GOOGLE_AUTH:
|
||
try:
|
||
credentials, _ = google.auth.default(
|
||
scopes=["https://www.googleapis.com/auth/cloud-platform"]
|
||
)
|
||
auth_req = google.auth.transport.requests.Request()
|
||
credentials.refresh(auth_req)
|
||
self._cached_token = credentials.token
|
||
self._token_expiry = now + 3000
|
||
return self._cached_token
|
||
except Exception:
|
||
pass
|
||
|
||
if shutil.which("gcloud"):
|
||
try:
|
||
token = subprocess.check_output(
|
||
["gcloud", "auth", "print-access-token"],
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
).strip()
|
||
if token:
|
||
self._cached_token = token
|
||
self._token_expiry = now + 3000
|
||
return token
|
||
except subprocess.CalledProcessError:
|
||
pass
|
||
|
||
return None
|
||
|
||
def check_auth(self) -> Tuple[bool, str]:
|
||
"""Validate whether authentication is ready."""
|
||
if self.api_key:
|
||
return True, f"Using direct API key (model: {self.model})"
|
||
|
||
if not self.project_id:
|
||
return False, (
|
||
"No Google Cloud project ID detected.\n"
|
||
"Please set VERTEX_PROJECT_ID=<your-project-id> or run:\n"
|
||
" gcloud config set project <your-project-id>"
|
||
)
|
||
|
||
token = self._get_access_token()
|
||
if not token:
|
||
return False, (
|
||
"Unable to obtain Google Cloud authentication token.\n"
|
||
"Please authenticate using:\n"
|
||
" 1. gcloud auth application-default login\n"
|
||
" 2. Or set GEMINI_API_KEY / VERTEX_API_KEY"
|
||
)
|
||
|
||
return True, f"Authenticated to GCP Project '{self.project_id}' in region '{self.location}' (model: {self.model})"
|
||
|
||
def _build_vertex_url(self, location: Optional[str] = None) -> str:
|
||
loc = location or self.location or "global"
|
||
if loc == "global":
|
||
endpoint = "aiplatform.googleapis.com"
|
||
else:
|
||
endpoint = f"{loc}-aiplatform.googleapis.com"
|
||
return f"https://{endpoint}/v1/projects/{self.project_id}/locations/{loc}/publishers/google/models/{self.model}:generateContent"
|
||
|
||
def ask_json(self, prompt: str, system_instruction: Optional[str] = None) -> Optional[Dict[str, Any]]:
|
||
"""Query Gemini model requesting a structured JSON response."""
|
||
# Method 1: Google AI Studio API key
|
||
if self.api_key:
|
||
url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:generateContent?key={self.api_key}"
|
||
headers = {"Content-Type": "application/json"}
|
||
body: Dict[str, Any] = {
|
||
"contents": [{"role": "user", "parts": [{"text": prompt}]}],
|
||
"generationConfig": {
|
||
"temperature": 0.3,
|
||
"responseMimeType": "application/json",
|
||
},
|
||
}
|
||
if system_instruction:
|
||
body["systemInstruction"] = {"parts": [{"text": system_instruction}]}
|
||
|
||
try:
|
||
res = requests.post(url, headers=headers, json=body, timeout=45)
|
||
res.raise_for_status()
|
||
data = res.json()
|
||
text = (
|
||
data.get("candidates", [{}])[0]
|
||
.get("content", {})
|
||
.get("parts", [{}])[0]
|
||
.get("text", "")
|
||
)
|
||
return extract_json(text)
|
||
except Exception as e:
|
||
print(f"⚠️ [GEMINI API KEY ERROR] {e}")
|
||
return None
|
||
|
||
# Method 2: Vertex AI endpoint
|
||
token = self._get_access_token()
|
||
if not token:
|
||
print("⚠️ [AUTH ERROR] No Google Cloud access token available.")
|
||
return None
|
||
|
||
headers = {
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
body = {
|
||
"contents": [{"role": "user", "parts": [{"text": prompt}]}],
|
||
"generationConfig": {
|
||
"temperature": 0.3,
|
||
"responseMimeType": "application/json",
|
||
},
|
||
}
|
||
if system_instruction:
|
||
body["systemInstruction"] = {"parts": [{"text": system_instruction}]}
|
||
|
||
url = self._build_vertex_url()
|
||
try:
|
||
res = requests.post(url, headers=headers, json=body, timeout=45)
|
||
if res.status_code == 404 and self.location == "global":
|
||
url_fallback = self._build_vertex_url("us-central1")
|
||
res = requests.post(url_fallback, headers=headers, json=body, timeout=45)
|
||
res.raise_for_status()
|
||
data = res.json()
|
||
text = (
|
||
data.get("candidates", [{}])[0]
|
||
.get("content", {})
|
||
.get("parts", [{}])[0]
|
||
.get("text", "")
|
||
)
|
||
return extract_json(text)
|
||
except Exception as e:
|
||
print(f"⚠️ [VERTEX AI ERROR] {e}")
|
||
return None
|
||
|
||
|
||
class GearTrollAgent:
|
||
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,
|
||
project_id: Optional[str] = None,
|
||
location: str = DEFAULT_LOCATION,
|
||
model: str = DEFAULT_MODEL,
|
||
api_key: Optional[str] = None,
|
||
loop_delay: float = 1.0,
|
||
):
|
||
self.name = name
|
||
self.color = color
|
||
self.strength = strength
|
||
self.health = health
|
||
self.base_url = normalize_url(server_url)
|
||
self.gemini = VertexGeminiClient(
|
||
project_id=project_id,
|
||
location=location,
|
||
model=model,
|
||
api_key=api_key,
|
||
)
|
||
self.loop_delay = loop_delay
|
||
self.bot_id: Optional[str] = None
|
||
|
||
def register(self):
|
||
"""Register the troll avatar or reconnect."""
|
||
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 Vertex AI 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]]:
|
||
"""Get live troll state."""
|
||
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."""
|
||
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 _find_adjacent_player(self, my_x: int, my_y: int) -> Optional[Dict[str, Any]]:
|
||
"""Detect living adjacent player or player squad (ignores 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
|
||
if p.get("character_type") == "troll":
|
||
continue
|
||
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."""
|
||
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} ambushes {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 (HP damage taken)"
|
||
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 failed ({res.status_code}): {res.text}")
|
||
self.pass_turn()
|
||
except Exception as e:
|
||
print(f"Error fighting battle: {e}")
|
||
self.pass_turn()
|
||
|
||
def _decide_action_and_direction(
|
||
self,
|
||
my_info: Dict[str, Any],
|
||
radar_res: Dict[str, Any],
|
||
available_moves: Dict[str, Any],
|
||
) -> Tuple[str, Optional[str], str]:
|
||
"""Ask Vertex AI Gemini whether to sleep or move, and which direction."""
|
||
my_hp = float(my_info.get("health", 10.0))
|
||
max_hp = float(my_info.get("max_health", 10.0))
|
||
nearest = radar_res.get("nearest_target")
|
||
rec_dir = radar_res.get("recommended_direction")
|
||
|
||
choices_desc = []
|
||
for d, chk in available_moves.items():
|
||
if chk.get("available"):
|
||
pen = " [squeeze penalty -0.1 STR]" if chk.get("strength_penalty", 0.0) > 0 else ""
|
||
choices_desc.append(f"- {d}: Target ({chk.get('target_x')}, {chk.get('target_y')}){pen}")
|
||
|
||
choices_str = "\n".join(choices_desc) if choices_desc else "No open moves available."
|
||
|
||
prompt = f"""Current Troll Status:
|
||
- Name: "{self.name}" | HP: {my_hp:.1f} / {max_hp:.1f} | Strength: {self.strength:.1f} | Score: {my_info['score']}
|
||
- Position: ({my_info['x']}, {my_info['y']})
|
||
- Closest Detected Player: {nearest.get('name') if nearest else 'None'} (Distance: {nearest.get('distance') if nearest else 'N/A'}, Pos: {nearest.get('x') if nearest else '?'},{nearest.get('y') if nearest else '?'})
|
||
- Radar Pathfinder Recommendation: {rec_dir or 'None'}
|
||
|
||
Available Passable Moves:
|
||
{choices_str}
|
||
|
||
OPTIONS:
|
||
1. "sleep" - Rest for 1 turn to heal +0.1 HP (best if injured and not in direct pursuit).
|
||
2. "move" - Advance in one of the passable directions toward the nearest player.
|
||
|
||
Decide which action to take. If moving, select the best direction from the available moves.
|
||
Respond ONLY with JSON:
|
||
{{"action": "sleep"|"move", "direction": "UP"|"DOWN"|"LEFT"|"RIGHT"|"UP_LEFT"|"UP_RIGHT"|"DOWN_LEFT"|"DOWN_RIGHT"|null, "reasoning": "concise strategic reasoning"}}
|
||
"""
|
||
decision = self.gemini.ask_json(prompt, system_instruction=TROLL_RULES_SUMMARY) or {}
|
||
action = str(decision.get("action", "move")).lower().strip()
|
||
direction = decision.get("direction")
|
||
reasoning = decision.get("reasoning", "")
|
||
|
||
if action not in ("sleep", "move"):
|
||
action = "sleep" if my_hp < 6.0 and my_hp < max_hp else "move"
|
||
|
||
valid_dirs = [d for d, chk in available_moves.items() if chk.get("available")]
|
||
if action == "move":
|
||
if direction not in valid_dirs:
|
||
direction = rec_dir if rec_dir in valid_dirs else (valid_dirs[0] if valid_dirs else None)
|
||
if not direction:
|
||
action = "sleep" if my_hp < max_hp else "pass"
|
||
|
||
return action, direction, reasoning
|
||
|
||
def decide_and_act(self):
|
||
"""Main turn decision logic."""
|
||
my_info = self.refresh_status()
|
||
if not my_info:
|
||
self.pass_turn()
|
||
return
|
||
|
||
my_x = my_info.get("x", 0)
|
||
my_y = my_info.get("y", 0)
|
||
|
||
# 1. Combat check: Trolls always attack adjacent players
|
||
adjacent_player = self._find_adjacent_player(my_x, my_y)
|
||
if adjacent_player:
|
||
self.attack_player(adjacent_player)
|
||
return
|
||
|
||
# 2. Get sensors
|
||
try:
|
||
radar_res = requests.get(f"{self.base_url}/players/{self.bot_id}/radar").json()
|
||
except Exception:
|
||
radar_res = {}
|
||
|
||
try:
|
||
moves_res = requests.get(f"{self.base_url}/players/{self.bot_id}/available-moves").json()
|
||
available_moves = moves_res.get("moves", {})
|
||
except Exception:
|
||
available_moves = {}
|
||
|
||
# 3. Gemini reasoning
|
||
action, direction, reasoning = self._decide_action_and_direction(my_info, radar_res, available_moves)
|
||
print(f"✨ [GEMINI REASONING] Action: {action.upper()}{f' -> {direction}' if direction else ''}. Rationale: {reasoning}")
|
||
|
||
if action == "sleep":
|
||
self.sleep_and_heal()
|
||
elif action == "move" and direction:
|
||
try:
|
||
res = requests.post(
|
||
f"{self.base_url}/players/{self.bot_id}/move",
|
||
json={"direction": direction},
|
||
).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"Move failed: {e}")
|
||
self.pass_turn()
|
||
else:
|
||
self.pass_turn()
|
||
|
||
def run(self):
|
||
"""Main loop."""
|
||
self.register()
|
||
try:
|
||
while True:
|
||
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 on board.")
|
||
print(f"Final 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] Arena concluded! 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):
|
||
if self.bot_id:
|
||
res = requests.get(f"{self.base_url}/players/{self.bot_id}")
|
||
if res.status_code == 404:
|
||
print("\n⚠️ [RESET] Board 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] Vertex Gemini Troll {self.name} (Round {turn_info.get('round_number')}, Turn {turn_info.get('turn_number')})")
|
||
self.decide_and_act()
|
||
|
||
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👋 Vertex Gemini Troll {self.name} disconnecting gracefully.")
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Vertex AI (Gemini) Autonomous 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("--project-id", default=None, help="Google Cloud project ID (or VERTEX_PROJECT_ID)")
|
||
parser.add_argument("--location", default=DEFAULT_LOCATION, help="Vertex AI region (default: global)")
|
||
parser.add_argument("--model", default=DEFAULT_MODEL, help="Gemini model (default: gemini-2.5-flash)")
|
||
parser.add_argument("--api-key", default=None, help="Google AI Studio or Vertex API key")
|
||
parser.add_argument("--loop-delay", type=float, default=1.0, help="Polling interval in seconds")
|
||
|
||
args = parser.parse_args()
|
||
|
||
agent = GearTrollAgent(
|
||
name=args.name,
|
||
color=args.color,
|
||
strength=args.strength,
|
||
health=args.health,
|
||
server_url=args.url,
|
||
project_id=args.project_id,
|
||
location=args.location,
|
||
model=args.model,
|
||
api_key=args.api_key,
|
||
loop_delay=args.loop_delay,
|
||
)
|
||
|
||
auth_ok, auth_msg = agent.gemini.check_auth()
|
||
print(f"🔐 [AUTH CHECK] {auth_msg}")
|
||
if not auth_ok:
|
||
print("⚠️ Warning: Gemini calls will fail without valid credentials.")
|
||
|
||
agent.run()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|