"""OCaml bridge for page helper SX functions. Replaces the deleted sx_ref.py imports. Uses OcamlSync to call SX functions defined in web/page-helpers.sx. """ from __future__ import annotations import os import logging from typing import Any _logger = logging.getLogger("sx.page_helpers") _bridge = None _loaded = False def _get_bridge(): """Get the shared OcamlSync bridge, loading page-helpers.sx on first use.""" global _bridge, _loaded from shared.sx.ocaml_sync import OcamlSync if _bridge is None: _bridge = OcamlSync() _bridge._ensure() if not _loaded: _loaded = True # Load files needed by page-helpers.sx base = os.path.join(os.path.dirname(__file__), "../../..") for f in ["spec/parser.sx", "spec/render.sx", "web/adapter-html.sx", "web/adapter-sx.sx", "web/web-forms.sx", "web/page-helpers.sx"]: path = os.path.join(base, f) if os.path.isfile(path): try: _bridge.load(path) except Exception as e: _logger.warning("Failed to load %s: %s", f, e) return _bridge def _py_to_sx(val: Any) -> str: """Serialize a Python value to SX source text.""" if val is None: return "nil" if isinstance(val, bool): return "true" if val else "false" if isinstance(val, (int, float)): return str(val) if isinstance(val, str): escaped = val.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") return f'"{escaped}"' if isinstance(val, (list, tuple)): items = " ".join(_py_to_sx(v) for v in val) return f"(list {items})" if isinstance(val, dict): pairs = " ".join(f":{k} {_py_to_sx(v)}" for k, v in val.items()) return "{" + pairs + "}" return f'"{val}"' def _sx_to_py(text: str) -> Any: """Parse an SX result back to Python. Handles strings, dicts, lists, nil.""" text = text.strip() if not text or text == "nil": return None if text == "true": return True if text == "false": return False # String result — the bridge already unescapes if text.startswith('"') and text.endswith('"'): return text[1:-1] # For complex results, parse as SX and convert from shared.sx.parser import parse_all from shared.sx.types import Keyword, Symbol, NIL exprs = parse_all(text) if not exprs: return text def _convert(val): if val is None or val is NIL: return None if isinstance(val, bool): return val if isinstance(val, (int, float)): return val if isinstance(val, str): return val if isinstance(val, Keyword): return val.name if isinstance(val, Symbol): return val.name if isinstance(val, dict): return {k: _convert(v) for k, v in val.items()} if isinstance(val, list): return [_convert(v) for v in val] return str(val) return _convert(exprs[0]) def call_sx(fn_name: str, *args: Any) -> Any: """Call an SX function by name with Python args, return Python result.""" bridge = _get_bridge() sx_args = " ".join(_py_to_sx(a) for a in args) sx_expr = f"({fn_name} {sx_args})" if sx_args else f"({fn_name})" result = bridge.eval(sx_expr) return _sx_to_py(result) def evaluate(source: str, env: Any = None) -> Any: """Evaluate SX source text, return result as SX string.""" bridge = _get_bridge() return bridge.eval(source) def load_file(path: str) -> None: """Load an .sx file into the bridge kernel.""" bridge = _get_bridge() bridge.load(path) def eval_in_env(expr_sx: str) -> str: """Evaluate an SX expression, return raw SX result string.""" bridge = _get_bridge() return bridge.eval(expr_sx) # ---- Drop-in replacements for sx_ref.py functions ---- def build_component_source(data: dict) -> str: return call_sx("build-component-source", data) or "" def categorize_special_forms(exprs: Any) -> dict: return call_sx("categorize-special-forms", exprs) or {} def build_reference_data(raw: dict, detail_keys: list | None = None) -> dict: return call_sx("build-reference-data", raw, detail_keys or []) or {} def build_attr_detail(slug: str) -> dict: return call_sx("build-attr-detail", slug) or {} def build_header_detail(slug: str) -> dict: return call_sx("build-header-detail", slug) or {} def build_event_detail(slug: str) -> dict: return call_sx("build-event-detail", slug) or {} def build_bundle_analysis(pages: Any, components: Any) -> dict: return call_sx("build-bundle-analysis", pages, components) or {} def build_routing_analysis(pages: Any, components: Any) -> dict: return call_sx("build-routing-analysis", pages, components) or {} def build_affinity_analysis(components: Any, pages: Any) -> dict: return call_sx("build-affinity-analysis", components, pages) or {} def make_env(): """No-op — env is managed by the OCaml bridge.""" return {} def eval_expr(expr, env=None): """Evaluate a parsed SX expression via OCaml bridge.""" from shared.sx.parser import serialize bridge = _get_bridge() sx_text = serialize(expr) if not isinstance(expr, str) else expr result = bridge.eval(sx_text) from shared.sx.parser import parse return parse(result) if result else None def trampoline(val): """No-op — OCaml bridge doesn't return thunks.""" return val def call_lambda(fn, args, env=None): """Call a lambda via the OCaml bridge.""" from shared.sx.parser import serialize bridge = _get_bridge() parts = [serialize(fn)] + [serialize(a) for a in args] result = bridge.eval(f"({' '.join(parts)})") from shared.sx.parser import parse return parse(result) if result else None def render_to_html(expr, env=None): """Render an SX expression to HTML via OCaml bridge.""" from shared.sx.parser import serialize bridge = _get_bridge() sx_text = serialize(expr) if not isinstance(expr, str) else expr return bridge.eval(f'(render-to-html (quote {sx_text}) (env))') # Router/deps/engine helpers — these are loaded from .sx files # and made available via eval. The try/except pattern in helpers.py # falls back to loading the .sx file directly, which works. # These stubs exist so the import doesn't fail. split_path_segments = None parse_route_pattern = None match_route_segments = None match_route = None find_matching_route = None make_route_segment = None scan_refs = None scan_components_from_source = None transitive_deps = None compute_all_deps = None components_needed = None page_component_bundle = None page_css_classes = None scan_io_refs = None transitive_io_refs = None compute_all_io_refs = None component_pure_p = None parse_time = None parse_trigger_spec = None default_trigger = None parse_swap_spec = None parse_retry_spec = None filter_params = None