LEDMatrix/scripts/dev_server.py

#!/usr/bin/env python3
"""
LEDMatrix Dev Preview Server

A standalone lightweight Flask app for rapid plugin development.
Pick a plugin, tweak its config, and instantly see the rendered display.

Usage:
    python scripts/dev_server.py
    python scripts/dev_server.py --port 5001
    python scripts/dev_server.py --extra-dir /path/to/custom-plugin

Opens at http://localhost:5001
"""

import sys
import os
import json
import re
import time
import argparse
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional

# Add project root to path
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))

# Prevent hardware imports
os.environ['EMULATOR'] = 'true'

from flask import Flask, render_template, request, jsonify

from src.common.path_safety import resolve_under, safe_path_component

app = Flask(__name__, template_folder=str(Path(__file__).parent / 'templates'))

logger = logging.getLogger(__name__)

# Will be set from CLI args
_extra_dirs: List[str] = []

# Render endpoint resource guards
MAX_WIDTH = 512
MAX_HEIGHT = 512
MIN_WIDTH = 1
MIN_HEIGHT = 1

# plugin_id arrives in request input and is used to build filesystem paths —
# allowlist it (same pattern the web UI's pages_v3 uses)
_SAFE_PLUGIN_ID_RE = re.compile(r'^[a-zA-Z0-9_-]{1,64}$')


# --------------------------------------------------------------------------
# Plugin discovery
# --------------------------------------------------------------------------

def get_search_dirs() -> List[Path]:
    """Get all directories to search for plugins."""
    dirs = [
        PROJECT_ROOT / 'plugins',
        PROJECT_ROOT / 'plugin-repos',
    ]
    for d in _extra_dirs:
        dirs.append(Path(d))
    return dirs


def discover_plugins() -> List[Dict[str, Any]]:
    """Discover all available plugins across search directories."""
    plugins: List[Dict[str, Any]] = []
    seen_ids: set = set()

    for search_dir in get_search_dirs():
        if not search_dir.exists():
            logger.debug("[Dev Server] Search dir missing, skipping: %s", search_dir)
            continue
        for item in sorted(search_dir.iterdir()):
            if item.name.startswith('.') or not item.is_dir():
                logger.debug("[Dev Server] Skipping non-plugin entry: %s", item)
                continue
            manifest_path = item / 'manifest.json'
            if not manifest_path.exists():
                logger.debug("[Dev Server] No manifest.json in %s, skipping", item)
                continue
            try:
                with open(manifest_path, 'r') as f:
                    manifest: Dict[str, Any] = json.load(f)
                plugin_id: str = manifest.get('id', item.name)
                if plugin_id in seen_ids:
                    logger.debug("[Dev Server] Duplicate plugin_id '%s' at %s, skipping", plugin_id, item)
                    continue
                seen_ids.add(plugin_id)
                logger.debug("[Dev Server] Discovered plugin id=%s name=%s", plugin_id, manifest.get('name', plugin_id))
                plugins.append({
                    'id': plugin_id,
                    'name': manifest.get('name', plugin_id),
                    'description': manifest.get('description', ''),
                    'author': manifest.get('author', ''),
                    'version': manifest.get('version', ''),
                    'source_dir': str(search_dir),
                    'plugin_dir': str(item),
                })
            except json.JSONDecodeError as e:
                logger.warning("[Dev Server] JSON decode error in %s: %s", manifest_path, e)
                continue
            except OSError as e:
                logger.warning("[Dev Server] OS error reading %s: %s", manifest_path, e)
                continue

    return plugins


def find_plugin_dir(plugin_id: str) -> Optional[Path]:
    """Find a plugin directory by ID.

    plugin_id comes from request input: it must pass an allowlist match,
    and the resulting directory is normalized and required to live inside
    one of the plugin search dirs, so a crafted id can never name a path
    outside them.
    """
    plugin_id = safe_path_component(plugin_id)
    if not plugin_id or not _SAFE_PLUGIN_ID_RE.match(plugin_id):
        return None
    from src.plugin_system.plugin_loader import PluginLoader
    loader = PluginLoader()
    for search_dir in get_search_dirs():
        if not search_dir.exists():
            continue
        result = loader.find_plugin_directory(plugin_id, search_dir)
        if not result:
            continue
        # Normalize WITHOUT following symlinks (dev plugins are often
        # symlinked into plugins/) and require lexical containment in the
        # search dir, so no id can ever name a path outside it.
        result_abs = os.path.abspath(str(result))
        root_abs = os.path.abspath(str(search_dir))
        if os.path.commonpath([result_abs, root_abs]) == root_abs:
            return Path(result_abs)
    return None


