"""Data model + JSON-backed store for the label evaluation platform. Entities (spec ยง37): Dataset, Scene, Sequence, Frame, Sensor, Object, Annotation, Track, Scenario, RareEvent, AnomalyDetection, ValidationResult, GroundTruth, Grader, GraderComparison, RegressionResult, Benchmark, BenchmarkResult, TriageDecision, ReviewTask, HumanReview, RelabelingAction, TrainingDataset, TrainingJob, Model, ModelEvaluation, ProcessUsage, AuditEvent. """ from __future__ import annotations import json import threading import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Literal, Optional from pydantic import BaseModel, Field STORE_DIR = Path("runs/labeleval") GroundTruthType = Literal[ "PSEUDO_GROUND_TRUTH", "VENDOR_GROUND_TRUTH", "HUMAN_VERIFIED_GROUND_TRUTH", "GOLD_STANDARD", ] TriageStatus = Literal["AUTO_GRADED", "FLAGGED", "VERIFIED", "REJECTED", "PENDING"] FAILURE_REASONS = [ "LOW_IOU", "POSITION_ERROR", "ORIENTATION_ERROR", "INSUFFICIENT_POINT_SUPPORT", "ANOMALY", "GRADER_DISAGREEMENT", "TRACK_FRAGMENTATION", "ID_SWITCH", "MODEL_REGRESSION", "LOW_CONFIDENCE", "SENSOR_DISAGREEMENT", ] CLASSES = ["pedestrian", "cyclist", "vehicle", "motorcycle", "truck"] SAFETY_CRITICAL_CLASSES = {"pedestrian", "cyclist", "motorcycle"} SCENARIO_TYPES = [ "near_collision", "extreme_ttc", "extreme_pet", "vru_interaction", "unusual_object_behavior", "severe_occlusion", "sensor_failure", "nighttime_glare", "adverse_weather", "unusual_trajectory", "unexpected_road_geometry", ] def now_iso() -> str: return datetime.now(timezone.utc).isoformat() def new_id(prefix: str) -> str: return f"{prefix}-{uuid.uuid4().hex[:10]}" # ------------------------------------------------------------------ sensor data class EgoState(BaseModel): x: float = 0.0 y: float = 0.0 z: float = 0.0 yaw: float = 0.0 speed_mps: float = 0.0 class GroundTruthBox(BaseModel): gt_id: str class_name: str bbox_3d: List[float] # [x, y, z, l, w, h, yaw] bbox_2d: Optional[List[float]] = None track_instance_id: str = "" gt_type: GroundTruthType = "VENDOR_GROUND_TRUTH" velocity: List[float] = Field(default_factory=lambda: [0.0, 0.0]) class Frame(BaseModel): frame_id: str dataset_id: str scene_id: str sequence_id: str index: int = 0 timestamp_us: int = 0 ego_pose: EgoState = Field(default_factory=EgoState) num_lidar_points: int = 0 camera_width: int = 800 camera_height: int = 450 gt_boxes: List[GroundTruthBox] = Field(default_factory=list) scenario_tags: List[str] = Field(default_factory=list) weather: str = "clear" time_of_day: str = "day" class Sequence(BaseModel): sequence_id: str scene_id: str dataset_id: str frame_ids: List[str] = Field(default_factory=list) class Scene(BaseModel): scene_id: str dataset_id: str sequence_ids: List[str] = Field(default_factory=list) description: str = "" class DatasetLineage(BaseModel): generated_from_model: Optional[str] = None corrected_by_review_batch: Optional[str] = None validated_by_policy: Optional[str] = None parent_dataset: Optional[str] = None class Dataset(BaseModel): dataset_id: str name: str version: str = "v1" created_at: str = Field(default_factory=now_iso) seed: int = 7 num_scenes: int = 0 num_sequences: int = 0 num_frames: int = 0 num_annotations: int = 0 gt_type: Optional[GroundTruthType] = "VENDOR_GROUND_TRUTH" gt_coverage: float = 0.0 lineage: DatasetLineage = Field(default_factory=DatasetLineage) status: str = "created" generation_params: Dict[str, Any] = Field(default_factory=dict) # ------------------------------------------------------------------ labels class Annotation(BaseModel): """Auto-generated label: a hypothesis requiring evaluation. Never assumed correct.""" annotation_id: str dataset_id: str frame_id: str object_id: str class_name: str confidence: float = 0.0 bbox_2d: Optional[List[float]] = None # [x, y, w, h] image px bbox_3d: Optional[List[float]] = None # [x, y, z, l, w, h, yaw] mask: Optional[str] = None track_id: Optional[str] = None model: str = "synthlab-detector" model_version: str = "model-v1" source: str = "auto" status: TriageStatus = "PENDING" matched_gt_id: Optional[str] = None injected_errors: List[str] = Field(default_factory=list) class Track(BaseModel): track_id: str dataset_id: str class_name: str = "vehicle" annotation_ids: List[str] = Field(default_factory=list) frame_ids: List[str] = Field(default_factory=list) class Scenario(BaseModel): scenario_id: str dataset_id: str scenario_type: str frame_ids: List[str] = Field(default_factory=list) description: str = "" class RareEvent(BaseModel): event_id: str dataset_id: str scenario_type: str severity: Literal["low", "medium", "high", "critical"] = "medium" rarity_score: float = 0.0 anomaly_score: float = 0.0 confidence: float = 0.0 evidence_frames: List[str] = Field(default_factory=list) sensor_evidence: Dict[str, str] = Field(default_factory=dict) description: str = "" verified: bool = False # ------------------------------------------------------------------ evaluation evidence class AnomalyDetection(BaseModel): annotation_id: str score: float = 0.0 is_anomaly: bool = False detector_scores: Dict[str, float] = Field(default_factory=dict) normalized_scores: Dict[str, float] = Field(default_factory=dict) ensemble_strategy: str = "weighted_average" ensemble_score: float = 0.0 decision_threshold: float = 0.7 class CheckLine(BaseModel): gate: str actual: Any = None threshold: Any = None passed: bool = True applicable: bool = True class ValidationResult(BaseModel): annotation_id: str passed: bool = True checks: List[CheckLine] = Field(default_factory=list) iou_3d: Optional[float] = None position_error: Optional[float] = None orientation_error_deg: Optional[float] = None dimension_error: Optional[float] = None point_density: Optional[float] = None point_in_box_ratio: Optional[float] = None ground_contact_error: Optional[float] = None sensor_consistent: bool = True class GraderComparison(BaseModel): annotation_id: str grader_count: int = 0 graders: List[str] = Field(default_factory=list) class_votes: Dict[str, str] = Field(default_factory=dict) class_agreement: Optional[float] = None spatial_agreement: Optional[float] = None temporal_agreement: Optional[float] = None consensus: Optional[float] = None disagreement_types: List[str] = Field(default_factory=list) kappa_stats: Dict[str, float] = Field(default_factory=dict) class TrackingEvidence(BaseModel): annotation_id: str id_switch: bool = False fragmentation: bool = False track_quality: Optional[float] = None class TriageDecision(BaseModel): decision_id: str annotation_id: str status: TriageStatus = "PENDING" failure_reasons: List[str] = Field(default_factory=list) primary_failure_reason: Optional[str] = None gate_lines: List[CheckLine] = Field(default_factory=list) policy_id: str = "" policy_values: Dict[str, float] = Field(default_factory=dict) decided_at: str = Field(default_factory=now_iso) # ------------------------------------------------------------------ review class RelabelingAction(BaseModel): action: str corrected_bbox_3d: Optional[List[float]] = None corrected_class: Optional[str] = None merge_with_track_id: Optional[str] = None note: Optional[str] = None class ReviewResolution(BaseModel): action: str corrected_bbox_3d: Optional[List[float]] = None corrected_class: Optional[str] = None revalidation_passed: Optional[bool] = None final_status: Optional[TriageStatus] = None reviewed_at: str = Field(default_factory=now_iso) class ReviewTask(BaseModel): task_id: str annotation_id: str frame_id: str dataset_id: str failure_reasons: List[str] = Field(default_factory=list) primary_failure_reason: Optional[str] = None status: Literal["open", "in_review", "resolved"] = "open" created_at: str = Field(default_factory=now_iso) review_batch: str = "" resolution: Optional[ReviewResolution] = None class HumanReview(BaseModel): review_id: str task_id: str annotation_id: str reviewer: str = "human-reviewer" action: RelabelingAction created_at: str = Field(default_factory=now_iso) # ------------------------------------------------------------------ regression / benchmark class RegressionMetricDelta(BaseModel): metric: str baseline: float current: float delta: float tolerance: float regressed: bool class RegressionResult(BaseModel): result_id: str model_version: str baseline_version: Optional[str] = None dataset_version: str = "" run_id: str = "" date: str = Field(default_factory=now_iso) regression_detected: bool = False affected_classes: List[str] = Field(default_factory=list) affected_scenarios: List[str] = Field(default_factory=list) deltas: List[RegressionMetricDelta] = Field(default_factory=list) kinds: List[str] = Field(default_factory=list) class BenchmarkResult(BaseModel): technique: str precision: float = 0.0 recall: float = 0.0 rare_recall: float = 0.0 f1: float = 0.0 box_error_3d: float = 0.0 id_swap_rate: float = 0.0 consensus: float = 0.0 process_units: int = 0 fp_rate: float = 0.0 class Benchmark(BaseModel): benchmark_id: str dataset_id: str = "" created_at: str = Field(default_factory=now_iso) rows: List[BenchmarkResult] = Field(default_factory=list) highlights: Dict[str, str] = Field(default_factory=dict) # ------------------------------------------------------------------ training / models class TrainingDataset(BaseModel): training_dataset_id: str source_dataset_id: str version: str num_verified_labels: int = 0 lineage: DatasetLineage = Field(default_factory=DatasetLineage) created_at: str = Field(default_factory=now_iso) class TrainingJob(BaseModel): job_id: str model_id: str model_version: str dataset_version: str status: Literal["queued", "running", "completed", "failed", "stopped"] = "queued" epoch: int = 0 total_epochs: int = 10 loss: float = 0.0 rare_recall: float = 0.0 safety_recall: float = 0.0 process_units: int = 0 logs: List[str] = Field(default_factory=list) started_at: str = Field(default_factory=now_iso) quality_policy: str = "" configuration: Dict[str, Any] = Field(default_factory=dict) lineage: DatasetLineage = Field(default_factory=DatasetLineage) class ModelEvaluation(BaseModel): precision: Optional[float] = None recall: Optional[float] = None f1: Optional[float] = None map_3d: Optional[float] = None safety_critical_recall: Optional[float] = None rare_recall: Optional[float] = None class Model(BaseModel): model_id: str model_version: str name: str = "synthlab-detector" trained_on_dataset: Optional[str] = None created_at: str = Field(default_factory=now_iso) status: str = "ready" metrics: ModelEvaluation = Field(default_factory=ModelEvaluation) regression_status: Literal["baseline", "improved", "regressed", "unknown"] = "unknown" # ------------------------------------------------------------------ ops class ProcessUsage(BaseModel): usage_id: str stage: str units: int = 0 run_id: str = "" created_at: str = Field(default_factory=now_iso) class AuditEvent(BaseModel): event_id: str timestamp: str = Field(default_factory=now_iso) actor: str = "system" action: str = "" entity_type: str = "" entity_id: str = "" detail: str = "" class Alert(BaseModel): alert_id: str kind: str severity: Literal["info", "warning", "critical"] = "warning" message: str = "" evidence_page: str = "overview" evidence_id: Optional[str] = None created_at: str = Field(default_factory=now_iso) # ------------------------------------------------------------------ store _COLLECTIONS = { "datasets": Dataset, "scenes": Scene, "sequences": Sequence, "frames": Frame, "annotations": Annotation, "tracks": Track, "scenarios": Scenario, "rare_events": RareEvent, "anomalies": AnomalyDetection, "validations": ValidationResult, "grader_comparisons": GraderComparison, "tracking_evidence": TrackingEvidence, "triage_decisions": TriageDecision, "review_tasks": ReviewTask, "human_reviews": HumanReview, "regressions": RegressionResult, "benchmarks": Benchmark, "training_datasets": TrainingDataset, "training_jobs": TrainingJob, "models": Model, "process_usage": ProcessUsage, "audit_events": AuditEvent, "alerts": Alert, } _KEY_FIELDS = { "datasets": "dataset_id", "scenes": "scene_id", "sequences": "sequence_id", "frames": "frame_id", "annotations": "annotation_id", "tracks": "track_id", "scenarios": "scenario_id", "rare_events": "event_id", "anomalies": "annotation_id", "validations": "annotation_id", "grader_comparisons": "annotation_id", "tracking_evidence": "annotation_id", "triage_decisions": "annotation_id", # latest decision per annotation "review_tasks": "task_id", "human_reviews": "review_id", "regressions": "result_id", "benchmarks": "benchmark_id", "training_datasets": "training_dataset_id", "training_jobs": "job_id", "models": "model_id", "process_usage": "usage_id", "audit_events": "event_id", "alerts": "alert_id", } class EvalStore: """Thread-safe in-memory store with JSON persistence. Determinism and explainability matter more than storage tech (spec). """ def __init__(self, base_dir: Path = STORE_DIR): self.base_dir = Path(base_dir) self.lock = threading.RLock() self.collections: Dict[str, Dict[str, BaseModel]] = {c: {} for c in _COLLECTIONS} self.meta: Dict[str, Any] = {} self._load() # -- persistence def _path(self) -> Path: return self.base_dir / "store.json" def _load(self) -> None: path = self._path() if not path.exists(): return try: with open(path) as f: raw = json.load(f) except Exception: return with self.lock: for name, model_cls in _COLLECTIONS.items(): items = raw.get(name, {}) parsed: Dict[str, BaseModel] = {} for key, payload in items.items(): try: parsed[key] = model_cls.model_validate(payload) except Exception: continue self.collections[name] = parsed self.meta = raw.get("_meta", {}) def save(self) -> None: with self.lock: payload = { name: {k: v.model_dump() for k, v in items.items()} for name, items in self.collections.items() } payload["_meta"] = self.meta self.base_dir.mkdir(parents=True, exist_ok=True) tmp = self._path().with_suffix(".tmp") with open(tmp, "w") as f: json.dump(payload, f) tmp.replace(self._path()) # -- generic access def put(self, collection: str, item: BaseModel) -> None: key = getattr(item, _KEY_FIELDS[collection]) with self.lock: self.collections[collection][key] = item def get(self, collection: str, key: str) -> Optional[Any]: with self.lock: return self.collections[collection].get(key) def all(self, collection: str) -> List[Any]: with self.lock: return list(self.collections[collection].values()) def where(self, collection: str, **filters: Any) -> List[Any]: with self.lock: items = list(self.collections[collection].values()) out = [] for it in items: if all(getattr(it, k, None) == v for k, v in filters.items()): out.append(it) return out def delete(self, collection: str, key: str) -> None: with self.lock: self.collections[collection].pop(key, None) def clear(self, collection: str) -> None: with self.lock: self.collections[collection] = {} # -- audit helper def audit(self, action: str, entity_type: str, entity_id: str, detail: str = "", actor: str = "system") -> None: evt = AuditEvent( event_id=new_id("audit"), actor=actor, action=action, entity_type=entity_type, entity_id=entity_id, detail=detail, ) self.put("audit_events", evt) _STORE: Optional[EvalStore] = None _STORE_LOCK = threading.Lock() def get_store() -> EvalStore: global _STORE with _STORE_LOCK: if _STORE is None: _STORE = EvalStore() return _STORE def reset_store(base_dir: Optional[Path] = None) -> EvalStore: """Used by tests to get a fresh isolated store.""" global _STORE with _STORE_LOCK: _STORE = EvalStore(base_dir or STORE_DIR) return _STORE