add more actions and improve to the next limit : import os
import time
import asyncio
import pyautogui
import google.generativeai as genai
import cv2
import numpy as np
from ultralytics import YOLO
import logging
from logging.handlers import RotatingFileHandler
import threading
import random
import string
from dotenv import load_dotenv
from abc import ABC, abstractmethod
from typing import Dict, List, Any, Callable
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from PIL import Image
import io
import re
import signal
import sys
from queue import PriorityQueue

# Load environment variables
load_dotenv()

# Load configuration from config.json
CONFIG_FILE = 'config.json'
if not os.path.exists(CONFIG_FILE):
    raise FileNotFoundError(f"Configuration file '{CONFIG_FILE}' not found.")

with open(CONFIG_FILE, 'r') as f:
    CONFIG = json.load(f)

# Set up logging with rotation
logger = logging.getLogger('AgentSystem')
logger.setLevel(logging.INFO)

log_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')

# General Log Handler
log_handler = RotatingFileHandler(
    CONFIG.get('log_file', 'agent_system.log'),
    maxBytes=CONFIG.get('max_log_size', 10 * 1024 * 1024),
    backupCount=CONFIG.get('backup_count', 5)
)
log_handler.setFormatter(log_formatter)
logger.addHandler(log_handler)

# Error Log Handler
error_handler = RotatingFileHandler(
    CONFIG.get('error_log_file', 'error.log'),
    maxBytes=CONFIG.get('max_log_size', 10 * 1024 * 1024),
    backupCount=CONFIG.get('backup_count', 5)
)
error_handler.setFormatter(log_formatter)
error_handler.setLevel(logging.ERROR)
logger.addHandler(error_handler)

# Console Handler
console_handler = logging.StreamHandler()
console_handler.setFormatter(log_formatter)
logger.addHandler(console_handler)

# Google API key setup
GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
if not GOOGLE_API_KEY:
    logger.error("GOOGLE_API_KEY environment variable is not set. Exiting.")
    sys.exit(1)

# Configure the Generative AI model
genai.configure(api_key=GOOGLE_API_KEY)

# Initialize Gemini model
model = genai.GenerativeModel('gemini-1.5-flash')

# YOLO model setup
YOLO_MODEL_PATH = CONFIG.get('yolo_model_path', 'yolov8n.pt')
if not os.path.exists(YOLO_MODEL_PATH):
    logger.error(f"YOLO model file '{YOLO_MODEL_PATH}' not found. Exiting.")
    sys.exit(1)

yolo = YOLO(YOLO_MODEL_PATH)

ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg"}

def allowed_file(filename: str) -> bool:
    """Check if the filename has an allowed extension."""
    _, ext = os.path.splitext(filename)
    return ext.lstrip('.').lower() in ALLOWED_EXTENSIONS

class PriorityTask:
    """A task with priority for the PriorityQueue."""
    def __init__(self, priority: int, task: str):
        self.priority = priority
        self.task = task

    def __lt__(self, other):
        return self.priority < other.priority

class Agent(ABC):
    """Abstract base class for all agents."""
    def __init__(self, name: str, role: str, system_prompt: str):
        self.name = name
        self.role = role
        self.system_prompt = system_prompt
        self.chat = model.start_chat(history=[])
        self.knowledge_base: Dict[str, Any] = {}
        self.lock = threading.Lock()
    def send_message(self, message: str, image: Image.Image = None) -> str:
        logger.info(f"{self.name}: Sending message to Gemini: {message}")
        try:
            if image:
                response = self.chat.send_message([message, image])
            else:
                response = self.chat.send_message(message)
            logger.info(f"{self.name}: Received response from Gemini: {response.text}")
            return response.text
        except Exception as e:
            logger.error(f"{self.name}: Error sending message to Gemini: {e}")
            return f"Error: {str(e)}"

    @abstractmethod
    def think(self) -> None:
        """Abstract method for agent's thinking process."""
        pass

    def update_knowledge(self, key: str, value: Any) -> None:
        """Update the agent's knowledge base."""
        with self.lock:
            self.knowledge_base[key] = value

    def get_knowledge(self, key: str) -> Any:
        """Retrieve information from the knowledge base."""
        return self.knowledge_base.get(key)

    def share_knowledge(self, other_agent: 'Agent') -> None:
        """Share knowledge with another agent."""
        with self.lock:
            other_agent.knowledge_base.update(self.knowledge_base)

