feat(02-07): add Pydantic v2 boundary-log schemas (OBS-01)

- Create src/gps_denied/obs/log_schemas.py with 3 models:
  MavlinkGpsInputEmitted (AC-4.3 + AC-1.4), ApiRequestCompleted,
  AnchorDecision (VERIFY-02)
- All models use ConfigDict(extra='forbid', frozen=True)
- SourceLabel Literal encodes AC-1.4 vocab; AnchorRejectReason encodes VERIFY-02
- Update gps_denied.obs barrel __init__.py to re-export all schemas + type aliases
This commit is contained in:
Yuzviak
2026-05-11 18:49:55 +03:00
parent e81b6fdfba
commit 94c1b76086
2 changed files with 130 additions and 0 deletions
+30
View File
@@ -0,0 +1,30 @@
"""Observability package — structlog spine + boundary log schemas.
Hot-path logging spine wired by ``configure_logging``; per-frame ``correlation_id``
(= frame_id) bound at ``pipeline/orchestrator.py:process_frame``.
Boundary log schemas (REST, FDR, anchor decisions) defined in ``log_schemas``.
"""
import structlog
from gps_denied.obs.log_schemas import (
AnchorDecision,
AnchorRejectReason,
ApiRequestCompleted,
MavlinkGpsInputEmitted,
SourceLabel,
)
from gps_denied.obs.logging_config import configure_logging
get_logger = structlog.get_logger # convenience re-export
__all__ = [
"configure_logging",
"get_logger",
# Boundary schemas
"MavlinkGpsInputEmitted",
"ApiRequestCompleted",
"AnchorDecision",
"SourceLabel",
"AnchorRejectReason",
]
+100
View File
@@ -0,0 +1,100 @@
"""Pydantic v2 schemas for boundary log records.
Boundary = a point where the hot path hands off to (or receives from) an external
system: MAVLink wire (AC-4.3), REST API, AnchorVerifier accept/reject (VERIFY-02),
FDR JSONL append (AC-NEW-3). Inside the hot path we use
``@dataclass(slots=True, frozen=True)`` per ARCH-02 — these Pydantic schemas are
NEVER used per-frame inside the pipeline body.
Usage pattern (Phase 5 / Phase 6 / Phase 3 wiring):
from gps_denied.obs import log_schemas, get_logger
log = get_logger(__name__)
rec = log_schemas.MavlinkGpsInputEmitted(
lat_deg=est.lat_deg, lon_deg=est.lon_deg, alt_m=est.alt_m,
fix_type=3, horiz_accuracy_m=est.cov_semi_major_m,
source_label=est.source_label, anchor_age_ms=est.anchor_age_ms,
cov_semi_major_m=est.cov_semi_major_m,
)
log.info("mavlink_emit_gps_input", **rec.model_dump(mode="json"))
# frame_id is auto-included via structlog merge_contextvars (Plan 02-06).
``mode="json"`` ensures datetimes / enums / numpy scalars are JSON-safe before the
structlog renderer fires. Validation cost on Pydantic v2 / pydantic-core is
~10-60 µs per record (RESEARCH.md §5.4 performance note); boundary call rates
(≤10 Hz MAVLink, ≤1 Hz API, ≤0.1 Hz failed-tile thumbnails) make this trivial
against the 400 ms / frame budget.
Field validation is intentionally tight (``extra='forbid'``, frozen=True) so adding
a producer-side field without updating the schema fails fast.
"""
from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
# Source-label vocabulary — AC-1.4. Matches the Phase-1 enum
# gps_denied.schemas.eskf.ConfidenceTier (kept in stdlib boundary schemas for
# legacy DB rows); new code uses this Literal.
SourceLabel = Literal["satellite_anchored", "vo_extrapolated", "dead_reckoned"]
# Reject-reason vocabulary — REQUIREMENTS.md VERIFY-02. Stable contract; Phase 3
# AnchorVerifier emits exactly these strings.
AnchorRejectReason = Literal[
"ok",
"too_few_inliers",
"mre_above_threshold",
"degenerate_homography",
"freshness_expired",
]
class _BoundaryLogRecord(BaseModel):
"""Common base. Tighten the contract:
- ``extra='forbid'`` — producer-side typo or schema drift fails validation.
- ``frozen=True`` — records are facts, not state. No mutation after build.
"""
model_config = ConfigDict(extra="forbid", frozen=True)
class MavlinkGpsInputEmitted(_BoundaryLogRecord):
"""Logged once per GPS_INPUT MAVLink message emitted on the wire (AC-4.3).
AC-1.4 categorical source_label + AC-1.3 anchor_age_ms + AC-NEW-8 cov_semi_major_m
are all required fields — the producer must compute them. Phase 5 (MAVOUT-01) wires.
"""
lat_deg: float
lon_deg: float
alt_m: float
fix_type: int = Field(ge=0, le=6) # MAVLink GPS fix_type: 0..6
horiz_accuracy_m: float = Field(ge=0)
source_label: SourceLabel # AC-1.4
anchor_age_ms: int = Field(ge=0) # AC-1.3
cov_semi_major_m: float = Field(ge=0) # AC-1.4 (95% covariance)
class ApiRequestCompleted(_BoundaryLogRecord):
"""Logged at the end of every FastAPI request. Phase 6 middleware consumer."""
path: str
method: Literal["GET", "POST", "PUT", "DELETE", "PATCH"]
status_code: int = Field(ge=100, lt=600)
duration_ms: float = Field(ge=0)
class AnchorDecision(_BoundaryLogRecord):
"""Logged on every AnchorVerifier accept/reject (VERIFY-02). Phase 3 consumer."""
decision: Literal["accept", "reject"]
reason: AnchorRejectReason
n_inliers: int = Field(ge=0)
mre_px: float = Field(ge=0)
__all__ = [
"SourceLabel",
"AnchorRejectReason",
"MavlinkGpsInputEmitted",
"ApiRequestCompleted",
"AnchorDecision",
]