diff options
Diffstat (limited to 'app')
| -rw-r--r-- | app/__init__.py | 0 | ||||
| -rw-r--r-- | app/bot.py | 137 | ||||
| -rw-r--r-- | app/main.py | 171 |
3 files changed, 308 insertions, 0 deletions
diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/app/__init__.py diff --git a/app/bot.py b/app/bot.py new file mode 100644 index 0000000..c289c85 --- /dev/null +++ b/app/bot.py @@ -0,0 +1,137 @@ +import logging +import random + +from codenames.v1.types_pb2 import Role, Team +from codenames.v1 import types_pb2, bot_pb2 + +logger = logging.getLogger(__name__) + + +class CodenamesBot: + """ + A bot that can play Codenames as either a spymaster or a guesser. + """ + + def __init__(self): + self.game_id: str = "" + self.my_team: Team = types_pb2.TEAM_UNSPECIFIED + self.my_role: Role = types_pb2.ROLE_UNSPECIFIED + self.words: list[str] = [] + + def handle_game_started( + self, request: bot_pb2.GameStartedRequest + ) -> bot_pb2.GameStartedResponse: + """ + Handle GameStarted request from the server. + + Initialize local game state from the board. + """ + if len(self.words) != 0: + logger.warning( + f"New game started while {self.game_id} was active, abandoning old game" + ) + + self.words = [card.word for card in request.board.cards] + + # Set the game state + self.game_id = request.game_id + self.my_team = request.your_team + self.my_role = request.your_role + + logger.info( + f"Game started: {self.game_id} for bot {request.bot_id}, team={self.my_team}, role={self.my_role}" + ) + + return bot_pb2.GameStartedResponse() + + def handle_give_clue( + self, _request: bot_pb2.GiveClueRequest + ) -> bot_pb2.GiveClueResponse: + """Handle GiveClue request as spymaster.""" + if len(self.words) == 0: + raise ValueError("No active game") + + # Get clue from strategy + word, count = self.get_clue() + + logger.info(f"Spymaster giving clue: {word} ({count})") + + response = bot_pb2.GiveClueResponse() + response.give_clue.word = word + response.give_clue.count = count + return response + + def get_clue(self) -> tuple[str, int]: + """Get a clue as the Spymaster.""" + CLUE_WORDS = [ + "think", + "hint", + "link", + "match", + "group", + "connect", + "relate", + "bridge", + "bond", + "chain", + "random", + "guess", + "word", + "game", + "play", + "team", + "card", + "board", + "code", + "name", + ] + + word = random.choice(CLUE_WORDS) + count = random.randint(1, 3) + return word, count + + def handle_make_guess( + self, _request: bot_pb2.MakeGuessRequest + ) -> bot_pb2.MakeGuessResponse: + """Handle MakeGuess request as guesser.""" + if len(self.words) == 0: + raise ValueError("No active game") + + # Get guess from strategy + word, should_guess = self.get_guess() + + response = bot_pb2.MakeGuessResponse() + + if should_guess and word: + logger.info(f"Guesser making guess: {word}") + response.guess_word.word = word + else: + logger.info("Guesser ending turn") + response.end_turn.SetInParent() + + return response + + def get_guess(self) -> tuple[str, bool]: + return random.choice(self.words), True + + def handle_game_ended( + self, request: bot_pb2.GameEndedRequest + ) -> bot_pb2.GameEndedResponse: + """ + Handle GameEnded request. + + Clean up game state after game completion. + """ + game_id = request.game_id + winner = request.winner + + result = "won" if winner == self.my_team else "lost" + logger.info(f"Game ended: {game_id}, result={result}") + + # Clear the game state + self.game_id = "" + self.words = [] + self.my_team = types_pb2.TEAM_UNSPECIFIED + self.my_role = types_pb2.ROLE_UNSPECIFIED + + return bot_pb2.GameEndedResponse() diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..b8e0fa3 --- /dev/null +++ b/app/main.py @@ -0,0 +1,171 @@ +import logging +import os +import uuid +from typing import Any + +from connectrpc.request import RequestContext +from codenames.v1.bot_connect import BotService, BotServiceASGIApplication +from codenames.v1.game_connect import GameServiceClient +from codenames.v1 import bot_pb2, game_pb2 + +from app.bot import CodenamesBot + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", +) +logger = logging.getLogger(__name__) + +if "PORT" not in os.environ: + raise RuntimeError("PORT environment variable must be set") +PORT = int(os.environ["PORT"]) +LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO") +IDENTITY_KEY = os.environ.get("IDENTITY_KEY", f"python-bot-{uuid.uuid4().hex[:8]}") +GAME_SERVER_URL = os.environ.get("GAME_SERVER_URL", "https://hackathon.evroc.dev/api") +TEAM_TOKEN_PATH = os.environ.get("TEAM_TOKEN_PATH", "/etc/codenames/game_key") + +logging.getLogger().setLevel(getattr(logging, LOG_LEVEL.upper())) +logging.getLogger("app").setLevel(getattr(logging, LOG_LEVEL.upper())) + +bot = CodenamesBot() + +logger.info(f"Identity Key: {IDENTITY_KEY}") + + +class BotServiceHandler(BotService): + async def game_started( + self, request: bot_pb2.GameStartedRequest, ctx: RequestContext + ) -> bot_pb2.GameStartedResponse: + return bot.handle_game_started(request) + + async def give_clue( + self, request: bot_pb2.GiveClueRequest, ctx: RequestContext + ) -> bot_pb2.GiveClueResponse: + return bot.handle_give_clue(request) + + async def make_guess( + self, request: bot_pb2.MakeGuessRequest, ctx: RequestContext + ) -> bot_pb2.MakeGuessResponse: + return bot.handle_make_guess(request) + + async def game_ended( + self, request: bot_pb2.GameEndedRequest, ctx: RequestContext + ) -> bot_pb2.GameEndedResponse: + return bot.handle_game_ended(request) + + +class LifespanASGIWrapper: + def __init__(self, connect_app: Any) -> None: + self.connect_app = connect_app + self.registered_bot_id: str | None = None + + async def register_bot(self) -> None: + import asyncio + + start_time = asyncio.get_event_loop().time() + max_duration = 30 # seconds + base_delay = 1 # seconds + + while True: + try: + if not os.path.exists(TEAM_TOKEN_PATH): + raise RuntimeError( + f"Team token not found at {TEAM_TOKEN_PATH}" + ) + + with open(TEAM_TOKEN_PATH) as f: + token = f.read().strip() + + if not token: + raise RuntimeError( + f"Team token file at {TEAM_TOKEN_PATH} is empty" + ) + + logger.info(f"Registering bot with identity_key={IDENTITY_KEY}") + + client = GameServiceClient(GAME_SERVER_URL) + request = game_pb2.RegisterBotRequest( + identity_key=IDENTITY_KEY, + port=PORT, + ) + + response = await client.register_bot( + request, + headers={"Authorization": f"Bearer {token}"}, + ) + self.registered_bot_id = response.bot.bot_id + logger.info(f"Bot registered successfully: {self.registered_bot_id}") + return + except Exception as e: + elapsed = asyncio.get_event_loop().time() - start_time + if elapsed >= max_duration: + raise RuntimeError( + f"Failed to register bot after {max_duration}s: {e}" + ) from e + + delay = min(base_delay * (2 ** int(elapsed / 5)), 5) # Exponential backoff, max 5s + logger.warning( + f"Registration failed (retrying in {delay}s): {e}" + ) + await asyncio.sleep(delay) + + async def unregister_bot(self) -> None: + try: + if not self.registered_bot_id: + logger.debug("No registered bot ID, skipping unregistration") + return + + if not os.path.exists(TEAM_TOKEN_PATH): + logger.warning( + f"Team token not found at {TEAM_TOKEN_PATH}, cannot unregister" + ) + return + + with open(TEAM_TOKEN_PATH) as f: + token = f.read().strip() + + if not token: + logger.warning( + f"Team token file at {TEAM_TOKEN_PATH} is empty, cannot unregister" + ) + return + + logger.info(f"Unregistering bot {self.registered_bot_id}") + + client = GameServiceClient(GAME_SERVER_URL) + request = game_pb2.UnregisterBotRequest( + bot_id=self.registered_bot_id, + ) + + await client.unregister_bot( + request, + headers={"Authorization": f"Bearer {token}"}, + ) + logger.info("Bot unregistered successfully") + except Exception as e: + logger.error(f"Failed to unregister bot: {e}") + + async def __call__( + self, scope: dict[str, Any], receive: Any, send: Any + ) -> None: + if scope.get("type") == "lifespan": + await self._handle_lifespan(scope, receive, send) + else: + await self.connect_app(scope, receive, send) + + async def _handle_lifespan( + self, scope: dict[str, Any], receive: Any, send: Any + ) -> None: + while True: + message = await receive() + if message["type"] == "lifespan.startup": + await self.register_bot() + await send({"type": "lifespan.startup.complete"}) + elif message["type"] == "lifespan.shutdown": + await self.unregister_bot() + await send({"type": "lifespan.shutdown.complete"}) + return + + +connect_app = BotServiceASGIApplication(BotServiceHandler()) +app = LifespanASGIWrapper(connect_app) |