class GeneralActionExecutor:
    """Executes mouse and keyboard actions based on commands."""
    def __init__(self):
        self.action_map: Dict[str, Callable[..., None]] = {
            "click": self.click,
            "double_click": self.double_click,
            "right_click": self.right_click,
            "move": self.move,
            "type": self.type_keys,
            "scroll": self.scroll,
            "drag": self.drag,
            "hotkey": self.hotkey,
            "image_click": self.image_click
            # Add more actions as needed
        }

    def execute(self, action: str, **kwargs) -> str:
        """Execute a given action with provided keyword arguments."""
        logger.info(f"Executing action: {action} with args: {kwargs}")
        action_func = self.action_map.get(action.lower())
        if not action_func:
            logger.error(f"Action '{action}' is not supported.")
            return f"Action '{action}' is not supported."
        try:
            action_func(**kwargs)
            return f"Action '{action}' executed successfully."
        except Exception as e:
            logger.error(f"Error executing action '{action}': {e}")
            return f"Error executing action '{action}': {e}"

    def click(self, x: int = None, y: int = None, button: str = 'left', num_clicks: int = 1, interval: float = 0.0) -> None:
        """Performs a mouse click."""
        pyautogui.click(x=x, y=y, button=button, clicks=num_clicks, interval=interval)

    def double_click(self, x: int = None, y: int = None, button: str = 'left') -> None:
        """Performs a mouse double-click."""
        pyautogui.doubleClick(x=x, y=y, button=button)

    def right_click(self, x: int = None, y: int = None) -> None:
        """Performs a right mouse click."""
        pyautogui.rightClick(x=x, y=y)

    def move(self, x: int, y: int, duration: float = 0.5, tween: Callable = pyautogui.linear) -> None:
        """Moves the mouse to a specific position."""
        pyautogui.moveTo(x, y, duration=duration, tween=tween)

    def type_keys(self, text: str, interval: float = 0.05) -> None:
        """Types the given text using the keyboard."""
        pyautogui.write(text, interval=interval)

    def scroll(self, clicks: int) -> None:
        """Scrolls the mouse."""
        pyautogui.scroll(clicks)

    def drag(self, start_x: int, start_y: int, end_x: int, end_y: int, duration: float = 0.5, button: str = 'left') -> None:
        """Drags the mouse from start to end coordinates."""
        pyautogui.moveTo(start_x, start_y, duration=duration)
        pyautogui.dragTo(end_x, end_y, duration=duration, button=button)

    def hotkey(self, keys: List[str], interval: float = 0.1) -> None:
        """Presses a combination of keys."""
        pyautogui.hotkey(*keys, interval=interval)

    def image_click(self, image_path: str, confidence: float = 0.8) -> None:
        """Clicks on the center of the detected image."""
        location = pyautogui.locateCenterOnScreen(image_path, confidence=confidence)
        if location:
            pyautogui.click(location.x, location.y)
        else:
            logger.error(f"Image '{image_path}' not found on screen.")
            raise FileNotFoundError(f"Image '{image_path}' not found on screen.")

