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!
107 lines
3.2 KiB
Python
107 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Data models for water monitoring system
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import Optional, List, Dict, Any
|
|
from enum import Enum
|
|
|
|
class DatabaseType(Enum):
|
|
SQLITE = "sqlite"
|
|
MYSQL = "mysql"
|
|
POSTGRESQL = "postgresql"
|
|
INFLUXDB = "influxdb"
|
|
VICTORIAMETRICS = "victoriametrics"
|
|
|
|
class StationStatus(Enum):
|
|
ACTIVE = "active"
|
|
INACTIVE = "inactive"
|
|
MAINTENANCE = "maintenance"
|
|
ERROR = "error"
|
|
|
|
@dataclass
|
|
class StationInfo:
|
|
"""Station information model"""
|
|
station_id: int
|
|
station_code: str
|
|
thai_name: str
|
|
english_name: str
|
|
latitude: Optional[float] = None
|
|
longitude: Optional[float] = None
|
|
geohash: Optional[str] = None
|
|
status: StationStatus = StationStatus.ACTIVE
|
|
|
|
@dataclass
|
|
class WaterMeasurement:
|
|
"""Water measurement data model"""
|
|
timestamp: datetime
|
|
station_info: StationInfo
|
|
water_level: float
|
|
discharge: float
|
|
water_level_unit: str = "m"
|
|
discharge_unit: str = "cms"
|
|
discharge_percent: Optional[float] = None
|
|
status: StationStatus = StationStatus.ACTIVE
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert to dictionary for database storage"""
|
|
return {
|
|
'timestamp': self.timestamp,
|
|
'station_id': self.station_info.station_id,
|
|
'station_code': self.station_info.station_code,
|
|
'station_name_en': self.station_info.english_name,
|
|
'station_name_th': self.station_info.thai_name,
|
|
'latitude': self.station_info.latitude,
|
|
'longitude': self.station_info.longitude,
|
|
'geohash': self.station_info.geohash,
|
|
'water_level': self.water_level,
|
|
'water_level_unit': self.water_level_unit,
|
|
'discharge': self.discharge,
|
|
'discharge_unit': self.discharge_unit,
|
|
'discharge_percent': self.discharge_percent,
|
|
'status': self.status.value
|
|
}
|
|
|
|
@dataclass
|
|
class DatabaseConfig:
|
|
"""Database configuration model"""
|
|
db_type: DatabaseType
|
|
connection_string: Optional[str] = None
|
|
host: Optional[str] = None
|
|
port: Optional[int] = None
|
|
database: Optional[str] = None
|
|
username: Optional[str] = None
|
|
password: Optional[str] = None
|
|
additional_params: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
@dataclass
|
|
class ScrapingResult:
|
|
"""Result of a scraping operation"""
|
|
success: bool
|
|
measurements_count: int
|
|
error_message: Optional[str] = None
|
|
timestamp: datetime = field(default_factory=datetime.now)
|
|
processing_time_seconds: Optional[float] = None
|
|
|
|
@dataclass
|
|
class StationCreateRequest:
|
|
"""Request model for creating a new station"""
|
|
station_code: str
|
|
thai_name: str
|
|
english_name: str
|
|
latitude: Optional[float] = None
|
|
longitude: Optional[float] = None
|
|
geohash: Optional[str] = None
|
|
status: StationStatus = StationStatus.ACTIVE
|
|
|
|
@dataclass
|
|
class StationUpdateRequest:
|
|
"""Request model for updating an existing station"""
|
|
thai_name: Optional[str] = None
|
|
english_name: Optional[str] = None
|
|
latitude: Optional[float] = None
|
|
longitude: Optional[float] = None
|
|
geohash: Optional[str] = None
|
|
status: Optional[StationStatus] = None |