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

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"Pixlet renderer initialized with binary: {self.pixlet_binary}")
        else:
            logger.warning("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
            cmd = [
                self.pixlet_binary,
                "render",
                star_file,
                "-o", output_path,
                "-m", str(magnify)
            ]

            # Add configuration parameters
            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"
                    else:
                        value_str = str(value)

                    # Validate value doesn't contain shell metacharacters
                    # Allow alphanumeric, spaces, and common safe chars: .-_:/@#,
                    if not re.match(r'^[a-zA-Z0-9 .\-_:/@#,{}"\[\]]*$', value_str):
                        logger.warning(f"Skipping config value with unsafe characters for key {key}: {value_str}")
                        continue

                    cmd.extend(["-c", f"{key}={value_str}"])

            logger.debug(f"Executing Pixlet: {' '.join(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 os.path.isfile(output_path):
                    logger.debug(f"Successfully rendered: {star_file} -> {output_path}")
                    return True, None
                else:
                    error = "Rendering succeeded but output file not found"
                    logger.error(error)
                    return False, error
            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"

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

        Args:
            star_file: Path to .star file

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

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

        try:
            # Use 'pixlet info' or 'pixlet serve' to extract schema
            # Note: Schema extraction may vary by Pixlet version
            cmd = [self.pixlet_binary, "serve", star_file, "--print-schema"]

            logger.debug(f"Extracting schema: {' '.join(cmd)}")

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

            if result.returncode == 0:
                # Parse JSON schema from output
                try:
                    schema = json.loads(result.stdout)
                    logger.debug(f"Extracted schema from: {star_file}")
                    return True, schema, None
                except json.JSONDecodeError as e:
                    error = f"Invalid schema JSON: {e}"
                    logger.warning(error)
                    return False, None, error
            else:
                # Schema extraction might not be supported
                logger.debug(f"Schema extraction not available or failed: {result.stderr}")
                return True, None, None  # Not an error, just no schema

        except subprocess.TimeoutExpired:
            error = "Schema extraction timeout"
            logger.warning(error)
            return False, None, error
        except (subprocess.SubprocessError, OSError):
            logger.exception("Schema extraction exception")
            return False, None, "Schema extraction failed - see logs for details"