pcb-afm-simulation-system/scripts/executor_config.py

"""Executor configuration loader (P5-M2).

Loads executor_config.json for the local headless executor.  The goal is
to make the packaged EXE configurable without re-building: web address,
model path, logging, poll interval and instance count all come from a
JSON sidecar that the operator can edit next to the EXE.

Priority (highest first):
  1. --config CLI path          (explicit)
  2. $EXECUTOR_CONFIG env var   (explicit)
  3. <exe_or_script_dir>/executor_config.json   (sidecar next to EXE)
  4. <repo_root>/executor_config.json           (source-tree template)
  5. built-in defaults

Single fields may still be overridden by env vars (highest):
  WEB_BASE_URL / MOTORCAD_MODEL / EXECUTOR_INSTANCES /
  EXECUTOR_POLL_INTERVAL / EXECUTOR_LOG_DIR / EXECUTOR_LOG_LEVEL /
  EXECUTOR_TOOL / EXECUTOR_MOCK

All code in this module is ASCII only (AGENTS.md constraint).
"""
import json
import os
import sys

DEFAULT_WEB_BASE_URL = "http://127.0.0.1:8000"
DEFAULT_POLL_INTERVAL = 5
DEFAULT_INSTANCES = 1
DEFAULT_LOG_DIR = "output/executor_logs"
DEFAULT_LOG_LEVEL = "INFO"
DEFAULT_TOOL = "motorcad"
DEFAULT_MODEL_REL = "models/MARS-12S10P_SSSR_D76-C150_V5.0-0819.mot"

VALID_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")


def default_config():
    """Return the built-in default configuration dict."""
    return {
        "web_base_url": DEFAULT_WEB_BASE_URL,
        "model_path": DEFAULT_MODEL_REL,
        "poll_interval": DEFAULT_POLL_INTERVAL,
        "instances": DEFAULT_INSTANCES,
        "log_dir": DEFAULT_LOG_DIR,
        "log_level": DEFAULT_LOG_LEVEL,
        "tool": DEFAULT_TOOL,
        "enable_mock": False,
        "enable_thermal": False,
    }


def _is_frozen():
    """True when running from a PyInstaller onefile bundle."""
    return bool(getattr(sys, "frozen", False))


def _base_dir():
    """Directory that should host the sidecar config for this runtime.

    - EXE mode: the folder containing the executable.
    - Script mode: the repository root (parent of the scripts/ folder).
    """
    if _is_frozen():
        return os.path.dirname(os.path.abspath(sys.executable))
    here = os.path.dirname(os.path.abspath(__file__))
    return os.path.dirname(here)


def repo_root():
    """Repository root (parent of scripts/) when running from source."""
    here = os.path.dirname(os.path.abspath(__file__))
    return os.path.dirname(here)


def find_config_path(cli_path=None):
    """Locate a config file path, or None when none exists.

    Args:
        cli_path: optional explicit path from --config.

    Returns:
        str path of the first existing config file, else None.
    """
    candidates = []
    if cli_path:
        candidates.append(cli_path)
    env_path = os.environ.get("EXECUTOR_CONFIG")
    if env_path:
        candidates.append(env_path)
    candidates.append(os.path.join(_base_dir(), "executor_config.json"))
    candidates.append(os.path.join(repo_root(), "executor_config.json"))
    for cand in candidates:
        if cand and os.path.isfile(cand):
            return cand
    return None


def _resolve_model_path(model_path, base_dir):
    """Turn a possibly-relative model_path into an absolute path.

    Relative paths resolve against the config base dir (EXE dir or repo
    root).  Empty / None values stay empty (EXE may rely on the operator
    to set it; script mode fills a default).
    """
    if not model_path:
        return None
    if os.path.isabs(model_path):
        return os.path.normpath(model_path)
    return os.path.normpath(os.path.join(base_dir, model_path))


