LEDMatrix/plugin-repos/starlark-apps/pixlet_renderer.py

"""
Pixlet Renderer Module for Starlark Apps

Handles execution of Pixlet CLI to render .star files into WebP animations.
Supports bundled binaries and system-installed Pixlet.
"""

import json
import logging
import os
import platform
import re
import shutil
import subprocess
from pathlib import Path
from typing import Dict, Any, Optional, Tuple, List

logger = logging.getLogger(__name__)


class PixletRenderer:
    """
    Wrapper for Pixlet CLI rendering.

    Handles:
    - Auto-detection of bundled or system Pixlet binary
    - Rendering .star files with configuration
    - Schema extraction from .star files
    - Timeout and error handling
    """

    def __init__(self, pixlet_path: Optional[str] = None, timeout: int = 30):
        """
        Initialize the Pixlet renderer.

        Args:
            pixlet_path: Optional explicit path to Pixlet binary
            timeout: Maximum seconds to wait for rendering
        """
        self.timeout = timeout
        self.pixlet_binary = self._find_pixlet_binary(pixlet_path)

        if self.pixlet_binary:
            logger.info(f"[Starlark Pixlet] Pixlet renderer initialized with binary: {self.pixlet_binary}")
        else:
            logger.warning("[Starlark Pixlet] Pixlet binary not found - rendering will fail")

    def _find_pixlet_binary(self, explicit_path: Optional[str] = None) -> Optional[str]:
        """
        Find Pixlet binary using the following priority:
        1. Explicit path provided
        2. Bundled binary for current architecture
        3. System PATH

        Args:
            explicit_path: User-specified path to Pixlet

        Returns:
            Path to Pixlet binary, or None if not found
        """
        # 1. Check explicit path
        if explicit_path and os.path.isfile(explicit_path):
            if os.access(explicit_path, os.X_OK):
                logger.debug(f"Using explicit Pixlet path: {explicit_path}")
                return explicit_path
            else:
                logger.warning(f"Explicit Pixlet path not executable: {explicit_path}")

        # 2. Check bundled binary
        try:
            bundled_path = self._get_bundled_binary_path()
            if bundled_path and os.path.isfile(bundled_path):
                # Ensure executable
                if not os.access(bundled_path, os.X_OK):
                    try:
                        os.chmod(bundled_path, 0o755)
                        logger.debug(f"Made bundled binary executable: {bundled_path}")
                    except OSError:
                        logger.exception(f"Could not make bundled binary executable: {bundled_path}")

                if os.access(bundled_path, os.X_OK):
                    logger.debug(f"Using bundled Pixlet binary: {bundled_path}")
                    return bundled_path
        except OSError:
            logger.exception("Could not locate bundled binary")

        # 3. Check system PATH
        system_pixlet = shutil.which("pixlet")
        if system_pixlet:
            logger.debug(f"Using system Pixlet: {system_pixlet}")
            return system_pixlet

        logger.error("Pixlet binary not found in any location")
        return None

    def _get_bundled_binary_path(self) -> Optional[str]:
        """
        Get path to bundled Pixlet binary for current architecture.

        Returns:
            Path to bundled binary, or None if not found
        """
        try:
            # Determine project root (parent of plugin-repos)
            current_dir = Path(__file__).resolve().parent
            project_root = current_dir.parent.parent
            bin_dir = project_root / "bin" / "pixlet"

            # Detect architecture
            system = platform.system().lower()
            machine = platform.machine().lower()

            # Map architecture to binary name
            if system == "linux":
                if "aarch64" in machine or "arm64" in machine:
                    binary_name = "pixlet-linux-arm64"
                elif "x86_64" in machine or "amd64" in machine:
                    binary_name = "pixlet-linux-amd64"
                else:
                    logger.warning(f"Unsupported Linux architecture: {machine}")
                    return None
            elif system == "darwin":
                if "arm64" in machine:
                    binary_name = "pixlet-darwin-arm64"
                else:
                    binary_name = "pixlet-darwin-amd64"
            elif system == "windows":
                binary_name = "pixlet-windows-amd64.exe"
            else:
                logger.warning(f"Unsupported system: {system}")
                return None

            binary_path = bin_dir / binary_name
            if binary_path.exists():
                return str(binary_path)

            logger.debug(f"Bundled binary not found at: {binary_path}")
            return None

        except OSError:
            logger.exception("Error finding bundled binary")
            return None

    def _get_safe_working_directory(self, star_file: str) -> Optional[str]:
        """
        Get a safe working directory for subprocess execution.

        Args:
            star_file: Path to .star file

        Returns:
            Resolved parent directory, or None if empty or invalid
        """
        try:
            resolved_parent = os.path.dirname(os.path.abspath(star_file))
            # Return None if empty string to avoid FileNotFoundError
            if not resolved_parent:
                logger.debug(f"Empty parent directory for star_file: {star_file}")
                return None
            return resolved_parent
        except (OSError, ValueError):
            logger.debug(f"Could not resolve working directory for: {star_file}")
            return None

    def is_available(self) -> bool:
        """
        Check if Pixlet is available and functional.

        Returns:
            True if Pixlet can be executed
        """
        if not self.pixlet_binary:
            return False

        try:
            result = subprocess.run(
                [self.pixlet_binary, "version"],
                capture_output=True,
                text=True,
                timeout=5
            )
            return result.returncode == 0
        except subprocess.TimeoutExpired:
            logger.debug("Pixlet version check timed out")
            return False
        except (subprocess.SubprocessError, OSError):
            logger.exception("Pixlet not available")
            return False

    def get_version(self) -> Optional[str]:
        """
        Get Pixlet version string.

        Returns:
            Version string, or None if unavailable
        """
        if not self.pixlet_binary:
            return None

        try:
            result = subprocess.run(
                [self.pixlet_binary, "version"],
                capture_output=True,
                text=True,
                timeout=5
            )
            if result.returncode == 0:
                return result.stdout.strip()
        except subprocess.TimeoutExpired:
            logger.debug("Pixlet version check timed out")
        except (subprocess.SubprocessError, OSError):
            logger.exception("Could not get Pixlet version")

        return None

    def render(
        self,
        star_file: str,
        output_path: str,
        config: Optional[Dict[str, Any]] = None,
        magnify: int = 1
    ) -> Tuple[bool, Optional[str]]:
        """
        Render a .star file to WebP output.

        Args:
            star_file: Path to .star file
            output_path: Where to save WebP output
            config: Configuration dictionary to pass to app
            magnify: Magnification factor (default 1)

        Returns:
            Tuple of (success: bool, error_message: Optional[str])
        """
        if not self.pixlet_binary:
            return False, "Pixlet binary not found"

        if not os.path.isfile(star_file):
            return False, f"Star file not found: {star_file}"

        try:
            # Build command - config params must be POSITIONAL between star_file and flags
            # Format: pixlet render <file.star> [key=value]... [flags]
            cmd = [
                self.pixlet_binary,
                "render",
                star_file
            ]

            # Add configuration parameters as positional arguments (BEFORE flags)
            if config:
                for key, value in config.items():
                    # Validate key format (alphanumeric + underscore only)
                    if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', key):
                        logger.warning(f"Skipping invalid config key: {key}")
                        continue

                    # Convert value to string for CLI
                    if isinstance(value, bool):
                        value_str = "true" if value else "false"
                    elif isinstance(value, str) and (value.startswith('{') or value.startswith('[')):
                        # JSON string - keep as-is, will be properly quoted by subprocess
                        value_str = value
                    else:
                        value_str = str(value)

                    # Validate value doesn't contain dangerous shell metacharacters.
                    # Kept as defence in depth only: cmd is a list and there is no
                    # shell=True below, so nothing here is ever interpreted by a
                    # shell. That made the list worth trimming rather than growing
                    # -- "|" is a normal character inside a config value, and apps
                    # do use it as a separator (a PennDOT sign id is
                    # "I-476 North|175659"). Blocking it dropped the whole key
                    # silently, and the app then rendered its own "not configured"
                    # screen with nothing to say why.
                    # Block: backticks, $(), redirects, semicolons, ampersands, null bytes
                    # Allow: most printable chars including spaces, quotes, brackets, braces, pipes
                    if re.search(r'[`$<>&;\x00]|\$\(', value_str):
                        logger.warning(f"Skipping config value with unsafe shell characters for key {key}: {value_str}")
                        continue

                    # Add as positional argument (not -c flag)
                    cmd.append(f"{key}={value_str}")

            # Add flags AFTER positional config arguments
            cmd.extend([
                "-o", output_path,
                "-m", str(magnify)
            ])

            # Build sanitized command for logging (redact sensitive values)
            sanitized_cmd = [self.pixlet_binary, "render", star_file]
            if config:
                config_keys = list(config.keys())
                sanitized_cmd.append(f"[{len(config_keys)} config entries: {', '.join(config_keys)}]")
            sanitized_cmd.extend(["-o", output_path, "-m", str(magnify)])
            logger.debug(f"Executing Pixlet: {' '.join(sanitized_cmd)}")

            # Execute rendering
            safe_cwd = self._get_safe_working_directory(star_file)
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=self.timeout,
                cwd=safe_cwd  # Run in .star file directory (or None if relative path)
            )

            if result.returncode == 0:
                if not os.path.isfile(output_path):
                    error = "Rendering succeeded but output file not found"
                    logger.error(error)
                    return False, error
                # Pixlet exits 0 and writes a 0-byte file when the app renders
                # nothing -- an app whose config leaves it with no content to
                # show does exactly that. Treating existence alone as success
                # handed the caller a file with no frames in it, which reads
                # downstream as a working app that draws a black panel.
                if os.path.getsize(output_path) == 0:
                    error = "Rendering produced an empty (0-byte) file - the app rendered no content"
                    logger.error(error)
                    return False, error
                logger.debug(f"Successfully rendered: {star_file} -> {output_path}")
                return True, None
            else:
                error = f"Pixlet failed (exit {result.returncode}): {result.stderr}"
                logger.error(error)
                return False, error

        except subprocess.TimeoutExpired:
            error = f"Rendering timeout after {self.timeout}s"
            logger.error(error)
            return False, error
        except (subprocess.SubprocessError, OSError):
            logger.exception("Rendering exception")
            return False, "Rendering failed - see logs for details"

    #: Schema extraction runs an app's own get_schema(), which may make a
    #: network call. Short enough that a hung app does not stall an upload,
    #: long enough for a real API round trip on a slow connection.
    SCHEMA_TIMEOUT = 20

    def extract_schema_via_pixlet(self, star_file: str) -> Optional[Dict[str, Any]]:
        """Ask Pixlet itself for the app's schema, or None if it cannot say.

        `pixlet schema` executes get_schema() instead of reading it, which is
        the only way to see options an app computes at runtime -- a dropdown
        whose choices come from a live API call has no option list anywhere in
        the source for the regex parser below to find, so that parser reports
        an empty dropdown and the config form offers nothing to pick.

        Pixlet's own field keys are remapped to the ones the rest of this
        plugin and the config UI already use ("typeOf"/"desc"), so the two
        extractors return the same shape and callers cannot tell them apart.
        """
        if not self.pixlet_binary:
            return None
        try:
            result = subprocess.run(
                [self.pixlet_binary, "schema", star_file],
                capture_output=True, text=True, timeout=self.SCHEMA_TIMEOUT,
                cwd=self._get_safe_working_directory(star_file),
            )
        except subprocess.TimeoutExpired:
            logger.warning(
                "pixlet schema timed out after %ss for %s - get_schema() may be "
                "making a slow network call", self.SCHEMA_TIMEOUT, star_file)
            return None
        except (subprocess.SubprocessError, OSError) as e:
            logger.warning(f"Could not run pixlet schema for {star_file}: {e}")
            return None

        if result.returncode != 0:
            # Not an error worth failing on: older Pixlet builds have no
            # `schema` subcommand at all, and the source parser still works.
            logger.debug(
                "pixlet schema exited %d for %s: %s",
                result.returncode, star_file, (result.stderr or '').strip()[:300])
            return None

        try:
            schema = json.loads(result.stdout)
        except (json.JSONDecodeError, ValueError) as e:
            logger.warning(f"pixlet schema returned unparseable output for {star_file}: {e}")
            return None

        if not isinstance(schema, dict) or not isinstance(schema.get("schema"), list):
            logger.warning(f"pixlet schema returned an unexpected shape for {star_file}")
            return None

        for field in schema["schema"]:
            if not isinstance(field, dict):
                continue
            if "type" in field and "typeOf" not in field:
                field["typeOf"] = field.pop("type")
            if "description" in field and "desc" not in field:
                field["desc"] = field.pop("description")
        return schema

    def extract_schema(self, star_file: str) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]:
        """
        Extract configuration schema from a .star file.

        Prefers `pixlet schema`, which runs the app and therefore sees options
        it computes at runtime. Falls back to parsing the source when Pixlet is
        unavailable, too old to have the subcommand, or the app fails to run --
        that parser handles:
        - Static field definitions (location, text, toggle, dropdown, color, datetime)
        - Variable-referenced dropdown options
        - Graceful degradation for unsupported field types

        Args:
            star_file: Path to .star file

        Returns:
            Tuple of (success: bool, schema: Optional[Dict], error: Optional[str])
        """
        if not os.path.isfile(star_file):
            return False, None, f"Star file not found: {star_file}"

        schema = self.extract_schema_via_pixlet(star_file)
        if schema is not None:
            logger.debug(
                "Extracted schema with %d field(s) from %s via pixlet schema",
                len(schema.get('schema', [])), star_file)
            return True, schema, None

        try:
            # Read .star file
            with open(star_file, 'r', encoding='utf-8') as f:
                content = f.read()

            # Parse schema from source
            schema = self._parse_schema_from_source(content, star_file)

            if schema:
                field_count = len(schema.get('schema', []))
                logger.debug(f"Extracted schema with {field_count} field(s) from: {star_file}")
                return True, schema, None
            else:
                # No schema found - not an error, app just doesn't have configuration
                logger.debug(f"No schema found in: {star_file}")
                return True, None, None

        except UnicodeDecodeError as e:
            error = f"File encoding error: {e}"
            logger.warning(error)
            return False, None, error
        except Exception as e:
            logger.exception(f"Schema extraction failed for {star_file}")
            return False, None, f"Schema extraction error: {str(e)}"

    def _parse_schema_from_source(self, content: str, file_path: str) -> Optional[Dict[str, Any]]:
        """
        Parse get_schema() function from Starlark source code.

        Args:
            content: .star file content
            file_path: Path to file (for logging)

        Returns:
            Schema dict with format {"version": "1", "schema": [...]}, or None
        """
        # Extract variable definitions (for dropdown options)
        var_table = self._extract_variable_definitions(content)

        # Extract get_schema() function body
        schema_body = self._extract_get_schema_body(content)
        if not schema_body:
            logger.debug(f"No get_schema() function found in {file_path}")
            return None

        # Extract version
        version_match = re.search(r'version\s*=\s*"([^"]+)"', schema_body)
        version = version_match.group(1) if version_match else "1"

        # Extract fields array from schema.Schema(...) - handle nested brackets
        fields_start_match = re.search(r'fields\s*=\s*\[', schema_body)
        if not fields_start_match:
            # Empty schema or no fields
            return {"version": version, "schema": []}

        # Find matching closing bracket
        bracket_count = 1
        i = fields_start_match.end()
        while i < len(schema_body) and bracket_count > 0:
            if schema_body[i] == '[':
                bracket_count += 1
            elif schema_body[i] == ']':
                bracket_count -= 1
            i += 1

        if bracket_count != 0:
            # Unmatched brackets
            logger.warning(f"Unmatched brackets in schema fields for {file_path}")
            return {"version": version, "schema": []}

        fields_text = schema_body[fields_start_match.end():i-1]

        # Parse individual fields
        schema_fields = []
        # Match schema.FieldType(...) patterns
        field_pattern = r'schema\.(\w+)\s*\((.*?)\)'

        # Find all field definitions (handle nested parentheses)
        pos = 0
        while pos < len(fields_text):
            match = re.search(field_pattern, fields_text[pos:], re.DOTALL)
            if not match:
                break

            field_type = match.group(1)
            field_start = pos + match.start()
            field_end = pos + match.end()

            # Handle nested parentheses properly
            paren_count = 1
            i = pos + match.start() + len(f'schema.{field_type}(')
            while i < len(fields_text) and paren_count > 0:
                if fields_text[i] == '(':
                    paren_count += 1
                elif fields_text[i] == ')':
                    paren_count -= 1
                i += 1

            field_params_text = fields_text[pos + match.start() + len(f'schema.{field_type}('):i-1]

            # Parse field
            field_dict = self._parse_schema_field(field_type, field_params_text, var_table)
            if field_dict:
                schema_fields.append(field_dict)

            pos = i

        return {
            "version": version,
            "schema": schema_fields
        }

    def _extract_variable_definitions(self, content: str) -> Dict[str, List[Dict]]:
        """
        Extract top-level variable assignments (for dropdown options).

        Args:
            content: .star file content

        Returns:
            Dict mapping variable names to their option lists
        """
        var_table = {}

        # Find variable definitions like: variableName = [schema.Option(...), ...]
        var_pattern = r'^(\w+)\s*=\s*\[(.*?schema\.Option.*?)\]'
        matches = re.finditer(var_pattern, content, re.MULTILINE | re.DOTALL)

        for match in matches:
            var_name = match.group(1)
            options_text = match.group(2)

            # Parse schema.Option entries
            options = self._parse_schema_options(options_text, {})
            if options:
                var_table[var_name] = options

        return var_table

    def _extract_get_schema_body(self, content: str) -> Optional[str]:
        """
        Extract get_schema() function body using indentation-aware parsing.

        Args:
            content: .star file content

        Returns:
            Function body text, or None if not found
        """
        # Find def get_schema(): line
        pattern = r'^(\s*)def\s+get_schema\s*\(\s*\)\s*:'
        match = re.search(pattern, content, re.MULTILINE)

        if not match:
            return None

        # Get the indentation level of the function definition
        func_indent = len(match.group(1))
        func_start = match.end()

        # Split content into lines starting after the function definition
        lines_after = content[func_start:].split('\n')
        body_lines = []

        for line in lines_after:
            # Skip empty lines
            if not line.strip():
                body_lines.append(line)
                continue

            # Calculate indentation of current line
            stripped = line.lstrip()
            line_indent = len(line) - len(stripped)

            # If line has same or less indentation than function def, check if it's a top-level def
            if line_indent <= func_indent:
                # This is a line at the same or outer level - check if it's a function
                if re.match(r'def\s+\w+', stripped):
                    # Found next top-level function, stop here
                    break
                # Otherwise it might be a comment or other top-level code, stop anyway
                break

            # Line is indented more than function def, so it's part of the body
            body_lines.append(line)

        if body_lines:
            return '\n'.join(body_lines)
        return None

    def _parse_schema_field(self, field_type: str, params_text: str, var_table: Dict) -> Optional[Dict[str, Any]]:
        """
        Parse individual schema field definition.

        Args:
            field_type: Field type (Location, Text, Toggle, etc.)
            params_text: Field parameters text
            var_table: Variable lookup table

        Returns:
            Field dict, or None if parse fails
        """
        # Map Pixlet field types to JSON typeOf
        type_mapping = {
            'Location': 'location',
            'Text': 'text',
            'Toggle': 'toggle',
            'Dropdown': 'dropdown',
            'Color': 'color',
            'DateTime': 'datetime',
            'OAuth2': 'oauth2',
            'PhotoSelect': 'photo_select',
            'LocationBased': 'location_based',
            'Typeahead': 'typeahead',
            'Generated': 'generated',
        }

        type_of = type_mapping.get(field_type, field_type.lower())

        # Skip Generated fields (invisible meta-fields)
        if type_of == 'generated':
            return None

        field_dict = {"typeOf": type_of}

        # Extract common parameters
        # id
        id_match = re.search(r'id\s*=\s*"([^"]+)"', params_text)
        if id_match:
            field_dict['id'] = id_match.group(1)
        else:
            # id is required, skip field if missing
            return None

        # name
        name_match = re.search(r'name\s*=\s*"([^"]+)"', params_text)
        if name_match:
            field_dict['name'] = name_match.group(1)

        # desc
        desc_match = re.search(r'desc\s*=\s*"([^"]+)"', params_text)
        if desc_match:
            field_dict['desc'] = desc_match.group(1)

        # icon
        icon_match = re.search(r'icon\s*=\s*"([^"]+)"', params_text)
        if icon_match:
            field_dict['icon'] = icon_match.group(1)

        # default (can be string, bool, or variable reference)
        # First try to match quoted strings (which may contain commas)
        default_match = re.search(r'default\s*=\s*"([^"]*)"', params_text)
        if not default_match:
            # Try single quotes
            default_match = re.search(r"default\s*=\s*'([^']*)'", params_text)
        if not default_match:
            # Fall back to unquoted value (stop at comma or closing paren)
            default_match = re.search(r'default\s*=\s*([^,\)]+)', params_text)

        if default_match:
            default_value = default_match.group(1).strip()
            # Handle boolean
            if default_value in ('True', 'False'):
                field_dict['default'] = default_value.lower()
            # Handle string literal from first two patterns (already extracted without quotes)
            elif re.search(r'default\s*=\s*["\']', params_text):
                # This was a quoted string, use the captured content directly
                field_dict['default'] = default_value
            # Handle variable reference (can't resolve, use as-is)
            else:
                # Try to extract just the value if it's like options[0].value
                if '.' in default_value or '[' in default_value:
                    # Complex expression, skip default
                    pass
                else:
                    field_dict['default'] = default_value

        # For dropdown, extract options
        if type_of == 'dropdown':
            options_match = re.search(r'options\s*=\s*([^,\)]+)', params_text)
            if options_match:
                options_ref = options_match.group(1).strip()
                # Check if it's a variable reference
                if options_ref in var_table:
                    field_dict['options'] = var_table[options_ref]
                # Or inline options
                elif options_ref.startswith('['):
                    # Find the full options array (handle nested brackets)
                    # This is tricky, for now try to extract inline options
                    inline_match = re.search(r'options\s*=\s*(\[.*?\])', params_text, re.DOTALL)
                    if inline_match:
                        options_text = inline_match.group(1)
                        field_dict['options'] = self._parse_schema_options(options_text, var_table)

        return field_dict

    def _parse_schema_options(self, options_text: str, var_table: Dict) -> List[Dict[str, str]]:
        """
        Parse schema.Option list.

        Args:
            options_text: Text containing schema.Option(...) entries
            var_table: Variable lookup table (not currently used)

        Returns:
            List of {"display": "...", "value": "..."} dicts
        """
        options = []

        # Match schema.Option(display = "...", value = "...")
        option_pattern = r'schema\.Option\s*\(\s*display\s*=\s*"([^"]+)"\s*,\s*value\s*=\s*"([^"]+)"\s*\)'
        matches = re.finditer(option_pattern, options_text)

        for match in matches:
            options.append({
                "display": match.group(1),
                "value": match.group(2)
            })

        return options