jan/autoqa/utils.py

import os
import logging
import subprocess
import psutil
import time
import pyautogui
import platform
from pathlib import Path

logger = logging.getLogger(__name__)

# Cross-platform window management
IS_LINUX = platform.system() == "Linux"
IS_WINDOWS = platform.system() == "Windows"
IS_MACOS = platform.system() == "Darwin"

if IS_WINDOWS:
    try:
        import pygetwindow as gw
    except ImportError:
        gw = None
        logger.warning("pygetwindow not available on this system")

def is_jan_running(jan_process_name="Jan.exe"):
    """
    Check if Jan application is currently running
    """
    for proc in psutil.process_iter(['pid', 'name']):
        try:
            if proc.info['name'] and jan_process_name.lower() in proc.info['name'].lower():
                return True
        except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
            pass
    return False

def force_close_jan(jan_process_name="Jan.exe"):
    """
    Force close Jan application if it's running
    """
    logger.info("Checking for running Jan processes...")
    closed_any = False
    
    for proc in psutil.process_iter(['pid', 'name']):
        try:
            if proc.info['name'] and jan_process_name.lower() in proc.info['name'].lower():
                logger.info(f"Force closing Jan process (PID: {proc.info['pid']})")
                proc.kill()
                closed_any = True
        except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
            pass
    
    if closed_any:
        logger.info("Waiting for Jan processes to terminate...")
        time.sleep(3)  # Wait for processes to fully terminate
    else:
        logger.info("No Jan processes found running")

def find_jan_window_linux():
    """
    Find Jan window on Linux using wmctrl
    """
    try:
        result = subprocess.run(['wmctrl', '-l'], capture_output=True, text=True, timeout=10)
        if result.returncode == 0:
            for line in result.stdout.split('\n'):
                if 'jan' in line.lower() or 'Jan' in line:
                    # Extract window ID (first column)
                    window_id = line.split()[0]
                    logger.info(f"Found Jan window with ID: {window_id}")
                    return window_id
    except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError) as e:
        logger.warning(f"wmctrl command failed: {e}")
    return None

def maximize_jan_window_linux():
    """
    Maximize Jan window on Linux using wmctrl
    """
    window_id = find_jan_window_linux()
    if window_id:
        try:
            # Maximize window using wmctrl
            subprocess.run(['wmctrl', '-i', '-r', window_id, '-b', 'add,maximized_vert,maximized_horz'], 
                         timeout=5)
            logger.info("Jan window maximized using wmctrl")
            return True
        except (subprocess.TimeoutExpired, subprocess.SubprocessError) as e:
            logger.warning(f"Failed to maximize with wmctrl: {e}")
    
    # Fallback: Try xdotool
    try:
        result = subprocess.run(['xdotool', 'search', '--name', 'Jan'], 
                              capture_output=True, text=True, timeout=5)
        if result.returncode == 0 and result.stdout.strip():
            window_id = result.stdout.strip().split('\n')[0]
            subprocess.run(['xdotool', 'windowactivate', window_id], timeout=5)
            subprocess.run(['xdotool', 'key', 'alt+F10'], timeout=5)  # Maximize shortcut
            logger.info("Jan window maximized using xdotool")
            return True
    except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError) as e:
        logger.warning(f"xdotool command failed: {e}")
    
    return False

def find_jan_window_macos():
    """
    Find Jan window on macOS using AppleScript
    """
    try:
        # AppleScript to find Jan window
        script = '''
        tell application "System Events"
            set janApps to (every process whose name contains "Jan")
            if length of janApps > 0 then
                return name of first item of janApps
            else
                return ""
            end if
        end tell
        '''
        result = subprocess.run(['osascript', '-e', script], 
                              capture_output=True, text=True, timeout=10)
        if result.returncode == 0 and result.stdout.strip():
            app_name = result.stdout.strip()
            logger.info(f"Found Jan app: {app_name}")
            return app_name
    except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError) as e:
        logger.warning(f"AppleScript command failed: {e}")
    return None

def maximize_jan_window_macos():
    """
    Maximize Jan window on macOS using AppleScript
    """
    app_name = find_jan_window_macos()
    if app_name:
        try:
            # AppleScript to maximize window
            script = f'''
            tell application "System Events"
                tell process "{app_name}"
                    set frontmost to true
                    tell window 1
                        set value of attribute "AXFullScreen" to true
                    end tell
                end tell
            end tell
            '''
            result = subprocess.run(['osascript', '-e', script], timeout=10)
            if result.returncode == 0:
                logger.info("Jan window maximized using AppleScript")
                return True
        except (subprocess.TimeoutExpired, subprocess.SubprocessError) as e:
            logger.warning(f"Failed to maximize with AppleScript: {e}")
    
    # Fallback: Try Command+M (fullscreen hotkey on macOS)
    try:
        logger.info("Trying Cmd+Ctrl+F hotkey to maximize")
        pyautogui.hotkey('cmd', 'ctrl', 'f')
        time.sleep(1)
        logger.info("Attempted to maximize using Cmd+Ctrl+F")
        return True
    except Exception as e:
        logger.warning(f"Hotkey maximize failed: {e}")
    
    return False

