grabowski
ae5d0a13d7
Add mapped river and ThaiWater sensor layers, PostgreSQL history charts, API endpoints, and dashboard tests.
62 lines
2.4 KiB
Python
62 lines
2.4 KiB
Python
"""Client for ThaiWater's public water-level sensor feed."""
|
|
|
|
from typing import Dict, List, Optional
|
|
|
|
import requests
|
|
|
|
|
|
class ThaiWaterClient:
|
|
API_URL = "https://twa-api-public.thaiwater.net/v2/waterlevel"
|
|
|
|
def __init__(self, session=None, api_key: Optional[str] = None, timeout: int = 30):
|
|
self.session = session or requests.Session()
|
|
self.api_key = api_key
|
|
self.timeout = timeout
|
|
|
|
def fetch_ping_sensors(self) -> List[Dict]:
|
|
if not self.api_key:
|
|
raise RuntimeError("THAIWATER_API_KEY is not configured")
|
|
|
|
response = self.session.get(
|
|
self.API_URL,
|
|
headers={"Accept-Language": "en", "x-api-key": self.api_key},
|
|
timeout=self.timeout,
|
|
)
|
|
response.raise_for_status()
|
|
return self._parse_ping_features(response.json())
|
|
|
|
@staticmethod
|
|
def _parse_ping_features(payload: Dict) -> List[Dict]:
|
|
sensors = []
|
|
for collection in payload.get("data", {}).values():
|
|
for feature in collection.get("features", []):
|
|
properties = feature.get("properties") or {}
|
|
basin = properties.get("basin") or {}
|
|
if basin.get("basin") != "Ping":
|
|
continue
|
|
|
|
geometry = feature.get("geometry") or {}
|
|
coordinates = geometry.get("coordinates") or []
|
|
if len(coordinates) < 2:
|
|
continue
|
|
|
|
station = properties.get("station") or {}
|
|
station_code = station.get("stationCode", "")
|
|
sensors.append(
|
|
{
|
|
"id": f"thaiwater:{properties.get('id')}",
|
|
"station_code": station_code.split("-", 1)[-1],
|
|
"station_name": station.get("station"),
|
|
"latitude": coordinates[1],
|
|
"longitude": coordinates[0],
|
|
"timestamp": properties.get("waterlevelDatetime"),
|
|
"water_level_msl": properties.get("waterlevelMsl"),
|
|
"bank_percent": properties.get("storagePercent"),
|
|
"distance_to_bank": properties.get("diffWlBank"),
|
|
"river_name": properties.get("riverName"),
|
|
"agency": (properties.get("agency") or {}).get("agencyShort"),
|
|
"source": "ThaiWater",
|
|
}
|
|
)
|
|
return sensors
|