[AZ-266] [AZ-269] [AZ-277] [AZ-280] Cross-cutting log/config + SE3/SHA256 helpers

AZ-266: schema-compliant JSON logging entrypoint, level normalisation,
handler-topology guard, format-error fallback (log_record_schema v1.0.0).
AZ-269: env > YAML > defaults config loader, frozen Config dataclass,
missing-var fail-fast with pointer to .env.example, component-block registry.
AZ-277: GTSAM-backed SE3Utils (matrix<->SE3 + exp/log/adjoint) with strict
orthogonality, dtype, and bottom-row contract enforcement.
AZ-280: atomicwrites-backed write_atomic + independent verify +
order-deterministic aggregate_hash; sidecar format strictness.
pyproject.toml pins gtsam>=4.2,<5.0 and atomicwrites>=1.4,<2.0
(named-backend deps per the AZ-277 / AZ-280 contracts).
139 unit tests pass (44 new). Review verdict: PASS_WITH_WARNINGS;
findings are perf-NFR + journald deferrals, no blocking issues.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Oleksandr Bezdieniezhnykh
2026-05-11 01:33:42 +03:00
parent b12db61444
commit 8e71f6c002
21 changed files with 2134 additions and 133 deletions
+144 -13
View File
@@ -1,26 +1,157 @@
"""Config dataclass schemas — STUB.
"""Config dataclasses for E-CC-CONF (AZ-269 / AZ-246).
Concrete YAML schema is owned by AZ-269. Bootstrap declares only the runtime-level
config container so the composition root can type its `compose_*` signatures.
The outer `Config` aggregates one frozen nested dataclass per top-level
config block. Cross-cutting blocks (`log`, `fdr`, `runtime`) live here;
per-component blocks live with their own component epic and are
registered into `Config.components` via `register_component_block`.
Public surface frozen by `_docs/02_document/contracts/shared_config/composition_root_protocol.md` v1.0.0.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from collections.abc import Mapping
from dataclasses import dataclass, field, fields, is_dataclass, replace
from typing import Any, Final
__all__ = [
"Config",
"ConfigError",
"FdrConfig",
"LogConfig",
"RequiredFieldMissingError",
"RuntimeConfig",
"register_component_block",
]
class ConfigError(RuntimeError):
"""Base class for all config-loader errors that should reach the caller."""
class RequiredFieldMissingError(ConfigError):
"""Raised when a required configuration value is absent from env + YAML + defaults.
Always names the missing variable and points at the runtime helper that
documents the full set (``.env.example``).
"""
@dataclass(frozen=True)
class LogConfig:
"""Cross-cutting logging block (E-CC-LOG)."""
level: str = "INFO"
tier: int = 1
sink: str = "console"
@dataclass(frozen=True)
class FdrConfig:
"""Cross-cutting Flight Data Recorder block (E-CC-FDR-CLIENT / AZ-247).
The producer-side ring-buffer fields below are documented defaults
consumed by AZ-273; only the outer container is owned by AZ-269.
"""
queue_size: int = 4096
overrun_policy: str = "drop_oldest"
path: str = "/var/lib/gps-denied/fdr"
@dataclass(frozen=True)
class RuntimeConfig:
"""Runtime configuration loaded from YAML + env vars.
The concrete field set is filled in by AZ-269. This stub is enough for the
composition root + tests to import the type.
"""
"""Top-level runtime descriptors that don't belong to a single component."""
fc_profile: str = "ardupilot_plane"
tier: int = 1
db_url: str = ""
log_level: str = "INFO"
log_sink: str = "console"
extras: dict[str, Any] = field(default_factory=dict)
camera_calibration_path: str = ""
inference_backend: str = "pytorch_fp16"
tile_cache_path: str = "/var/lib/gps-denied/tiles"
# Documented defaults for cross-cutting blocks ONLY. Per-component defaults
# live with their own component epic. The registry below is the single
# source of truth so two components cannot silently claim the same key.
_DEFAULT_BLOCKS: Final[dict[str, type]] = {
"log": LogConfig,
"fdr": FdrConfig,
"runtime": RuntimeConfig,
}
# Registry for per-component nested dataclasses. Each component epic
# calls ``register_component_block("c5_state", C5StateConfig)`` from its
# package import path; the composition root drives those imports before
# calling ``load_config``.
_COMPONENT_REGISTRY: dict[str, type] = {}
def register_component_block(slug: str, dataclass_type: type) -> None:
"""Register a per-component frozen dataclass under its component slug."""
if not is_dataclass(dataclass_type):
raise TypeError(
f"register_component_block({slug!r}, ...): block must be a dataclass; "
f"got {dataclass_type!r}"
)
existing = _COMPONENT_REGISTRY.get(slug)
if existing is not None and existing is not dataclass_type:
raise ConfigError(
f"duplicate component config registration for slug {slug!r}: "
f"{existing!r} vs {dataclass_type!r}"
)
_COMPONENT_REGISTRY[slug] = dataclass_type
def _resolve_default_blocks() -> dict[str, Any]:
"""Instantiate every documented cross-cutting block with its defaults."""
return {name: cls() for name, cls in _DEFAULT_BLOCKS.items()}
def _resolve_component_blocks() -> dict[str, Any]:
"""Instantiate every registered per-component block with its defaults."""
return {slug: cls() for slug, cls in _COMPONENT_REGISTRY.items()}
@dataclass(frozen=True)
class Config:
"""Outer composition-root config (frozen end-to-end).
Components consume only their own slice via ``config.components[slug]``;
the runtime / log / fdr cross-cutting blocks are read directly via
attribute access by the composition root.
"""
runtime: RuntimeConfig = field(default_factory=RuntimeConfig)
log: LogConfig = field(default_factory=LogConfig)
fdr: FdrConfig = field(default_factory=FdrConfig)
components: Mapping[str, Any] = field(default_factory=dict)
@classmethod
def with_blocks(cls, **blocks: Any) -> Config:
"""Build a `Config` from a flat name-to-instance map.
Cross-cutting names (``log``, ``fdr``, ``runtime``) become attributes;
every other key is treated as a component slug and goes into
``components``.
"""
runtime = blocks.pop("runtime", RuntimeConfig())
log = blocks.pop("log", LogConfig())
fdr = blocks.pop("fdr", FdrConfig())
return cls(runtime=runtime, log=log, fdr=fdr, components=dict(blocks))
def _block_field_names(block: Any) -> tuple[str, ...]:
return tuple(f.name for f in fields(block))
def _replace_block(block: Any, overrides: Mapping[str, Any]) -> Any:
"""Return ``replace(block, **overrides)`` after filtering unknown keys."""
if not overrides:
return block
known = set(_block_field_names(block))
filtered = {k: v for k, v in overrides.items() if k in known}
if not filtered:
return block
return replace(block, **filtered)