""" Component dependency analysis. Thin host wrapper over bootstrapped deps module from shared/sx/ref/deps.sx. The canonical logic lives in the spec; this module provides Python-typed entry points for the rest of the codebase. """ from __future__ import annotations import os from typing import Any from .types import Component, Macro, Symbol def _use_ref() -> bool: return os.environ.get("SX_USE_REF") == "1" # --------------------------------------------------------------------------- # Hand-written fallback (used when SX_USE_REF != 1) # --------------------------------------------------------------------------- def _scan_ast(node: Any) -> set[str]: refs: set[str] = set() _walk(node, refs) return refs def _walk(node: Any, refs: set[str]) -> None: if isinstance(node, Symbol): if node.name.startswith("~"): refs.add(node.name) return if isinstance(node, list): for item in node: _walk(item, refs) return if isinstance(node, dict): for v in node.values(): _walk(v, refs) return def _transitive_deps_fallback(name: str, env: dict[str, Any]) -> set[str]: seen: set[str] = set() def walk(n: str) -> None: if n in seen: return seen.add(n) val = env.get(n) if isinstance(val, Component): for dep in _scan_ast(val.body): walk(dep) elif isinstance(val, Macro): for dep in _scan_ast(val.body): walk(dep) key = name if name.startswith("~") else f"~{name}" walk(key) return seen - {key} def _compute_all_deps_fallback(env: dict[str, Any]) -> None: for key, val in env.items(): if isinstance(val, Component): val.deps = _transitive_deps_fallback(key, env) def _scan_io_refs_fallback(node: Any, io_names: set[str]) -> set[str]: """Scan an AST node for references to IO primitive names.""" refs: set[str] = set() _walk_io(node, io_names, refs) return refs def _walk_io(node: Any, io_names: set[str], refs: set[str]) -> None: if isinstance(node, Symbol): if node.name in io_names: refs.add(node.name) return if isinstance(node, list): for item in node: _walk_io(item, io_names, refs) return if isinstance(node, dict): for v in node.values(): _walk_io(v, io_names, refs) return def _transitive_io_refs_fallback( name: str, env: dict[str, Any], io_names: set[str] ) -> set[str]: """Compute transitive IO primitive references for a component.""" all_refs: set[str] = set() seen: set[str] = set() def walk(n: str) -> None: if n in seen: return seen.add(n) val = env.get(n) if isinstance(val, Component): all_refs.update(_scan_io_refs_fallback(val.body, io_names)) for dep in _scan_ast(val.body): walk(dep) elif isinstance(val, Macro): all_refs.update(_scan_io_refs_fallback(val.body, io_names)) for dep in _scan_ast(val.body): walk(dep) key = name if name.startswith("~") else f"~{name}" walk(key) return all_refs def _compute_all_io_refs_fallback( env: dict[str, Any], io_names: set[str] ) -> None: for key, val in env.items(): if isinstance(val, Component): val.io_refs = _transitive_io_refs_fallback(key, env, io_names) def _scan_components_from_sx_fallback(source: str) -> set[str]: import re return {f"~{m}" for m in re.findall(r'\(~([a-zA-Z_][a-zA-Z0-9_\-]*)', source)} def _components_needed_fallback(page_sx: str, env: dict[str, Any]) -> set[str]: direct = _scan_components_from_sx_fallback(page_sx) all_needed: set[str] = set() for name in direct: all_needed.add(name) val = env.get(name) if isinstance(val, Component) and val.deps: all_needed.update(val.deps) else: all_needed.update(_transitive_deps_fallback(name, env)) return all_needed # --------------------------------------------------------------------------- # Public API — dispatches to bootstrapped or fallback # --------------------------------------------------------------------------- def transitive_deps(name: str, env: dict[str, Any]) -> set[str]: """Compute transitive component dependencies for *name*. Returns the set of all component names (with ~ prefix) that *name* can transitively render, NOT including *name* itself. """ if _use_ref(): from .ref.sx_ref import transitive_deps as _ref_td return set(_ref_td(name, env)) return _transitive_deps_fallback(name, env) def compute_all_deps(env: dict[str, Any]) -> None: """Compute and cache deps for all Component entries in *env*.""" if _use_ref(): from .ref.sx_ref import compute_all_deps as _ref_cad _ref_cad(env) return _compute_all_deps_fallback(env) def scan_components_from_sx(source: str) -> set[str]: """Extract component names referenced in SX source text. Returns names with ~ prefix, e.g. {"~card", "~nav-link"}. """ if _use_ref(): from .ref.sx_ref import scan_components_from_source as _ref_sc return set(_ref_sc(source)) return _scan_components_from_sx_fallback(source) def components_needed(page_sx: str, env: dict[str, Any]) -> set[str]: """Compute the full set of component names needed for a page. Returns names with ~ prefix. """ if _use_ref(): from .ref.sx_ref import components_needed as _ref_cn return set(_ref_cn(page_sx, env)) return _components_needed_fallback(page_sx, env) def compute_all_io_refs(env: dict[str, Any], io_names: set[str]) -> None: """Compute and cache transitive IO refs for all Component entries in *env*.""" if _use_ref(): from .ref.sx_ref import compute_all_io_refs as _ref_cio _ref_cio(env, list(io_names)) return _compute_all_io_refs_fallback(env, io_names) def get_all_io_names() -> set[str]: """Build the complete set of IO primitive names from all boundary tiers. Includes: core IO (primitives_io.py handlers), plus all page helper names from every service boundary. """ from .primitives_io import IO_PRIMITIVES from .boundary import declared_helpers names = set(IO_PRIMITIVES) for _svc, helper_names in declared_helpers().items(): names.update(helper_names) return names