def validate_config(cfg):
    """Validate a config dict; raise ValueError on any violation.

    Checks: types, ranges and known enum values.  Empty model_path is
    allowed (EXE may be pointed at a model only at run time).
    """
    if not isinstance(cfg, dict):
        raise ValueError("config must be a dict")
    web = cfg.get("web_base_url")
    if not isinstance(web, str) or not web.startswith("http"):
        raise ValueError("web_base_url must be an http(s) URL string")
    instances = cfg.get("instances")
    if not isinstance(instances, int) or instances < 1:
        raise ValueError("instances must be an int >= 1")
    interval = cfg.get("poll_interval")
    if not isinstance(interval, (int, float)) or interval <= 0:
        raise ValueError("poll_interval must be a positive number")
    level = cfg.get("log_level")
    if level not in VALID_LOG_LEVELS:
        raise ValueError("log_level must be one of %s" % (VALID_LOG_LEVELS,))
    tool = cfg.get("tool")
    if not isinstance(tool, str) or not tool:
        raise ValueError("tool must be a non-empty string")
    mock = cfg.get("enable_mock")
    if not isinstance(mock, bool):
        raise ValueError("enable_mock must be a boolean")
    thermal = cfg.get("enable_thermal")
    if not isinstance(thermal, bool):
        raise ValueError("enable_thermal must be a boolean")
    return cfg


def load_config(cli_path=None, env_overrides=True):
    """Load and validate the effective configuration.

    Args:
        cli_path: optional --config path.
        env_overrides: when True, let env vars override file fields.

    Returns:
        dict with resolved absolute model_path/log_dir and the source
        string under key "config_source".
    """
    cfg = default_config()
    source = "defaults"

    cfg_path = find_config_path(cli_path)
    if cfg_path:
        try:
            with open(cfg_path, "r", encoding="utf-8") as fh:
                file_cfg = json.load(fh)
            if not isinstance(file_cfg, dict):
                raise ValueError("config file must contain a JSON object")
            cfg.update({k: v for k, v in file_cfg.items() if v is not None})
            source = cfg_path
        except (OSError, ValueError) as exc:
            # A malformed explicit config must not be silently ignored:
            # the operator asked for it, so surface the error.
            raise ValueError("failed to load config %s: %s" % (cfg_path, exc))

    if env_overrides:
        env_map = {
            "WEB_BASE_URL": "web_base_url",
            "MOTORCAD_MODEL": "model_path",
            "EXECUTOR_INSTANCES": "instances",
            "EXECUTOR_POLL_INTERVAL": "poll_interval",
            "EXECUTOR_LOG_DIR": "log_dir",
            "EXECUTOR_LOG_LEVEL": "log_level",
            "EXECUTOR_TOOL": "tool",
            "EXECUTOR_MOCK": "enable_mock",
            "EXECUTOR_THERMAL": "enable_thermal",
        }
        for env_key, cfg_key in env_map.items():
            raw = os.environ.get(env_key)
            if raw is None or raw == "":
                continue
            if cfg_key in ("instances", "poll_interval"):
                try:
                    cfg[cfg_key] = float(raw) if "." in raw else int(raw)
                except ValueError:
                    raise ValueError("env %s must be numeric, got %r" % (env_key, raw))
            elif cfg_key in ("enable_mock", "enable_thermal"):
                cfg[cfg_key] = raw.strip().lower() in ("1", "true", "yes", "on")
            else:
                cfg[cfg_key] = raw

    # Coerce numeric fields read from JSON (json gives int/float already).
    if not isinstance(cfg["instances"], int):
        cfg["instances"] = int(cfg["instances"])
    if not isinstance(cfg["poll_interval"], (int, float)):
        cfg["poll_interval"] = float(cfg["poll_interval"])

    base_dir = _base_dir() if _is_frozen() else repo_root()
    model_path = _resolve_model_path(cfg.get("model_path"), base_dir)
    log_dir = cfg.get("log_dir")
    if not log_dir:
        log_dir = DEFAULT_LOG_DIR
    if os.path.isabs(log_dir):
        log_dir = os.path.normpath(log_dir)
    else:
        log_dir = os.path.normpath(os.path.join(base_dir, log_dir))

    resolved = dict(cfg)
    resolved["model_path"] = model_path
    resolved["log_dir"] = log_dir
    resolved["config_source"] = source
    validate_config(resolved)
    return resolved