Spaces:
Running on Zero
Running on Zero
| """Phase 7 — Safety Risk Fusion orchestrator. | |
| Connects the completed probability models without retraining them: | |
| - Phase 4 near-miss/conflict model | |
| (models/near_miss/random_forest_D_calibrated.joblib) | |
| via the existing near_miss.probability_pipeline.ProbabilityPipeline | |
| -> calibrated class-1 probability. | |
| - Phase 6 calibrated accident detector | |
| (models/accident/calibrated_accident_detector.pth) | |
| via accident.calibrated_detector.CalibratedAccidentDetector | |
| -> calibrated accident probability. | |
| Model D is the deployment-realistic near-miss model. It does NOT require: | |
| - PET | |
| - FE_inv_PET | |
| - FE_safety_index | |
| The near-miss probability and accident probability remain separate | |
| and are passed to risk_engine.assess_risk together with caller-supplied | |
| severity and event age. | |
| This module: | |
| - does NOT retrain models | |
| - does NOT control traffic signals | |
| - does NOT treat the risk score as a probability | |
| - does NOT substitute raw accident detector confidence for probability | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import sys | |
| from typing import Any | |
| import pandas as pd | |
| _SRC_DIR = os.path.dirname(__file__) | |
| if _SRC_DIR not in sys.path: | |
| sys.path.insert(0, _SRC_DIR) | |
| try: | |
| # When src/ is directly on PYTHONPATH | |
| import config as _config | |
| from risk_engine import assess_risk | |
| except ImportError: | |
| # Package-style import | |
| from . import config as _config | |
| from .risk_engine import assess_risk | |
| # --------------------------------------------------------------------- | |
| # Version | |
| # --------------------------------------------------------------------- | |
| ENGINE_VERSION = "safety_risk_fusion_v1" | |
| # --------------------------------------------------------------------- | |
| # Phase 6 detector output keys | |
| # --------------------------------------------------------------------- | |
| _DETECTOR_PROBABILITY_KEY = "calibrated_accident_probability" | |
| _DETECTOR_RAW_SIGNAL_KEY = "accident_detection_confidence" | |
| # --------------------------------------------------------------------- | |
| # Expected Model D features | |
| # --------------------------------------------------------------------- | |
| MODEL_D_FEATURES = [ | |
| "speed_object_1_kph", | |
| "speed_object_2_kph", | |
| "angle_degrees", | |
| "acceleration_obj1_mps2", | |
| "acceleration_obj2_mps2", | |
| "min_dist_dual_check_m", | |
| "object_count_at_frame1", | |
| "target_dist_px", | |
| "FE_log_mesafe", | |
| "FE_dist_squared", | |
| "class_object_1", | |
| "class_object_2", | |
| ] | |
| class SafetyIntelligence: | |
| """Lazy-loading fusion of near-miss Model D and accident models.""" | |
| def __init__( | |
| self, | |
| near_miss_artifact: str | None = None, | |
| accident_checkpoint: str | None = None, | |
| ): | |
| """ | |
| Parameters | |
| ---------- | |
| near_miss_artifact: | |
| Optional explicit path to the calibrated near-miss model. | |
| accident_checkpoint: | |
| Optional explicit path to the calibrated accident detector. | |
| If paths are not supplied, values are taken from config.py. | |
| """ | |
| self.near_miss_artifact = near_miss_artifact | |
| self.accident_checkpoint = accident_checkpoint | |
| # Models are loaded lazily. | |
| self._near_miss = None | |
| self._accident = None | |
| # ================================================================= | |
| # MODEL LOADING | |
| # ================================================================= | |
| def _ensure_near_miss(self): | |
| """Load the calibrated Model D artifact once.""" | |
| if self._near_miss is None: | |
| try: | |
| from near_miss.probability_pipeline import ProbabilityPipeline | |
| except ModuleNotFoundError: | |
| # Repository-root pytest imports this module as ``src``. | |
| from src.near_miss.probability_pipeline import ProbabilityPipeline | |
| path = ( | |
| self.near_miss_artifact | |
| or _config.NEAR_MISS_MODEL_PATH | |
| ) | |
| self._near_miss = ProbabilityPipeline(path) | |
| # Safety check: | |
| # Make sure the configured artifact is actually Model D. | |
| actual_features = list(self._near_miss.features) | |
| if actual_features != MODEL_D_FEATURES: | |
| raise ValueError( | |
| "The configured near-miss artifact does not match " | |
| "the deployment-realistic Model D feature schema.\n\n" | |
| f"Expected {len(MODEL_D_FEATURES)} features:\n" | |
| f"{MODEL_D_FEATURES}\n\n" | |
| f"Loaded {len(actual_features)} features:\n" | |
| f"{actual_features}\n\n" | |
| "Make sure config.py points to:\n" | |
| "models/near_miss/random_forest_D_calibrated.joblib" | |
| ) | |
| return self._near_miss | |
| def _ensure_accident(self): | |
| """Load the calibrated accident detector once.""" | |
| if self._accident is None: | |
| from accident.calibrated_detector import ( | |
| CalibratedAccidentDetector | |
| ) | |
| path = ( | |
| self.accident_checkpoint | |
| or _config.ACCIDENT_DETECTOR_PATH | |
| ) | |
| self._accident = CalibratedAccidentDetector(path) | |
| return self._accident | |
| # ================================================================= | |
| # MODEL STATUS | |
| # ================================================================= | |
| def near_miss_model_loaded(self) -> bool: | |
| """Return True if the near-miss model has been loaded.""" | |
| return self._near_miss is not None | |
| def accident_model_loaded(self) -> bool: | |
| """Return True if the accident model has been loaded.""" | |
| return self._accident is not None | |
| # ================================================================= | |
| # NEAR-MISS PROBABILITY | |
| # ================================================================= | |
| def _near_miss_probability( | |
| self, | |
| features: pd.DataFrame | dict[str, Any] | None, | |
| ) -> float | None: | |
| """ | |
| Calculate calibrated Model D class-1 probability. | |
| Parameters | |
| ---------- | |
| features: | |
| Either: | |
| - pandas DataFrame containing Model D features | |
| - dictionary containing one Model D feature vector | |
| - None | |
| Returns | |
| ------- | |
| float | None | |
| Calibrated class-1 probability. | |
| """ | |
| if features is None: | |
| return None | |
| pipe = self._ensure_near_miss() | |
| # Convert dictionary into one-row DataFrame. | |
| if isinstance(features, pd.DataFrame): | |
| frame = features.copy() | |
| elif isinstance(features, dict): | |
| frame = pd.DataFrame([features]) | |
| else: | |
| raise TypeError( | |
| "near_miss_features must be a pandas DataFrame, " | |
| "dictionary, or None." | |
| ) | |
| # ------------------------------------------------------------- | |
| # Validate required Model D features | |
| # ------------------------------------------------------------- | |
| missing = [ | |
| feature | |
| for feature in pipe.features | |
| if feature not in frame.columns | |
| ] | |
| if missing: | |
| raise ValueError( | |
| "Input missing required Model D columns: " | |
| f"{missing}" | |
| ) | |
| # ------------------------------------------------------------- | |
| # Select ONLY model features | |
| # ------------------------------------------------------------- | |
| model_frame = frame[pipe.features] | |
| # ------------------------------------------------------------- | |
| # Run calibrated probability model | |
| # ------------------------------------------------------------- | |
| probabilities = pipe.predict_proba(model_frame) | |
| probability = float(probabilities[0, 1]) | |
| # Probability should always be in [0, 1]. | |
| if not 0.0 <= probability <= 1.0: | |
| raise ValueError( | |
| f"Near-miss model returned invalid probability: " | |
| f"{probability}" | |
| ) | |
| return round(probability, 6) | |
| # ================================================================= | |
| # ACCIDENT PROBABILITY | |
| # ================================================================= | |
| def _accident_probability( | |
| self, | |
| accident_detector_output: dict[str, Any] | None, | |
| accident_probability: float | None, | |
| ) -> tuple[float | None, Any]: | |
| """ | |
| Resolve calibrated accident probability. | |
| Explicit accident_probability has priority. | |
| If detector output is supplied, only: | |
| calibrated_accident_probability | |
| is treated as a probability. | |
| The raw: | |
| accident_detection_confidence | |
| value is preserved separately and NEVER treated as a probability. | |
| """ | |
| raw_signal = None | |
| # ------------------------------------------------------------- | |
| # Explicit probability supplied by caller | |
| # ------------------------------------------------------------- | |
| if accident_probability is not None: | |
| resolved = float(accident_probability) | |
| # ------------------------------------------------------------- | |
| # Probability supplied through detector output | |
| # ------------------------------------------------------------- | |
| elif accident_detector_output is not None: | |
| if _DETECTOR_PROBABILITY_KEY not in accident_detector_output: | |
| raise ValueError( | |
| "accident_detector_output is missing " | |
| f"{_DETECTOR_PROBABILITY_KEY!r}.\n" | |
| "The raw detector confidence is NOT a probability " | |
| "and cannot be substituted." | |
| ) | |
| resolved = float( | |
| accident_detector_output[ | |
| _DETECTOR_PROBABILITY_KEY | |
| ] | |
| ) | |
| raw_signal = accident_detector_output.get( | |
| _DETECTOR_RAW_SIGNAL_KEY | |
| ) | |
| # ------------------------------------------------------------- | |
| # No accident evidence supplied | |
| # ------------------------------------------------------------- | |
| else: | |
| resolved = None | |
| # ------------------------------------------------------------- | |
| # Validate probability | |
| # ------------------------------------------------------------- | |
| if resolved is not None: | |
| if not 0.0 <= resolved <= 1.0: | |
| raise ValueError( | |
| "Accident probability must be between 0 and 1. " | |
| f"Received: {resolved}" | |
| ) | |
| return resolved, raw_signal | |
| # ================================================================= | |
| # PUBLIC EVENT API | |
| # ================================================================= | |
| def assess_event( | |
| self, | |
| near_miss_features: ( | |
| pd.DataFrame | |
| | dict[str, Any] | |
| | None | |
| ) = None, | |
| accident_detector_output: dict[str, Any] | None = None, | |
| accident_probability: float | None = None, | |
| severity: float | None = None, | |
| event_age_hours: float | None = None, | |
| risk_config: dict[str, Any] | None = None, | |
| ) -> dict[str, Any]: | |
| """ | |
| Fuse near-miss and accident evidence into a safety assessment. | |
| Parameters | |
| ---------- | |
| near_miss_features: | |
| Model D feature vector. | |
| accident_detector_output: | |
| Output dictionary from Phase 6 accident detector. | |
| accident_probability: | |
| Explicit calibrated accident probability. | |
| severity: | |
| Caller-supplied event severity in [0, 1]. | |
| event_age_hours: | |
| Age of the event in hours. | |
| risk_config: | |
| Optional risk-engine configuration. | |
| Returns | |
| ------- | |
| dict | |
| JSON-serializable safety assessment. | |
| """ | |
| # ------------------------------------------------------------- | |
| # Near-miss model | |
| # ------------------------------------------------------------- | |
| near_miss_probability = self._near_miss_probability( | |
| near_miss_features | |
| ) | |
| # ------------------------------------------------------------- | |
| # Accident model | |
| # ------------------------------------------------------------- | |
| resolved_accident, raw_signal = self._accident_probability( | |
| accident_detector_output, | |
| accident_probability, | |
| ) | |
| # ------------------------------------------------------------- | |
| # Risk fusion | |
| # ------------------------------------------------------------- | |
| assessment = assess_risk( | |
| near_miss_probability=near_miss_probability, | |
| accident_probability=resolved_accident, | |
| severity=severity, | |
| event_age_hours=event_age_hours, | |
| config=risk_config, | |
| ) | |
| # ------------------------------------------------------------- | |
| # Build final output | |
| # ------------------------------------------------------------- | |
| output: dict[str, Any] = { | |
| "engine_version": ENGINE_VERSION, | |
| "near_miss_probability": ( | |
| assessment["near_miss_probability"] | |
| ), | |
| "accident_probability": ( | |
| assessment["accident_probability"] | |
| ), | |
| "risk_score": assessment["risk_score"], | |
| "risk_level": assessment["risk_level"], | |
| "severity": assessment["severity"], | |
| "event_age_hours": assessment["event_age_hours"], | |
| "explanation": list( | |
| assessment["explanation"] | |
| ), | |
| "present_inputs": assessment["present_inputs"], | |
| "missing_inputs": assessment["missing_inputs"], | |
| "model_provenance": { | |
| "near_miss_artifact": str( | |
| self.near_miss_artifact | |
| or _config.NEAR_MISS_MODEL_PATH | |
| ), | |
| "near_miss_model_loaded": ( | |
| self.near_miss_model_loaded | |
| ), | |
| "near_miss_feature_configuration": ( | |
| "D_deployment_realistic" | |
| ), | |
| "near_miss_features": MODEL_D_FEATURES, | |
| "accident_checkpoint": str( | |
| self.accident_checkpoint | |
| or _config.ACCIDENT_DETECTOR_PATH | |
| ), | |
| "accident_model_loaded": ( | |
| self.accident_model_loaded | |
| ), | |
| "retrained": False, | |
| }, | |
| } | |
| # ------------------------------------------------------------- | |
| # Preserve raw accident detector confidence separately | |
| # ------------------------------------------------------------- | |
| if raw_signal is not None: | |
| output["detector_raw_signal"] = { | |
| "accident_detection_confidence": raw_signal, | |
| "note": ( | |
| "Raw detector confidence is a ranking signal " | |
| "aggregated from accident-class box scores. " | |
| "It is NOT a probability. The accident probability " | |
| "is the calibrated 'accident_probability' value." | |
| ), | |
| } | |
| return output | |
| # ================================================================= | |
| # IMAGE-BASED ACCIDENT + SAFETY ASSESSMENT | |
| # ================================================================= | |
| def assess_image( | |
| self, | |
| image: Any, | |
| near_miss_features: ( | |
| pd.DataFrame | |
| | dict[str, Any] | |
| | None | |
| ) = None, | |
| severity: float | None = None, | |
| event_age_hours: float | None = None, | |
| score_threshold: float | None = None, | |
| risk_config: dict[str, Any] | None = None, | |
| ) -> dict[str, Any]: | |
| """ | |
| Run the Phase 6 accident detector on an image and fuse | |
| the result with near-miss evidence. | |
| """ | |
| # ------------------------------------------------------------- | |
| # Load accident detector | |
| # ------------------------------------------------------------- | |
| detector = self._ensure_accident() | |
| # ------------------------------------------------------------- | |
| # Run detector | |
| # ------------------------------------------------------------- | |
| detector_output = detector.predict( | |
| image, | |
| score_threshold=score_threshold, | |
| ) | |
| # ------------------------------------------------------------- | |
| # Fuse safety information | |
| # ------------------------------------------------------------- | |
| output = self.assess_event( | |
| near_miss_features=near_miss_features, | |
| accident_detector_output=detector_output, | |
| severity=severity, | |
| event_age_hours=event_age_hours, | |
| risk_config=risk_config, | |
| ) | |
| # ------------------------------------------------------------- | |
| # Add detector information | |
| # ------------------------------------------------------------- | |
| output["detector_output"] = { | |
| "detector_version": detector_output.get( | |
| "detector_version" | |
| ), | |
| "image_size": detector_output.get( | |
| "image_size" | |
| ), | |
| "n_detections": len( | |
| detector_output.get( | |
| "detections", | |
| [] | |
| ) | |
| ), | |
| "detections": detector_output.get( | |
| "detections", | |
| [] | |
| ), | |
| "calibration": detector_output.get( | |
| "calibration" | |
| ), | |
| } | |
| return output | |
| # ===================================================================== | |
| # JSON SERIALIZATION | |
| # ===================================================================== | |
| def to_json( | |
| output: dict[str, Any], | |
| indent: int | None = 2, | |
| ) -> str: | |
| """ | |
| Convert safety assessment into JSON. | |
| """ | |
| return json.dumps( | |
| output, | |
| indent=indent, | |
| ) |