Spaces:
Paused
Paused
File size: 3,577 Bytes
729628a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | """
Accident Analysis Engine Layer
Performs temporal pattern detection, geospatial hotspot clustering, demographic reviews, and caches results.
"""
import logging
from typing import Dict, Any, List, Optional
import pandas as pd
from crash_analyzer import CrashDataAnalyzer
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class MetadataAnalyzer:
"""Performs schema, time, and severity breakdown analysis on accident records."""
def __init__(self, analyzer: CrashDataAnalyzer):
self.analyzer = analyzer
def analyze(self) -> Dict[str, Any]:
"""Runs temporal and severity breakdown metrics."""
temporal = self.analyzer.analyze_temporal_patterns()
severity = self.analyzer.analyze_severity()
collision = self.analyzer.analyze_collision_types()
return {
"temporal_patterns": temporal,
"severity_distribution": severity,
"collision_types": collision
}
class GeospatialAnalyzer:
"""Performs geographic analysis and hotspot identification."""
def __init__(self, analyzer: CrashDataAnalyzer):
self.analyzer = analyzer
def analyze(self, grid_size: float = 0.01) -> Dict[str, Any]:
"""Clusters coordinates to locate hotspot sectors."""
summary = self.analyzer.get_geographic_summary()
hotspots = self.analyzer.identify_hotspots(grid_size=grid_size)
return {
"geographic_summary": summary,
"hotspots": hotspots
}
class DemographicsAnalyzer:
"""Performs party, vehicle type, and traveler demographic analysis."""
def __init__(self, analyzer: CrashDataAnalyzer):
self.analyzer = analyzer
def analyze(self) -> Dict[str, Any]:
"""Parses driver profiles and vehicle risk attributes."""
vehicles = self.analyzer.analyze_vehicle_types()
demographics = self.analyzer.analyze_demographics()
return {
"vehicle_risk": vehicles,
"demographics": demographics
}
class AccidentAnalysisEngine:
"""Orchestrates temporal, geospatial, and demographic analysis while caching results."""
def __init__(self, accidents_df: pd.DataFrame, vehicles_df: Optional[pd.DataFrame] = None,
parties_df: Optional[pd.DataFrame] = None):
self.analyzer = CrashDataAnalyzer(accidents_df, vehicles_df, parties_df)
self.metadata_analyzer = MetadataAnalyzer(self.analyzer)
self.geospatial_analyzer = GeospatialAnalyzer(self.analyzer)
self.demographics_analyzer = DemographicsAnalyzer(self.analyzer)
self._cache: Dict[str, Any] = {}
def run_full_pipeline(self, bypass_cache: bool = False) -> Dict[str, Any]:
"""
Executes all sub-analyzers and returns cached or fresh analysis results.
Args:
bypass_cache: Force fresh computation.
Returns:
Dict containing full analysis indicators.
"""
if "full_metrics" in self._cache and not bypass_cache:
logger.info("Returning cached analysis results.")
return self._cache["full_metrics"]
logger.info("Computing fresh analysis metrics...")
meta = self.metadata_analyzer.analyze()
geo = self.geospatial_analyzer.analyze()
demo = self.demographics_analyzer.analyze()
results = {
"metadata": meta,
"geospatial": geo,
"demographics": demo
}
self._cache["full_metrics"] = results
return results
|