Spaces:
Paused
Paused
File size: 3,513 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 | """
Accident Data Validation Layer
Validates required columns, data types, coordinates, date ranges, and calculates a data quality score.
"""
import logging
from typing import Dict, Any, List
import pandas as pd
from data_extractor import DataValidator, ValidationReport
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AccidentDataValidator:
"""Validates accident dataframes for schema compliance and geographic/temporal correctness."""
def __init__(self, dataframe: pd.DataFrame):
self.df = dataframe.copy()
self.validator = DataValidator(self.df)
def validate_schema(self) -> ValidationReport:
"""
Runs standard validation checks on columns, types, coordinates, and nulls.
Returns:
ValidationReport outlining validity state, errors, and warnings.
"""
logger.info("Starting accident schema validation...")
# 1. Required Columns
required = ["case_id", "accident_date", "severity", "latitude", "longitude"]
self.validator.validate_required_columns(required)
# 2. Check no null values in primary keys
self.validator.validate_no_nulls(["case_id"])
# 3. Data Types Check
type_map = {
"case_id": "object",
"latitude": "float64",
"longitude": "float64"
}
self.validator.validate_data_types(type_map)
# 4. Check Coordinate Ranges
if "latitude" in self.df.columns and "longitude" in self.df.columns:
# Valid US lat/lon boundaries approx
lat_invalid = self.df[(self.df["latitude"] < 24.0) | (self.df["latitude"] > 49.0)]
lon_invalid = self.df[(self.df["longitude"] < -125.0) | (self.df["longitude"] > -66.0)]
if not lat_invalid.empty:
self.validator.errors.append(f"Latitude out of valid US range (24-49): {len(lat_invalid)} violations")
if not lon_invalid.empty:
self.validator.errors.append(f"Longitude out of valid US range (-125 to -66): {len(lon_invalid)} violations")
# 5. Check Date Ranges
if "accident_date" in self.df.columns:
try:
parsed_dates = pd.to_datetime(self.df["accident_date"], errors="coerce")
null_dates = parsed_dates.isna().sum()
if null_dates > 0:
self.validator.errors.append(f"accident_date: {null_dates} rows contain unparseable dates")
# Check for future dates
future_dates = (parsed_dates > pd.Timestamp.now()).sum()
if future_dates > 0:
self.validator.errors.append(f"accident_date: {future_dates} rows contain future dates")
except Exception as e:
self.validator.errors.append(f"Failed to validate date column: {str(e)}")
# 6. Calculate Quality Score
report = self.validator.get_report()
# Compute dynamic quality score (percentage of successful validation elements)
total_checks = 6
failed_checks = len(report.errors)
quality_score = max(0.0, min(1.0, (total_checks - failed_checks) / total_checks))
# Inject quality score attribute
report.quality_score = quality_score
logger.info(f"Validation complete. Valid: {report.valid}, Quality Score: {quality_score * 100:.1f}%")
return report
|