def maximize_jan_window():
    """
    Find and maximize Jan window (cross-platform)
    """
    try:
        # Wait a bit for window to appear
        time.sleep(2)
        
        if IS_LINUX:
            return maximize_jan_window_linux()
        
        elif IS_MACOS:
            return maximize_jan_window_macos()
        
        elif IS_WINDOWS and gw:
            # Method 1: Try to find window by title containing "Jan"
            windows = gw.getWindowsWithTitle("Jan")
            if windows:
                jan_window = windows[0]
                logger.info(f"Found Jan window: {jan_window.title}")
                jan_window.maximize()
                logger.info("Jan window maximized using pygetwindow")
                return True
        
        # Fallback methods for both platforms
        # Method 2: Try Alt+Space then X (maximize hotkey) - works on both platforms
        logger.info("Trying Alt+Space+X hotkey to maximize")
        pyautogui.hotkey('alt', 'space')
        time.sleep(0.5)
        pyautogui.press('x')
        logger.info("Attempted to maximize using Alt+Space+X")
        return True
        
    except Exception as e:
        logger.warning(f"Could not maximize Jan window: {e}")
        
        # Method 3: Platform-specific fallback
        try:
            if IS_WINDOWS:
                logger.info("Trying Windows+Up arrow to maximize")
                pyautogui.hotkey('win', 'up')
            elif IS_LINUX:
                logger.info("Trying Alt+F10 to maximize")
                pyautogui.hotkey('alt', 'F10')
            elif IS_MACOS:
                logger.info("Trying macOS specific maximize")
                pyautogui.hotkey('cmd', 'tab')  # Switch to Jan if it's running
                time.sleep(0.5)
            return True
        except Exception as e2:
            logger.warning(f"All maximize methods failed: {e2}")
            return False

def start_jan_app(jan_app_path=None):
    """
    Start Jan application in maximized window (cross-platform)
    """
    # Set default path based on platform
    if jan_app_path is None:
        if IS_WINDOWS:
            jan_app_path = os.path.expanduser(r"~\AppData\Local\Programs\jan\Jan.exe")
        elif IS_LINUX:
            jan_app_path = "/usr/bin/Jan"  # or "/usr/bin/Jan" for regular
        elif IS_MACOS:
            jan_app_path = "/Applications/Jan.app/Contents/MacOS/Jan"  # Default macOS path
        else:
            raise NotImplementedError(f"Platform {platform.system()} not supported")
    
    logger.info(f"Starting Jan application from: {jan_app_path}")
    
    if not os.path.exists(jan_app_path):
        logger.error(f"Jan executable not found at: {jan_app_path}")
        raise FileNotFoundError(f"Jan app not found at {jan_app_path}")
    
    try:
        # Start the Jan application
        if IS_WINDOWS:
            subprocess.Popen([jan_app_path], shell=True)
        elif IS_LINUX:
            # On Linux, start with DISPLAY environment variable
            env = os.environ.copy()
            subprocess.Popen([jan_app_path], env=env)
        elif IS_MACOS:
            # On macOS, use 'open' command to launch .app bundle properly
            if jan_app_path.endswith('.app/Contents/MacOS/Jan'):
                # Use the .app bundle path instead
                app_bundle = jan_app_path.replace('/Contents/MacOS/Jan', '')
                subprocess.Popen(['open', app_bundle])
            elif jan_app_path.endswith('.app'):
                # Direct .app bundle
                subprocess.Popen(['open', jan_app_path])
            elif '/Contents/MacOS/' in jan_app_path:
                # Extract app bundle from full executable path
                app_bundle = jan_app_path.split('/Contents/MacOS/')[0]
                subprocess.Popen(['open', app_bundle])
            else:
                # Fallback: try to execute directly
                subprocess.Popen([jan_app_path])
        else:
            raise NotImplementedError(f"Platform {platform.system()} not supported")
        logger.info("Jan application started")
        
        # Wait for app to fully load
        logger.info("Waiting for Jan application to initialize...")
        time.sleep(5)
        
        # Try to maximize the window
        if maximize_jan_window():
            logger.info("Jan application maximized successfully")
        else:
            logger.warning("Could not maximize Jan application window")
        
        # Wait a bit more after maximizing
        time.sleep(10)
        logger.info("Jan application should be ready, waiting for additional setup...")
        time.sleep(10)  # Additional wait to ensure everything is ready
        
    except Exception as e:
        logger.error(f"Error starting Jan application: {e}")
        raise

def scan_test_files(tests_dir="tests"):
    """
    Scan tests folder and find all .txt files
    Returns list with format [{'path': 'relative_path', 'prompt': 'file_content'}]
    """
    test_files = []
    tests_path = Path(tests_dir)
    
    if not tests_path.exists():
        logger.error(f"Tests directory {tests_dir} does not exist!")
        return test_files
    
    # Scan all .txt files in folder and subfolders
    for txt_file in tests_path.rglob("*.txt"):
        try:
            # Read file content
            with open(txt_file, 'r', encoding='utf-8') as f:
                content = f.read().strip()
            
            # Get relative path
            relative_path = txt_file.relative_to(tests_path)
            
            test_files.append({
                'path': str(relative_path),
                'prompt': content
            })
            logger.info(f"Found test file: {relative_path}")
        except Exception as e:
            logger.error(f"Error reading file {txt_file}: {e}")
    
    return test_files

def get_latest_trajectory_folder(trajectory_base_path):
    """
    Get the latest created folder in trajectory base path
    """
    if not os.path.exists(trajectory_base_path):
        logger.warning(f"Trajectory base path not found: {trajectory_base_path}")
        return None
    
    # Get all folders and sort by creation time (latest first)
    folders = [f for f in os.listdir(trajectory_base_path) 
               if os.path.isdir(os.path.join(trajectory_base_path, f))]
    
    if not folders:
        logger.warning(f"No trajectory folders found in: {trajectory_base_path}")
        return None
    
    # Sort by folder name (assuming timestamp format like 20250715_100443)
    folders.sort(reverse=True)
    latest_folder = folders[0]
    
    full_path = os.path.join(trajectory_base_path, latest_folder)
    logger.info(f"Found latest trajectory folder: {full_path}")
    return full_path