class MasterAgent(Agent):
    """Master agent that manages tasks, takes screenshots, and executes actions."""
    def __init__(self, name: str = "Master"):
        super().__init__(
            name=name,
            role="Control, delegate, and execute actions",
            system_prompt="You are the master agent. Your role is to analyze tasks, execute actions, and manage operations."
        )
        self.action_executor = GeneralActionExecutor()
        self.task_queue: PriorityQueue = PriorityQueue()
        self.executor = ThreadPoolExecutor(max_workers=CONFIG.get('max_workers', 10))
        self.screenshot_interval: int = CONFIG.get('screenshot_interval', 60)  # Screenshot interval in seconds
        self.yolo_model = yolo
        self.running = True
        self.screenshot_counter: int = 0
        self.screenshot_lock = threading.Lock()

    def delegate_task(self, task_description: str, priority: int = 5) -> str:
        """Delegate a task based on its description with a given priority."""
        logger.info(f"{self.name}: Received task: {task_description} with priority {priority}")
        self.task_queue.put(PriorityTask(priority, task_description))
        return "Task added to queue for processing."

    def think(self) -> None:
        """Process the task queue and execute tasks."""
        logger.info(f"{self.name}: Thinking...")
        self.process_task_queue()

    def process_task_queue(self) -> None:
        """Process tasks in the queue using a thread pool."""
        logger.info(f"{self.name}: Processing task queue. Current queue size: {self.task_queue.qsize()}")
        tasks = []
        while not self.task_queue.empty():
            priority_task = self.task_queue.get()
            task = priority_task.task
            logger.info(f"{self.name}: Processing task: {task} with priority {priority_task.priority}")
            tasks.append(self.executor.submit(self.process_task, task))

        for future in as_completed(tasks):
            try:
                result = future.result()
                logger.info(f"{self.name}: Task result: {result}")
            except Exception as e:
                logger.error(f"{self.name}: Task execution error: {e}")

    def process_task(self, task: str) -> str:
        """Process an individual task."""
        logger.info(f"{self.name}: Starting task processing: {task}")
        if task.startswith("action:"):
            action_command = task[len("action:"):].strip()
            response = self.execute_action_command(action_command)
            logger.info(f"{self.name}: Action response: {response}")
            return response
        elif task.startswith("screenshot_analysis:"):
            analysis = task[len("screenshot_analysis:"):].strip()
            self.handle_screenshot_analysis(analysis)
            return "Screenshot analysis handled."
        else:
            response = self.send_message(f"Analyze this task and suggest actions: {task}")
            logger.info(f"{self.name}: Task analysis: {response}")
            self.parse_and_enqueue_actions(response)
            return "Task analyzed and actions enqueued."

    def execute_action_command(self, command: str) -> str:
        """Parse and execute an action command."""
        try:
            if ':' not in command:
                return "Invalid action command format."
            action_type, params = command.split(':', 1)
            param_dict = dict(re.findall(r'(\w+)=([\w\.\-\/\\]+)', params))

            # Convert parameter values to appropriate types
            for key, value in param_dict.items():
                if value.isdigit():
                    param_dict[key] = int(value)
                else:
                    try:
                        param_dict[key] = float(value)
                    except ValueError:
                        pass  # Keep as string
            logger.info(f"{self.name}: Executing action '{action_type}' with params {param_dict}")
            return self.action_executor.execute(action_type, **param_dict)
        except Exception as e:
            logger.error(f"{self.name}: Error parsing action command '{command}': {e}")
            return f"Error parsing action command: {e}"

    def parse_and_enqueue_actions(self, analysis: str) -> None:
        """Parse the analysis from the model and enqueue actionable tasks."""
        lines = analysis.split('\n')
        for line in lines:
            if line.lower().startswith("execute action:"):
                action_command = line[len("execute action:"):].strip()
                self.task_queue.put(PriorityTask(priority=5, task=f"action:{action_command}"))

    def handle_screenshot_analysis(self, analysis: str) -> None:
        """Handle the results of a screenshot analysis."""
        logger.info(f"{self.name}: Handling screenshot analysis: {analysis}")
        # Example: If certain text is found, enqueue specific tasks
        if "error" in analysis.lower():
            self.delegate_task("action:notify_user:message=Error detected in screenshot", priority=1)
        elif "button" in analysis.lower():
            self.delegate_task("action:click:x=100:y=200", priority=3)
        # Add more conditional handling as needed

    def start_screenshot_thread(self) -> None:
        """Start a background thread to take periodic screenshots."""
        logger.info(f"{self.name}: Starting screenshot thread")
        thread = threading.Thread(target=self.take_screenshots, daemon=True)
        thread.start()

    def take_screenshots(self) -> None:
        """Continuously take screenshots at specified intervals."""
        logger.info(f"{self.name}: Screenshot thread started with interval {self.screenshot_interval} seconds")
        while self.running:
            start_time = time.time()
            try:
                logger.info(f"{self.name}: Taking screenshot")
                with self.screenshot_lock:
                    screenshot = pyautogui.screenshot()
                # Process screenshot in-memory
                screenshot_bytes = io.BytesIO()
                screenshot.save(screenshot_bytes, format='PNG')
                screenshot_bytes = screenshot_bytes.getvalue()
                analysis = self.analyze_screenshot_bytes(screenshot_bytes)
                self.task_queue.put(PriorityTask(priority=4, task=f"screenshot_analysis:{analysis}"))
            except Exception as e:
                logger.error(f"{self.name}: Error taking screenshot: {e}")
                self.task_queue.put(PriorityTask(priority=2, task=f"Handle failed screenshot analysis: {e}"))
            # Sleep until next interval
            elapsed = time.time() - start_time
            sleep_time = max(0, self.screenshot_interval - elapsed)
            time.sleep(sleep_time)

    def analyze_screenshot_bytes(self, image_bytes: bytes) -> str:
        """Analyze a screenshot using both Gemini and YOLO."""
        logger.info(f"{self.name}: Analyzing screenshot with Gemini and YOLO")
        try:
            image = Image.open(io.BytesIO(image_bytes))
            # Textual analysis with Gemini
            text_analysis = self.send_message("Analyze this screenshot and describe its contents:", image)
            # Visual analysis with YOLO
            np_image = np.array(image)
            results = self.yolo_model(np_image)
            yolo_analysis = results_to_text(results)
            combined_analysis = f"Text Analysis:\n{text_analysis}\n\nObject Detection:\n{yolo_analysis}"
            logger.info(f"{self.name}: Combined analysis: {combined_analysis}")
            return combined_analysis
        except Exception as e:
            logger.error(f"{self.name}: Error analyzing screenshot: {e}")
            return f"Error analyzing screenshot: {e}"

    def stop(self) -> None:
        """Stop the agent's operations gracefully."""
        logger.info(f"{self.name}: Stopping agent.")
        self.running = False
        self.executor.shutdown(wait=True)
        logger.info(f"{self.name}: Agent stopped.")

