MP-Server/macro_manager.py

# Macro management and execution with command sequence support

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
        macro_data = {
            "name": name,
            "type": "sequence",
            "commands": 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
        self.macros[macro_id] = {
            "name": name,
            "type": "sequence",
            "commands": 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()