Ability to play Uno implemented in early stages!
This commit is contained in:
@@ -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; \
|
||||
|
||||
@@ -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)
|
||||
|
||||
195
bot/commands/uno.py
Normal file
195
bot/commands/uno.py
Normal file
@@ -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 <room_code>
|
||||
"""
|
||||
if not room_code:
|
||||
await message.channel.send("🎴 Please provide a room code! Usage: `!uno join <ROOM_CODE>`")
|
||||
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 <code>` 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 <subcommand> [args]
|
||||
|
||||
Subcommands:
|
||||
!uno join <code> - 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 <ROOM_CODE>`")
|
||||
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 <CODE>` to make me join!\n4. I'll play automatically and trash talk in chat! 🎶",
|
||||
color=discord.Color.green()
|
||||
)
|
||||
|
||||
commands = [
|
||||
("!uno join <CODE>", "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)
|
||||
34
bot/setup_uno_playwright.sh
Executable file
34
bot/setup_uno_playwright.sh
Executable file
@@ -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
|
||||
@@ -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
|
||||
|
||||
448
bot/utils/uno_game.py
Normal file
448
bot/utils/uno_game.py
Normal file
@@ -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}")
|
||||
Reference in New Issue
Block a user