MP-Server/web_server.py

# FastAPI web server for MacroPad

import os
import sys
import asyncio
from typing import List, Optional
from contextlib import asynccontextmanager

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, UploadFile, File, HTTPException
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, HTMLResponse
from pydantic import BaseModel
import uvicorn

from config import DEFAULT_PORT, VERSION


def get_resource_path(relative_path):
    """Get the path to a bundled resource file."""
    if getattr(sys, 'frozen', False):
        base_path = sys._MEIPASS
    else:
        base_path = os.path.dirname(os.path.abspath(__file__))
    return os.path.join(base_path, relative_path)


class Command(BaseModel):
    """Single command in a macro sequence."""
    type: str  # text, key, hotkey, wait, app
    value: Optional[str] = None
    keys: Optional[List[str]] = None
    ms: Optional[int] = None
    command: Optional[str] = None


class MacroCreate(BaseModel):
    """Request model for creating a macro."""
    name: str
    commands: List[Command]
    category: Optional[str] = ""


class MacroUpdate(BaseModel):
    """Request model for updating a macro."""
    name: str
    commands: List[Command]
    category: Optional[str] = ""


class ExecuteRequest(BaseModel):
    """Request model for executing a macro."""
    macro_id: str


class TabCreate(BaseModel):
    """Request model for creating a tab."""
    name: str


class ConnectionManager:
    """Manages WebSocket connections for real-time updates."""

    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        if websocket in self.active_connections:
            self.active_connections.remove(websocket)

    async def broadcast(self, message: dict):
        """Send message to all connected clients."""
        for connection in self.active_connections:
            try:
                await connection.send_json(message)
            except Exception:
                pass


