1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
|
"""DWD WFS API - Utility module to update warncells.md."""
from __future__ import annotations
import urllib.parse
import requests
AREAS = {
"Gemeinden": "dwd:Warngebiete_Gemeinden",
"Landkreise": "dwd:Warngebiete_Kreise",
"Binnenseen": "dwd:Warngebiete_Binnenseen",
"Küste": "dwd:Warngebiete_Kueste",
}
def fetch_json(layer: str) -> dict:
"""Fetch data via json."""
url = (
"https://maps.dwd.de/geoserver/dwd/ows?service=WFS"
+ "&version=2.0.0&request=GetFeature&typeName="
+ urllib.parse.quote(layer)
+ "&OutputFormat=application/json"
)
response = requests.get(url=url, timeout=10.0)
return response.json()
def parse_json(section: str, data: dict) -> tuple[str, str, str]:
"""Parse json data."""
stations = []
for x in data["features"]:
cellid = str(x["properties"]["WARNCELLID"])
name = x["properties"]["NAME"]
stations.append((cellid, section, name))
return stations
# Fetch and sort all available stations
all_stations = []
for category, area in AREAS.items():
print(f"Fetching {category}")
all_stations.extend(parse_json(category, fetch_json(area)))
all_stations.sort(key=lambda x: x[0])
print("Updating warncells.md")
with open("warncells.md", "w", encoding="utf-8") as f:
print("| Warncell ID | Gebietstyp | Name |", file=f)
print("|-------------|------------|------|", file=f)
for warncell in all_stations:
output = "| " + warncell[0] + " | " + warncell[1] + " | " + warncell[2] + " |"
print(output, file=f)
f.close()
|