def load_config_defaults(plugin_dir: 'str | Path') -> Dict[str, Any]:
    """Extract default values from config_schema.json.

    The same extraction a device and the plugin harness use
    (src/plugin_system/testing/loading.py), nested defaults included.
    """
    from src.plugin_system.testing.loading import (
        load_config_defaults as _load_config_defaults,
    )
    schema_path = resolve_under(plugin_dir, 'config_schema.json')
    if schema_path is None or not schema_path.exists():
        return {}
    return _load_config_defaults(schema_path.parent)


# --------------------------------------------------------------------------
# Routes
# --------------------------------------------------------------------------

@app.route('/')
def index():
    """Serve the dev preview page."""
    return render_template('dev_preview.html')


@app.route('/api/plugins')
def api_plugins():
    """List all available plugins."""
    return jsonify({'plugins': discover_plugins()})


@app.route('/api/plugins/<plugin_id>/schema')
def api_plugin_schema(plugin_id):
    """Get a plugin's config_schema.json."""
    plugin_dir = find_plugin_dir(plugin_id)
    if not plugin_dir:
        return jsonify({'error': f'Plugin not found: {plugin_id}'}), 404

    schema_path = resolve_under(plugin_dir, 'config_schema.json')
    if schema_path is None or not schema_path.exists():
        return jsonify({'schema': {'type': 'object', 'properties': {}}})

    with open(schema_path, 'r') as f:
        schema = json.load(f)
    return jsonify({'schema': schema})


@app.route('/api/plugins/<plugin_id>/defaults')
def api_plugin_defaults(plugin_id):
    """Get default config values from the schema."""
    plugin_dir = find_plugin_dir(plugin_id)
    if not plugin_dir:
        return jsonify({'error': f'Plugin not found: {plugin_id}'}), 404

    defaults = load_config_defaults(plugin_dir)
    defaults['enabled'] = True
    return jsonify({'defaults': defaults})


#: /api/render "vegas" values: the plugin's block of the Vegas strip, built
#: from its live elements (falling back to its Vegas content, as the ticker
#: does) or from its ordinary Vegas content only.
VEGAS_VIEWS = ('live', 'plain')


def _render_once(plugin_id, plugin_dir, manifest, config, mock_data, width, height,
                 skip_update, vegas=None):
    """Render one plugin at one size. Returns the /api/render response dict.

    A fresh plugin instance per call, mirroring the safety harness, so sizes
    never share state. With ``vegas`` set ('live' or 'plain') the image is the
    plugin's block of the Vegas strip instead of its display(), laid out by
    the ticker's own code (src/plugin_system/testing/vegas.py), and the
    response lists where each live element sits in it.
    """
    from src.plugin_system.testing import VisualTestDisplayManager, MockCacheManager, MockPluginManager
    from src.plugin_system.plugin_loader import PluginLoader

    display_manager = VisualTestDisplayManager(width=width, height=height)
    cache_manager = MockCacheManager()
    plugin_manager = MockPluginManager()

    # Pre-populate cache with mock data
    for key, value in mock_data.items():
        cache_manager.set(key, value)

    loader = PluginLoader()
    errors = []
    warnings = []

    plugin_instance, _module = loader.load_plugin(
        plugin_id=plugin_id,
        manifest=manifest,
        plugin_dir=plugin_dir,
        config=config,
        display_manager=display_manager,
        cache_manager=cache_manager,
        plugin_manager=plugin_manager,
        install_deps=False,
    )

    start_time = time.time()

    # Run update()
    if not skip_update:
        try:
            plugin_instance.update()
        except Exception as e:
            logger.warning("update() raised for plugin %s", plugin_id, exc_info=True)
            warnings.append(f"update() raised: {type(e).__name__} — see server log")

    if vegas:
        return _vegas_response(plugin_id, plugin_instance, display_manager, vegas,
                               start_time, errors, warnings)

    # Run display()
    try:
        plugin_instance.display(force_clear=True)
    except Exception as e:
        logger.warning("display() raised for plugin %s", plugin_id, exc_info=True)
        errors.append(f"display() raised: {type(e).__name__} — see server log")

    render_time_ms = round((time.time() - start_time) * 1000, 1)

    return {
        'image': f'data:image/png;base64,{display_manager.get_image_base64()}',
        'width': width,
        'height': height,
        'render_time_ms': render_time_ms,
        'errors': errors,
        'warnings': warnings,
    }


