Modernize application to v0.9.0 with PySide6, FastAPI, and PWA support
## Major Changes ### Build System - Replace requirements.txt with pyproject.toml for modern dependency management - Support for uv package manager alongside pip - Update PyInstaller spec files for new dependencies and structure ### Desktop GUI (Tkinter → PySide6) - Complete rewrite of UI using PySide6/Qt6 - New modular structure in gui/ directory: - main_window.py: Main application window - macro_editor.py: Macro creation/editing dialog - command_builder.py: Visual command sequence builder - Modern dark theme with consistent styling - System tray integration ### Web Server (Flask → FastAPI) - Migrate from Flask/Waitress to FastAPI/Uvicorn - Add WebSocket support for real-time updates - Full CRUD API for macro management - Image upload endpoint ### Web Interface → PWA - New web/ directory with standalone static files - PWA manifest and service worker for installability - Offline caching support - Full macro editing from web interface - Responsive mobile-first design - Command builder UI matching desktop functionality ### Macro System Enhancement - New command sequence model replacing simple text/app types - Command types: text, key, hotkey, wait, app - Support for delays between commands (wait in ms) - Support for key presses between commands (enter, tab, etc.) - Automatic migration of existing macros to new format - Backward compatibility maintained ### Files Added - pyproject.toml - gui/__init__.py, main_window.py, macro_editor.py, command_builder.py - gui/widgets/__init__.py - web/index.html, manifest.json, service-worker.js - web/css/styles.css, web/js/app.js - web/icons/icon-192.png, icon-512.png ### Files Removed - requirements.txt (replaced by pyproject.toml) - ui_components.py (replaced by gui/ modules) - web_templates.py (replaced by web/ static files) - main.spec (consolidated into platform-specific specs) ### Files Modified - main.py: Simplified entry point for PySide6 - macro_manager.py: Command sequence model and migration - web_server.py: FastAPI implementation - config.py: Version bump to 0.9.0 - All .spec files: Updated for PySide6 and new structure - README.md: Complete rewrite for v0.9.0 - .gitea/workflows/release.yml: Disabled pending build testing 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
343
web_server.py
343
web_server.py
@@ -1,99 +1,272 @@
|
||||
# Web server component for MacroPad
|
||||
# FastAPI web server for MacroPad
|
||||
|
||||
from flask import Flask, render_template_string, request, jsonify, send_file
|
||||
from waitress import serve
|
||||
import logging
|
||||
import os
|
||||
from web_templates import INDEX_HTML
|
||||
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
|
||||
|
||||
|
||||
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:
|
||||
def __init__(self, macro_manager, app_dir, port=40000):
|
||||
"""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.app_dir = app_dir
|
||||
self.port = port
|
||||
self.app = None
|
||||
|
||||
def create_app(self):
|
||||
"""Create and configure Flask application"""
|
||||
app = Flask(__name__)
|
||||
|
||||
# Disable Flask's logging except for errors
|
||||
log = logging.getLogger('werkzeug')
|
||||
log.setLevel(logging.ERROR)
|
||||
|
||||
@app.route('/')
|
||||
def index():
|
||||
return render_template_string(INDEX_HTML)
|
||||
|
||||
@app.route('/api/tabs')
|
||||
def get_tabs():
|
||||
"""Get all available tabs (similar to setup_tabs logic)"""
|
||||
tabs = ["All"]
|
||||
|
||||
# Add tabs based on macro types and custom categories
|
||||
unique_types = set()
|
||||
for macro in self.macro_manager.macros.values():
|
||||
if macro.get("type"):
|
||||
unique_types.add(macro["type"].title())
|
||||
# Check for custom category
|
||||
if macro.get("category"):
|
||||
unique_types.add(macro["category"])
|
||||
|
||||
for tab_type in sorted(unique_types):
|
||||
if tab_type not in ["All"]:
|
||||
tabs.append(tab_type)
|
||||
|
||||
return jsonify(tabs)
|
||||
|
||||
@app.route('/api/macros')
|
||||
def get_macros():
|
||||
return jsonify(self.macro_manager.macros)
|
||||
|
||||
@app.route('/api/macros/<tab_name>')
|
||||
def get_macros_by_tab(tab_name):
|
||||
"""Filter macros by tab (similar to filter_macros_by_tab logic)"""
|
||||
if tab_name == "All":
|
||||
return jsonify(self.macro_manager.macros)
|
||||
|
||||
filtered_macros = {}
|
||||
for macro_id, macro in self.macro_manager.macros.items():
|
||||
# Check type match
|
||||
if macro.get("type", "").title() == tab_name:
|
||||
filtered_macros[macro_id] = macro
|
||||
# Check custom category match
|
||||
elif macro.get("category") == tab_name:
|
||||
filtered_macros[macro_id] = macro
|
||||
|
||||
return jsonify(filtered_macros)
|
||||
|
||||
@app.route('/api/image/<path:image_path>')
|
||||
def get_image(image_path):
|
||||
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
|
||||
web_dir = os.path.join(self.app_dir, "web")
|
||||
if os.path.exists(web_dir):
|
||||
app.mount("/static", StaticFiles(directory=web_dir), name="static")
|
||||
|
||||
# Serve macro images
|
||||
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:
|
||||
image_path = os.path.join(self.app_dir, image_path)
|
||||
return send_file(image_path)
|
||||
except Exception as e:
|
||||
return str(e), 404
|
||||
|
||||
@app.route('/api/execute', methods=['POST'])
|
||||
def execute_macro():
|
||||
data = request.get_json()
|
||||
if not data or 'macro_id' not in data:
|
||||
return jsonify({"success": False, "error": "Invalid request"})
|
||||
|
||||
macro_id = data['macro_id']
|
||||
success = self.macro_manager.execute_macro(macro_id)
|
||||
return jsonify({"success": success})
|
||||
|
||||
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"""
|
||||
"""Run the web server (blocking)."""
|
||||
if not self.app:
|
||||
self.create_app()
|
||||
|
||||
try:
|
||||
serve(self.app, host='0.0.0.0', port=self.port, threads=4)
|
||||
except Exception as e:
|
||||
raise e
|
||||
|
||||
config = uvicorn.Config(
|
||||
self.app,
|
||||
host="0.0.0.0",
|
||||
port=self.port,
|
||||
log_level="warning"
|
||||
)
|
||||
server = uvicorn.Server(config)
|
||||
server.run()
|
||||
|
||||
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"
|
||||
)
|
||||
self.server = uvicorn.Server(config)
|
||||
await self.server.serve()
|
||||
|
||||
Reference in New Issue
Block a user