From 0a9145728eb097c8ad75f422be4388f406afbc89 Mon Sep 17 00:00:00 2001 From: koko210Serve Date: Fri, 30 Jan 2026 21:43:20 +0200 Subject: [PATCH] Ability to play Uno implemented in early stages! --- bot/Dockerfile | 4 +- bot/bot.py | 6 + bot/commands/uno.py | 195 ++++++++++++++++ bot/setup_uno_playwright.sh | 34 +++ bot/utils/logger.py | 1 + bot/utils/uno_game.py | 448 ++++++++++++++++++++++++++++++++++++ 6 files changed, 687 insertions(+), 1 deletion(-) create mode 100644 bot/commands/uno.py create mode 100755 bot/setup_uno_playwright.sh create mode 100644 bot/utils/uno_game.py diff --git a/bot/Dockerfile b/bot/Dockerfile index 5374fd9..1070ad9 100644 --- a/bot/Dockerfile +++ b/bot/Dockerfile @@ -4,7 +4,6 @@ WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt -RUN playwright install # Install system dependencies # ffmpeg: video/audio processing for media handling @@ -21,6 +20,9 @@ RUN apt-get update && apt-get install -y \ && apt-get clean \ && rm -rf /var/lib/apt/lists/* +# Install Playwright browsers with system dependencies (for UNO automation) +RUN playwright install --with-deps chromium + # Install Docker CLI and docker compose plugin so the bot can build/create the face detector container RUN set -eux; \ curl -fsSL https://download.docker.com/linux/debian/gpg | gpg --dearmor -o /usr/share/keyrings/docker-archive-keyring.gpg; \ diff --git a/bot/bot.py b/bot/bot.py index 775e613..e853e61 100644 --- a/bot/bot.py +++ b/bot/bot.py @@ -144,6 +144,12 @@ async def on_message(message): await handle_voice_command(message, cmd, args) return + # Check for UNO commands (!uno create, !uno join, !uno list, !uno quit, !uno help) + if message.content.strip().lower().startswith('!uno'): + from commands.uno import handle_uno_command + await handle_uno_command(message) + return + # Block all text responses when voice session is active if globals.VOICE_SESSION_ACTIVE: # Queue the message for later processing (optional) diff --git a/bot/commands/uno.py b/bot/commands/uno.py new file mode 100644 index 0000000..e2ec97b --- /dev/null +++ b/bot/commands/uno.py @@ -0,0 +1,195 @@ +""" +UNO Game Commands for Miku +Allows Miku to play UNO games via Discord +""" +import discord +import asyncio +import requests +import json +import logging +from typing import Optional, Dict, Any +from utils.logger import get_logger + +logger = get_logger('uno') + +# UNO game server configuration (use host IP from container) +UNO_SERVER_URL = "http://192.168.1.2:5000" +UNO_CLIENT_URL = "http://192.168.1.2:3002" + +# Active games tracking +active_uno_games: Dict[str, Dict[str, Any]] = {} + + +async def join_uno_game(message: discord.Message, room_code: str): + """ + Miku joins an UNO game as Player 2 + Usage: !uno join + """ + if not room_code: + await message.channel.send("🎴 Please provide a room code! Usage: `!uno join `") + return + + room_code = room_code.strip() # Keep exact case - don't convert to uppercase! + + # Check if already in a game + if room_code in active_uno_games: + await message.channel.send(f"🎴 I'm already playing in room **{room_code}**! Let me finish this game first~ 🎶") + return + + await message.channel.send(f"🎤 Joining UNO game **{room_code}** as Player 2! Time to show you how it's done! ✨") + + try: + # Import here to avoid circular imports + from utils.uno_game import MikuUnoPlayer + + # Define cleanup callback to remove from active games + async def cleanup_game(code: str): + if code in active_uno_games: + logger.info(f"[UNO] Removing room {code} from active games") + del active_uno_games[code] + + # Create Miku's player instance with cleanup callback + player = MikuUnoPlayer(room_code, message.channel, cleanup_callback=cleanup_game) + + # Join the game (this will open browser and join) + success = await player.join_game() + + if success: + active_uno_games[room_code] = { + 'player': player, + 'channel': message.channel, + 'started_by': message.author.id + } + + await message.channel.send(f"✅ Joined room **{room_code}**! Waiting for Player 1 to start the game... 🎮") + + # Start the game loop + asyncio.create_task(player.play_game()) + else: + await message.channel.send(f"❌ Couldn't join room **{room_code}**. Make sure the room exists and has space!") + + except Exception as e: + logger.error(f"Error joining UNO game: {e}", exc_info=True) + await message.channel.send(f"❌ Oops! Something went wrong: {str(e)}") + + +async def list_uno_games(message: discord.Message): + """ + List active UNO games Miku is in + Usage: !uno list + """ + if not active_uno_games: + await message.channel.send("🎴 I'm not in any UNO games right now! Create a room and use `!uno join ` to make me play! 🎤") + return + + embed = discord.Embed( + title="🎴 Active UNO Games", + description="Here are the games I'm currently playing:", + color=discord.Color.blue() + ) + + for room_code, game_info in active_uno_games.items(): + player = game_info['player'] + status = "🎮 Playing" if player.is_game_active() else "⏸️ Waiting" + embed.add_field( + name=f"Room: {room_code}", + value=f"Status: {status}\nChannel: <#{game_info['channel'].id}>", + inline=False + ) + + await message.channel.send(embed=embed) + + +async def quit_uno_game(message: discord.Message, room_code: Optional[str] = None): + """ + Miku quits an UNO game + Usage: !uno quit [room_code] + """ + if not room_code: + # Quit all games + if not active_uno_games: + await message.channel.send("🎴 I'm not in any games right now!") + return + + for code, game_info in list(active_uno_games.items()): + await game_info['player'].quit_game() + del active_uno_games[code] + + await message.channel.send("👋 I quit all my UNO games! See you next time~ 🎶") + return + + room_code = room_code.strip() # Keep exact case + + if room_code not in active_uno_games: + await message.channel.send(f"🤔 I'm not in room **{room_code}**!") + return + + game_info = active_uno_games[room_code] + await game_info['player'].quit_game() + del active_uno_games[room_code] + + await message.channel.send(f"👋 I left room **{room_code}**! That was fun~ 🎤") + + +async def handle_uno_command(message: discord.Message): + """ + Main UNO command router + Usage: !uno [args] + + Subcommands: + !uno join - Join an existing game as Player 2 + !uno list - List active games + !uno quit [code] - Quit a game (or all games) + !uno help - Show this help + """ + content = message.content.strip() + parts = content.split() + + if len(parts) == 1: + # Just !uno + await show_uno_help(message) + return + + subcommand = parts[1].lower() + + if subcommand == "join": + if len(parts) < 3: + await message.channel.send("❌ Please provide a room code! Usage: `!uno join `") + return + await join_uno_game(message, parts[2]) + + elif subcommand == "list": + await list_uno_games(message) + + elif subcommand == "quit" or subcommand == "leave": + room_code = parts[2] if len(parts) > 2 else None + await quit_uno_game(message, room_code) + + elif subcommand == "help": + await show_uno_help(message) + + else: + await message.channel.send(f"❌ Unknown command: `{subcommand}`. Use `!uno help` to see available commands!") + + +async def show_uno_help(message: discord.Message): + """Show UNO command help""" + embed = discord.Embed( + title="🎴 Miku's UNO Commands", + description="Play UNO with me! I'll join as Player 2 and use my AI to make strategic moves~ 🎤✨\n\n**How to play:**\n1. Create a room at http://192.168.1.2:3002\n2. Copy the room code\n3. Use `!uno join ` to make me join!\n4. I'll play automatically and trash talk in chat! 🎶", + color=discord.Color.green() + ) + + commands = [ + ("!uno join ", "Make me join your UNO game as Player 2"), + ("!uno list", "List all active games I'm playing"), + ("!uno quit [CODE]", "Make me quit a game (or all games if no code)"), + ("!uno help", "Show this help message"), + ] + + for cmd, desc in commands: + embed.add_field(name=cmd, value=desc, inline=False) + + embed.set_footer(text="I'll trash talk and celebrate in chat during games! 🎶") + + await message.channel.send(embed=embed) diff --git a/bot/setup_uno_playwright.sh b/bot/setup_uno_playwright.sh new file mode 100755 index 0000000..890c249 --- /dev/null +++ b/bot/setup_uno_playwright.sh @@ -0,0 +1,34 @@ +#!/bin/bash +# setup_uno_playwright.sh +# Sets up Playwright browsers for UNO bot automation + +echo "🎮 Setting up Playwright for Miku UNO Bot..." +echo "" + +# Check if we're in the bot directory +if [ ! -f "bot.py" ]; then + echo "❌ Error: Please run this script from the bot directory" + echo " cd /home/koko210Serve/docker/miku-discord/bot" + exit 1 +fi + +# Install Playwright browsers +echo "📦 Installing Playwright browsers..." +python -m playwright install chromium + +if [ $? -eq 0 ]; then + echo "✅ Playwright browsers installed successfully!" + echo "" + echo "🎮 You can now use the UNO commands:" + echo " !uno create - Create a new game" + echo " !uno join CODE - Join an existing game" + echo " !uno list - List active games" + echo " !uno quit CODE - Quit a game" + echo " !uno help - Show help" + echo "" + echo "📚 See UNO_BOT_SETUP.md for more details" +else + echo "❌ Failed to install Playwright browsers" + echo " Try running manually: python -m playwright install chromium" + exit 1 +fi diff --git a/bot/utils/logger.py b/bot/utils/logger.py index d37ce6a..e7d56c9 100644 --- a/bot/utils/logger.py +++ b/bot/utils/logger.py @@ -64,6 +64,7 @@ COMPONENTS = { 'voice_audio': 'Voice audio streaming and TTS', 'container_manager': 'Docker container lifecycle management', 'error_handler': 'Error detection and webhook notifications', + 'uno': 'UNO game automation and commands', } # Global configuration diff --git a/bot/utils/uno_game.py b/bot/utils/uno_game.py new file mode 100644 index 0000000..2165d1d --- /dev/null +++ b/bot/utils/uno_game.py @@ -0,0 +1,448 @@ +""" +Miku UNO Player - Browser automation and AI strategy +Handles joining games via Playwright and making LLM-powered decisions +""" +import asyncio +import json +import requests +from typing import Optional, Dict, Any, List +from playwright.async_api import async_playwright, Page, Browser +from utils.llm import query_llama +from utils.logger import get_logger +import globals + +logger = get_logger('uno') + +# Configuration +# Use host.docker.internal to reach host machine from inside container +# Fallback to 192.168.1.2 if host.docker.internal doesn't work +UNO_SERVER_URL = "http://192.168.1.2:5000" +UNO_CLIENT_URL = "http://192.168.1.2:3002" +POLL_INTERVAL = 2 # seconds between checking for turn + + +class MikuUnoPlayer: + """Miku's UNO player with browser automation and AI strategy""" + + def __init__(self, room_code: str, discord_channel, cleanup_callback=None): + self.room_code = room_code + self.discord_channel = discord_channel + self.browser: Optional[Browser] = None + self.page: Optional[Page] = None + self.playwright = None + self.is_playing = False + self.game_started = False + self.last_card_count = 7 + self.last_turn_processed = None # Track last turn we processed to avoid duplicate moves + self.cleanup_callback = cleanup_callback # Callback to remove from active_uno_games + + async def join_game(self) -> bool: + """Join an existing UNO game as Player 2 via browser automation""" + try: + logger.info(f"[UNO] Joining game: {self.room_code}") + + # Launch browser + self.playwright = await async_playwright().start() + self.browser = await self.playwright.chromium.launch(headless=True) + self.page = await self.browser.new_page() + + # Enable console logging to debug (filter out verbose game state logs) + def log_console(msg): + text = msg.text + # Skip verbose game state logs but keep important ones + if "FULL GAME STATE" in text or "JSON for Bot API" in text: + return + logger.debug(f"[Browser] {text[:150]}...") # Truncate to 150 chars + + self.page.on("console", log_console) + self.page.on("pageerror", lambda err: logger.error(f"[Browser Error] {err}")) + + # Navigate to homepage + logger.info(f"[UNO] Navigating to: {UNO_CLIENT_URL}") + await self.page.goto(UNO_CLIENT_URL) + await asyncio.sleep(2) + + # Find and fill the room code input + try: + # Look for input field and fill with room code + input_field = await self.page.query_selector('input[type="text"]') + if not input_field: + logger.error("[UNO] Could not find input field") + return False + + await input_field.fill(self.room_code) + logger.info(f"[UNO] Filled room code: {self.room_code}") + await asyncio.sleep(0.5) + + # Click the "Join Room" button + buttons = await self.page.query_selector_all('button') + join_clicked = False + for button in buttons: + text = await button.inner_text() + if 'JOIN' in text.upper(): + logger.info(f"[UNO] Found join button, clicking...") + await button.click() + join_clicked = True + break + + if not join_clicked: + logger.error("[UNO] Could not find join button") + return False + + # Wait for navigation to /play + logger.info("[UNO] Waiting for navigation to game page...") + await asyncio.sleep(3) + + # Verify we're on the play page + current_url = self.page.url + logger.info(f"[UNO] Current URL after click: {current_url}") + + if '/play' not in current_url: + logger.error(f"[UNO] Did not navigate to game page, still on: {current_url}") + return False + + # Wait longer for Socket.IO connection and game setup + logger.info("[UNO] Waiting for Socket.IO connection and game initialization...") + await asyncio.sleep(5) + + # Take a screenshot for debugging + try: + screenshot_path = f"/app/memory/uno_debug_{self.room_code}.png" + await self.page.screenshot(path=screenshot_path) + logger.info(f"[UNO] Screenshot saved to {screenshot_path}") + except Exception as e: + logger.error(f"[UNO] Could not save screenshot: {e}") + + # Get page content for debugging + content = await self.page.content() + logger.debug(f"[UNO] Page content length: {len(content)} chars") + + # Check current URL + current_url = self.page.url + logger.info(f"[UNO] Current URL: {current_url}") + + # Check if we're actually in the game by looking for game elements + game_element = await self.page.query_selector('.game-screen, .player-deck, .uno-card') + if game_element: + logger.info(f"[UNO] Successfully joined room {self.room_code} as Player 2 - game elements found") + else: + logger.warning(f"[UNO] Joined room {self.room_code} but game elements not found yet") + + return True + + except Exception as e: + logger.error(f"[UNO] Error during join process: {e}", exc_info=True) + return False + + except Exception as e: + logger.error(f"[UNO] Error joining game: {e}", exc_info=True) + await self.cleanup() + return False + + async def play_game(self): + """Main game loop - poll for turns and make moves""" + self.is_playing = True + logger.info(f"Starting game loop for room {self.room_code}") + + try: + while self.is_playing: + # Get current game state + game_state = await self.get_game_state() + + if not game_state: + await asyncio.sleep(POLL_INTERVAL) + continue + + # Check if game started + if not self.game_started and game_state['game'].get('currentTurn'): + self.game_started = True + await self.discord_channel.send("🎮 Game started! Let's do this! 🎤✨") + + # Check if game over + if is_over: + # Game has ended + winner = game_state.get('game', {}).get('winner') + if winner == 2: + await self.discord_channel.send(f"🎉 **I WON!** That was too easy! GG! 🎤✨") + else: + await self.discord_channel.send(f"😤 You got lucky this time... I'll win next time! 💢") + + logger.info(f"[UNO] Game over in room {self.room_code}. Winner: Player {winner}") + + # Call cleanup callback to remove from active_uno_games + if self.cleanup_callback: + await self.cleanup_callback(self.room_code) + + break + + # Check if it's Miku's turn + if game_state['game']['currentTurn'] == 'Player 2': + # Create a unique turn identifier combining multiple factors + # This handles cases where bot's turn comes twice in a row (after Skip, etc) + turn_id = f"{game_state['game']['turnNumber']}_{game_state['player2']['cardCount']}_{len(game_state['currentCard'])}" + + if turn_id != self.last_turn_processed: + logger.info("It's Miku's turn!") + self.last_turn_processed = turn_id + await self.make_move(game_state) + else: + # Same turn state, but check if it's been more than 5 seconds (might be stuck) + # For now just skip to avoid duplicate moves + pass + + # Wait before next check + await asyncio.sleep(POLL_INTERVAL) + + except Exception as e: + logger.error(f"Error in game loop: {e}", exc_info=True) + await self.discord_channel.send(f"❌ Oops! Something went wrong in the game: {str(e)}") + finally: + await self.cleanup() + + async def get_game_state(self) -> Optional[Dict[str, Any]]: + """Get current game state from server""" + try: + response = requests.get( + f"{UNO_SERVER_URL}/api/game/{self.room_code}/state", + timeout=5 + ) + + if response.status_code == 200: + data = response.json() + if data.get('success'): + return data['gameState'] + + return None + + except Exception as e: + logger.error(f"Error getting game state: {e}") + return None + + async def make_move(self, game_state: Dict[str, Any]): + """Use LLM to decide and execute a move""" + try: + # Check if bot can play any cards + can_play = len(game_state['player2']['playableCards']) > 0 + + # Get Miku's decision from LLM + action = await self.get_miku_decision(game_state) + + if not action: + logger.warning("No action from LLM, drawing card") + action = {"action": "draw"} + + logger.info(f"🎮 Miku's decision: {json.dumps(action)}") + + # Send trash talk before move + await self.send_trash_talk(game_state, action) + + # Execute the action + success = await self.send_action(action) + + if success: + # Check for UNO situation + current_cards = game_state['player2']['cardCount'] + if action['action'] == 'play' and current_cards == 2: + await self.discord_channel.send("🔥 **UNO!!** One more card and I win! 🎤") + + logger.info(f"✅ Action executed successfully") + + # Reset turn tracker after successful action so we can process next turn + self.last_turn_processed = None + + # Brief wait for socket sync (now that useEffect dependencies are fixed, this can be much shorter) + await asyncio.sleep(0.5) + + else: + logger.warning(f"⚠️ Action failed (invalid move), will try different action next turn") + # Don't reset turn tracker - let it skip this turn state + # The game state will update and we'll try again with updated info + + except Exception as e: + logger.error(f"Error making move: {e}", exc_info=True) + + async def get_miku_decision(self, game_state: Dict[str, Any]) -> Optional[Dict[str, Any]]: + """Use Miku's LLM to decide the best move""" + try: + # Build strategic prompt + prompt = self.build_strategy_prompt(game_state) + + # Query LLM with required parameters (query_llama is already async) + guild_id = self.discord_channel.guild.id if hasattr(self.discord_channel, 'guild') and self.discord_channel.guild else None + response = await query_llama( + user_prompt=prompt, + user_id="uno_bot", + guild_id=guild_id, + response_type="uno_strategy", + author_name="Miku UNO Bot" + ) + + # Extract JSON from response + action = self.parse_llm_response(response) + + return action + + except Exception as e: + logger.error(f"Error getting LLM decision: {e}", exc_info=True) + return None + + def build_strategy_prompt(self, game_state: Dict[str, Any]) -> str: + """Build a prompt for Miku to make strategic decisions""" + current_card = game_state['currentCard'] + my_cards = game_state['player2']['cards'] + playable_cards = game_state['player2']['playableCards'] + opponent_cards = game_state['player1']['cardCount'] + my_card_count = game_state['player2']['cardCount'] + + # Build card list + my_cards_str = ", ".join([f"{c['displayName']} ({c['code']})" for c in my_cards]) + playable_str = ", ".join([f"{c['displayName']} ({c['code']})" for c in playable_cards]) + + prompt = f"""You are Hatsune Miku, the cheerful virtual idol! You're playing UNO and it's your turn. + +GAME STATE: +- Current card on table: {current_card['displayName']} ({current_card['code']}) +- Your cards ({my_card_count}): {my_cards_str} +- Playable cards: {playable_str if playable_str else "NONE - must draw"} +- Opponent has {opponent_cards} cards + +STRATEGY: +- If opponent has 1-2 cards, play attack cards (Draw 2, Draw 4, Skip) to stop them! +- Play Draw 2/Draw 4 aggressively to disrupt opponent +- Save Wild cards for when you have no other options +- When playing Wild cards, choose the color you have most of +- Call UNO when you have 2 cards and are about to play one + +YOUR TASK: +Respond with ONLY a valid JSON action. No explanation, just the JSON. + +ACTION FORMAT: +1. To play a card: {{"action": "play", "card": "CODE"}} +2. To play a Wild: {{"action": "play", "card": "W", "color": "R/G/B/Y"}} +3. To play Wild Draw 4: {{"action": "play", "card": "D4W", "color": "R/G/B/Y"}} +4. To draw a card: {{"action": "draw"}} +5. To play + call UNO: {{"action": "play", "card": "CODE", "callUno": true}} + +VALID CARD CODES: +{playable_str if playable_str else "No playable cards - must draw"} + +Choose wisely! What's your move? + +RESPONSE (JSON only):""" + + return prompt + + def parse_llm_response(self, response: str) -> Optional[Dict[str, Any]]: + """Parse LLM response to extract JSON action""" + try: + # Try to find JSON in response + import re + + # Look for JSON object + json_match = re.search(r'\{[^}]+\}', response) + if json_match: + json_str = json_match.group(0) + action = json.loads(json_str) + + # Validate action format + if 'action' in action: + return action + + logger.warning(f"Could not parse LLM response: {response}") + return None + + except Exception as e: + logger.error(f"Error parsing LLM response: {e}") + return None + + async def send_trash_talk(self, game_state: Dict[str, Any], action: Dict[str, Any]): + """Send personality-driven trash talk before moves""" + try: + opponent_cards = game_state['player1']['cardCount'] + my_cards = game_state['player2']['cardCount'] + + # Special trash talk for different situations + if action['action'] == 'play': + card_code = action.get('card', '') + + if 'D4W' in card_code: + messages = [ + "Wild Draw 4! Take that! 😈", + "Draw 4 cards! Ahahaha! 🌈💥", + "This is what happens when you challenge me! +4! 💫" + ] + elif 'D2' in card_code: + messages = [ + "Draw 2! Better luck next time~ 🎵", + "Here, have some extra cards! 📥", + "+2 for you! Hope you like drawing! 😊" + ] + elif 'skip' in card_code: + messages = [ + "Skip! You lose your turn! ⏭️", + "Not so fast! Skipped! 🎤", + "Your turn? Nope! Skipped! ✨" + ] + elif 'W' in card_code: + color_names = {'R': 'Red', 'G': 'Green', 'B': 'Blue', 'Y': 'Yellow'} + chosen_color = color_names.get(action.get('color', 'R'), 'Red') + messages = [ + f"Wild card! Changing to {chosen_color}! 🌈", + f"Let's go {chosen_color}! Time to mix things up! 💫" + ] + else: + if my_cards == 2: + messages = ["Almost there... one more card! 🎯"] + elif opponent_cards <= 2: + messages = ["Not gonna let you win! 😤", "I see you getting close... not on my watch! 💢"] + else: + messages = ["Hehe, perfect card! ✨", "This is too easy~ 🎤", "Watch and learn! 🎶"] + + import random + await self.discord_channel.send(random.choice(messages)) + + except Exception as e: + logger.error(f"Error sending trash talk: {e}") + + async def send_action(self, action: Dict[str, Any]) -> bool: + """Send action to game server""" + try: + response = requests.post( + f"{UNO_SERVER_URL}/api/game/{self.room_code}/action", + json=action, + headers={'Content-Type': 'application/json'}, + timeout=5 + ) + + if response.status_code == 200: + data = response.json() + return data.get('success', False) + + return False + + except Exception as e: + logger.error(f"Error sending action: {e}") + return False + + def is_game_active(self) -> bool: + """Check if game is currently active""" + return self.is_playing + + async def quit_game(self): + """Quit the game and cleanup""" + self.is_playing = False + await self.cleanup() + + async def cleanup(self): + """Cleanup browser resources""" + try: + if self.page: + await self.page.close() + if self.browser: + await self.browser.close() + if self.playwright: + await self.playwright.stop() + + logger.info(f"Cleaned up resources for room {self.room_code}") + except Exception as e: + logger.error(f"Error during cleanup: {e}")