MP-Server/macro_manager.py

# Macro management and execution with command sequence support

import copy
import json
import os
import uuid
import pyautogui
import subprocess
import time
from PIL import Image
from typing import Optional


class MacroManager:
    """Manages macro storage, migration, and execution with command sequences."""

    def __init__(self, data_file: str, images_dir: str, app_dir: str):
        self.data_file = data_file
        self.images_dir = images_dir
        self.app_dir = app_dir
        self.macros = {}
        self.load_macros()

    def load_macros(self):
        """Load macros from JSON file and migrate if needed."""
        try:
            if os.path.exists(self.data_file):
                with open(self.data_file, "r") as file:
                    self.macros = json.load(file)

                # Migrate old format macros
                migrated = False
                for macro_id, macro in list(self.macros.items()):
                    if macro.get("type") != "sequence":
                        self.macros[macro_id] = self._migrate_macro(macro)
                        migrated = True

                if migrated:
                    self.save_macros()
                    print("Migrated macros to new command sequence format")

        except Exception as e:
            print(f"Error loading macros: {e}")
            self.macros = {}

    def _migrate_macro(self, old_macro: dict) -> dict:
        """Convert old macro format to new command sequence format."""
        if old_macro.get("type") == "sequence":
            return old_macro

        commands = []
        modifiers = old_macro.get("modifiers", {})

        if old_macro.get("type") == "text":
            # Build held keys list
            held_keys = []
            for mod in ["ctrl", "alt", "shift"]:
                if modifiers.get(mod):
                    held_keys.append(mod)

            if held_keys:
                # Use hotkey for modified text
                commands.append({
                    "type": "hotkey",
                    "keys": held_keys + [old_macro.get("command", "")]
                })
            else:
                # Plain text
                commands.append({
                    "type": "text",
                    "value": old_macro.get("command", "")
                })

            # Add enter if requested
            if modifiers.get("enter"):
                commands.append({"type": "key", "value": "enter"})

        elif old_macro.get("type") == "app":
            commands.append({
                "type": "app",
                "command": old_macro.get("command", "")
            })

        return {
            "name": old_macro.get("name", "Unnamed"),
            "type": "sequence",
            "commands": commands,
            "category": old_macro.get("category", ""),
            "image_path": old_macro.get("image_path", ""),
            "last_used": old_macro.get("last_used", 0)
        }

    def save_macros(self):
        """Save macros to JSON file."""
        try:
            with open(self.data_file, "w") as file:
                json.dump(self.macros, file, indent=4)
        except Exception as e:
            print(f"Error saving macros: {e}")

    def get_sorted_macros(self, sort_by: str = "name"):
        """Get macros sorted by specified criteria."""
        macro_list = list(self.macros.items())

        if sort_by == "name":
            macro_list.sort(key=lambda x: x[1]["name"].lower())
        elif sort_by == "type":
            # Sort by first command type in sequence
            def get_first_type(macro):
                cmds = macro.get("commands", [])
                return cmds[0].get("type", "") if cmds else ""
            macro_list.sort(key=lambda x: (get_first_type(x[1]), x[1]["name"].lower()))
        elif sort_by == "recent":
            macro_list.sort(
                key=lambda x: (x[1].get("last_used", 0), x[1]["name"].lower()),
                reverse=True
            )

        return macro_list

    def filter_macros_by_tab(self, macro_list: list, tab_name: str):
        """Filter macros based on tab/category name."""
        if tab_name == "All":
            return macro_list

        filtered = []
        for macro_id, macro in macro_list:
            # Check category match
            if macro.get("category") == tab_name:
                filtered.append((macro_id, macro))
            # Check first command type match
            elif macro.get("commands"):
                first_type = macro["commands"][0].get("type", "").title()
                if first_type == tab_name:
                    filtered.append((macro_id, macro))

        return filtered

    def get_unique_tabs(self) -> list:
        """Get list of unique tabs based on categories."""
        tabs = ["All"]
        categories = set()

        for macro in self.macros.values():
            if macro.get("category"):
                categories.add(macro["category"])

        for category in sorted(categories):
            if category and category not in tabs:
                tabs.append(category)

        return tabs

    def add_macro(
        self,
        name: str,
        commands: list,
        category: str = "",
        image_path: str = ""
    ) -> str:
        """Add a new macro with command sequence."""
        macro_id = str(uuid.uuid4())

        # Process image if provided
        image_path_reference = self._process_image(image_path) if image_path else ""

        # Create macro data (deep copy commands to avoid reference issues)
        macro_data = {
            "name": name,
            "type": "sequence",
            "commands": copy.deepcopy(commands),
            "category": category,
            "image_path": image_path_reference,
            "last_used": 0
        }

        self.macros[macro_id] = macro_data
        self.save_macros()
        return macro_id

    def update_macro(
        self,
        macro_id: str,
        name: str,
        commands: list,
        category: str = "",
        image_path: Optional[str] = None
    ) -> bool:
        """Update an existing macro."""
        if macro_id not in self.macros:
            return False

        macro = self.macros[macro_id]

        # Keep old image or update with new one
        if image_path is not None:
            image_path_reference = self._process_image(image_path) if image_path else ""
        else:
            image_path_reference = macro.get("image_path", "")

        # Update macro data (deep copy commands to avoid reference issues)
        self.macros[macro_id] = {
            "name": name,
            "type": "sequence",
            "commands": copy.deepcopy(commands),
            "category": category,
            "image_path": image_path_reference,
            "last_used": macro.get("last_used", 0)
        }

        self.save_macros()
        return True

    def _process_image(self, image_path: str) -> str:
        """Process and store an image for a macro."""
        if not image_path:
            return ""

        try:
            file_ext = os.path.splitext(image_path)[1].lower()
            unique_filename = f"{uuid.uuid4().hex}{file_ext}"
            dest_path = os.path.join(self.images_dir, unique_filename)

            # Resize image to max 256x256
            with Image.open(image_path) as img:
                img.thumbnail((256, 256))
                img.save(dest_path)

            return os.path.join("macro_images", unique_filename)
        except Exception as e:
            print(f"Error processing image: {e}")
            return ""

    def delete_macro(self, macro_id: str) -> bool:
        """Delete a macro."""
        if macro_id not in self.macros:
            return False

        macro = self.macros[macro_id]

        # Delete associated image file
        if macro.get("image_path"):
            try:
                img_path = os.path.join(self.app_dir, macro["image_path"])
                if os.path.exists(img_path):
                    os.remove(img_path)
            except Exception as e:
                print(f"Error removing image file: {e}")

        del self.macros[macro_id]
        self.save_macros()
        return True

    def execute_macro(self, macro_id: str) -> bool:
        """Execute a macro by ID."""
        if macro_id not in self.macros:
            return False

        macro = self.macros[macro_id]

        # Update last_used timestamp
        self.macros[macro_id]["last_used"] = time.time()
        self.save_macros()

        try:
            commands = macro.get("commands", [])
            for cmd in commands:
                self._execute_command(cmd)
            return True
        except Exception as e:
            print(f"Error executing macro: {e}")
            return False

    def _execute_command(self, cmd: dict):
        """Execute a single command from a sequence."""
        cmd_type = cmd.get("type", "")

        if cmd_type == "text":
            # Type text string using clipboard for Unicode support
            value = cmd.get("value", "")
            if value:
                try:
                    import pyperclip
                    # Save current clipboard, paste text, restore (optional)
                    pyperclip.copy(value)
                    pyautogui.hotkey("ctrl", "v")
                    time.sleep(0.05)  # Small delay after paste
                except ImportError:
                    # Fallback to typewrite for ASCII-only
                    if len(value) == 1:
                        pyautogui.press(value)
                    else:
                        pyautogui.typewrite(value, interval=0.02)

        elif cmd_type == "key":
            # Press a single key
            key = cmd.get("value", "")
            if key:
                pyautogui.press(key)

        elif cmd_type == "hotkey":
            # Press key combination
            keys = cmd.get("keys", [])
            if keys:
                pyautogui.hotkey(*keys)

        elif cmd_type == "wait":
            # Delay in milliseconds
            ms = cmd.get("ms", 0)
            if ms > 0:
                time.sleep(ms / 1000.0)

        elif cmd_type == "app":
            # Launch application
            command = cmd.get("command", "")
            if command:
                subprocess.Popen(command, shell=True)

    # Legacy API compatibility methods
    def add_macro_legacy(
        self,
        name: str,
        macro_type: str,
        command: str,
        category: str = "",
        modifiers: Optional[dict] = None,
        image_path: str = ""
    ) -> str:
        """Add macro using legacy format (auto-converts to sequence)."""
        if modifiers is None:
            modifiers = {"ctrl": False, "alt": False, "shift": False, "enter": False}

        # Build command sequence
        commands = []
        held_keys = []
        for mod in ["ctrl", "alt", "shift"]:
            if modifiers.get(mod):
                held_keys.append(mod)

        if macro_type == "text":
            if held_keys:
                commands.append({"type": "hotkey", "keys": held_keys + [command]})
            else:
                commands.append({"type": "text", "value": command})

            if modifiers.get("enter"):
                commands.append({"type": "key", "value": "enter"})
        elif macro_type == "app":
            commands.append({"type": "app", "command": command})

        return self.add_macro(name, commands, category, image_path)

    def get_macro(self, macro_id: str) -> Optional[dict]:
        """Get a macro by ID."""
        return self.macros.get(macro_id)

    def get_all_macros(self) -> dict:
        """Get all macros."""
        return self.macros.copy()