def results_to_text(results) -> str:
    """Convert YOLO results to a readable text format."""
    detected_objects = []
    for result in results:
        boxes = result.boxes
        for box in boxes:
            cls_id = int(box.cls[0])
            cls_name = result.names[cls_id] if hasattr(result, 'names') else f"Class {cls_id}"
            score = box.conf[0]
            detected_objects.append(f"{cls_name} ({score:.2f}) at {box.xyxy[0].tolist()}")
    return "\n".join(detected_objects) if detected_objects else "No objects detected."

class AgentCoordinator:
    """Coordinator that manages one or multiple MasterAgents."""
    def __init__(self, num_agents: int = 1):
        self.master_agents: List[MasterAgent] = [MasterAgent(name=f"Master_{i+1}") for i in range(num_agents)]
        self.running = True

    def run(self) -> None:
        """Run the coordinator to manage agents."""
        logger.info("AgentCoordinator: Starting coordinator.")
        try:
            for master in self.master_agents:
                master.start_screenshot_thread()

            loop = asyncio.get_event_loop()
            loop.run_until_complete(self.main_loop())
        except KeyboardInterrupt:
            logger.info("AgentCoordinator: Received KeyboardInterrupt. Shutting down.")
        except Exception as e:
            logger.error(f"AgentCoordinator: Encountered an error: {e}")
        finally:
            self.cleanup()

    async def main_loop(self):
        """Main loop to periodically invoke agents' think methods."""
        while self.running:
            for master in self.master_agents:
                master.think()
            await asyncio.sleep(5)  # Adjust the sleep duration as needed

    def cleanup(self) -> None:
        """Clean up all master agents."""
        logger.info("AgentCoordinator: Cleaning up agents.")
        for master in self.master_agents:
            master.stop()
        logger.info("AgentCoordinator: All agents have been cleaned up.")

def signal_handler(sig, frame, coordinator: AgentCoordinator):
    logger.info(f"Received signal {sig}. Initiating shutdown.")
    coordinator.running = False
    coordinator.cleanup()
    sys.exit(0)

def main() -> None:
    """Main entry point for the agent system."""
    num_agents = CONFIG.get('num_agents', 1)  # Configurable number of MasterAgents
    coordinator = AgentCoordinator(num_agents=num_agents)

    # Register signal handlers for graceful shutdown
    signal.signal(signal.SIGINT, lambda sig, frame: signal_handler(sig, frame, coordinator))
    signal.signal(signal.SIGTERM, lambda sig, frame: signal_handler(sig, frame, coordinator))

    coordinator.run()

if __name__ == "__main__":
    main()