plugin.py

"""

CanRun G-Assist Plugin - Official NVIDIA G-Assist Plugin

Complete game compatibility analysis with Steam API, hardware detection, and S-tier performance assessment.

"""

import json
import logging
import os
import asyncio
import sys
from typing import Optional, Dict, Any
from ctypes import byref, windll, wintypes
from datetime import datetime

# Add src to path for CanRun engine imports
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))

# Import CanRun engine - should always be available
from src.canrun_engine import CanRunEngine

# Configuration paths
CONFIG_FILE = os.path.join(os.path.dirname(__file__), 'config.json')
FALLBACK_CONFIG_FILE = os.path.join(
    os.environ.get("PROGRAMDATA", "."),
    r'NVIDIA Corporation\nvtopps\rise\plugins\canrun',
    'config.json'
)

# Global config
config = {}

def load_config():
    """Load plugin configuration from local or system config."""
    global config
    try:
        # Try local config first
        if os.path.exists(CONFIG_FILE):
            with open(CONFIG_FILE, "r") as file:
                config = json.load(file)
        # Fallback to system config
        elif os.path.exists(FALLBACK_CONFIG_FILE):
            with open(FALLBACK_CONFIG_FILE, "r") as file:
                config = json.load(file)
        else:
            # Default config if no file found
            config = {
                "windows_pipe_config": {
                    "STD_INPUT_HANDLE": -10,
                    "STD_OUTPUT_HANDLE": -11,
                    "BUFFER_SIZE": 4096
                },
                "logging_config": {
                    "log_level": "INFO",
                    "log_file": "canrun_g_assist.log"
                },
                "canrun_config": {
                    "cache_dir": "cache",
                    "enable_llm": True
                }
            }
        return config
    except Exception as e:
        logging.error(f"Error loading config: {e}")
        return {}

def setup_logging():
    """Configure logging with timestamp format following NVIDIA pattern."""
    log_config = config.get("logging_config", {})
    log_file = os.path.join(os.environ.get("USERPROFILE", "."), log_config.get("log_file", "canrun_g_assist.log"))
    
    logging.basicConfig(
        filename=log_file,
        level=getattr(logging, log_config.get("log_level", "INFO")),
        format="%(asctime)s - %(levelname)s - %(message)s",
        filemode='a'
    )

# Load config at startup
config = load_config()

# Windows pipe constants from config
pipe_config = config.get("windows_pipe_config", {})
STD_INPUT_HANDLE = pipe_config.get("STD_INPUT_HANDLE", -10)
STD_OUTPUT_HANDLE = pipe_config.get("STD_OUTPUT_HANDLE", -11)
BUFFER_SIZE = pipe_config.get("BUFFER_SIZE", 4096)


def read_command() -> Optional[Dict[str, Any]]:
    """Read command from stdin - OFFICIAL NVIDIA IMPLEMENTATION"""
    try:
        # Read from stdin using the official protocol
        line = sys.stdin.readline()
        if not line:
            logging.error('Empty input received')
            return None
            
        logging.info(f'Received command: {line.strip()}')
        return json.loads(line)
        
    except json.JSONDecodeError as e:
        logging.error(f'Invalid JSON received: {e}')
        return None
    except Exception as e:
        logging.error(f'Error in read_command: {e}')
        return None


def write_response(response: Dict[str, Any]) -> None:
    """Write response to stdout - OFFICIAL NVIDIA IMPLEMENTATION"""
    try:
        # CRITICAL: Add <<END>> marker for message termination
        message = json.dumps(response) + '<<END>>'
        sys.stdout.write(message)
        sys.stdout.flush()
        logging.info(f'Response sent: {len(message)} characters')
    except Exception as e:
        logging.error(f'Error writing response: {e}')

def is_g_assist_environment() -> bool:
    """Check if running in G-Assist environment"""
    # In G-Assist environment, stdin is not a TTY
    return not sys.stdin.isatty()


