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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
"""Thin wrapper around the UK Environment Agent Real-time Flood Monitoring API."""
import enum
from typing import Any, Dict, List
import aiohttp
class Status(enum.Enum):
"""The status of the monitoring station."""
ACTIVE = "Active"
CLOSED = "Closed"
SUSPENDED = "Suspended"
async def get_stations(
session: aiohttp.ClientSession,
parameter_name: str = None,
parameter: str = None,
qualifier: str = None,
label: str = None,
town: str = None,
river_name: str = None,
station: str = None,
status: Status = Status.ACTIVE,
) -> List[Dict[str, Any]]:
"""Returns a list of stations."""
params = {}
if parameter_name:
params["parameterName"] = parameter_name
if parameter:
params["parameter"] = parameter
if qualifier:
params["qualifier"] = qualifier
if label:
params["label"] = label
if town:
params["town"] = town
if river_name:
params["riverName"] = river_name
if station:
params["stationReference"] = station
if status:
params["status"] = status.value
response = await session.get(
"http://environment.data.gov.uk/flood-monitoring/id/stations",
raise_for_status=True,
timeout=aiohttp.ClientTimeout(total=30),
params=params,
)
results = await response.json()
return results["items"]
async def get_station(session: aiohttp.ClientSession, station: str) -> Dict[str, Any]:
"""Returns all data for a given station."""
response = await session.get(
f"http://environment.data.gov.uk/flood-monitoring/id/stations/{station}",
raise_for_status=True,
timeout=aiohttp.ClientTimeout(total=15),
)
results = await response.json()
return results["items"]
|
