grabowski
6eafb353b1
POST app.rid.go.th/reservoir/api/dams (open, archive >=2009) collected hourly into rid_dams + rid_reservoir_daily; backfill script fetches only missing days so reruns repair holes and are safe alongside the live collector. /api/stats counts the new table via an engine fallback that works when HII collection is disabled. Mae Ngat (DAM_ID 200103) hit 113% usable capacity with ~19 MCM/day inflow in the Oct 2024 flood — candidate features for the next retrain.
300 lines
12 KiB
Python
300 lines
12 KiB
Python
import os
|
|
from typing import Any, Dict
|
|
|
|
# Load environment variables from .env file
|
|
try:
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
except ImportError:
|
|
# python-dotenv not installed, continue without it
|
|
pass
|
|
|
|
try:
|
|
from .exceptions import ConfigurationError
|
|
from .models import DatabaseType
|
|
except ImportError:
|
|
# Handle case when running as standalone script
|
|
class ConfigurationError(Exception):
|
|
pass
|
|
|
|
from enum import Enum
|
|
|
|
class DatabaseType(Enum):
|
|
SQLITE = "sqlite"
|
|
MYSQL = "mysql"
|
|
POSTGRESQL = "postgresql"
|
|
INFLUXDB = "influxdb"
|
|
VICTORIAMETRICS = "victoriametrics"
|
|
|
|
|
|
class Config:
|
|
"""Configuration class for the Water Level Monitor"""
|
|
|
|
# Database settings
|
|
DATABASE_PATH = os.getenv("WATER_DB_PATH", "water_levels.db")
|
|
|
|
# Website settings
|
|
TARGET_URL = "https://hyd-app-db.rid.go.th/hydro1h.html"
|
|
API_URL = "https://hyd-app-db.rid.go.th/webservice/getGroupHourlyWaterLevelReportAllHL.ashx"
|
|
THAIWATER_API_KEY = os.getenv("THAIWATER_API_KEY")
|
|
REQUEST_TIMEOUT = int(os.getenv("REQUEST_TIMEOUT", "30"))
|
|
USER_AGENT = (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
|
|
)
|
|
|
|
# Database configuration
|
|
# When DB_TYPE is not set explicitly, a configured Postgres connection wins over the sqlite default
|
|
DB_TYPE = os.getenv(
|
|
"DB_TYPE",
|
|
"postgresql" if os.getenv("POSTGRES_CONNECTION_STRING") else "sqlite",
|
|
).lower()
|
|
|
|
# VictoriaMetrics settings
|
|
# Default to localhost; set VM_HOST in the environment for real deployments
|
|
# (avoids committing infrastructure hostnames to the repo).
|
|
VM_HOST = os.getenv("VM_HOST", "localhost")
|
|
VM_PORT = int(os.getenv("VM_PORT", "443"))
|
|
|
|
# Support for HTTPS URLs (e.g., behind reverse proxy)
|
|
VM_URL = os.getenv("VM_URL") # Full URL override (e.g., https://vm.example.com)
|
|
|
|
# InfluxDB settings
|
|
INFLUX_HOST = os.getenv("INFLUX_HOST", "localhost")
|
|
INFLUX_PORT = int(os.getenv("INFLUX_PORT", "8086"))
|
|
INFLUX_DATABASE = os.getenv("INFLUX_DATABASE", "water_monitoring")
|
|
INFLUX_USERNAME = os.getenv("INFLUX_USERNAME")
|
|
INFLUX_PASSWORD = os.getenv("INFLUX_PASSWORD")
|
|
|
|
# PostgreSQL settings
|
|
POSTGRES_CONNECTION_STRING = os.getenv("POSTGRES_CONNECTION_STRING")
|
|
POSTGRES_HOST = os.getenv("POSTGRES_HOST", "localhost")
|
|
POSTGRES_PORT = int(os.getenv("POSTGRES_PORT", "5432"))
|
|
POSTGRES_DB = os.getenv("POSTGRES_DB", "water_monitoring")
|
|
POSTGRES_USER = os.getenv("POSTGRES_USER", "postgres")
|
|
POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD")
|
|
|
|
# MySQL settings
|
|
MYSQL_CONNECTION_STRING = os.getenv("MYSQL_CONNECTION_STRING")
|
|
|
|
# HII/ThaiWater open api-v3 collection (rainfall + backup water level)
|
|
# See docs/DATA_SOURCES.md. Requires a SQL DB_TYPE (sqlite/postgresql/mysql).
|
|
ENABLE_HII_COLLECTION = os.getenv("ENABLE_HII_COLLECTION", "true").lower() in (
|
|
"1",
|
|
"true",
|
|
"yes",
|
|
)
|
|
HII_BASIN_CODE = int(os.getenv("HII_BASIN_CODE", "6")) # 6 = Ping Basin
|
|
|
|
# RID large-dam daily status (app.rid.go.th/reservoir) — Mae Ngat et al.
|
|
ENABLE_RESERVOIR_COLLECTION = os.getenv(
|
|
"ENABLE_RESERVOIR_COLLECTION", "true"
|
|
).lower() in ("1", "true", "yes")
|
|
# TTL for the /api/hii/*/latest response cache; source data changes hourly
|
|
HII_CACHE_TTL_SECONDS = int(os.getenv("HII_CACHE_TTL_SECONDS", "120"))
|
|
# TTL for the /measurements/latest response cache (hottest endpoint)
|
|
LATEST_CACHE_TTL_SECONDS = int(os.getenv("LATEST_CACHE_TTL_SECONDS", "45"))
|
|
|
|
# TTL for /health check results (includes an external RID-API probe)
|
|
HEALTH_CACHE_TTL_SECONDS = int(os.getenv("HEALTH_CACHE_TTL_SECONDS", "30"))
|
|
|
|
# Thread-pool size for blocking work in the web process (DB queries,
|
|
# inference, health probes). Waiting threads are cheap; starving the pool
|
|
# stalls every endpoint that needs a thread.
|
|
EXECUTOR_THREADS = int(os.getenv("EXECUTOR_THREADS", "48"))
|
|
|
|
# Web server worker processes. Above 1, uvicorn forks workers and a
|
|
# localhost lock port elects a single background-collection leader.
|
|
WEB_WORKERS = int(os.getenv("WEB_WORKERS", "2"))
|
|
COLLECTION_LEADER_PORT = int(os.getenv("COLLECTION_LEADER_PORT", "8901"))
|
|
|
|
# Umami analytics (self-hosted). The website id is public (it ships in the
|
|
# dashboard <script> tag); server-side API tracking posts to /api/send.
|
|
UMAMI_API_URL = os.getenv("UMAMI_API_URL", "https://stats.buildfor.life/api/send")
|
|
UMAMI_WEBSITE_ID = os.getenv(
|
|
"UMAMI_WEBSITE_ID", "00b2be73-8f5f-4400-9029-3be852eb08f7"
|
|
)
|
|
UMAMI_TRACK_API = os.getenv("UMAMI_TRACK_API", "true").lower() in (
|
|
"1",
|
|
"true",
|
|
"yes",
|
|
)
|
|
|
|
# Scheduler settings
|
|
SCRAPING_INTERVAL_HOURS = int(os.getenv("SCRAPING_INTERVAL_HOURS", "1"))
|
|
|
|
# Logging settings
|
|
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
|
|
LOG_FILE = os.getenv("LOG_FILE", "water_monitor.log")
|
|
LOG_FORMAT = "%(asctime)s - %(levelname)s - %(message)s"
|
|
|
|
# Data retention
|
|
DATA_RETENTION_DAYS = int(os.getenv("DATA_RETENTION_DAYS", "365"))
|
|
|
|
# Retry settings
|
|
MAX_RETRIES = int(os.getenv("MAX_RETRIES", "3"))
|
|
RETRY_DELAY_SECONDS = int(os.getenv("RETRY_DELAY_SECONDS", "60"))
|
|
|
|
# Station configuration
|
|
# Runtime-writable JSON file that persists the station mapping across restarts
|
|
# (station CRUD via the API writes here). If it does not exist, the bundled
|
|
# defaults in src/data/stations.json are used to seed it.
|
|
STATION_CONFIG_PATH = os.getenv("STATION_CONFIG_PATH", "stations.json")
|
|
|
|
# Web API / CORS settings
|
|
# Comma-separated list of allowed origins. Defaults to none (same-origin only);
|
|
# set CORS_ALLOW_ORIGINS to a specific list of front-end origins in production.
|
|
# Credentials are only enabled when explicit (non-wildcard) origins are set,
|
|
# because "*" + credentials is rejected by browsers and unsafe.
|
|
CORS_ALLOW_ORIGINS = [
|
|
origin.strip()
|
|
for origin in os.getenv("CORS_ALLOW_ORIGINS", "").split(",")
|
|
if origin.strip()
|
|
]
|
|
|
|
@classmethod
|
|
def validate_config(cls) -> bool:
|
|
"""Validate configuration settings"""
|
|
errors = []
|
|
|
|
# Validate database type
|
|
try:
|
|
DatabaseType(cls.DB_TYPE)
|
|
except ValueError:
|
|
errors.append(f"Invalid DB_TYPE: {cls.DB_TYPE}")
|
|
|
|
# Validate database-specific settings
|
|
if cls.DB_TYPE == "victoriametrics":
|
|
if not cls.VM_HOST:
|
|
errors.append("VM_HOST is required for VictoriaMetrics")
|
|
if not isinstance(cls.VM_PORT, int) or cls.VM_PORT <= 0:
|
|
errors.append("VM_PORT must be a positive integer")
|
|
|
|
elif cls.DB_TYPE == "influxdb":
|
|
if not cls.INFLUX_HOST:
|
|
errors.append("INFLUX_HOST is required for InfluxDB")
|
|
if not cls.INFLUX_DATABASE:
|
|
errors.append("INFLUX_DATABASE is required for InfluxDB")
|
|
|
|
elif cls.DB_TYPE in ["postgresql", "mysql"]:
|
|
if cls.DB_TYPE == "postgresql":
|
|
# Check if either connection string or individual components are provided
|
|
if not cls.POSTGRES_CONNECTION_STRING:
|
|
# If no connection string, check individual components
|
|
if not cls.POSTGRES_HOST:
|
|
errors.append("POSTGRES_HOST is required for PostgreSQL")
|
|
if not cls.POSTGRES_USER:
|
|
errors.append("POSTGRES_USER is required for PostgreSQL")
|
|
if not cls.POSTGRES_PASSWORD:
|
|
errors.append("POSTGRES_PASSWORD is required for PostgreSQL")
|
|
if not cls.POSTGRES_DB:
|
|
errors.append("POSTGRES_DB is required for PostgreSQL")
|
|
else: # mysql
|
|
if not cls.MYSQL_CONNECTION_STRING:
|
|
errors.append("MYSQL_CONNECTION_STRING is required for MySQL")
|
|
|
|
# Validate numeric settings
|
|
if cls.SCRAPING_INTERVAL_HOURS <= 0:
|
|
errors.append("SCRAPING_INTERVAL_HOURS must be positive")
|
|
|
|
if cls.DATA_RETENTION_DAYS <= 0:
|
|
errors.append("DATA_RETENTION_DAYS must be positive")
|
|
|
|
if errors:
|
|
raise ConfigurationError(f"Configuration errors: {'; '.join(errors)}")
|
|
|
|
return True
|
|
|
|
@classmethod
|
|
def get_database_config(cls) -> Dict[str, Any]:
|
|
"""Returns database configuration based on DB_TYPE"""
|
|
if cls.DB_TYPE == "victoriametrics":
|
|
return {"type": "victoriametrics", "host": cls.VM_HOST, "port": cls.VM_PORT}
|
|
elif cls.DB_TYPE == "influxdb":
|
|
return {
|
|
"type": "influxdb",
|
|
"host": cls.INFLUX_HOST,
|
|
"port": cls.INFLUX_PORT,
|
|
"database": cls.INFLUX_DATABASE,
|
|
"username": cls.INFLUX_USERNAME,
|
|
"password": cls.INFLUX_PASSWORD,
|
|
}
|
|
elif cls.DB_TYPE == "postgresql":
|
|
# Use individual components if POSTGRES_CONNECTION_STRING is not provided
|
|
if cls.POSTGRES_CONNECTION_STRING:
|
|
return {
|
|
"type": "postgresql",
|
|
"connection_string": cls.POSTGRES_CONNECTION_STRING,
|
|
}
|
|
else:
|
|
# Build connection string from components (automatically URL-encodes password)
|
|
import urllib.parse
|
|
|
|
if not cls.POSTGRES_PASSWORD:
|
|
raise ConfigurationError(
|
|
"POSTGRES_PASSWORD is required for PostgreSQL (no default is provided)"
|
|
)
|
|
password = urllib.parse.quote(cls.POSTGRES_PASSWORD, safe="")
|
|
connection_string = (
|
|
f"postgresql://{cls.POSTGRES_USER}:{password}"
|
|
f"@{cls.POSTGRES_HOST}:{cls.POSTGRES_PORT}/{cls.POSTGRES_DB}"
|
|
)
|
|
return {"type": "postgresql", "connection_string": connection_string}
|
|
elif cls.DB_TYPE == "mysql":
|
|
if not cls.MYSQL_CONNECTION_STRING:
|
|
raise ConfigurationError(
|
|
"MYSQL_CONNECTION_STRING is required for MySQL (no default is provided)"
|
|
)
|
|
return {"type": "mysql", "connection_string": cls.MYSQL_CONNECTION_STRING}
|
|
else: # sqlite
|
|
return {
|
|
"type": "sqlite",
|
|
"connection_string": f"sqlite:///{cls.DATABASE_PATH}",
|
|
}
|
|
|
|
@classmethod
|
|
def get_all_settings(cls) -> Dict[str, Any]:
|
|
"""Returns all configuration settings"""
|
|
return {
|
|
"DB_TYPE": cls.DB_TYPE,
|
|
"DATABASE_PATH": cls.DATABASE_PATH,
|
|
"TARGET_URL": cls.TARGET_URL,
|
|
"API_URL": cls.API_URL,
|
|
"REQUEST_TIMEOUT": cls.REQUEST_TIMEOUT,
|
|
"SCRAPING_INTERVAL_HOURS": cls.SCRAPING_INTERVAL_HOURS,
|
|
"LOG_LEVEL": cls.LOG_LEVEL,
|
|
"LOG_FILE": cls.LOG_FILE,
|
|
"DATA_RETENTION_DAYS": cls.DATA_RETENTION_DAYS,
|
|
"MAX_RETRIES": cls.MAX_RETRIES,
|
|
"RETRY_DELAY_SECONDS": cls.RETRY_DELAY_SECONDS,
|
|
"VM_HOST": cls.VM_HOST,
|
|
"VM_PORT": cls.VM_PORT,
|
|
"INFLUX_HOST": cls.INFLUX_HOST,
|
|
"INFLUX_PORT": cls.INFLUX_PORT,
|
|
"INFLUX_DATABASE": cls.INFLUX_DATABASE,
|
|
}
|
|
|
|
@classmethod
|
|
def print_settings(cls):
|
|
"""Prints all current settings"""
|
|
print("=== Water Level Monitor Configuration ===")
|
|
for key, value in cls.get_all_settings().items():
|
|
# Hide sensitive information
|
|
if "PASSWORD" in key and value:
|
|
value = "*" * len(str(value))
|
|
print(f"{key}: {value}")
|
|
print("=" * 45)
|
|
|
|
print("\nDatabase Configuration:")
|
|
db_config = cls.get_database_config()
|
|
for key, value in db_config.items():
|
|
if "password" in key and value:
|
|
value = "*" * len(str(value))
|
|
print(f" {key}: {value}")
|
|
print("=" * 45)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
Config.print_settings()
|