def _vegas_response(plugin_id, plugin_instance, display_manager, vegas, start_time,
                    errors, warnings):
    """The /api/render response for the Vegas strip view."""
    import base64
    import io

    from src.plugin_system.testing.vegas import render_vegas_strip

    block, layout = None, []
    try:
        block, layout = render_vegas_strip(plugin_instance, plugin_id, display_manager,
                                           live=(vegas == 'live'))
    except Exception as e:
        logger.warning("Vegas render raised for plugin %s", plugin_id, exc_info=True)
        errors.append(f"Vegas render raised: {type(e).__name__} — see server log")
    if block is None:
        if not errors:
            errors.append("The plugin has no Vegas content")
        block = display_manager.image
    elif vegas == 'live' and not layout:
        warnings.append("No live elements: this is the plugin's ordinary Vegas content")
    buffer = io.BytesIO()
    block.convert('RGB').save(buffer, format='PNG')
    return {
        'image': 'data:image/png;base64,' + base64.b64encode(buffer.getvalue()).decode('ascii'),
        'width': block.width,
        'height': block.height,
        'render_time_ms': round((time.time() - start_time) * 1000, 1),
        'errors': errors,
        'warnings': warnings,
        'live_elements': [{'key': key, 'x': x, 'width': width} for x, key, width in layout],
    }


def _trusted_plugin_dir(plugin_dir: Path) -> Optional[Path]:
    """Re-derive a plugin directory from the search dirs' own listings.

    Path-injection barrier: unlike ``Path.iterdir()`` (which CodeQL doesn't
    recognize as a taint-clearing enumeration), ``os.scandir()`` is. The
    returned Path is built from a trusted root plus a name the filesystem
    itself produced under that root via scandir — request-derived strings
    never enter its construction — so a crafted plugin id can never make
    downstream file access leave the plugin search dirs. Comparison is by
    name, deliberately without symlink resolution (dev plugins are
    commonly symlinked into plugins/).
    """
    wanted_name = Path(os.path.normpath(str(plugin_dir))).name
    for search_dir in get_search_dirs():
        search_dir_str = str(search_dir)
        try:
            with os.scandir(search_dir_str) as entries:
                for entry in entries:
                    if entry.name == wanted_name and entry.is_dir():
                        return Path(search_dir_str) / entry.name
        except OSError:
            continue
    return None


def _parse_render_request(data):
    """Shared /api/render* request prep. Returns (plugin_dir, manifest, config,
    mock_data, skip_update) or raises ValueError with a client message."""
    plugin_id = data['plugin_id']
    candidate_dir = find_plugin_dir(plugin_id)
    # Never reuse `candidate_dir` past this point: it's built from
    # request-derived input, and a variable reassigned only on some paths
    # isn't a barrier CodeQL's flow analysis honors. `trusted_dir` is the
    # sole name used below, always the scandir-sourced result.
    trusted_dir = _trusted_plugin_dir(candidate_dir) if candidate_dir else None
    if not trusted_dir:
        raise LookupError(f'Plugin not found: {plugin_id}')

    manifest_path = trusted_dir / 'manifest.json'
    with open(manifest_path, 'r') as f:
        manifest = json.load(f)

    # Build config the way a device would: schema defaults under a forced
    # enabled, with the user's overrides deep-merged on top
    from src.plugin_system.testing.loading import build_config
    overrides = data.get('config') or {}
    if not isinstance(overrides, dict):
        raise ValueError('config must be a JSON object')
    config = build_config(trusted_dir, overrides)

    return trusted_dir, manifest, config, data.get('mock_data', {}), data.get('skip_update', False)


