grabowski
af62cfef0b
Security & Dependency Updates / Dependency Security Scan (push) Successful in 29s
Security & Dependency Updates / Docker Security Scan (push) Failing after 53s
Security & Dependency Updates / License Compliance (push) Successful in 13s
Security & Dependency Updates / Check for Dependency Updates (push) Successful in 19s
Security & Dependency Updates / Code Quality Metrics (push) Successful in 11s
Security & Dependency Updates / Security Summary (push) Successful in 7s
Features: - Real-time water level monitoring for Ping River Basin (16 stations) - Coverage from Chiang Dao to Nakhon Sawan in Northern Thailand - FastAPI web interface with interactive dashboard and station management - Multi-database support (SQLite, MySQL, PostgreSQL, InfluxDB, VictoriaMetrics) - Comprehensive monitoring with health checks and metrics collection - Docker deployment with Grafana integration - Production-ready architecture with enterprise-grade observability CI/CD & Automation: - Complete Gitea Actions workflows for CI/CD, security, and releases - Multi-Python version testing (3.9-3.12) - Multi-architecture Docker builds (amd64, arm64) - Daily security scanning and dependency monitoring - Automated documentation generation - Performance testing and validation Production Ready: - Type safety with Pydantic models and comprehensive type hints - Data validation layer with range checking and error handling - Rate limiting and request tracking for API protection - Enhanced logging with rotation, colors, and performance metrics - Station management API for dynamic CRUD operations - Comprehensive documentation and deployment guides Technical Stack: - Python 3.9+ with FastAPI and Pydantic - Multi-database architecture with adapter pattern - Docker containerization with multi-stage builds - Grafana dashboards for visualization - Gitea Actions for CI/CD automation - Enterprise monitoring and alerting Ready for deployment to B4L infrastructure!
116 lines
4.5 KiB
Python
116 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Data validation utilities for water monitoring system
|
|
"""
|
|
|
|
from typing import List, Dict, Any, Optional
|
|
from datetime import datetime
|
|
import logging
|
|
from .exceptions import DataValidationError
|
|
from .models import WaterMeasurement, StationInfo
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class DataValidator:
|
|
"""Validates water measurement data"""
|
|
|
|
# Reasonable ranges for water measurements
|
|
WATER_LEVEL_MIN = -10.0 # meters
|
|
WATER_LEVEL_MAX = 50.0 # meters
|
|
DISCHARGE_MIN = 0.0 # cms
|
|
DISCHARGE_MAX = 10000.0 # cms
|
|
DISCHARGE_PERCENT_MIN = 0.0
|
|
DISCHARGE_PERCENT_MAX = 200.0 # Allow some overflow
|
|
|
|
@classmethod
|
|
def validate_measurement(cls, measurement: Dict[str, Any]) -> bool:
|
|
"""Validate a single measurement"""
|
|
try:
|
|
# Check required fields
|
|
required_fields = ['timestamp', 'station_id', 'water_level', 'discharge']
|
|
for field in required_fields:
|
|
if field not in measurement:
|
|
logger.warning(f"Missing required field: {field}")
|
|
return False
|
|
|
|
# Validate timestamp
|
|
if not isinstance(measurement['timestamp'], datetime):
|
|
logger.warning(f"Invalid timestamp type: {type(measurement['timestamp'])}")
|
|
return False
|
|
|
|
# Validate water level
|
|
water_level = float(measurement['water_level'])
|
|
if not (cls.WATER_LEVEL_MIN <= water_level <= cls.WATER_LEVEL_MAX):
|
|
logger.warning(f"Water level out of range: {water_level}")
|
|
return False
|
|
|
|
# Validate discharge
|
|
discharge = float(measurement['discharge'])
|
|
if not (cls.DISCHARGE_MIN <= discharge <= cls.DISCHARGE_MAX):
|
|
logger.warning(f"Discharge out of range: {discharge}")
|
|
return False
|
|
|
|
# Validate discharge percent if present
|
|
if measurement.get('discharge_percent') is not None:
|
|
discharge_percent = float(measurement['discharge_percent'])
|
|
if not (cls.DISCHARGE_PERCENT_MIN <= discharge_percent <= cls.DISCHARGE_PERCENT_MAX):
|
|
logger.warning(f"Discharge percent out of range: {discharge_percent}")
|
|
return False
|
|
|
|
# Validate station ID
|
|
station_id = measurement['station_id']
|
|
if not isinstance(station_id, int) or station_id < 1 or station_id > 16:
|
|
logger.warning(f"Invalid station ID: {station_id}")
|
|
return False
|
|
|
|
return True
|
|
|
|
except (ValueError, TypeError) as e:
|
|
logger.warning(f"Data validation error: {e}")
|
|
return False
|
|
|
|
@classmethod
|
|
def validate_measurements(cls, measurements: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
"""Validate and filter a list of measurements"""
|
|
valid_measurements = []
|
|
invalid_count = 0
|
|
|
|
for measurement in measurements:
|
|
if cls.validate_measurement(measurement):
|
|
valid_measurements.append(measurement)
|
|
else:
|
|
invalid_count += 1
|
|
|
|
if invalid_count > 0:
|
|
logger.warning(f"Filtered out {invalid_count} invalid measurements")
|
|
|
|
return valid_measurements
|
|
|
|
@classmethod
|
|
def validate_station_info(cls, station_info: Dict[str, Any]) -> bool:
|
|
"""Validate station information"""
|
|
try:
|
|
required_fields = ['station_id', 'station_code', 'thai_name', 'english_name']
|
|
for field in required_fields:
|
|
if field not in station_info or not station_info[field]:
|
|
logger.warning(f"Missing or empty station field: {field}")
|
|
return False
|
|
|
|
# Validate coordinates if present
|
|
if station_info.get('latitude') is not None:
|
|
lat = float(station_info['latitude'])
|
|
if not (-90 <= lat <= 90):
|
|
logger.warning(f"Invalid latitude: {lat}")
|
|
return False
|
|
|
|
if station_info.get('longitude') is not None:
|
|
lon = float(station_info['longitude'])
|
|
if not (-180 <= lon <= 180):
|
|
logger.warning(f"Invalid longitude: {lon}")
|
|
return False
|
|
|
|
return True
|
|
|
|
except (ValueError, TypeError) as e:
|
|
logger.warning(f"Station validation error: {e}")
|
|
return False |