Spaces:
Running on Zero
Running on Zero
File size: 6,125 Bytes
a74054f | 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 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | """
Data validation utilities.
Single Responsibility: Validate data quality and consistency.
"""
import pandas as pd
import numpy as np
from typing import List, Dict, Optional
from datetime import datetime
class DataValidator:
"""
Validates hydrological and meteorological datasets.
Checks for data quality issues, temporal consistency, and physical constraints.
"""
def __init__(self):
"""Initialize data validator."""
self.validation_results = {}
def validate(
self,
df: pd.DataFrame,
date_col: str = "date",
station_col: str = "station_id",
required_cols: Optional[List[str]] = None
) -> Dict[str, bool]:
"""
Run all validation checks.
Args:
df: DataFrame to validate
date_col: Name of date column
station_col: Name of station ID column
required_cols: List of required columns
Returns:
Dictionary with validation results
"""
results = {}
# Schema validation
results["has_required_columns"] = self._validate_schema(df, required_cols)
# Date validation
if date_col in df.columns:
results["dates_valid"] = self._validate_dates(df, date_col)
results["no_temporal_gaps"] = self._check_temporal_continuity(
df, date_col, station_col
)
# Data range validation (physical constraints)
results["values_in_valid_range"] = self._validate_ranges(df)
# Consistency checks
if station_col in df.columns:
results["no_duplicate_station_dates"] = self._check_duplicates(
df, date_col, station_col
)
self.validation_results = results
return results
def _validate_schema(
self,
df: pd.DataFrame,
required_cols: Optional[List[str]]
) -> bool:
"""Check if required columns exist."""
if required_cols is None:
return True
return all(col in df.columns for col in required_cols)
def _validate_dates(self, df: pd.DataFrame, date_col: str) -> bool:
"""Validate date column format and range."""
try:
dates = pd.to_datetime(df[date_col])
# Check if dates are in reasonable range (1900-2030)
min_date = pd.Timestamp("1900-01-01")
max_date = pd.Timestamp("2030-12-31")
return (dates >= min_date).all() and (dates <= max_date).all()
except Exception:
return False
def _check_temporal_continuity(
self,
df: pd.DataFrame,
date_col: str,
station_col: str,
max_gap_days: int = 7
) -> bool:
"""
Check for large temporal gaps in time series.
Args:
df: DataFrame
date_col: Date column name
station_col: Station column name
max_gap_days: Maximum allowed gap in days
Returns:
True if no large gaps exist
"""
if station_col not in df.columns:
return True
df_sorted = df.sort_values([station_col, date_col])
df_sorted[date_col] = pd.to_datetime(df_sorted[date_col])
# Check gaps per station
for station in df_sorted[station_col].unique():
station_data = df_sorted[df_sorted[station_col] == station]
dates = station_data[date_col]
if len(dates) > 1:
gaps = dates.diff().dt.days.dropna()
if (gaps > max_gap_days).any():
return False
return True
def _validate_ranges(self, df: pd.DataFrame) -> bool:
"""
Validate that values are within physically meaningful ranges.
Hydrological constraints.
"""
# Physical constraints for hydrological variables
constraints = {
# Discharge (m³/s): must be >= 0
"QmnJ": (0, None),
"QIXnJ": (0, None),
"QmM": (0, None),
"QIXM": (0, None),
# Water level (m): must be >= 0
"HIXnJ": (0, None),
"HIXM": (0, None),
# Temperature (°C): reasonable range
"T_Q": (-40, 50),
# Humidity (%): 0-100
"HU_Q": (0, 100),
# Precipitation (mm): >= 0
"PRELIQ_Q": (0, None),
"PRENEI_Q": (0, None),
# Radiation (W/m²): >= 0
"DLI_Q": (0, None),
"SSI_Q": (0, None),
# ETP (mm): >= 0
"ETP_Q": (0, None),
# Wind speed (m/s): >= 0
"FF_Q": (0, None)
}
for col, (min_val, max_val) in constraints.items():
if col in df.columns:
values = df[col].dropna()
if min_val is not None and (values < min_val).any():
return False
if max_val is not None and (values > max_val).any():
return False
return True
def _check_duplicates(
self,
df: pd.DataFrame,
date_col: str,
station_col: str
) -> bool:
"""Check for duplicate station-date combinations."""
if date_col in df.columns and station_col in df.columns:
return not df.duplicated(subset=[date_col, station_col]).any()
return True
def get_validation_summary(self) -> str:
"""Get human-readable validation summary."""
if not self.validation_results:
return "No validation performed yet"
passed = sum(self.validation_results.values())
total = len(self.validation_results)
summary = f"Validation: {passed}/{total} checks passed\n"
for check, result in self.validation_results.items():
status = "✓" if result else "✗"
summary += f" {status} {check}\n"
return summary
|