94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
from xml.etree import ElementTree as ET
|
|
|
|
from app.cache import cache_get, cache_set
|
|
from app.http_client import get_client
|
|
from app.config import CACHE_TTL_WMS_TIMES, DWD_WMS_URL
|
|
|
|
NS = {"wms": "http://www.opengis.net/wms"}
|
|
|
|
|
|
def _parse_iso_duration_minutes(duration: str) -> int:
|
|
match = re.fullmatch(r"PT(\d+)M", duration.strip().upper())
|
|
if not match:
|
|
return 5
|
|
return int(match.group(1))
|
|
|
|
|
|
def _parse_time_dimension(value: str) -> tuple[datetime, datetime, int]:
|
|
start_raw, end_raw, step_raw = value.split("/")
|
|
start = datetime.fromisoformat(start_raw.replace("Z", "+00:00"))
|
|
end = datetime.fromisoformat(end_raw.replace("Z", "+00:00"))
|
|
step_minutes = _parse_iso_duration_minutes(step_raw)
|
|
return start, end, step_minutes
|
|
|
|
|
|
def _generate_time_steps(start: datetime, end: datetime, step_minutes: int) -> list[str]:
|
|
steps: list[str] = []
|
|
current = start
|
|
delta = timedelta(minutes=step_minutes)
|
|
while current <= end:
|
|
steps.append(current.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z"))
|
|
current += delta
|
|
return steps
|
|
|
|
|
|
def _find_layer_dimension(xml_text: str, layer_name: str) -> str | None:
|
|
root = ET.fromstring(xml_text)
|
|
target = layer_name.split(":")[-1]
|
|
for layer in root.iter():
|
|
if not layer.tag.endswith("Layer"):
|
|
continue
|
|
name_el = layer.find("wms:Name", NS) or layer.find("Name")
|
|
if name_el is None or name_el.text != target:
|
|
continue
|
|
for dim in layer:
|
|
if dim.tag.endswith("Dimension") and (dim.attrib.get("name") or "").lower() == "time":
|
|
return (dim.text or "").strip()
|
|
return None
|
|
|
|
|
|
async def fetch_wms_time_steps(layer: str, minutes: int = 120) -> dict:
|
|
cache_key = f"wms:times:{layer}:{minutes}"
|
|
cached = await cache_get(cache_key)
|
|
if cached:
|
|
return cached
|
|
|
|
params = {"service": "WMS", "version": "1.3.0", "request": "GetCapabilities"}
|
|
client = get_client()
|
|
response = await client.get(DWD_WMS_URL, params=params, timeout=15.0)
|
|
response.raise_for_status()
|
|
dimension = _find_layer_dimension(response.text, layer)
|
|
if not dimension:
|
|
raise ValueError(f"No TIME dimension found for layer {layer}")
|
|
|
|
if "/" in dimension:
|
|
start, end, step_minutes = _parse_time_dimension(dimension)
|
|
all_steps = _generate_time_steps(start, end, step_minutes)
|
|
else:
|
|
all_steps = [item.strip() for item in dimension.split(",") if item.strip()]
|
|
|
|
cutoff = datetime.now(timezone.utc) - timedelta(minutes=minutes)
|
|
filtered = []
|
|
for step in all_steps:
|
|
try:
|
|
ts = datetime.fromisoformat(step.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
continue
|
|
if ts >= cutoff:
|
|
filtered.append(step)
|
|
|
|
payload = {
|
|
"layer": layer,
|
|
"minutes": minutes,
|
|
"step_minutes": _parse_iso_duration_minutes("PT5M") if "/" in dimension else 5,
|
|
"count": len(filtered),
|
|
"times": filtered,
|
|
"latest": filtered[-1] if filtered else None,
|
|
}
|
|
await cache_set(cache_key, payload, CACHE_TTL_WMS_TIMES)
|
|
return payload
|