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
from functools import lru_cache
import hashlib
import importlib
from importlib import metadata
from pathlib import Path
import re
from typing import Any, Callable, Mapping
 
from .error import HpoError
 
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 the strategy pipeline, with a separately cached capability canary."""
 
    generated_cpp: str | None
    inputs: tuple[dict[str, Any], ...] = ()
    strategy_params: Mapping[str, Any] = field(default_factory=dict)
    diagnostics: tuple[TranspileDiagnostic, ...] = ()
    input_kind_schema: int | None = None
    request_feed_inventory: Mapping[str, Any] | None = None
 
    @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(HpoError, 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)
        self.with_failure("hpo_transpile_failed", {})
        if any(
            diagnostic.phase == "CONFIGURATION" for diagnostic in result.diagnostics
        ):
            self.with_failure("hpo_toolchain_unavailable", {"reason": "configuration"})
        elif any(diagnostic.phase == "INTERNAL" for diagnostic in result.diagnostics):
            self.with_failure(
                "hpo_toolchain_unavailable", {"reason": "transpile_internal"}
            )
 
 
@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. When available, the inventory producer replaces ``transpile_full``
    for this source. It stays chart-unbound; requests for the chart's timeframe
    therefore remain unknown. The input-kind canary still uses its own source.
    """
 
    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,
                ),
            ),
        )
 
    produces_inventory = codegen_supports_request_inventory()
    strategy_transpiler = (
        pineforge_codegen.transpile_with_request_inventory
        if produces_inventory
        else transpiler
    )
    try:
        payload = strategy_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))
 
    inventory = payload.get("request_feed_inventory") if produces_inventory else None
    if produces_inventory and (
        not isinstance(inventory, Mapping)
        or inventory.get("schema") != REQUEST_FEED_INVENTORY_SCHEMA
    ):
        return TranspileResult(
            generated_cpp=None,
            diagnostics=(
                TranspileDiagnostic(
                    severity="error",
                    phase="INTERNAL",
                    message=(
                        "transpile_with_request_inventory returned an invalid "
                        "request_feed_inventory object or schema"
                    ),
                    filename=filename,
                ),
            ),
        )
 
    return TranspileResult(
        generated_cpp=cpp,
        inputs=tuple(normalized_inputs),
        strategy_params=dict(strategy_params),
        diagnostics=_diagnostics_from_payload(payload),
        input_kind_schema=(
            1
            if isinstance(payload.get("requests"), list)
            and all(isinstance(item, Mapping) for item in payload["requests"])
            and _input_kind_canary(transpiler)
            else None
        ),
        request_feed_inventory=inventory,
    )
 
 
def codegen_supports_request_inventory() -> bool:
    """Feature-detect the loaded producer, including older supported codegens."""
 
    try:
        _load_codegen()
    except ModuleNotFoundError:
        return False
    return callable(
        getattr(pineforge_codegen, "transpile_with_request_inventory", None)
    )
 
 
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),
    )