gps-denied-onboard/src/gps_denied_onboard/runtime_root/vio_factory.py

"""C1 VIO strategy composition-root factory (AZ-331).

:func:`build_vio_strategy` selects exactly one strategy by
``config.components['c1_vio'].strategy`` and respects compile-time
``BUILD_*`` gating: requesting a strategy whose flag is OFF raises
:class:`StrategyNotAvailableError` at composition time (NOT at first
frame).

Concrete strategy modules (``okvis2``, ``vins_mono``, ``klt_ransac``)
are imported lazily — a Tier-0 workstation build with
``BUILD_OKVIS2=OFF`` MUST NOT load ``c1_vio.okvis2`` (Risk-2 / I-5;
verifiable via ``sys.modules``).
"""

from __future__ import annotations

import inspect
import os
from typing import TYPE_CHECKING

from gps_denied_onboard.runtime_root.errors import StrategyNotAvailableError

if TYPE_CHECKING:
    from gps_denied_onboard.components.c1_vio import (
        C1VioConfig,
        VioStrategy,
    )
    from gps_denied_onboard.components.c13_fdr import FdrClient
    from gps_denied_onboard.config.schema import Config
    from gps_denied_onboard.helpers.altitude_provider import AltitudeProvider

__all__ = ["build_vio_strategy"]


_STRATEGY_TO_BUILD_FLAG: dict[str, str] = {
    "okvis2": "BUILD_OKVIS2",
    "vins_mono": "BUILD_VINS_MONO",
    "klt_ransac": "BUILD_KLT_RANSAC",
}

_STRATEGY_TO_MODULE: dict[str, tuple[str, str]] = {
    "okvis2": ("gps_denied_onboard.components.c1_vio.okvis2", "Okvis2Strategy"),
    "vins_mono": (
        "gps_denied_onboard.components.c1_vio.vins_mono",
        "VinsMonoStrategy",
    ),
    "klt_ransac": (
        "gps_denied_onboard.components.c1_vio.klt_ransac",
        "KltRansacStrategy",
    ),
}


def _is_build_flag_on(flag_name: str) -> bool:
    """Read a compile-time ``BUILD_*`` flag from the environment.

    ``ON`` / ``1`` / ``true`` / ``yes`` (case-insensitive) → ``True``;
    anything else (including unset) → ``False``. Defaults to OFF so
    test environments must opt-in explicitly per strategy.
    """
    raw = os.environ.get(flag_name, "")
    return raw.strip().lower() in {"on", "1", "true", "yes"}


def _c1_config(config: "Config") -> "C1VioConfig":
    """Pull the registered C1 config block.

    ``c1_vio.__init__`` registers it on import; a missing
    registration is a developer error and surfaces as ``KeyError``
    rather than a silent fallback.
    """
    return config.components["c1_vio"]


def build_vio_strategy(
    config: "Config",
    *,
    fdr_client: "FdrClient",
    altitude_provider: "AltitudeProvider | None" = None,
) -> "VioStrategy":
    """Construct the :class:`VioStrategy` impl selected by config.

    1. Reads ``config.components['c1_vio'].strategy``.
    2. Checks the matching ``BUILD_*`` flag — if OFF, raises
       :class:`StrategyNotAvailableError` BEFORE any import.
    3. Lazily imports the concrete strategy module.
    4. Constructs and returns the strategy instance, passing
       ``config`` and ``fdr_client`` and (if the strategy's
       ``__init__`` accepts it) ``altitude_provider`` — AZ-919.

    Raises :class:`StrategyNotAvailableError` when the compile-time
    flag is OFF (canonical Tier-0 path) or when the concrete strategy
    module has not been built yet (AZ-332 / AZ-333 / AZ-334 pending).

    The ``altitude_provider`` kwarg is introspected against the
    selected strategy's ``__init__`` signature so OKVIS2 / VINS-Mono
    do not need to grow a parameter they do not yet consume; the
    KLT/RANSAC strategy accepts it (per AZ-919 AC-3) and will use it
    for GSD-based scale recovery in AZ-920.
    """
    block = _c1_config(config)
    strategy = block.strategy
    flag_name = _STRATEGY_TO_BUILD_FLAG.get(strategy)
    module_info = _STRATEGY_TO_MODULE.get(strategy)
    if flag_name is None or module_info is None:
        raise StrategyNotAvailableError(
            f"VioStrategy {strategy!r} is not buildable in this binary."
        )
    if not _is_build_flag_on(flag_name):
        raise StrategyNotAvailableError(
            f"VioStrategy {strategy!r} requires {flag_name}=ON in this "
            "binary; the flag is OFF."
        )
    module_name, class_name = module_info
    try:
        module = __import__(module_name, fromlist=[class_name])
    except ModuleNotFoundError as exc:
        raise StrategyNotAvailableError(
            f"VioStrategy {strategy!r} is configured but its concrete impl "
            f"module {module_name!r} has not been built into this binary "
            "yet (AZ-332 / AZ-333 / AZ-334 pending)."
        ) from exc
    strategy_cls = getattr(module, class_name)
    kwargs: dict[str, object] = {"fdr_client": fdr_client}
    if "altitude_provider" in inspect.signature(strategy_cls).parameters:
        kwargs["altitude_provider"] = altitude_provider
    return strategy_cls(config, **kwargs)