sensorflow/rotr/models.py

"""Typed data contracts for the ROTR capability.

Design rules (docs/architecture/rotr-architecture.md §7):
* every persisted record carries a Provenance block;
* tri-state evidence is an enum, never a boolean;
* no defaults that could fabricate evidence — missing evidence is UNKNOWN.
"""

from __future__ import annotations

from datetime import datetime, timezone
from typing import Dict, List, Optional

from pydantic import BaseModel, Field

# ---------------------------------------------------------------- enums

CAUSAL_LAYERS = ["perception", "prediction", "planning", "localization",
                 "map", "control", "policy_rule", "data_label"]

EVIDENCE_STATUSES = ["SUPPORTED", "RULED_OUT", "UNKNOWN"]

CONSEQUENCE_CLASSES = ["NO_MATERIAL_CONSEQUENCE", "DEGRADED_COMFORT",
                       "PLANNER_INTERVENTION", "SAFETY_CRITICAL"]

DATASET_ROLES = ["TRAIN", "VALIDATION", "TEST", "REGRESSION", "LAUNCH",
                 "MONITORING"]
# Membership in these roles blocks training eligibility without an override.
PROTECTED_ROLES = {"REGRESSION", "LAUNCH"}

RIGHT_OF_WAY = ["HAS_ROW", "MUST_YIELD", "NONE"]
INTENTS = ["CROSSING", "MERGING", "YIELDING", "PROCEEDING", "WAITING"]

SURROGATE_CAVEAT = (
    "TTC/PET/DRAC are surrogate conflict measures: they do not alone "
    "establish collision risk (controllability, evasive options and actor "
    "reaction are unmodeled). They prioritize triage; they never decide a "
    "gate by themselves.")


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


# ---------------------------------------------------------------- provenance


class Provenance(BaseModel):
    scenario_id: Optional[str] = None
    dataset_version: str = ""
    model_version: str = ""
    software_version: str = ""
    calibration_version: str = ""
    timestamp: str = Field(default_factory=now_iso)
    source: str = "SYNTHETIC"          # SYNTHETIC | LOG | COUNTERFACTUAL
    confidence: Optional[float] = None


# ---------------------------------------------------------------- world state


class EgoState(BaseModel):
    t: float
    x: float
    y: float
    yaw: float = 0.0
    speed: float = 0.0
    accel: float = 0.0
    lane_id: Optional[str] = None            # true lane association
    believed_x: Optional[float] = None       # localization view (None=UNKNOWN)
    believed_y: Optional[float] = None
    believed_lane_id: Optional[str] = None


class ActorObservation(BaseModel):
    """The stack's view of one actor at one frame (None = not detected)."""

    detected: bool = False
    x: Optional[float] = None
    y: Optional[float] = None
    class_name: Optional[str] = None


class ActorState(BaseModel):
    t: float
    x: float
    y: float
    vx: float = 0.0
    vy: float = 0.0
    yaw: float = 0.0
    occluded: bool = False
    observation: Optional[ActorObservation] = None


class Actor(BaseModel):
    actor_id: str
    class_name: str                          # pedestrian|cyclist|vehicle|bus
    dims: List[float] = Field(default_factory=lambda: [4.5, 1.9, 1.6])
    right_of_way: str = "NONE"               # relative to ego at the conflict
    intent: str = "PROCEEDING"               # ground-truth intent
    predicted_intent: Optional[str] = None   # stack's prediction (None=UNKNOWN)
    states: List[ActorState] = Field(default_factory=list)


class Lane(BaseModel):
    lane_id: str
    center_y: float
    direction: str = "+x"
    permitted_maneuvers: List[str] = Field(
        default_factory=lambda: ["STRAIGHT", "LANE_CHANGE"])
    restricted_to: Optional[str] = None      # e.g. "bus" — None = unrestricted


class Crosswalk(BaseModel):
    crosswalk_id: str
    x_min: float
    x_max: float


class RoadContext(BaseModel):
    context_id: str = "ctx"
    intersection_type: str = "none"          # none|uncontrolled|controlled
    control: str = "none"                    # none|signal|stop_sign|yield_sign
    signal_state_for_ego: Optional[str] = None   # green|red (controlled only)
    intersection_x_min: Optional[float] = None
    intersection_x_max: Optional[float] = None
    stop_line_x: Optional[float] = None
    lanes: List[Lane] = Field(default_factory=list)
    crosswalks: List[Crosswalk] = Field(default_factory=list)
    speed_limit_mps: float = 13.4


class Environment(BaseModel):
    visibility: str = "clear"                # clear|low
    weather: str = "clear"                   # clear|rain
    lighting: str = "day"                    # day|dusk|night


class PlannedPoint(BaseModel):
    t: float
    x: float
    y: float
    v: float


class PlantedTruth(BaseModel):
    """Ground truth planted by the generator, used ONLY for evaluation."""

    kind: str
    is_violation_opportunity: bool = False
    committed: bool = False                  # did this stack commit it?
    expected_rule_id: Optional[str] = None
    cause_layer: Optional[str] = None
    notes: str = ""


