transpile.py

Go to the documentation of this file.

"""One-pass PineScript transpiler bridge with structured diagnostics."""
 
from __future__ import annotations
 
from dataclasses import dataclass, field
import hashlib
import importlib
from importlib import metadata
from pathlib import Path
from typing import Any, Mapping
 
try:
    import pineforge_codegen
    from pineforge_codegen import transpile_full
    from pineforge_codegen.errors import CompileError
except ModuleNotFoundError:
    pineforge_codegen = None
    transpile_full = None
    CompileError = None
 
 
@dataclass(frozen=True)
class TranspileDiagnostic:
    """A stable diagnostic envelope independent of codegen's Python classes."""
 
    severity: str
    phase: str
    message: str
    filename: str | None = None
    line: int | None = None
    column: int | None = None
    end_column: int | None = None
    hint: str | None = None
 
    def to_dict(self) -> dict[str, Any]:
        return {
            "severity": self.severityseverity,
            "phase": self.phasephase,
            "message": self.messagemessage,
            "filename": self.filenamefilename,
            "line": self.lineline,
            "column": self.columncolumn,
            "end_column": self.end_columnend_column,
            "hint": self.hinthint,
        }
 
 
@dataclass(frozen=True)
class TranspileResult:
    """Result of exactly one ``transpile_full`` pipeline invocation."""
 
    generated_cpp: str | None
    inputs: tuple[dict[str, Any], ...] = ()
    strategy_params: Mapping[str, Any] = field(default_factory=dict)
    diagnostics: tuple[TranspileDiagnostic, ...] = ()
 
    @property
    def ok(self) -> bool:
        return self.generated_cpp is not None and not any(
            diagnostic.severity == "error" for diagnostic in self.diagnostics
        )
 
    def require_success(self) -> str:
        if not self.ok or self.generated_cpp is None:
            raise TranspileFailure(self)
        return self.generated_cpp
 
 
class TranspileFailure(RuntimeError):
    """Raised when an artifact build cannot proceed past transpilation."""
 
    def __init__(self, result: TranspileResult):
        self.result = result
        summary = "; ".join(d.message for d in result.diagnostics) or "transpile failed"
        super().__init__(summary)
 
 
@dataclass(frozen=True)
class CodegenIdentity:
    """Installed codegen version, source hash, and resolved module location."""
 
    version: str
    implementation_sha256: str
    module_path: str
 
    def to_dict(self) -> dict[str, str]:
        return {
            "version": self.versionversion,
            "implementation_sha256": self.implementation_sha256implementation_sha256,
            "module_path": self.module_pathmodule_path,
        }
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
def transpile_source(pine_source: str, *, filename: str = "<input>") -> TranspileResult:
    """Transpile source once and normalize all expected diagnostics.
 
    Pine language errors are returned in ``TranspileResult`` rather than being
    conflated with native compiler failures. Unexpected bridge failures are also
    represented as an ``INTERNAL`` diagnostic so a CLI can emit one stable JSON
    envelope.
    """
 
    try:
        transpiler, compile_error = _load_codegen()
    except ModuleNotFoundError:
        return TranspileResult(
            generated_cpp=None,
            diagnostics=(
                TranspileDiagnostic(
                    severity="error",
                    phase="CONFIGURATION",
                    message=(
                        "pineforge-codegen is required to compile PineScript; "
                        "install pineforge-hpo[transpile] or provide a precompiled artifact"
                    ),
                    filename=filename,
                ),
            ),
        )
 
    try:
        payload = transpiler(pine_source, filename=filename)
    except compile_error as error:
        return TranspileResult(
            generated_cpp=None,
            diagnostics=tuple(_diagnostic_from_codegen(d) for d in error.diagnostics),
        )
    except Exception as error:  # keep the initialization error JSON-serializable
        return TranspileResult(
            generated_cpp=None,
            diagnostics=(
                TranspileDiagnostic(
                    severity="error",
                    phase="INTERNAL",
                    message=f"{type(error).__name__}: {error}",
                    filename=filename,
                ),
            ),
        )
 
    if not isinstance(payload, Mapping):
        return TranspileResult(
            generated_cpp=None,
            diagnostics=(
                TranspileDiagnostic(
                    severity="error",
                    phase="INTERNAL",
                    message="transpile_full returned a non-object result",
                    filename=filename,
                ),
            ),
        )
 
    cpp = payload.get("cpp")
    inputs = payload.get("inputs", [])
    strategy_params = payload.get("strategyParams", {})
    if (
        not isinstance(cpp, str)
        or not isinstance(inputs, list)
        or not isinstance(strategy_params, Mapping)
    ):
        return TranspileResult(
            generated_cpp=None,
            diagnostics=(
                TranspileDiagnostic(
                    severity="error",
                    phase="INTERNAL",
                    message="transpile_full returned an invalid result shape",
                    filename=filename,
                ),
            ),
        )
 
    normalized_inputs: list[dict[str, Any]] = []
    for item in inputs:
        if not isinstance(item, Mapping):
            return TranspileResult(
                generated_cpp=None,
                diagnostics=(
                    TranspileDiagnostic(
                        severity="error",
                        phase="INTERNAL",
                        message="transpile_full input manifest contains a non-object entry",
                        filename=filename,
                    ),
                ),
            )
        normalized_inputs.append(dict(item))
 
    return TranspileResult(
        generated_cpp=cpp,
        inputs=tuple(normalized_inputs),
        strategy_params=dict(strategy_params),
        diagnostics=_diagnostics_from_payload(payload),
    )
 
 
def codegen_identity() -> CodegenIdentity:
    """Return version plus a hash that detects dirty/editable codegen installs."""
 
    _transpiler, _compile_error = _load_codegen()
    assert pineforge_codegen is not None
 
    try:
        version = metadata.version("pineforge-codegen")
    except metadata.PackageNotFoundError:
        version = str(getattr(pineforge_codegen, "__version__", "unknown"))
 
    module_file = Path(pineforge_codegen.__file__ or "").resolve()
    package_root = module_file.parent
    digest = hashlib.sha256()
    files = (
        sorted(package_root.rglob("*.py")) if package_root.is_dir() else [module_file]
    )
    for path in files:
        if not path.is_file():
            continue
        try:
            relative = path.relative_to(package_root).as_posix()
        except ValueError:
            relative = path.name
        digest.update(relative.encode("utf-8"))
        digest.update(b"\0")
        digest.update(path.read_bytes())
        digest.update(b"\0")
    return CodegenIdentity(
        version=version,
        implementation_sha256=digest.hexdigest(),
        module_path=str(module_file),
    )