You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
93 lines
2.5 KiB
Python
93 lines
2.5 KiB
Python
import datetime
|
|
from typing import Any, Dict, Iterable, List, NamedTuple, Optional
|
|
|
|
import requests
|
|
|
|
|
|
class Measurement(NamedTuple):
|
|
timestamp: datetime.datetime # in UTC
|
|
unit: str
|
|
value: float
|
|
|
|
|
|
class Location(NamedTuple):
|
|
lat: Optional[float]
|
|
lon: Optional[float]
|
|
alt: Optional[float]
|
|
|
|
|
|
class Station(NamedTuple):
|
|
id: str # might contain not only digits so no int
|
|
name: str
|
|
operator: Optional[str]
|
|
location: Location
|
|
flow: Optional[Dict[str, Measurement]]
|
|
water_level: Dict[str, Measurement]
|
|
raw_data: Dict[str, Any]
|
|
|
|
|
|
def _parse_measurements(values: dict, key: str) -> Optional[Dict[str, Measurement]]:
|
|
if key not in values:
|
|
return None
|
|
return {
|
|
name: Measurement(
|
|
timestamp=datetime.datetime.fromtimestamp(measurement["dt"] / 1000),
|
|
unit=measurement["unit"],
|
|
value=measurement["v"],
|
|
)
|
|
for name, measurement in values[key].items()
|
|
if measurement != {}
|
|
}
|
|
|
|
|
|
def get_hydris_data(url: str) -> List[Station]:
|
|
data = requests.get(url).json()
|
|
return [
|
|
Station(
|
|
id=station["number"],
|
|
name=station["name"],
|
|
operator=station.get("Betreiber"),
|
|
location=Location(
|
|
lat=station.get("latlng", (None, None))[0],
|
|
lon=station.get("latlng", (None, None))[1],
|
|
alt=station.get("altitude"),
|
|
),
|
|
flow=_parse_measurements(station["values"], "Q"),
|
|
water_level=_parse_measurements(station["values"], "W"),
|
|
raw_data=station,
|
|
)
|
|
for station in data
|
|
]
|
|
|
|
|
|
def get_station(data: Iterable[Station], station_id: str) -> Optional[Station]:
|
|
result = filter(lambda station: station.id == station_id, data)
|
|
return None if not result else list(result)[0]
|
|
|
|
|
|
def main() -> None:
|
|
data = get_hydris_data("https://www.salzburg.gv.at/wasser/hydro/grafiken/data.json")
|
|
|
|
numbers = [204180, 204032, 203323, 204198, 203570, 203539]
|
|
for number in numbers:
|
|
station = get_station(data, station_id=str(number))
|
|
|
|
# Salzburg_Nonntaler Brücke / Salzach
|
|
print(station.name)
|
|
|
|
# HD Salzburg
|
|
print(station.operator)
|
|
|
|
# Location(lat=47.79805653502927, lon=13.054023495047648, alt=412.8)
|
|
print(station.location)
|
|
|
|
__import__("pprint").pprint(station.flow)
|
|
__import__("pprint").pprint(station.water_level)
|
|
|
|
# Raw station JSON data
|
|
# print(station.raw_data)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|