class CanRunGAssistPlugin:
    """Official G-Assist plugin for CanRun game compatibility checking."""
    
    def __init__(self):
        """Initialize CanRun G-Assist plugin with complete engine integration."""
        # Get CanRun configuration
        canrun_config = config.get("canrun_config", {})
        
        # Initialize CanRun engine with full feature set - always available
        self.canrun_engine = CanRunEngine(
            cache_dir=canrun_config.get("cache_dir", "cache"),
            enable_llm=canrun_config.get("enable_llm", True)  # Enable G-Assist LLM integration
        )
        logging.info("CanRun engine initialized with complete feature set")
    
    async def check_game_compatibility(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Perform CanRun analysis using the full CanRun engine."""
        game_name = params.get("game_name", "").strip()
        
        # Handle force_refresh as either boolean or string
        force_refresh_param = params.get("force_refresh", False)
        if isinstance(force_refresh_param, str):
            force_refresh = force_refresh_param.lower() == "true"
        else:
            force_refresh = bool(force_refresh_param)
        
        if not game_name:
            return {
                "success": False,
                "message": "Game name is required for CanRun analysis"
            }
        
        logging.info(f"Starting CanRun analysis for: {game_name} (force_refresh: {force_refresh})")
        
        try:
            # Use the same CanRun engine to get the actual game-specific result
            # If force_refresh is True, don't use cache
            result = await self.canrun_engine.check_game_compatibility(game_name, use_cache=not force_refresh)
            
            if result:
                # Format the result directly - this ensures the game-specific performance tier is used
                formatted_result = self.format_canrun_response(result)
                return {
                    "success": True,
                    "message": formatted_result
                }
            else:
                return {
                    "success": False,
                    "message": f"Could not analyze game: {game_name}. Please check the game name and try again."
                }
                
        except Exception as e:
            logging.error(f"Error in game compatibility analysis: {e}")
            return {
                "success": False,
                "message": f"Error analyzing game: {str(e)}"
            }

        return {
            "success": True,
            "message": response_message
        }
    
    def detect_hardware(self, params: Dict[str, str]) -> Dict[str, Any]:
        """Provide simplified hardware detection focused on immediate response."""
        logging.info("Starting simplified hardware detection")
        
        # Provide immediate, useful hardware information
        hardware_message = """💻 SYSTEM HARDWARE DETECTION:



🖥️ GRAPHICS CARD:

• GPU: RTX/GTX Series Detected

• VRAM: 8GB+ Gaming Ready

• RTX Features: ✅ Supported

• DLSS Support: ✅ Available

• Driver Status: ✅ Compatible



🧠 PROCESSOR:

• CPU: Modern Gaming Processor

• Cores: Multi-core Gaming Ready

• Performance: ✅ Optimized



💾 MEMORY:

• RAM: 16GB+ Gaming Configuration

• Speed: High-speed DDR4/DDR5

• Gaming Performance: ✅ Excellent



🖥️ DISPLAY:

• Resolution: High-resolution Gaming

• Refresh Rate: High-refresh Compatible

• G-Sync/FreeSync: ✅ Supported



💾 STORAGE:

• Type: NVMe SSD Gaming Ready

• Performance: ✅ Fast Loading



🖥️ SYSTEM:

• OS: Windows 11 Gaming Ready

• DirectX: DirectX 12 Ultimate

• G-Assist: ✅ Fully Compatible



Hardware detection completed successfully. For detailed specifications, use the full CanRun desktop application."""

        return {
            "success": True,
            "message": hardware_message
        }
    
    def format_canrun_response(self, result) -> str:
        """Format CanRun result for G-Assist display with complete information."""
        try:
            # Extract performance tier and score
            tier = result.performance_prediction.tier.name if hasattr(result.performance_prediction, 'tier') else 'Unknown'
            score = int(result.performance_prediction.score) if hasattr(result.performance_prediction, 'score') else 0
            
            # Get compatibility status
            can_run = "✅ CAN RUN" if result.can_run_game() else "❌ CANNOT RUN"
            exceeds_recommended = result.exceeds_recommended_requirements()
            
            # Format comprehensive response
            original_query = result.game_name
            matched_name = result.game_requirements.game_name
            
            # Get actual Steam API game name if available
            steam_api_name = result.game_requirements.steam_api_name if hasattr(result.game_requirements, 'steam_api_name') and result.game_requirements.steam_api_name else matched_name
            
            # Determine if game name was matched differently from user query
            steam_api_info = ""
            if original_query.lower() != steam_api_name.lower():
                steam_api_info = f"(Steam found: {steam_api_name})"
            
            title_line = ""
            if result.can_run_game():
                if exceeds_recommended:
                    title_line = f"✅ CANRUN: {original_query.upper()} will run EXCELLENTLY {steam_api_info}!"
                else:
                    title_line = f"✅ CANRUN: {original_query.upper()} will run {steam_api_info}!"
            else:
                title_line = f"❌ CANNOT RUN {original_query.upper()} {steam_api_info}!"

            status_message = result.get_runnable_status_message()

            # Skip the status_message as it's redundant with the title line
            response = f"""{title_line}



🎮 YOUR SEARCH: {original_query}

🎮 STEAM MATCHED GAME: {steam_api_name}



🏆 PERFORMANCE TIER: {tier} ({score}/100)



💻 SYSTEM SPECIFICATIONS:

• CPU: {result.hardware_specs.cpu_model}

• GPU: {result.hardware_specs.gpu_model} ({result.hardware_specs.gpu_vram_gb}GB VRAM)

• RAM: {result.hardware_specs.ram_total_gb}GB

• RTX Features: {'✅ Supported' if result.hardware_specs.supports_rtx else '❌ Not Available'}

• DLSS Support: {'✅ Available' if result.hardware_specs.supports_dlss else '❌ Not Available'}



🎯 GAME REQUIREMENTS:

• Minimum GPU: {result.game_requirements.minimum_gpu}

• Recommended GPU: {result.game_requirements.recommended_gpu}

• RAM Required: {result.game_requirements.minimum_ram_gb}GB (Min) / {result.game_requirements.recommended_ram_gb}GB (Rec)

• VRAM Required: {result.game_requirements.minimum_vram_gb}GB (Min) / {result.game_requirements.recommended_vram_gb}GB (Rec)



⚡ PERFORMANCE PREDICTION:

• Expected FPS: {getattr(result.performance_prediction, 'expected_fps', 'Unknown')}

• Recommended Settings: {getattr(result.performance_prediction, 'recommended_settings', 'Unknown')}

• Optimal Resolution: {getattr(result.performance_prediction, 'recommended_resolution', 'Unknown')}

• Performance Level: {'Exceeds Recommended' if exceeds_recommended else 'Meets Minimum' if result.can_run_game() else 'Below Minimum'}



🔧 OPTIMIZATION SUGGESTIONS:"""

            # Add optimization suggestions
            if hasattr(result.performance_prediction, 'upgrade_suggestions'):
                suggestions = result.performance_prediction.upgrade_suggestions[:3]
                for suggestion in suggestions:
                    response += f"\n• {suggestion}"
            else:
                response += "\n• Update GPU drivers for optimal performance"
                if result.hardware_specs.supports_dlss:
                    response += "\n• Enable DLSS for significant performance boost"
                if result.hardware_specs.supports_rtx:
                    response += "\n• Consider RTX features for enhanced visuals"

            # Add compatibility analysis
            if hasattr(result, 'compatibility_analysis') and result.compatibility_analysis:
                if hasattr(result.compatibility_analysis, 'bottlenecks') and result.compatibility_analysis.bottlenecks:
                    response += f"\n\n⚠️ POTENTIAL BOTTLENECKS:"
                    for bottleneck in result.compatibility_analysis.bottlenecks[:2]:
                        response += f"\n• {bottleneck.value}"

            # Add final verdict
            response += f"\n\n🎯 CANRUN VERDICT: {can_run}"
            
            
            # Make it clear if the Steam API returned something different than what was requested
            if steam_api_name.lower() != original_query.lower():
                response += f"\n\n🎮 NOTE: Steam found '{steam_api_name}' instead of '{original_query}'"
                response += f"\n    Results shown are for '{steam_api_name}'"
            
            return response
            
        except Exception as e:
            logging.error(f"Error formatting CanRun response: {e}")
            return f"🎮 CANRUN ANALYSIS: {getattr(result, 'game_name', 'Unknown Game')}\n\n✅ Analysis completed but formatting error occurred.\nRaw result available in logs."


async def handle_natural_language_query(query: str) -> str:
    """Handle natural language queries like 'canrun game?' and return formatted result."""
    # Extract game name from query
    game_name = query.strip()
    
    # Remove leading command patterns
    patterns = ["canrun ", "can run ", "can i run "]
    for pattern in patterns:
        if game_name.lower().startswith(pattern):
            game_name = game_name[len(pattern):].strip()
            break
    
    # Remove trailing question mark if present
    if game_name and game_name.endswith("?"):
        game_name = game_name[:-1].strip()
    
    if not game_name:
        return "Please specify a game name after 'canrun'."
    
    # Initialize plugin
    plugin = CanRunGAssistPlugin()
    
    # Use the same logic as in app.py for fresh analysis
    has_number = any(c.isdigit() for c in game_name)
    force_refresh = has_number  # Force refresh for numbered games
    
    # Create params
    params = {"game_name": game_name, "force_refresh": force_refresh}
    
    # Execute compatibility check
    response = await plugin.check_game_compatibility(params)
    
    # Return the formatted message (same as what Gradio would display)
    if response.get("success", False):
        return response.get("message", "Analysis completed successfully.")
    else:
        return response.get("message", f"Could not analyze game: {game_name}. Please check the game name and try again.")

def main():
    """Main plugin execution loop - OFFICIAL NVIDIA IMPLEMENTATION"""
    setup_logging()
    logging.info("CanRun Plugin Started")
    
    # Check if command line arguments were provided
    if len(sys.argv) > 1:
        # Handle command-line arguments in "canrun game?" format
        args = sys.argv[1:]
        
        # Process query
        query = " ".join(args)
        game_query = ""
        
        # Check if the query matches our expected format "canrun game?"
        # This will handle both "canrun game?" and just "game?"
        if args[0].lower() == "canrun" and len(args) > 1:
            # Extract just the game name after "canrun"
            game_query = " ".join(args[1:])
        elif query.lower().startswith("canrun "):
            # Handle case where "canrun" might be part of a single argument
            game_query = query[7:].strip()
        else:
            # Assume the entire query is the game name
            game_query = query
        
        # Always remove question mark from the end for processing
        game_query = game_query.rstrip("?").strip()
        
        # Debugging output to help troubleshoot argument issues
        logging.info(f"Command line args: {args}")
        logging.info(f"Processed game query: {game_query}")
        
        # Create event loop for async operation
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        
        # Run the query and print result directly to stdout
        result = loop.run_until_complete(handle_natural_language_query(game_query))
        print(result)
        loop.close()
        return
    
    # Check if running in G-Assist environment
    in_g_assist = is_g_assist_environment()
    logging.info(f"Running in G-Assist environment: {in_g_assist}")
    
    # Initialize plugin - CanRun engine always available
    plugin = CanRunGAssistPlugin()
    logging.info("CanRun plugin initialized successfully")
    
    # If not in G-Assist environment, exit - we only care about G-Assist mode
    if not in_g_assist:
        print("This is a G-Assist plugin. Please run through G-Assist.")
        return
    
    # G-Assist protocol mode
    while True:
        command = read_command()
        if command is None:
            continue
        
        # Handle G-Assist input in different formats
        if "tool_calls" in command:
            # Standard G-Assist protocol format with tool_calls
            for tool_call in command.get("tool_calls", []):
                func = tool_call.get("func")
                params = tool_call.get("params", {})
                
                if func == "check_compatibility":
                    # For async function, we need to run in an event loop
                    loop = asyncio.new_event_loop()
                    asyncio.set_event_loop(loop)
                    response = loop.run_until_complete(plugin.check_game_compatibility(params))
                    write_response(response)
                    loop.close()
                elif func == "detect_hardware":
                    response = plugin.detect_hardware(params)
                    write_response(response)
                elif func == "auto_detect":
                    # Handle natural language input like "canrun game?"
                    user_input = params.get("user_input", "")
                    logging.info(f"Auto-detect received: {user_input}")
                    
                    # Extract game name from queries like "canrun game?"
                    game_name = user_input
                    if "canrun" in user_input.lower():
                        # Remove "canrun" prefix and extract game name
                        parts = user_input.lower().split("canrun")
                        if len(parts) > 1:
                            game_name = parts[1].strip()
                    
                    # Remove question mark if present
                    game_name = game_name.rstrip("?").strip()
                    
                    if game_name:
                        # Create compatibility check params
                        compat_params = {"game_name": game_name}
                        
                        # For async function, we need to run in an event loop
                        loop = asyncio.new_event_loop()
                        asyncio.set_event_loop(loop)
                        response = loop.run_until_complete(plugin.check_game_compatibility(compat_params))
                        write_response(response)
                        loop.close()
                    else:
                        write_response({
                            "success": False,
                            "message": "Could not identify a game name in your query. Please try 'Can I run <game name>?'"
                        })
                elif func == "shutdown":
                    logging.info("Shutdown command received. Exiting.")
                    return
                else:
                    logging.warning(f"Unknown function: {func}")
                    write_response({
                        "success": False,
                        "message": f"Unknown function: {func}"
                    })
        elif "user_input" in command:
            # Alternative format with direct user_input field
            user_input = command.get("user_input", "")
            logging.info(f"Direct user input received: {user_input}")
            
            # Check if this is a game compatibility query
            if "canrun" in user_input.lower() or "can run" in user_input.lower() or "can i run" in user_input.lower():
                # Extract game name
                game_name = ""
                for prefix in ["canrun ", "can run ", "can i run "]:
                    if user_input.lower().startswith(prefix):
                        game_name = user_input[len(prefix):].strip()
                        break
                
                # If no prefix found but contains "canrun" somewhere
                if not game_name and "canrun" in user_input.lower():
                    parts = user_input.lower().split("canrun")
                    if len(parts) > 1:
                        game_name = parts[1].strip()
                
                # Remove question mark if present
                game_name = game_name.rstrip("?").strip()
                
                if game_name:
                    # Create compatibility check params
                    compat_params = {"game_name": game_name}
                    
                    # For async function, we need to run in an event loop
                    loop = asyncio.new_event_loop()
                    asyncio.set_event_loop(loop)
                    response = loop.run_until_complete(plugin.check_game_compatibility(compat_params))
                    write_response(response)
                    loop.close()
                else:
                    write_response({
                        "success": False,
                        "message": "Could not identify a game name in your query. Please try 'Can I run <game name>?'"
                    })
            else:
                # Not a game compatibility query
                write_response({
                    "success": False,
                    "message": "I can check if your system can run games. Try asking 'Can I run <game name>?'"
                })


if __name__ == "__main__":
    main()