rose-ash/shared/config.py

# shared/config.py — SX-first config loader with YAML fallback
from __future__ import annotations

import asyncio
import os
from types import MappingProxyType
from typing import Any, Optional
import copy

# Default config paths (override with APP_CONFIG_FILE)
_DEFAULT_YAML_PATH = os.environ.get(
    "APP_CONFIG_FILE",
    os.path.join(os.getcwd(), "config/app-config.yaml"),
)
_DEFAULT_SX_PATH = os.environ.get(
    "APP_CONFIG_SX_FILE",
    os.path.join(os.getcwd(), "config/app-config.sx"),
)

# Module state
_init_lock = asyncio.Lock()
_data_frozen: Any = None   # read-only view (mappingproxy / tuples / frozensets)
_data_plain: Any = None    # plain builtins for pretty-print / logging

# ---------------- utils ----------------
def _freeze(obj: Any) -> Any:
    """Deep-freeze containers to read-only equivalents."""
    if isinstance(obj, dict):
        return MappingProxyType({k: _freeze(v) for k, v in obj.items()})
    if isinstance(obj, list):
        return tuple(_freeze(v) for v in obj)
    if isinstance(obj, set):
        return frozenset(_freeze(v) for v in obj)
    if isinstance(obj, tuple):
        return tuple(_freeze(v) for v in obj)
    return obj


def _sx_to_dict(expr: Any) -> Any:
    """Convert parsed SX config values to plain Python dicts/lists.

    - Keyword keys become strings (kebab-case preserved, also aliased to
      underscore form for backward compatibility with YAML consumers).
    - (env-get "VAR") calls are resolved to os.environ.
    - Lists become plain Python lists.
    - Everything else passes through as-is.
    """
    from shared.sx.types import Keyword, Symbol

    # (env-get "VAR") → os.environ.get("VAR")
    if isinstance(expr, list) and len(expr) == 2:
        head = expr[0]
        if isinstance(head, Symbol) and head.name == "env-get":
            var_name = str(expr[1])
            return os.environ.get(var_name)

    # dict with keyword keys
    if isinstance(expr, dict):
        result: dict[str, Any] = {}
        for k, v in expr.items():
            key = k if isinstance(k, str) else str(k)
            val = _sx_to_dict(v)
            result[key] = val
            # Alias kebab-case → underscore for backward compat
            underscore = key.replace("-", "_")
            if underscore != key:
                result[underscore] = val
        return result

    if isinstance(expr, list):
        # Check for (env-get ...) first (already handled above for len==2)
        return [_sx_to_dict(item) for item in expr]

    if isinstance(expr, tuple):
        return [_sx_to_dict(item) for item in expr]

    if isinstance(expr, Keyword):
        return str(expr)

    if isinstance(expr, Symbol):
        name = expr.name
        if name == "true":
            return True
        if name == "false":
            return False
        if name == "nil":
            return None
        return name

    return expr


def _load_sx_config(path: str) -> dict:
    """Load an SX config file and return a plain dict.

    Expects a single (defconfig name body) form.
    """
    from shared.sx.parser import parse_all

    with open(path, "r", encoding="utf-8") as f:
        source = f.read()

    exprs = parse_all(source)
    for expr in exprs:
        if (isinstance(expr, list) and len(expr) >= 3
                and hasattr(expr[0], 'name') and expr[0].name == "defconfig"):
            # (defconfig name {body})
            body = expr[2]
            return _sx_to_dict(body)

    raise ValueError(f"No (defconfig ...) form found in {path}")


# ---------------- API ----------------
async def init_config(path: Optional[str] = None, *, force: bool = False) -> None:
    """
    Load config and cache both a frozen (read-only) and a plain copy.

    Prefers SX config (app-config.sx) when available, falls back to YAML.
    Idempotent; pass force=True to reload.
    """
    global _data_frozen, _data_plain

    if _data_frozen is not None and not force:
        return

    async with _init_lock:
        if _data_frozen is not None and not force:
            return

        # Try SX first, then YAML
        sx_path = path if (path and path.endswith(".sx")) else _DEFAULT_SX_PATH
        yaml_path = path if (path and not path.endswith(".sx")) else _DEFAULT_YAML_PATH

        if os.path.exists(sx_path):
            raw = _load_sx_config(sx_path)
        elif os.path.exists(yaml_path):
            import yaml
            with open(yaml_path, "r", encoding="utf-8") as f:
                raw = yaml.safe_load(f)
        else:
            raise FileNotFoundError(
                f"No config found: tried {sx_path} and {yaml_path}")

        _data_plain = raw
        _data_frozen = _freeze(raw)

def config() -> Any:
    """
    Return the read-only (frozen) config. Call init_config() first.
    """
    if _data_frozen is None:
        raise RuntimeError("init_config() has not been awaited yet.")
    return _data_frozen

def as_plain() -> Any:
    """
    Return a deep copy of the plain config for safe external use/pretty printing.
    """
    if _data_plain is None:
        raise RuntimeError("init_config() has not been awaited yet.")
    return copy.deepcopy(_data_plain)

def pretty() -> str:
    """
    Pretty string for logging. Uses YAML if available, else pprint.
    """
    if _data_plain is None:
        raise RuntimeError("init_config() has not been awaited yet.")
    try:
        import yaml
        return yaml.safe_dump(_data_plain, sort_keys=False, allow_unicode=True)
    except ImportError:
        import pprint
        return pprint.pformat(_data_plain)