class ROTRScenario(BaseModel):
    scenario_id: str
    bank_id: str = ""
    seed: int = 0
    description: str = ""
    environment: Environment = Field(default_factory=Environment)
    map_context: RoadContext = Field(default_factory=RoadContext)   # as mapped
    actual_context: RoadContext = Field(default_factory=RoadContext)  # as built
    ego: List[EgoState] = Field(default_factory=list)     # executed trajectory
    planned: List[PlannedPoint] = Field(default_factory=list)
    actors: List[Actor] = Field(default_factory=list)
    planted: PlantedTruth = Field(default_factory=lambda: PlantedTruth(kind=""))
    provenance: Provenance = Field(default_factory=Provenance)


# ---------------------------------------------------------------- detection


class ROTRViolation(BaseModel):
    violation_id: str
    scenario_id: str
    rule_id: str
    rule_version: str
    description: str = ""
    actor_ids: List[str] = Field(default_factory=list)
    t_start: float = 0.0
    t_end: float = 0.0
    evidence: Dict = Field(default_factory=dict)
    confidence: float = 0.0
    taxonomy: Dict[str, str] = Field(default_factory=dict)
    provenance: Provenance = Field(default_factory=Provenance)


# ---------------------------------------------------------------- attribution


class LayerEvidence(BaseModel):
    layer: str
    status: str = "UNKNOWN"                  # SUPPORTED|RULED_OUT|UNKNOWN
    evidence: str = ""
    confidence: float = 0.0


class FailureAttribution(BaseModel):
    violation_id: str
    scenario_id: str
    layers: Dict[str, LayerEvidence] = Field(default_factory=dict)
    primary_layer: Optional[str] = None      # None = triage to HITL
    note: str = ""
    provenance: Provenance = Field(default_factory=Provenance)


# ---------------------------------------------------------------- consequence


class SafetyAssessment(BaseModel):
    min_ttc_s: Optional[float] = None
    pet_s: Optional[float] = None
    min_clearance_m: Optional[float] = None
    stopping_distance_m: Optional[float] = None
    max_braking_mps2: float = 0.0
    max_lateral_deviation_m: float = 0.0
    collision: bool = False
    surrogate_caveat: str = SURROGATE_CAVEAT


class PlannerEvaluation(BaseModel):
    engine: str = ""                         # which replay planner produced this
    observed_trajectory: List[Dict] = Field(default_factory=list)
    corrected_trajectory: List[Dict] = Field(default_factory=list)
    max_position_divergence_m: float = 0.0
    max_speed_divergence_mps: float = 0.0
    corrected_max_braking_mps2: float = 0.0
    corrected_intervention: bool = False


class CounterfactualScenario(BaseModel):
    counterfactual_id: str
    violation_id: str
    scenario_id: str
    corrected_layers: List[str] = Field(default_factory=list)
    consequence_class: str = "NO_MATERIAL_CONSEQUENCE"
    planner_evaluation: PlannerEvaluation = Field(default_factory=PlannerEvaluation)
    observed_safety: SafetyAssessment = Field(default_factory=SafetyAssessment)
    corrected_safety: SafetyAssessment = Field(default_factory=SafetyAssessment)
    provenance: Provenance = Field(default_factory=Provenance)


# ---------------------------------------------------------------- metrics


class RegressionResult(BaseModel):
    regression_id: str
    baseline_run_id: str
    candidate_run_id: str
    baseline_model: str = ""
    candidate_model: str = ""
    metric_deltas: Dict = Field(default_factory=dict)
    six_outcomes: Dict[str, bool] = Field(default_factory=dict)
    primary_outcome: str = ""
    seqeval: Optional[Dict] = None           # delegation payload (None=unavailable)
    distribution_note: str = ""
    provenance: Provenance = Field(default_factory=Provenance)


# ---------------------------------------------------------------- governance


class HITLReview(BaseModel):
    review_id: str
    run_id: str
    violation_id: str
    cluster_id: Optional[str] = None
    status: str = "PENDING"                  # PENDING|VALIDATED|REJECTED
    action: Optional[str] = None
    actor: Optional[str] = None
    notes: str = ""
    provenance: Provenance = Field(default_factory=Provenance)


class TrainingCandidate(BaseModel):
    candidate_id: str
    run_id: str
    violation_id: str
    cluster_id: Optional[str] = None
    dataset_role: str = "REGRESSION"
    training_eligible: bool = False
    guard_state: str = "PROTECTED"           # PROTECTED|ELIGIBLE|OVERRIDDEN
    override: Optional[Dict] = None          # {actor, reason, timestamp}
    provenance: Provenance = Field(default_factory=Provenance)


class ReleaseGateResult(BaseModel):
    gate_id: str
    run_id: str
    policy_version: str
    outcome: str = "GO"                      # GO|NO_GO
    events: List[Dict] = Field(default_factory=list)
    agentic_advisory: Optional[Dict] = None  # forwarded policy-engine outcome
    provenance: Provenance = Field(default_factory=Provenance)