@app.route('/api/render', methods=['POST'])
def api_render():
    """Render a plugin and return the display as base64 PNG."""
    data = request.get_json()
    if not data or 'plugin_id' not in data:
        return jsonify({'error': 'plugin_id is required'}), 400

    try:
        width = int(data.get('width', 128))
        height = int(data.get('height', 32))
    except (TypeError, ValueError):
        return jsonify({'error': 'width and height must be integers'}), 400

    if not (MIN_WIDTH <= width <= MAX_WIDTH):
        return jsonify({'error': f'width must be between {MIN_WIDTH} and {MAX_WIDTH}'}), 400
    if not (MIN_HEIGHT <= height <= MAX_HEIGHT):
        return jsonify({'error': f'height must be between {MIN_HEIGHT} and {MAX_HEIGHT}'}), 400

    vegas = data.get('vegas') or None
    if vegas is not None and vegas not in VEGAS_VIEWS:
        return jsonify({'error': f'vegas must be one of {", ".join(VEGAS_VIEWS)}'}), 400

    try:
        plugin_dir, manifest, config, mock_data, skip_update = _parse_render_request(data)
    except LookupError:
        return jsonify({'error': f"Plugin not found: {data['plugin_id']}"}), 404
    except Exception:
        # Bad manifest.json / schema / fixture — details go to the dev's
        # console, not the HTTP response
        app.logger.exception('render request preparation failed')
        return jsonify({'error': 'Could not prepare render request; see server log'}), 400

    try:
        result = _render_once(data['plugin_id'], plugin_dir, manifest, config,
                              mock_data, width, height, skip_update, vegas=vegas)
    except Exception:
        app.logger.exception('plugin load failed during render')
        return jsonify({'error': 'Failed to load plugin; see server log'}), 500
    return jsonify(result)


@app.route('/api/sizes')
def api_sizes():
    """The representative panel-size sample the safety harness renders at."""
    from src.plugin_system.testing.sizes import DEFAULT_TEST_SIZES
    return jsonify({'sizes': [list(s) for s in DEFAULT_TEST_SIZES]})


MAX_MATRIX_SIZES = 12


@app.route('/api/render-matrix', methods=['POST'])
def api_render_matrix():
    """Render a plugin at a list of sizes (default: the harness sample) so the
    UI can show a side-by-side multi-resolution gallery."""
    data = request.get_json()
    if not data or 'plugin_id' not in data:
        return jsonify({'error': 'plugin_id is required'}), 400

    from src.plugin_system.testing.sizes import DEFAULT_TEST_SIZES
    sizes = data.get('sizes') or [list(s) for s in DEFAULT_TEST_SIZES]
    if len(sizes) > MAX_MATRIX_SIZES:
        return jsonify({'error': f'at most {MAX_MATRIX_SIZES} sizes per request'}), 400
    parsed_sizes = []
    for pair in sizes:
        try:
            w, h = int(pair[0]), int(pair[1])
        except (TypeError, ValueError, IndexError):
            return jsonify({'error': f'invalid size entry {pair!r} (expected [w, h])'}), 400
        if not (MIN_WIDTH <= w <= MAX_WIDTH and MIN_HEIGHT <= h <= MAX_HEIGHT):
            return jsonify({'error': f'size {w}x{h} out of bounds'}), 400
        parsed_sizes.append((w, h))

    try:
        plugin_dir, manifest, config, mock_data, skip_update = _parse_render_request(data)
    except LookupError:
        return jsonify({'error': f"Plugin not found: {data['plugin_id']}"}), 404
    except Exception:
        app.logger.exception('render request preparation failed')
        return jsonify({'error': 'Could not prepare render request; see server log'}), 400

    results = []
    for w, h in parsed_sizes:
        try:
            results.append(_render_once(data['plugin_id'], plugin_dir, manifest,
                                        config, mock_data, w, h, skip_update))
        except Exception:
            app.logger.exception('plugin load failed during %dx%d render', w, h)
            results.append({'image': None, 'width': w, 'height': h,
                            'render_time_ms': 0,
                            'errors': ['Failed to load plugin; see server log'],
                            'warnings': []})
    return jsonify({'results': results})


# --------------------------------------------------------------------------
# Main
# --------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(description='LEDMatrix Dev Preview Server')
    parser.add_argument('--port', type=int, default=5001, help='Port to run on (default: 5001)')
    parser.add_argument('--host', default='127.0.0.1', help='Host to bind to (default: 127.0.0.1)')
    parser.add_argument('--extra-dir', action='append', default=[],
                        help='Extra plugin directory to search (can be repeated)')
    parser.add_argument('--debug', action='store_true', help='Enable Flask debug mode')

    args = parser.parse_args()

    global _extra_dirs
    _extra_dirs = args.extra_dir

    print("LEDMatrix Dev Preview Server")
    print(f"Open http://{args.host}:{args.port} in your browser")
    print(f"Plugin search dirs: {[str(d) for d in get_search_dirs()]}")
    print()

    app.run(host=args.host, port=args.port, debug=args.debug)


if __name__ == '__main__':
    main()