class WebServer:
    """FastAPI-based web server for MacroPad."""

    def __init__(self, macro_manager, app_dir: str, port: int = DEFAULT_PORT):
        self.macro_manager = macro_manager
        self.app_dir = app_dir
        self.port = port
        self.app = None
        self.manager = ConnectionManager()
        self.server = None

    def create_app(self) -> FastAPI:
        """Create FastAPI application."""

        @asynccontextmanager
        async def lifespan(app: FastAPI):
            yield

        app = FastAPI(
            title="MacroPad Server",
            version=VERSION,
            lifespan=lifespan
        )

        # Serve static files from web directory (bundled with app)
        web_dir = get_resource_path("web")
        if os.path.exists(web_dir):
            app.mount("/static", StaticFiles(directory=web_dir), name="static")

        # Serve macro images (user data directory)
        images_dir = os.path.join(self.app_dir, "macro_images")
        if os.path.exists(images_dir):
            app.mount("/images", StaticFiles(directory=images_dir), name="images")

        @app.get("/", response_class=HTMLResponse)
        async def index():
            """Serve the main PWA page."""
            index_path = os.path.join(web_dir, "index.html")
            if os.path.exists(index_path):
                return FileResponse(index_path, media_type="text/html")
            return HTMLResponse("<h1>MacroPad Server</h1><p>Web interface not found.</p>")

        @app.get("/manifest.json")
        async def manifest():
            """Serve PWA manifest."""
            manifest_path = os.path.join(web_dir, "manifest.json")
            if os.path.exists(manifest_path):
                return FileResponse(manifest_path, media_type="application/json")
            raise HTTPException(status_code=404, detail="Manifest not found")

        @app.get("/service-worker.js")
        async def service_worker():
            """Serve service worker."""
            sw_path = os.path.join(web_dir, "service-worker.js")
            if os.path.exists(sw_path):
                return FileResponse(sw_path, media_type="application/javascript")
            raise HTTPException(status_code=404, detail="Service worker not found")

        @app.get("/api/tabs")
        async def get_tabs():
            """Get available tab categories."""
            return {"tabs": self.macro_manager.get_unique_tabs()}

        @app.get("/api/macros")
        async def get_macros():
            """Get all macros."""
            macros = self.macro_manager.get_all_macros()
            return {"macros": macros}

        @app.get("/api/macros/{tab}")
        async def get_macros_by_tab(tab: str):
            """Get macros filtered by tab/category."""
            all_macros = self.macro_manager.get_sorted_macros()
            filtered = self.macro_manager.filter_macros_by_tab(all_macros, tab)
            return {"macros": dict(filtered)}

        @app.get("/api/macro/{macro_id}")
        async def get_macro(macro_id: str):
            """Get a single macro by ID."""
            macro = self.macro_manager.get_macro(macro_id)
            if macro:
                return {"macro": macro}
            raise HTTPException(status_code=404, detail="Macro not found")

        @app.post("/api/execute")
        async def execute_macro(request: ExecuteRequest):
            """Execute a macro by ID."""
            success = self.macro_manager.execute_macro(request.macro_id)
            if success:
                # Broadcast execution to all connected clients
                await self.manager.broadcast({
                    "type": "executed",
                    "macro_id": request.macro_id
                })
                return {"success": True}
            raise HTTPException(status_code=404, detail="Macro not found or execution failed")

        @app.post("/api/macros")
        async def create_macro(macro: MacroCreate):
            """Create a new macro."""
            commands = [cmd.model_dump(exclude_none=True) for cmd in macro.commands]
            macro_id = self.macro_manager.add_macro(
                name=macro.name,
                commands=commands,
                category=macro.category or ""
            )
            # Broadcast update
            await self.manager.broadcast({"type": "macro_created", "macro_id": macro_id})
            return {"success": True, "macro_id": macro_id}

        @app.put("/api/macros/{macro_id}")
        async def update_macro(macro_id: str, macro: MacroUpdate):
            """Update an existing macro."""
            commands = [cmd.model_dump(exclude_none=True) for cmd in macro.commands]
            success = self.macro_manager.update_macro(
                macro_id=macro_id,
                name=macro.name,
                commands=commands,
                category=macro.category or ""
            )
            if success:
                await self.manager.broadcast({"type": "macro_updated", "macro_id": macro_id})
                return {"success": True}
            raise HTTPException(status_code=404, detail="Macro not found")

        @app.delete("/api/macros/{macro_id}")
        async def delete_macro(macro_id: str):
            """Delete a macro."""
            success = self.macro_manager.delete_macro(macro_id)
            if success:
                await self.manager.broadcast({"type": "macro_deleted", "macro_id": macro_id})
                return {"success": True}
            raise HTTPException(status_code=404, detail="Macro not found")

        @app.post("/api/upload-image")
        async def upload_image(file: UploadFile = File(...)):
            """Upload an image for a macro."""
            if not file.content_type or not file.content_type.startswith("image/"):
                raise HTTPException(status_code=400, detail="File must be an image")

            # Save to temp location
            import tempfile
            import shutil

            ext = os.path.splitext(file.filename)[1] if file.filename else ".png"
            with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp:
                shutil.copyfileobj(file.file, tmp)
                return {"path": tmp.name}

        @app.get("/api/image/{image_path:path}")
        async def get_image(image_path: str):
            """Get macro image (legacy compatibility)."""
            full_path = os.path.join(self.app_dir, image_path)
            if os.path.exists(full_path):
                return FileResponse(full_path)
            raise HTTPException(status_code=404, detail="Image not found")

        @app.websocket("/ws")
        async def websocket_endpoint(websocket: WebSocket):
            """WebSocket for real-time updates."""
            await self.manager.connect(websocket)
            try:
                while True:
                    data = await websocket.receive_json()
                    # Handle incoming messages if needed
                    if data.get("type") == "ping":
                        await websocket.send_json({"type": "pong"})
            except WebSocketDisconnect:
                self.manager.disconnect(websocket)

        self.app = app
        return app

    def run(self):
        """Run the web server (blocking)."""
        if not self.app:
            self.create_app()

        config = uvicorn.Config(
            self.app,
            host="0.0.0.0",
            port=self.port,
            log_level="warning",
            log_config=None  # Disable default logging config for PyInstaller compatibility
        )
        self.server = uvicorn.Server(config)
        self.server.run()

    def stop(self):
        """Stop the web server."""
        if self.server:
            self.server.should_exit = True

    async def run_async(self):
        """Run the web server asynchronously."""
        if not self.app:
            self.create_app()

        config = uvicorn.Config(
            self.app,
            host="0.0.0.0",
            port=self.port,
            log_level="warning",
            log_config=None  # Disable default logging config for PyInstaller compatibility
        )
        self.server = uvicorn.Server(config)
        await self.server.serve()