diff --git a/.gitignore b/.gitignore index d903ae8..587886c 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,8 @@ +game_stats.json +input_mode.txt .env .venv +*.db* default_config.json data *.db diff --git a/main.py b/main.py index e74d607..dd7c62e 100644 --- a/main.py +++ b/main.py @@ -114,6 +114,7 @@ def main() -> None: parser.add_argument('--admin-users', type=str, help='Comma-separated list of additional admin users who have the same permissions as the channel owner') parser.add_argument('--config', type=str, default=DEFAULT_CONFIG_PATH, help=f'Path to configuration file (default: {DEFAULT_CONFIG_PATH})') + parser.add_argument('--force-new-token', action='store_true', help='Force generation of a new token even if a cached one exists') args: argparse.Namespace = parser.parse_args() # Load configuration from file if it exists @@ -180,7 +181,24 @@ def main() -> None: admin_users = config["admin_users"] print(f"Additional admin users from config file: {', '.join(admin_users)}") - # Create bot instance + # Make sure the cache directory exists + cache_dir = os.path.dirname(args.token_cache) + os.makedirs(cache_dir, exist_ok=True) + + print(f"Using token cache file: {args.token_cache}") + if args.force_new_token: + print("Forcing new token generation (any cached token will be ignored)") + # Remove cached token if forcing new one + if os.path.exists(args.token_cache): + try: + os.remove(args.token_cache) + print("Removed existing token cache file") + except Exception as e: + print(f"Warning: Could not remove existing token cache: {e}") + else: + print("Using cached token if available") + + # Create bot instance - explicitly pass None for access_token to avoid using env vars bot: TwitchBot = TwitchBot( username=username, client_id=client_id, @@ -188,7 +206,10 @@ def main() -> None: channel=channel, use_queue=args.use_queue, token_cache_file=args.token_cache, - admin_users=admin_users + admin_users=admin_users, + force_new_token=args.force_new_token, + access_token=None, # Explicitly set to None to avoid using env vars + refresh_token=None # Explicitly set to None to avoid using env vars ) # Register basic commands @@ -206,9 +227,13 @@ def main() -> None: # Start the Huey consumer if requested if args.start_consumer: try: - from src.queue.server import start_consumer + from src.queue.server import start_consumer, set_input_mode, get_input_mode import threading + # Make sure input mode is set before starting consumer + if args.game_control and args.input_type: + print(f"Consumer will use input mode: {get_input_mode()}") + consumer_thread = threading.Thread(target=start_consumer) consumer_thread.daemon = True consumer_thread.start() @@ -251,6 +276,18 @@ def main() -> None: print("Falling back to keyboard/mouse input.") args.input_type = 'keyboard' + # Persist the input type to make it available to the queue consumer process + try: + from src.queue.server import set_input_mode, INPUT_MODE_KEYBOARD, INPUT_MODE_CONTROLLER + if args.input_type == 'keyboard': + set_input_mode(INPUT_MODE_KEYBOARD) + print(f"Set persistent input mode to keyboard (this will be used by queue consumer)") + else: + set_input_mode(INPUT_MODE_CONTROLLER) + print(f"Set persistent input mode to controller (this will be used by queue consumer)") + except ImportError as e: + print(f"Warning: Could not set persistent input mode: {e}") + print(f"Enabling game control in {args.game_mode} mode with {args.input_type} input...") if args.input_type == 'keyboard' and not PYNPUT_AVAILABLE: @@ -282,6 +319,7 @@ def main() -> None: if args.input_type == 'controller': print(" pip install vgamepad # For controller support") + print("xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxinput type: ", args.input_type) # Start the bot print(f"Starting bot for channel #{channel}") print("Basic commands: !hello, !dice [sides], !echo [message], !8ball") @@ -290,6 +328,9 @@ def main() -> None: if args.use_queue: print("Queue mode enabled, commands will be processed by the queue system") print("Use !qstats to see queue statistics") + + if args.force_new_token: + print("Note: Forcing generation of a new authentication token") try: # Make sure requests library is installed diff --git a/scripts/setup_twitch_auth.py b/scripts/setup_twitch_auth.py index 4c6b8b8..e316300 100644 --- a/scripts/setup_twitch_auth.py +++ b/scripts/setup_twitch_auth.py @@ -7,9 +7,7 @@ This script helps users generate a Twitch user access token with proper IRC scop import os import sys import logging -import json import argparse -import time from dotenv import load_dotenv # Load environment variables if available @@ -25,43 +23,7 @@ from src.core.auth import TwitchAuth logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s') logger = logging.getLogger('twitch-auth-setup') -# Path to default config -DEFAULT_CONFIG_PATH = os.path.join("data", "default_config.json") - -def load_default_config(): - """Load default configuration from JSON file""" - if os.path.exists(DEFAULT_CONFIG_PATH): - try: - with open(DEFAULT_CONFIG_PATH, 'r') as f: - return json.load(f) - except json.JSONDecodeError: - logger.warning(f"Invalid JSON in {DEFAULT_CONFIG_PATH}, ignoring") - return {} - -def save_default_config(config_data): - """Save configuration to default config JSON file""" - # Ensure directory exists - os.makedirs(os.path.dirname(DEFAULT_CONFIG_PATH), exist_ok=True) - - # If file exists, load current data and update it - existing_data = {} - if os.path.exists(DEFAULT_CONFIG_PATH): - try: - with open(DEFAULT_CONFIG_PATH, 'r') as f: - existing_data = json.load(f) - except json.JSONDecodeError: - logger.warning(f"Invalid JSON in {DEFAULT_CONFIG_PATH}, overwriting") - - # Update with new data - existing_data.update(config_data) - - # Save back to file - with open(DEFAULT_CONFIG_PATH, 'w') as f: - json.dump(existing_data, f, indent=2) - - logger.info(f"Configuration saved to {DEFAULT_CONFIG_PATH}") - -def setup_twitch_auth(client_id=None, client_secret=None, scopes=None, manual_mode=False, force_new_token=False, use_cached_token=False): +def setup_twitch_auth(client_id=None, client_secret=None, scopes=None, manual_mode=False, force_new_token=False): """ Setup Twitch authentication by generating a user access token @@ -71,46 +33,21 @@ def setup_twitch_auth(client_id=None, client_secret=None, scopes=None, manual_mo scopes: Space-separated list of scopes manual_mode: Whether to use manual mode (no browser) force_new_token: Force generation of a new token even if a cached one exists - use_cached_token: Whether to use a cached token if available (default: False) Returns: bool: True if successful, False otherwise """ try: - # Load default config - default_config = load_default_config() + # Use provided values, then env vars + client_id = client_id or os.environ.get('TWITCH_CLIENT_ID') + client_secret = client_secret or os.environ.get('TWITCH_CLIENT_SECRET') - # Priority of values: - # 1. Function arguments - # 2. Environment variables - # 3. Default config - - # Use provided values, then env vars, then default config - client_id = client_id or os.environ.get('TWITCH_CLIENT_ID') or default_config.get('TWITCH_CLIENT_ID') - client_secret = client_secret or os.environ.get('TWITCH_CLIENT_SECRET') or default_config.get('TWITCH_CLIENT_SECRET') - - if not client_id: - client_id = input("Enter your Twitch Client ID: ") - - if not client_secret: - client_secret = input("Enter your Twitch Client Secret: ") - - if not client_id or not client_secret: - logger.error("Client ID and Client Secret are required") - return False - - # Default scopes for IRC chat - prioritize env var, then default config + # Default scopes for IRC chat default_scopes = "chat:read chat:edit" - scopes = scopes or os.environ.get('TWITCH_SCOPES') or default_config.get('TWITCH_SCOPES', default_scopes) + scopes = scopes or os.environ.get('TWITCH_SCOPES', default_scopes) # Define redirect URI - redirect_uri = os.environ.get('TWITCH_REDIRECT_URI') or default_config.get('TWITCH_REDIRECT_URI', "https://localhost:3000") - - print("\n===== Twitch Authentication Setup =====") - print(f"Client ID: {client_id}") - print(f"Redirect URI: {redirect_uri} (must match your Twitch app settings)") - print(f"Requested Scopes: {scopes}") - print("=====================================\n") + redirect_uri = os.environ.get('TWITCH_REDIRECT_URI', "https://localhost:3000") # Make sure the cache directory exists cache_dir = os.path.join("data", "cache") @@ -118,11 +55,6 @@ def setup_twitch_auth(client_id=None, client_secret=None, scopes=None, manual_mo # Create the token cache file path token_cache_file = os.path.join(cache_dir, "token_cache.json") - - # By default, clear any cached tokens unless explicitly told to use them - if (not use_cached_token or force_new_token) and os.path.exists(token_cache_file): - logger.info("Clearing cached token to generate fresh credentials") - os.remove(token_cache_file) # Create auth handler auth = TwitchAuth( @@ -133,97 +65,12 @@ def setup_twitch_auth(client_id=None, client_secret=None, scopes=None, manual_mo scopes=scopes ) - # Check if a valid token is already available - has_valid_token = False - if use_cached_token and auth.access_token and auth.token_expiry and time.time() < (auth.token_expiry - 60): - has_valid_token = True - # Using cached token - if not force_new_token: - print("\nUsing cached token (valid and not expired).") - print("If you want to force a new token, run with --force-new-token") - try: - # Try to validate the token to make sure it's working - token_info = auth.validate_token() - print(f"Token belongs to user: {token_info.get('login', 'unknown')}") - expiry_hours = (auth.token_expiry - time.time()) / 3600 - print(f"Token expires in {expiry_hours:.1f} hours") - - # Skip browser flow and continue with cached token - oauth_token = f"oauth:{auth.access_token}" - except Exception as e: - logger.warning(f"Error validating cached token: {e}") - logger.warning("Will attempt to get a new token") - has_valid_token = False - - # If no valid token, go through the auth flow - if not has_valid_token or force_new_token: - # Show security warning about certificates - if not manual_mode: - print("\n⚠️ IMPORTANT SECURITY NOTE ⚠️") - print("This application uses a self-signed certificate for HTTPS, which is required by Twitch.") - print("When your browser opens, you will see a security warning.") - print("This is expected for local development. Please proceed to the site anyway:") - print(" • In Chrome: Click 'Advanced' and then 'Proceed to localhost (unsafe)'") - print(" • In Firefox: Click 'Advanced', then 'Accept the Risk and Continue'") - print(" • In Edge: Click 'Details' and then 'Go on to the webpage'\n") - print("NOTE: If the browser doesn't open automatically, a URL will be provided for you to copy and paste.") - input("Press Enter to continue...") - - # Start OAuth flow - logger.info("Starting OAuth authentication flow...") - oauth_token = auth.get_oauth_token(manual_auth=manual_mode) - - # At this point, we should have a valid token - if auth.access_token: - # Get token info - try: - token_info = auth.validate_token() - - print("\n===== Authentication Successful =====") - print(f"Access Token: {auth.access_token}") - if auth.refresh_token: - print(f"Refresh Token: {auth.refresh_token}") - print(f"Token Scopes: {token_info.get('scopes', [])}") - print(f"User Name: {token_info.get('login', 'unknown')}") - print("=====================================\n") - except Exception as e: - logger.error(f"Failed to validate token: {e}") - print("\n===== Authentication Status =====") - print(f"Access Token: {auth.access_token}") - if auth.refresh_token: - print(f"Refresh Token: {auth.refresh_token}") - print("Warning: Could not validate token") - print("============================\n") - - # Store token info in default_config.json - if input("Would you like to save these credentials to default_config.json? (y/n): ").lower() == 'y': - config_data = { - "TWITCH_CLIENT_ID": client_id, - "TWITCH_CLIENT_SECRET": client_secret, - } - - # Try to use the validated user info - if 'token_info' in locals(): - username = token_info.get('login', '') - config_data["TWITCH_USERNAME"] = username - - # Get channel name - default_channel = username if 'username' in locals() else os.environ.get('TWITCH_CHANNEL', '') or default_config.get('TWITCH_CHANNEL', '') - channel = input(f"Enter Twitch channel to join (default: {default_channel}): ") or default_channel - config_data["TWITCH_CHANNEL"] = channel - - # Save access and refresh tokens - config_data["TWITCH_ACCESS_TOKEN"] = auth.access_token - if auth.refresh_token: - config_data["TWITCH_REFRESH_TOKEN"] = auth.refresh_token - - # Save to default_config.json - save_default_config(config_data) - + # Get the token, which will automatically trigger setup if needed + try: + oauth_token = auth.get_oauth_token(manual_auth=manual_mode, force_new_token=force_new_token) return True - - else: - logger.error("Failed to get OAuth token") + except ValueError as e: + logger.error(f"Authentication failed: {e}") return False except Exception as e: @@ -238,11 +85,10 @@ def main(): parser.add_argument("--scopes", help="Space-separated list of scopes") parser.add_argument("--manual", action="store_true", help="Use manual mode (no browser)") parser.add_argument("--force-new-token", action="store_true", help="Force generation of a new token even if a cached one exists") - parser.add_argument("--use-cached-token", action="store_true", help="Use cached token if available (default is to clear cached tokens)") args = parser.parse_args() - if setup_twitch_auth(args.client_id, args.client_secret, args.scopes, args.manual, args.force_new_token, args.use_cached_token): + if setup_twitch_auth(args.client_id, args.client_secret, args.scopes, args.manual, args.force_new_token): print("Setup completed successfully!") sys.exit(0) else: diff --git a/src/core/auth.py b/src/core/auth.py index bd3677a..f111a53 100644 --- a/src/core/auth.py +++ b/src/core/auth.py @@ -175,38 +175,49 @@ class TwitchAuth: logger.error(f"Failed to exchange authorization code: {e}") return False, f"Failed to exchange authorization code: {str(e)}" - def get_oauth_token(self, manual_auth: bool = False) -> str: + def _setup_auth(self, manual_mode: bool = False) -> bool: """ - Get OAuth token for Twitch authentication - - If no valid token is available: - 1. Try to refresh the token if a refresh token is available - 2. If refresh fails or no refresh token, require user auth + Interactive setup for Twitch authentication Args: - manual_auth: If True, provide instructions for manual authentication - instead of automatic browser launch - - Returns: - str: The OAuth token with 'oauth:' prefix - - Raises: - ValueError: If authentication fails - """ - # Check if we have a valid token - if self.access_token and self.token_expiry and time.time() < (self.token_expiry - 60): - expiry_min = (self.token_expiry - time.time()) / 60 - logger.debug(f"Using cached token, expires in {expiry_min:.1f} minutes") - return f"oauth:{self.access_token}" - - # Try to refresh the token - if self.refresh_token and self._refresh_token(): - return f"oauth:{self.access_token}" + manual_mode: Whether to use manual mode (no browser) - # Need user authentication - use our auth server to handle the flow - logger.info("Starting OAuth authentication flow") + Returns: + bool: True if successful, False otherwise + """ + # Verify we have the required credentials + if not self.client_id: + self.client_id = input("Enter your Twitch Client ID: ") - if manual_auth: + if not self.client_secret: + self.client_secret = input("Enter your Twitch Client Secret: ") + + if not self.client_id or not self.client_secret: + logger.error("Client ID and Client Secret are required") + return False + + print("\n===== Twitch Authentication Setup =====") + print(f"Client ID: {self.client_id}") + print(f"Redirect URI: {self.redirect_uri} (must match your Twitch app settings)") + print(f"Requested Scopes: {self.scopes}") + print("=====================================\n") + + # Show security warning about certificates + if not manual_mode: + print("\n⚠️ IMPORTANT SECURITY NOTE ⚠️") + print("This application uses a self-signed certificate for HTTPS, which is required by Twitch.") + print("When your browser opens, you will see a security warning.") + print("This is expected for local development. Please proceed to the site anyway:") + print(" • In Chrome: Click 'Advanced' and then 'Proceed to localhost (unsafe)'") + print(" • In Firefox: Click 'Advanced', then 'Accept the Risk and Continue'") + print(" • In Edge: Click 'Details' and then 'Go on to the webpage'\n") + print("NOTE: If the browser doesn't open automatically, a URL will be provided for you to copy and paste.") + input("Press Enter to continue...") + + # Start OAuth flow + logger.info("Starting OAuth authentication flow...") + + if manual_mode: # For environments where browser launch isn't possible auth_url = ( f"https://id.twitch.tv/oauth2/authorize" @@ -231,14 +242,82 @@ class TwitchAuth: ) if not auth_code: - raise ValueError("Failed to get authorization code") + logger.error("Failed to get authorization code") + return False # Exchange the auth code for tokens success, message = self._handle_auth_code(auth_code) if success: - return f"oauth:{self.access_token}" + # At this point, we should have a valid token + try: + token_info = self.validate_token() + + print("\n===== Authentication Successful =====") + print(f"Token Scopes: {token_info.get('scopes', [])}") + print(f"User Name: {token_info.get('login', 'unknown')}") + print("=====================================\n") + return True + except Exception as e: + logger.error(f"Failed to validate token: {e}") + print("\n===== Authentication Status =====") + print("Warning: Could not validate token") + print("============================\n") + return False else: - raise ValueError(f"Authentication failed: {message}") + logger.error(f"Authentication failed: {message}") + return False + + def get_oauth_token(self, manual_auth: bool = False, force_new_token: bool = False) -> str: + """ + Get OAuth token for Twitch authentication + + If no valid token is available: + 1. Try to refresh the token if a refresh token is available + 2. If refresh fails or no refresh token, automatically trigger setup + + Args: + manual_auth: If True, provide instructions for manual authentication + instead of automatic browser launch + force_new_token: Force generation of a new token even if a cached one exists + + Returns: + str: The OAuth token with 'oauth:' prefix + + Raises: + ValueError: If authentication fails + """ + # If force_new_token is set, clear any cached tokens first + if force_new_token and os.path.exists(self.token_cache_file): + logger.info("Clearing cached token to generate fresh credentials") + os.remove(self.token_cache_file) + self.access_token = None + self.refresh_token = None + self.token_expiry = None + + # Check if we have a valid token (with an extra safety margin of 5 minutes) + if not force_new_token and self.access_token and self.token_expiry: + # Add extra safety margin for token expiry to avoid edge cases + safety_margin = 300 # 5 minutes in seconds + if time.time() < (self.token_expiry - safety_margin): + expiry_min = (self.token_expiry - time.time()) / 60 + logger.debug(f"Using cached token, expires in {expiry_min:.1f} minutes") + return f"oauth:{self.access_token}" + else: + logger.warning(f"Token expiring soon (within safety margin), attempting refresh") + + # Try to refresh the token + if not force_new_token and self.refresh_token and self._refresh_token(): + logger.info("Successfully refreshed token") + return f"oauth:{self.access_token}" + + # If token refresh failed or no refresh token is available, trigger setup + logger.info("No valid token available, starting authentication flow") + if self._setup_auth(manual_mode=manual_auth): + if self.access_token: + return f"oauth:{self.access_token}" + + # If we got here, authentication failed + raise ValueError("Failed to obtain a valid OAuth token") def validate_token(self) -> Dict[str, Any]: """ diff --git a/src/core/twitch.py b/src/core/twitch.py index 52da207..b1d32c0 100644 --- a/src/core/twitch.py +++ b/src/core/twitch.py @@ -25,7 +25,8 @@ class TwitchBot: access_token: str = None, refresh_token: str = None, channel: str = None, use_queue: bool = False, token_cache_file: str = "data/cache/token_cache.json", - admin_users: List[str] = None) -> None: + admin_users: List[str] = None, + force_new_token: bool = False) -> None: self.username: str = username.lower() self.client_id: Optional[str] = client_id or os.environ.get("TWITCH_CLIENT_ID") self.client_secret: Optional[str] = client_secret or os.environ.get("TWITCH_CLIENT_SECRET") @@ -36,26 +37,28 @@ class TwitchBot: self.commands: Dict[str, CommandCallback] = {} self.running: bool = False self.last_command: str = "" + self.force_new_token: bool = force_new_token + self.connection_attempts: int = 0 # Admin users who have the same privileges as the channel owner self.admin_users: List[str] = [self.channel] # Channel owner is always an admin if admin_users: self.admin_users.extend([user.lower() for user in admin_users]) - # Use access_token from args or environment - direct_access_token = access_token or os.environ.get("TWITCH_ACCESS_TOKEN") - direct_refresh_token = refresh_token or os.environ.get("TWITCH_REFRESH_TOKEN") + # Make sure the cache directory exists + os.makedirs(os.path.dirname(token_cache_file), exist_ok=True) - # Auth token handling + # Initialize auth handler - only use token cache for tokens self.auth = TwitchAuth(self.client_id, self.client_secret, token_cache_file) self.token_cache_file: str = token_cache_file - # If tokens provided directly, update the auth handler - if direct_access_token: - logger.info("Using provided access token") + # Only use directly provided tokens (not from env vars) for backward compatibility + # This is only used during testing or if explicitly passed to the constructor + if access_token is not None: + logger.info("Using provided access token parameter (not from environment)") self.auth.update_manually( - access_token=direct_access_token, - refresh_token=direct_refresh_token + access_token=access_token, + refresh_token=refresh_token ) # Queue integration @@ -75,19 +78,93 @@ class TwitchBot: def token_expiry(self) -> Optional[float]: """Get the token expiry from the auth handler""" return self.auth.token_expiry + + def validate_token_with_twitch(self) -> bool: + """ + Validate the current token with Twitch API to ensure it's still valid + + Returns: + bool: True if token is valid, False otherwise + """ + if not self.auth.access_token: + logger.warning("No access token available to validate") + return False + + try: + validate_url = "https://id.twitch.tv/oauth2/validate" + headers = { + "Authorization": f"OAuth {self.auth.access_token}" + } + + response = requests.get(validate_url, headers=headers) + if response.status_code == 200: + token_info = response.json() + logger.info(f"Token valid for user: {token_info.get('login', 'unknown')}") + return True + else: + logger.warning(f"Token validation failed with status {response.status_code}: {response.text}") + return False + except Exception as e: + logger.error(f"Error validating token with Twitch: {e}") + return False - def get_oauth_token(self) -> str: - """Get an OAuth token for authenticating with Twitch""" - return self.auth.get_oauth_token() + def get_oauth_token(self, force_refresh: bool = False) -> str: + """ + Get an OAuth token for authenticating with Twitch + + Args: + force_refresh: Force a token refresh regardless of expiry + + Returns: + str: The OAuth token in 'oauth:xxx' format + """ + try: + # If forcing a refresh, temporarily set force_new_token + original_force_new_token = self.force_new_token + if force_refresh: + logger.info("Forcing token refresh due to connection issues") + self.force_new_token = True + + # The updated TwitchAuth will automatically handle token refresh or setup if needed + token = self.auth.get_oauth_token(force_new_token=self.force_new_token) + + # Restore original force_new_token setting + if force_refresh: + self.force_new_token = original_force_new_token + + # Reset connection attempts counter on successful token retrieval + self.connection_attempts = 0 + return token + + except Exception as e: + logger.error(f"Failed to get OAuth token: {e}") + raise def connect(self) -> bool: """Connect to Twitch IRC server""" try: - # Get the OAuth token - oauth_token = self.get_oauth_token() + # Increment connection attempts + self.connection_attempts += 1 + + # If we've had multiple connection attempts, validate the token with Twitch first + if self.connection_attempts > 1 and self.auth.access_token: + if not self.validate_token_with_twitch(): + logger.warning("Token validation failed, forcing refresh before connection attempt") + # Force a token refresh since validation failed + oauth_token = self.get_oauth_token(force_refresh=True) + else: + logger.info("Token validation successful, proceeding with connection") + oauth_token = self.get_oauth_token() + else: + # First attempt, use normal token flow + oauth_token = self.get_oauth_token() + self.oauth_token = oauth_token # Store for queue use + # Create a new socket for each connection attempt + self.socket = socket.socket() self.socket.connect((self.server, self.port)) + # Add a timeout so we can process keyboard interrupts self.socket.settimeout(1.0) self.socket.send(f"PASS {oauth_token}\n".encode('utf-8')) @@ -96,8 +173,30 @@ class TwitchBot: # Wait for confirmation data = self.socket.recv(2048).decode('utf-8') + + # Check for authentication error + if ":tmi.twitch.tv NOTICE * :Login authentication failed" in data: + logger.error("Authentication failed! Forcing token refresh for next attempt") + # Force a new token on the next attempt - clear the token cache + if os.path.exists(self.token_cache_file): + try: + os.remove(self.token_cache_file) + logger.info(f"Removed invalid token cache: {self.token_cache_file}") + except Exception as e: + logger.error(f"Failed to remove token cache: {e}") + + # If we've tried multiple times, let's be more explicit about the error + if self.connection_attempts >= 3: + logger.error("Multiple authentication failures. Please check your Twitch credentials and permissions.") + logger.error("You may need to revoke any existing tokens in the Twitch Developer Console.") + + return False + + # Check for successful connection if "Welcome" in data or "You are in a maze of twisty passages" in data: logger.info(f"Connected to {self.channel}'s chat") + # Reset connection attempts counter on successful connection + self.connection_attempts = 0 return True else: logger.error(f"Failed to connect: {data}") @@ -138,49 +237,50 @@ class TwitchBot: content: str = message_data['content'].strip() username: str = message_data['username'] - # Check if we should use the queue - if self.use_queue: - # If this is a regular message, add it to the message queue - if not content.startswith('!'): - enqueue_message(username, content) - return - - # Check if message is a command (starts with !) - if content.startswith('!'): - parts: List[str] = content.split() - command: str = parts[0][1:].lower() # Remove ! and convert to lowercase - args: List[str] = parts[1:] if len(parts) > 1 else [] - - # Store the last executed command - self.last_command = command - - # Check if we should use the queue + # If this is a regular message (not a command), handle it + if not content.startswith('!'): if self.use_queue: - # Add the command to the queue and return - # We need to create a simple command registry to pass to the queue - command_registry = {cmd: cmd for cmd in self.commands.keys()} - - # Make sure we have an OAuth token - if not self.oauth_token: - self.oauth_token = self.get_oauth_token() - - enqueue_command( - username, - command, - args, - channel=self.channel, - oauth_token=self.oauth_token, - bot_username=self.username, - command_registry=command_registry - ) - return + # If using queue, send the message to the queue + enqueue_message(username, content) + return + + # At this point, we know this is a command (starts with !) + parts: List[str] = content.split() + command: str = parts[0][1:].lower() # Remove ! and convert to lowercase + args: List[str] = parts[1:] if len(parts) > 1 else [] + + # Store the last executed command + self.last_command = command + + # Log if this is a controller command + if command in self.commands: + logger.info(f"Recognized command: !{command}") + + # If using queue, just enqueue the command without any processing + if self.use_queue: + # Make sure we have an OAuth token + if not self.oauth_token: + self.oauth_token = self.get_oauth_token() - # Direct execution (no queue) - if command in self.commands: - try: - self.commands[command](username, args, self) - except Exception as e: - logger.error(f"Error executing command {command}: {e}") + # Simply pass the command name and let the queue worker handle all processing + logger.info(f"Enqueueing command '{command}' for processing by queue worker") + + enqueue_command( + username, + command, + args, + channel=self.channel, + oauth_token=self.oauth_token, + bot_username=self.username + ) + return + + # Direct execution (no queue) + if command in self.commands: + try: + self.commands[command](username, args, self) + except Exception as e: + logger.error(f"Error executing command {command}: {e}") def start(self, use_queue: bool = False) -> None: """Start the bot and listen for messages""" @@ -188,9 +288,44 @@ class TwitchBot: self.use_queue = use_queue logger.info(f"Starting bot with queue {'enabled' if use_queue else 'disabled'}") - if not self.connect(): - logger.error("Failed to start bot: Connection error") - return + # Try to connect with up to 3 attempts, possibly with token refreshes + max_attempts = 3 + for attempt in range(1, max_attempts + 1): + logger.info(f"Connection attempt {attempt}/{max_attempts}") + if self.connect(): + break + elif attempt < max_attempts: + # Wait a moment before retrying + time.sleep(2) + logger.info("Retrying connection...") + else: + logger.error(f"Failed to connect after {max_attempts} attempts") + + # Clean up any existing token cache + if os.path.exists(self.token_cache_file): + try: + os.remove(self.token_cache_file) + logger.info(f"Removed token cache after failed connection attempts: {self.token_cache_file}") + except Exception as e: + logger.error(f"Failed to remove token cache: {e}") + + # Force a completely new authentication process + logger.info("Running authentication setup after failed connection attempts...") + try: + # Force a new token and run the setup process + self.oauth_token = self.get_oauth_token(force_refresh=True) + + # Try one final connection + logger.info("Attempting connection with fresh authentication...") + if self.connect(): + logger.info("Successfully connected with new authentication!") + else: + logger.error("Still unable to connect even with fresh authentication") + logger.error("Please check your Twitch credentials and network connection") + return + except Exception as e: + logger.error(f"Authentication setup failed: {e}") + return self.running = True buffer: str = "" diff --git a/src/game/controller.py b/src/game/controller.py index 370c0c9..4678ada 100644 --- a/src/game/controller.py +++ b/src/game/controller.py @@ -17,8 +17,9 @@ except ImportError: # Try to import the VirtualController try: - from src.game.controller import VirtualController, VGAMEPAD_AVAILABLE -except ImportError: + from src.game.input.gamepad import VirtualController, VGAMEPAD_AVAILABLE +except ImportError as e: + print(e) VGAMEPAD_AVAILABLE = False print("controller_support.py not found or error importing it.") print("Virtual controller support will be disabled.") @@ -105,10 +106,10 @@ class GameController: "shoot": self.controller.press_right_trigger, # D-pad for items/quick slots - "item1": self.controller.press_dpad_up, - "item2": self.controller.press_dpad_right, - "item3": self.controller.press_dpad_down, - "item4": self.controller.press_dpad_left, + "dpad_up": self.controller.press_dpad_up, + "dpad_right": self.controller.press_dpad_right, + "dpad_down": self.controller.press_dpad_down, + "dpad_left": self.controller.press_dpad_left, # Menu buttons "start": self.controller.press_start, diff --git a/src/queue/consumer.py b/src/queue/consumer.py index 338d771..a2851f2 100644 --- a/src/queue/consumer.py +++ b/src/queue/consumer.py @@ -32,6 +32,7 @@ class QueueConsumer: self.running = False self.consumer_thread = None self.handlers = {} + self.input_type = None # Track the current input type: "keyboard" or "controller" logger.info("Queue consumer initialized") @@ -46,6 +47,42 @@ class QueueConsumer: self.handlers[task_type] = handler logger.info(f"Registered handler for {task_type}") + def set_input_type(self, input_type: str) -> None: + """ + Set the current input type + + Args: + input_type: Either "keyboard" or "controller" + """ + if input_type not in ["keyboard", "controller"]: + logger.error(f"Invalid input type: {input_type}") + return + + self.input_type = input_type + logger.info(f"Input type set to: {input_type}") + + def validate_command(self, command: str, requested_type: str) -> bool: + """ + Validate if a command can be executed with the requested input type + + Args: + command: The command to validate + requested_type: The input type being requested ("keyboard" or "controller") + + Returns: + bool: True if the command is valid for the requested input type + """ + # If no input type is set yet, allow the request to establish it + if self.input_type is None: + return True + + # Ensure command matches current input type + if self.input_type != requested_type: + logger.warning(f"Command type mismatch: Current input type is {self.input_type}, but {requested_type} command received") + return False + + return True + def start(self) -> None: """Start the consumer in a background thread""" if self.running: diff --git a/src/queue/server.py b/src/queue/server.py index 56bb22b..098fea4 100644 --- a/src/queue/server.py +++ b/src/queue/server.py @@ -3,18 +3,74 @@ import os import logging import socket import time -from typing import Dict, Any, Optional, List, Callable, Union +import random +import cloudpickle +from typing import Dict, Any, Optional, List, Callable, Union, Tuple # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s') logger = logging.getLogger('twitch-queue') +# Define constants for input modes +INPUT_MODE_KEYBOARD = "keyboard" +INPUT_MODE_CONTROLLER = "controller" +INPUT_MODE_FILE = "input_mode.txt" + +# Function to get/set the current input mode using persistent storage +def get_input_mode(): + """Read the current input mode from persistent storage""" + try: + if os.path.exists(INPUT_MODE_FILE): + with open(INPUT_MODE_FILE, 'r') as f: + mode = f.read().strip() + if mode in [INPUT_MODE_KEYBOARD, INPUT_MODE_CONTROLLER]: + return mode + except Exception as e: + logger.error(f"Error reading input mode: {e}") + + # Default to keyboard mode if not set or error + return INPUT_MODE_KEYBOARD + +def set_input_mode(mode): + """Save the current input mode to persistent storage""" + if mode not in [INPUT_MODE_KEYBOARD, INPUT_MODE_CONTROLLER, None]: + logger.error(f"Invalid input mode: {mode}") + return False + + try: + if mode is None: + # Remove the file if mode is None (reset) + if os.path.exists(INPUT_MODE_FILE): + os.remove(INPUT_MODE_FILE) + return True + + # Write mode to file + with open(INPUT_MODE_FILE, 'w') as f: + f.write(mode) + return True + except Exception as e: + logger.error(f"Error writing input mode: {e}") + return False + +# CloudPickle serializer for Huey - wraps cloudpickle to provide the interface Huey expects +class CloudPickleSerializer: + @staticmethod + def serialize(data): + return cloudpickle.dumps(data) + + @staticmethod + def deserialize(data): + if data: + return cloudpickle.loads(data) + return None + # Create a Huey instance with SQLite storage huey = SqliteHuey( name='twitch_commands', filename='twitch_queue.db', - results=True, # Store task results - immediate=False # Don't execute tasks immediately (use consumer) + results=False, # Store task results + immediate=False, # Don't execute tasks immediately (use consumer) + serializer=CloudPickleSerializer # Use cloudpickle for better serialization ) # Queue statistics @@ -60,120 +116,190 @@ queue_stats = QueueStats() IRC_SERVER = 'irc.chat.twitch.tv' IRC_PORT = 6667 -# Task definitions -@huey.task() -def process_chat_command(username: str, command: str, args: List[str], - channel: str, oauth_token: str, bot_username: str, - command_registry: Dict[str, str]) -> Dict[str, Any]: - """ - Process a chat command and return the result +# Command handler type +CommandHandler = Callable[[str, List[str], str, str, str], bool] + +# Command registry - maps command names to handler functions +command_registry: Dict[str, CommandHandler] = {} + +# Try to import the game controller module for controller support +try: + # Since we're running in a separate process, we need to initialize our own controller + # Avoid importing GameController directly to prevent circular imports + from src.game.input.keyboard import KeyboardController + from src.game.input.gamepad import VirtualController, VGAMEPAD_AVAILABLE - This task will be executed by the Huey consumer - """ - logger.info(f"Processing command '{command}' from {username} with args: {args}") + # Create controller instances - these will be initialized when needed + keyboard_controller = None + gamepad_controller = None - # Record stats - queue_stats.log_processed(f"command:{command}") + CONTROLLER_SUPPORT = True + logger.info("Game controller support loaded in queue worker") +except ImportError as e: + logger.warning(f"Game controller support not available in queue worker: {e}") + CONTROLLER_SUPPORT = False + +# ---- CONTROLLER MANAGEMENT ---- + +def initialize_controller(controller_type=INPUT_MODE_KEYBOARD): + """Initialize the appropriate controller type""" + global keyboard_controller, gamepad_controller - # Check if the command is registered - if command not in command_registry: - logger.warning(f"Command '{command}' not found in registry") - return { - 'success': False, - 'error': f"Command '{command}' not found", - 'username': username, - 'command': command, - 'processed_at': time.time() - } + if not CONTROLLER_SUPPORT: + logger.warning("Controller support not available") + return False + + try: + if controller_type == INPUT_MODE_KEYBOARD: + if keyboard_controller is None: + logger.info("Initializing keyboard controller in queue worker") + keyboard_controller = KeyboardController() + return keyboard_controller is not None + elif controller_type == INPUT_MODE_CONTROLLER: + if not VGAMEPAD_AVAILABLE: + logger.warning("vgamepad not available, can't initialize controller") + return False + + if gamepad_controller is None: + logger.info("Initializing gamepad controller in queue worker") + gamepad_controller = VirtualController() + # Give it a moment to initialize + time.sleep(0.2) + + # Check if controller was properly initialized + if not gamepad_controller.is_available(): + logger.error("VirtualController was created but is not available") + return False + + logger.info("Virtual gamepad controller successfully initialized") + + return gamepad_controller is not None and gamepad_controller.is_available() + else: + logger.warning(f"Unknown controller type: {controller_type}") + return False + except Exception as e: + logger.error(f"Error initializing controller: {e}") + return False + +# Maintain the current input mode - either "keyboard" or "controller" +# This global variable is no longer used - see get_input_mode() and set_input_mode() functions +# for the persistent file-based approach that works across processes + +def execute_game_command(command, controller_type=INPUT_MODE_KEYBOARD): + """Execute a game controller command""" + logger.info(f"execute_game_command called for '{command}' with controller_type={controller_type}") + + if not CONTROLLER_SUPPORT: + logger.warning("Controller support not available") + return False try: - # For built-in commands we handle here - if command == "hello": - send_twitch_message(channel, oauth_token, bot_username, f"Hello, {username}!") - elif command == "dice": - import random - sides = 6 # Default to 6-sided dice - if args and args[0].isdigit(): - sides = int(args[0]) - result = random.randint(1, sides) - send_twitch_message(channel, oauth_token, bot_username, f"@{username} rolled a {result} (d{sides})") - elif command == "echo": - if args: - message = " ".join(args) - send_twitch_message(channel, oauth_token, bot_username, f"Echo: {message}") + if controller_type == INPUT_MODE_KEYBOARD: + # Make sure keyboard controller is initialized + if keyboard_controller is None and not initialize_controller(INPUT_MODE_KEYBOARD): + logger.error("Failed to initialize keyboard controller") + return False + + if keyboard_controller is None: + logger.error("Keyboard controller initialization failed") + return False + + # Map commands to keyboard controller methods + command_mapping = { + "up": lambda: keyboard_controller.press_key('w'), + "down": lambda: keyboard_controller.press_key('s'), + "left": lambda: keyboard_controller.press_key('a'), + "right": lambda: keyboard_controller.press_key('d'), + "jump": lambda: keyboard_controller.press_key('space'), + "attack": lambda: keyboard_controller.press_mouse_button('left'), + "interact": lambda: keyboard_controller.press_key('e'), + "inventory": lambda: keyboard_controller.press_key('i'), + "skill1": lambda: keyboard_controller.press_key('1'), + "skill2": lambda: keyboard_controller.press_key('2'), + "skill3": lambda: keyboard_controller.press_key('3'), + "ultimate": lambda: keyboard_controller.press_key('r'), + } + + if command in command_mapping: + logger.info(f"Executing keyboard command: {command}") + try: + command_mapping[command]() + # Release key after a short delay + time.sleep(0.1) + keyboard_controller.release_all() + return True + except Exception as e: + logger.error(f"Error executing keyboard command {command}: {e}") + return False else: - send_twitch_message(channel, oauth_token, bot_username, f"@{username}, you didn't provide a message to echo!") - elif command == "8ball": - import random - responses = [ - "It is certain.", - "It is decidedly so.", - "Without a doubt.", - "Yes, definitely.", - "You may rely on it.", - "As I see it, yes.", - "Most likely.", - "Outlook good.", - "Yes.", - "Signs point to yes.", - "Reply hazy, try again.", - "Ask again later.", - "Better not tell you now.", - "Cannot predict now.", - "Concentrate and ask again.", - "Don't count on it.", - "My reply is no.", - "My sources say no.", - "Outlook not so good.", - "Very doubtful." - ] - send_twitch_message(channel, oauth_token, bot_username, f"@{username}, {random.choice(responses)}") - elif command == "qstats": - stats = get_queue_stats() - send_twitch_message(channel, oauth_token, bot_username, - f"Queue stats: {stats['total_processed']}/{stats['total_enqueued']} processed, {stats['pending']} pending") + logger.warning(f"Unknown keyboard command: {command}") + return False + + elif controller_type == INPUT_MODE_CONTROLLER: + # Make sure gamepad controller is initialized + if gamepad_controller is None and not initialize_controller(INPUT_MODE_CONTROLLER): + logger.error("Failed to initialize gamepad controller") + return False + + if gamepad_controller is None: + logger.error("Gamepad controller initialization failed") + return False + + # Map commands to gamepad controller methods + command_mapping = { + "up": lambda: gamepad_controller.move_left_stick_up(), + "down": lambda: gamepad_controller.move_left_stick_down(), + "left": lambda: gamepad_controller.move_left_stick_left(), + "right": lambda: gamepad_controller.move_left_stick_right(), + "look_up": lambda: gamepad_controller.move_right_stick_up(), + "look_down": lambda: gamepad_controller.move_right_stick_down(), + "look_left": lambda: gamepad_controller.move_right_stick_left(), + "look_right": lambda: gamepad_controller.move_right_stick_right(), + "jump": lambda: gamepad_controller.press_a(), + "action": lambda: gamepad_controller.press_b(), + "interact": lambda: gamepad_controller.press_x(), + "menu": lambda: gamepad_controller.press_y(), + "block": lambda: gamepad_controller.press_left_shoulder(), + "attack": lambda: gamepad_controller.press_right_shoulder(), + "aim": lambda: gamepad_controller.press_left_trigger(), + "shoot": lambda: gamepad_controller.press_right_trigger(), + "dup": lambda: gamepad_controller.press_dpad_up(), + "dright": lambda: gamepad_controller.press_dpad_right(), + "ddown": lambda: gamepad_controller.press_dpad_down(), + "dleft": lambda: gamepad_controller.press_dpad_left(), + "start": lambda: gamepad_controller.press_start(), + "select": lambda: gamepad_controller.press_back(), + } + + if command in command_mapping: + logger.info(f"Executing controller command: {command}") + try: + command_mapping[command]() + # Release buttons after a short delay + time.sleep(0.1) + gamepad_controller.reset() + return True + except Exception as e: + logger.error(f"Error executing controller command {command}: {e}") + return False + else: + logger.warning(f"Unknown controller command: {command}") + return False else: - # For any custom commands, we'd need to implement them here - send_twitch_message(channel, oauth_token, bot_username, - f"Command '{command}' is registered but not implemented in the queue worker.") - - return { - 'success': True, - 'username': username, - 'command': command, - 'args': args, - 'processed_at': time.time() - } + logger.warning(f"Unknown controller type: {controller_type}") + return False except Exception as e: - logger.error(f"Error executing command {command}: {e}") - return { - 'success': False, - 'error': str(e), - 'username': username, - 'command': command, - 'args': args, - 'processed_at': time.time() - } + logger.error(f"Error executing game command: {e}") + return False -@huey.task() -def process_chat_message(username: str, message: str) -> Dict[str, Any]: - """ - Process a regular chat message - - This task will be executed by the Huey consumer - """ - logger.info(f"Processing message from {username}: {message}") - - # Record stats - queue_stats.log_processed("message") - - # Return information about the processed message - return { - 'success': True, - 'username': username, - 'message': message, - 'processed_at': time.time() - } +# Function to reset input mode (e.g., when switching games) +def reset_input_mode(): + """Reset the current input mode to allow switching between keyboard and controller""" + # Use persistent storage for input mode + set_input_mode(None) + logger.info("Input mode has been reset") + +# ---- COMMUNICATION UTILITIES ---- def send_twitch_message(channel: str, oauth_token: str, username: str, message: str) -> bool: """ @@ -204,20 +330,508 @@ def send_twitch_message(channel: str, oauth_token: str, username: str, message: logger.error(f"Error sending message to Twitch: {e}") return False +# ---- COMMAND HANDLERS ---- + +# Define command handler for each specific command +# Each will be registered as a separate Huey task + +@huey.task() +def handle_hello_task(username: str, args: List[str], channel: str, oauth_token: str, bot_username: str) -> Dict[str, Any]: + """Handler for !hello command""" + success = send_twitch_message(channel, oauth_token, bot_username, f"Hello, {username}!") + return { + 'success': success, + 'username': username, + 'command': 'hello', + 'timestamp': time.time() + } + +@huey.task() +def handle_dice_task(username: str, args: List[str], channel: str, oauth_token: str, bot_username: str) -> Dict[str, Any]: + """Handler for !dice command""" + sides = 6 # Default to 6-sided dice + if args and args[0].isdigit(): + sides = int(args[0]) + result = random.randint(1, sides) + success = send_twitch_message(channel, oauth_token, bot_username, f"@{username} rolled a {result} (d{sides})") + return { + 'success': success, + 'username': username, + 'command': 'dice', + 'result': result, + 'timestamp': time.time() + } + +@huey.task() +def handle_echo_task(username: str, args: List[str], channel: str, oauth_token: str, bot_username: str) -> Dict[str, Any]: + """Handler for !echo command""" + if args: + message = " ".join(args) + success = send_twitch_message(channel, oauth_token, bot_username, f"Echo: {message}") + else: + success = send_twitch_message(channel, oauth_token, bot_username, f"@{username}, you didn't provide a message to echo!") + + return { + 'success': success, + 'username': username, + 'command': 'echo', + 'timestamp': time.time() + } + +@huey.task() +def handle_8ball_task(username: str, args: List[str], channel: str, oauth_token: str, bot_username: str) -> Dict[str, Any]: + """Handler for !8ball command""" + responses = [ + "It is certain.", + "It is decidedly so.", + "Without a doubt.", + "Yes, definitely.", + "You may rely on it.", + "As I see it, yes.", + "Most likely.", + "Outlook good.", + "Yes.", + "Signs point to yes.", + "Reply hazy, try again.", + "Ask again later.", + "Better not tell you now.", + "Cannot predict now.", + "Concentrate and ask again.", + "Don't count on it.", + "My reply is no.", + "My sources say no.", + "Outlook not so good.", + "Very doubtful." + ] + response = random.choice(responses) + success = send_twitch_message(channel, oauth_token, bot_username, f"@{username}, {response}") + return { + 'success': success, + 'username': username, + 'command': '8ball', + 'response': response, + 'timestamp': time.time() + } + +@huey.task() +def handle_qstats_task(username: str, args: List[str], channel: str, oauth_token: str, bot_username: str) -> Dict[str, Any]: + """Handler for !qstats command""" + stats = get_queue_stats() + success = send_twitch_message(channel, oauth_token, bot_username, + f"Queue stats: {stats['total_processed']}/{stats['total_enqueued']} processed, {stats['pending']} pending") + + # Only return simple picklable stats, not the whole QueueStats object + safe_stats = { + 'total_enqueued': stats['total_enqueued'], + 'total_processed': stats['total_processed'], + 'pending': stats['pending'] + } + + return { + 'success': success, + 'username': username, + 'command': 'qstats', + 'stats': safe_stats, + 'timestamp': time.time() + } + +@huey.task() +def handle_keyboard_command_task(username: str, args: List[str], channel: str, oauth_token: str, bot_username: str, command: str) -> Dict[str, Any]: + """Handler for keyboard mode game commands""" + try: + # Check if current input mode is compatible + current_mode = get_input_mode() + if current_mode != INPUT_MODE_KEYBOARD: + message = f"@{username}, unable to use keyboard commands while controller mode is active" + send_twitch_message(channel, oauth_token, bot_username, message) + return { + 'success': False, + 'username': username, + 'command': command, + 'command_type': 'keyboard', + 'error': 'input_mode_mismatch', + 'timestamp': time.time() + } + + # Keep the controller operations completely isolated + execute_success = False + if initialize_controller(INPUT_MODE_KEYBOARD): + # Execute the command + execute_success = execute_game_command(command, INPUT_MODE_KEYBOARD) + if execute_success: + send_twitch_message(channel, oauth_token, bot_username, f"@{username} used {command}") + else: + send_twitch_message(channel, oauth_token, bot_username, f"@{username}, failed to execute {command}") + else: + send_twitch_message(channel, oauth_token, bot_username, f"@{username}, keyboard controller not available") + except Exception as e: + logger.error(f"Error in keyboard command task: {e}") + execute_success = False + send_twitch_message(channel, oauth_token, bot_username, f"@{username}, error executing {command}: {str(e)}") + + # Only return simple, picklable data with no references to any controller objects + return { + 'success': execute_success, + 'username': username, + 'command': command, + 'command_type': 'keyboard', + 'timestamp': time.time() + } + +@huey.task() +def handle_controller_command_task(username: str, args: List[str], channel: str, oauth_token: str, bot_username: str, command: str) -> Dict[str, Any]: + """Handler for controller mode game commands""" + logger.info(f"Controller command task executing for '{command}'") + + try: + # Check if current input mode is compatible + current_mode = get_input_mode() + if current_mode != INPUT_MODE_CONTROLLER: + message = f"@{username}, unable to use controller commands while keyboard mode is active" + send_twitch_message(channel, oauth_token, bot_username, message) + return { + 'success': False, + 'username': username, + 'command': command, + 'command_type': 'controller', + 'error': 'input_mode_mismatch', + 'timestamp': time.time() + } + + # Keep the controller operations completely isolated + execute_success = False + # Regular command flow for all controller commands + if initialize_controller(INPUT_MODE_CONTROLLER): + execute_success = execute_game_command(command, INPUT_MODE_CONTROLLER) + if execute_success: + send_twitch_message(channel, oauth_token, bot_username, f"@{username} used {command}") + else: + send_twitch_message(channel, oauth_token, bot_username, f"@{username}, failed to execute {command}") + else: + send_twitch_message(channel, oauth_token, bot_username, f"@{username}, controller support not available") + except Exception as e: + logger.error(f"Error in controller command task: {e}") + execute_success = False + send_twitch_message(channel, oauth_token, bot_username, f"@{username}, error executing {command}: {str(e)}") + + # Only return simple, picklable data with no references to any controller objects + return { + 'success': execute_success, + 'username': username, + 'command': command, + 'command_type': 'controller', + 'timestamp': time.time() + } + +@huey.task() +def handle_unknown_command_task(username: str, command: str, args: List[str], channel: str, oauth_token: str, bot_username: str) -> Dict[str, Any]: + """Handle commands that aren't explicitly registered""" + logger.info(f"Attempting to handle unknown command: '{command}'") + + # Define valid command lists + keyboard_commands = [ + "up", "down", "left", "right", "jump", "attack", "interact", + "inventory", "skill1", "skill2", "skill3", "ultimate" + ] + + controller_commands = [ + "look_up", "look_down", "look_left", "look_right", + "action", "menu", "block", "aim", "shoot", + "dup", "dright", "ddown", "dleft", "start", "select" + ] + + # Check if it's a keyboard command + if command in keyboard_commands and CONTROLLER_SUPPORT: + return handle_keyboard_command_task(username, args, channel, oauth_token, bot_username, command) + + # Check if it's a controller command + if command in controller_commands and CONTROLLER_SUPPORT: + return handle_controller_command_task(username, args, channel, oauth_token, bot_username, command) + + # Handle a few standard commands that might not be registered + if command == "help": + send_twitch_message(channel, oauth_token, bot_username, + f"@{username}, available commands: !hello, !dice, !echo, !8ball, !qstats") + if CONTROLLER_SUPPORT: + send_twitch_message(channel, oauth_token, bot_username, + f"Game commands: !up, !down, !left, !right, !jump, etc.") + send_twitch_message(channel, oauth_token, bot_username, + f"Admins can use !switchmode [keyboard|controller|reset] to change input mode") + return { + 'success': True, + 'username': username, + 'command': 'help', + 'timestamp': time.time() + } + + elif command == "ping": + send_twitch_message(channel, oauth_token, bot_username, f"@{username}, Pong!") + return { + 'success': True, + 'username': username, + 'command': 'ping', + 'timestamp': time.time() + } + + # Command not recognized + logger.warning(f"Command '{command}' not recognized as a known command") + send_twitch_message(channel, oauth_token, bot_username, f"Sorry @{username}, the command !{command} is not supported.") + return { + 'success': False, + 'error': f"Command '{command}' not found", + 'username': username, + 'command': command, + 'timestamp': time.time() + } + +@huey.task() +def handle_switchmode_task(username: str, args: List[str], channel: str, oauth_token: str, bot_username: str) -> Dict[str, Any]: + """Handler for !switchmode command to switch between keyboard and controller modes""" + # Only allow admins to switch modes + from src.core.twitch import is_admin + if not is_admin(username, channel): + message = f"@{username}, only channel admins can switch input modes" + send_twitch_message(channel, oauth_token, bot_username, message) + return { + 'success': False, + 'username': username, + 'command': 'switchmode', + 'error': 'not_admin', + 'timestamp': time.time() + } + + # Check if a new mode is specified + if not args: + current = get_input_mode() + message = f"@{username}, current input mode is: {current}. Use !switchmode keyboard or !switchmode controller to change" + send_twitch_message(channel, oauth_token, bot_username, message) + return { + 'success': True, + 'username': username, + 'command': 'switchmode', + 'current_mode': current, + 'timestamp': time.time() + } + + # Get the requested mode + requested_mode = args[0].lower() + if requested_mode not in [INPUT_MODE_KEYBOARD, INPUT_MODE_CONTROLLER, "reset"]: + message = f"@{username}, invalid mode: {requested_mode}. Use 'keyboard', 'controller', or 'reset'" + send_twitch_message(channel, oauth_token, bot_username, message) + return { + 'success': False, + 'username': username, + 'command': 'switchmode', + 'error': 'invalid_mode', + 'timestamp': time.time() + } + + # If "reset" is specified, reset the mode + if requested_mode == "reset": + reset_input_mode() + message = f"@{username} reset input mode. Users can now use either keyboard or controller commands" + send_twitch_message(channel, oauth_token, bot_username, message) + return { + 'success': True, + 'username': username, + 'command': 'switchmode', + 'new_mode': None, + 'timestamp': time.time() + } + + # Set the new mode + old_mode = get_input_mode() + logger.info(f"Switching input mode from {old_mode} to {requested_mode}") + + # First, save the mode to the persistent storage + set_input_mode(requested_mode) + + # Then initialize the controller based on the saved mode + initialize_success = initialize_controller(requested_mode) + + if initialize_success: + message = f"@{username} switched input mode to {requested_mode}" + send_twitch_message(channel, oauth_token, bot_username, message) + return { + 'success': True, + 'username': username, + 'command': 'switchmode', + 'old_mode': old_mode, + 'new_mode': requested_mode, + 'timestamp': time.time() + } + else: + # If initialization failed, revert to the old mode + set_input_mode(old_mode) + message = f"@{username}, failed to switch to {requested_mode} mode. Staying with {old_mode or 'no'} mode" + send_twitch_message(channel, oauth_token, bot_username, message) + return { + 'success': False, + 'username': username, + 'command': 'switchmode', + 'error': 'initialization_failed', + 'timestamp': time.time() + } + +# ---- TASK PROCESSING ---- + +@huey.task() +def process_chat_command(username: str, command: str, args: List[str], + channel: str, oauth_token: str, bot_username: str) -> Dict[str, Any]: + """ + Process a chat command and return the result + + This task will be executed by the Huey consumer + """ + logger.info(f"Processing command '{command}' from {username} with args: {args}") + + # Record stats + queue_stats.log_processed(f"command:{command}") + + # Determine the current input mode from persistent storage + mode = get_input_mode() + logger.info(f"Current input mode: {mode}") + + # Create command map based on the current mode + command_task_map = { + # Standard commands - always available regardless of mode + "hello": handle_hello_task, + "dice": handle_dice_task, + "echo": handle_echo_task, + "8ball": handle_8ball_task, + "qstats": handle_qstats_task, + "switchmode": handle_switchmode_task, + } + + # Add mode-specific commands + if mode == INPUT_MODE_KEYBOARD: + # Add keyboard-specific commands + keyboard_commands = { + # Directional and common commands + "up": lambda u, a, c, o, b: handle_keyboard_command_task(u, a, c, o, b, "up"), + "down": lambda u, a, c, o, b: handle_keyboard_command_task(u, a, c, o, b, "down"), + "left": lambda u, a, c, o, b: handle_keyboard_command_task(u, a, c, o, b, "left"), + "right": lambda u, a, c, o, b: handle_keyboard_command_task(u, a, c, o, b, "right"), + "jump": lambda u, a, c, o, b: handle_keyboard_command_task(u, a, c, o, b, "jump"), + "attack": lambda u, a, c, o, b: handle_keyboard_command_task(u, a, c, o, b, "attack"), + "interact": lambda u, a, c, o, b: handle_keyboard_command_task(u, a, c, o, b, "interact"), + + # Keyboard-only commands + "inventory": lambda u, a, c, o, b: handle_keyboard_command_task(u, a, c, o, b, "inventory"), + "skill1": lambda u, a, c, o, b: handle_keyboard_command_task(u, a, c, o, b, "skill1"), + "skill2": lambda u, a, c, o, b: handle_keyboard_command_task(u, a, c, o, b, "skill2"), + "skill3": lambda u, a, c, o, b: handle_keyboard_command_task(u, a, c, o, b, "skill3"), + "ultimate": lambda u, a, c, o, b: handle_keyboard_command_task(u, a, c, o, b, "ultimate"), + } + command_task_map.update(keyboard_commands) + else: # mode == INPUT_MODE_CONTROLLER + # Add controller-specific commands + controller_commands = { + # Directional and common commands + "up": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "up"), + "down": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "down"), + "left": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "left"), + "right": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "right"), + "jump": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "jump"), + "attack": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "attack"), + "interact": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "interact"), + + # Controller-only commands + "look_up": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "look_up"), + "look_down": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "look_down"), + "look_left": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "look_left"), + "look_right": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "look_right"), + "action": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "action"), + "menu": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "menu"), + "block": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "block"), + "aim": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "aim"), + "shoot": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "shoot"), + "dup": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "dup"), + "dright": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "dright"), + "ddown": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "ddown"), + "dleft": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "dleft"), + "start": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "start"), + "select": lambda u, a, c, o, b: handle_controller_command_task(u, a, c, o, b, "select"), + } + command_task_map.update(controller_commands) + + try: + # Check if this is a known command + if command in command_task_map: + task_handler = command_task_map[command] + try: + # Execute the dedicated task handler + logger.info(f"Found handler for '{command}' in {mode} mode, executing...") + return task_handler(username, args, channel, oauth_token, bot_username) + except Exception as e: + logger.error(f"Error executing command {command}: {e}") + return { + 'success': False, + 'error': str(e), + 'username': username, + 'command': command, + 'args': args, + 'timestamp': time.time() + } + else: + # Unknown command - try the fallback handler + try: + return handle_unknown_command_task(username, command, args, channel, oauth_token, bot_username) + except Exception as e: + logger.error(f"Error handling unknown command {command}: {e}") + return { + 'success': False, + 'error': str(e), + 'username': username, + 'command': command, + 'args': args, + 'timestamp': time.time() + } + except Exception as e: + logger.error(f"Unexpected error in process_chat_command: {e}") + # Make sure we return a safely picklable object + return { + 'success': False, + 'error': str(e), + 'username': username, + 'command': command, + 'timestamp': time.time() + } + +@huey.task() +def process_chat_message(username: str, message: str) -> Dict[str, Any]: + """ + Process a regular chat message + + This task will be executed by the Huey consumer + """ + logger.info(f"Processing message from {username}: {message}") + + # Record stats + queue_stats.log_processed("message") + + # Return information about the processed message + return { + 'success': True, + 'username': username, + 'message': message, + 'timestamp': time.time() + } + # Helper functions for enqueueing tasks def enqueue_command(username: str, command: str, args: List[str], channel: str = None, oauth_token: str = None, - bot_username: str = None, command_registry: Dict[str, str] = None) -> None: + bot_username: str = None) -> None: """Add a command to the processing queue""" logger.info(f"Enqueueing command '{command}' from {username} with args: {args}") queue_stats.log_enqueue(f"command:{command}", username) - if not all([channel, oauth_token, bot_username, command_registry]): + if not all([channel, oauth_token, bot_username]): logger.error("Missing required parameters for enqueue_command") return None return process_chat_command( - username, command, args, channel, oauth_token, bot_username, command_registry + username, command, args, channel, oauth_token, bot_username ) def enqueue_message(username: str, message: str) -> None: @@ -239,6 +853,11 @@ def start_consumer() -> None: python -m huey.bin.huey_consumer src.queue.server.huey """ from huey.consumer import Consumer + + # Log current input mode at startup + current_mode = get_input_mode() + logger.info(f"Starting consumer with input mode: {current_mode}") + consumer = Consumer(huey) consumer.start()