Add flood_alarm/warning_level and flood_alarm/warning sensors

imgw_pib/__init__.py

@@ -7,7 +7,13 @@
 
 from aiohttp import ClientSession
 
-from .const import API_HYDROLOGICAL_ENDPOINT, API_WEATHER_ENDPOINT, HEADERS, TIMEOUT
+from .const import (
+    API_HYDROLOGICAL_DETAILS_ENDPOINT,
+    API_HYDROLOGICAL_ENDPOINT,
+    API_WEATHER_ENDPOINT,
+    HEADERS,
+    TIMEOUT,
+)
 from .exceptions import ApiError
 from .model import ApiNames, HydrologicalData, SensorData, Units, WeatherData
 
@@ -27,6 +33,8 @@ def __init__(
         self._session = session
         self._weather_station_list: dict[str, str] = {}
         self._hydrological_station_list: dict[str, str] = {}
+        self._alarm_water_level: float | None = None
+        self._warning_water_level: float | None = None
 
         self.weather_station_id = weather_station_id
         self.hydrological_station_id = hydrological_station_id
@@ -72,6 +80,8 @@ async def initialize(self: Self) -> None:
                 msg = f"Invalid hydrological station ID: {self.hydrological_station_id}"
                 raise ApiError(msg)
 
+            await self._update_hydrological_details()
+
     async def update_weather_stations(self: Self) -> None:
         """Update list of weather stations."""
         url = API_WEATHER_ENDPOINT
@@ -108,6 +118,17 @@ async def update_hydrological_stations(self: Self) -> None:
             for station in stations_data
         }
 
+    async def _update_hydrological_details(self: Self) -> None:
+        """Update hydrological details."""
+        url = API_HYDROLOGICAL_DETAILS_ENDPOINT.format(
+            hydrological_station_id=self.hydrological_station_id
+        )
+
+        hydrological_details = await self._http_request(url)
+
+        self._warning_water_level = hydrological_details["status"]["warningValue"]
+        self._alarm_water_level = hydrological_details["status"]["alarmValue"]
+
     async def get_hydrological_data(self: Self) -> HydrologicalData:
         """Get hydrological data."""
         if self.hydrological_station_id is None:
@@ -208,6 +229,20 @@ def _parse_hydrological_data(self: Self, data: dict[str, Any]) -> HydrologicalDa
             data[ApiNames.WATER_LEVEL_MEASUREMENT_DATE],
             "%Y-%m-%d %H:%M:%S",
         )
+        flood_warning_level_sensor = SensorData(
+            name="Flood Warning Level",
+            value=self._warning_water_level,
+            unit=Units.CENTIMETERS.value
+            if self._warning_water_level is not None
+            else None,
+        )
+        flood_alarm_level_sensor = SensorData(
+            name="Flood Alarm Level",
+            value=self._alarm_water_level,
+            unit=Units.CENTIMETERS.value
+            if self._alarm_water_level is not None
+            else None,
+        )
         water_temperature = data[ApiNames.WATER_TEMPERATURE]
         water_temperature_sensor = SensorData(
             name="Water Temperature",
@@ -221,6 +256,8 @@ def _parse_hydrological_data(self: Self, data: dict[str, Any]) -> HydrologicalDa
 
         return HydrologicalData(
             water_level=water_level_sensor,
+            flood_warning_level=flood_warning_level_sensor,
+            flood_alarm_level=flood_alarm_level_sensor,
             water_temperature=water_temperature_sensor,
             station=data[ApiNames.STATION],
             river=data[ApiNames.RIVER],

imgw_pib/const.py

@@ -5,6 +5,9 @@
 API_BASE_ENDPOINT = "https://danepubliczne.imgw.pl/api/data"
 API_HYDROLOGICAL_ENDPOINT = f"{API_BASE_ENDPOINT}/hydro"
 API_WEATHER_ENDPOINT = f"{API_BASE_ENDPOINT}/synop"
+API_HYDROLOGICAL_DETAILS_ENDPOINT = (
+    "https://hydro-back.imgw.pl/station/hydro/status?id={hydrological_station_id}"
+)
 
 HEADERS = {"Content-Type": "application/json"}
 TIMEOUT = ClientTimeout(total=20)

imgw_pib/model.py

@@ -3,6 +3,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from enum import StrEnum
+from typing import Self
 
 
 @dataclass
@@ -29,8 +30,10 @@ class WeatherData(ImgwPibData):
     wind_speed: SensorData
     wind_direction: SensorData
     precipitation: SensorData
+
     station: str
     station_id: str
+
     measurement_date: datetime | None
 
 
@@ -39,13 +42,32 @@ class HydrologicalData(ImgwPibData):
     """Hudrological Data class for IMGW-PIB."""
 
     water_level: SensorData
+    flood_alarm_level: SensorData
+    flood_warning_level: SensorData
     water_temperature: SensorData
-    station: str
+
     river: str
     station_id: str
+    station: str
+
     water_level_measurement_date: datetime | None
     water_temperature_measurement_date: datetime | None
 
+    flood_alarm: bool | None = None
+    flood_warning: bool | None = None
+
+    def __post_init__(self: Self) -> None:
+        """Call after initialization."""
+        if self.water_level.value is not None:
+            if self.flood_warning_level.value is not None:
+                self.flood_alarm = (
+                    self.water_level.value >= self.flood_warning_level.value
+                )
+            if self.flood_alarm_level.value is not None:
+                self.flood_warning = (
+                    self.water_level.value >= self.flood_alarm_level.value
+                )
+
 
 class ApiNames(StrEnum):
     """Names type for API."""

tests/conftest.py

@@ -42,6 +42,15 @@ def hydrological_station() -> dict[str, Any]:
         return cast(dict[str, Any], json.load(file))
 
 
+@pytest.fixture()
+def hydrological_details() -> dict[str, Any]:
+    """Return hydrological details from the fixture file."""
+    with Path.open(
+        "tests/fixtures/hydrological_details.json", encoding="utf-8"
+    ) as file:
+        return cast(dict[str, Any], json.load(file))
+
+
 @pytest.fixture()
 def snapshot(snapshot: SnapshotAssertion) -> SnapshotAssertion:
     """Return snapshot assertion fixture."""

tests/fixtures/hydrological_details.json

@@ -0,0 +1,52 @@
+{
+    "properties": {
+        "maximumStateValue": 700,
+        "maximumStateDate": "2009-10-14",
+        "minimumStateValue": 375,
+        "minimumStateDate": "1999-12-04",
+        "province": null,
+        "dataClause": null
+    },
+    "stateCode": "high",
+    "info": null,
+    "area": null,
+    "status": {
+        "id": 4693,
+        "name": "154190050",
+        "description": "NOWAKOWO",
+        "river": "Zalew Wi\u015blany (5119)",
+        "parameter": "B00020S",
+        "currentState": {
+            "date": "2024-04-23T12:40:00Z",
+            "value": 539.0
+        },
+        "previousState": {
+            "date": "2024-04-23T12:30:00Z",
+            "value": 539.0
+        },
+        "lastState": null,
+        "status": 180,
+        "outdatedState": false,
+        "riverCourseKm": 2.75,
+        "catchmentArea": 1449.07,
+        "trend": 10,
+        "province": "warmi\u0144sko-mazurskie",
+        "notes": [
+            "Stany wody okresowo sztucznie pi\u0119trzone",
+            "W pobli\u017cu stacji hydrologicznej prace budowlane w korycie rzeki lub w jego s\u0105siedztwie",
+            "Stany wody pod wp\u0142ywem waha\u0144 poziomu zalewu",
+            "Przerwa w danych spowodowana czynnikami niezale\u017cnymi od IMGW-PIB",
+            "Stacja hydrologiczna w zasi\u0119gu cofki recypienta"
+        ],
+        "waterGaugeZeroOrdinate": null,
+        "waterGaugeZeroOrdinateReferenceSystem": null,
+        "alarmValue": 630.0,
+        "warningValue": 590.0
+    },
+    "id": "154190050",
+    "coordinates": {
+        "y": 54.2305565,
+        "x": 19.3630562
+    },
+    "mapSettings": null
+}

tests/snapshots/test_init.ambr

@@ -1,6 +1,6 @@
 # serializer version: 1
 # name: test_hydrological_station
-  HydrologicalData(water_level=SensorData(name='Water Level', value=568.0, unit='cm'), water_temperature=SensorData(name='Water Temperature', value=8.1, unit='°C'), station='Nowe Batorowo', river='Zalew Wiślany', station_id='154190050', water_level_measurement_date=datetime.datetime(2024, 4, 22, 10, 0, tzinfo=datetime.timezone.utc), water_temperature_measurement_date=datetime.datetime(2024, 4, 22, 10, 10, tzinfo=datetime.timezone.utc))
+  HydrologicalData(water_level=SensorData(name='Water Level', value=568.0, unit='cm'), flood_alarm_level=SensorData(name='Flood Alarm Level', value=630.0, unit='cm'), flood_warning_level=SensorData(name='Flood Warning Level', value=590.0, unit='cm'), water_temperature=SensorData(name='Water Temperature', value=8.1, unit='°C'), river='Zalew Wiślany', station_id='154190050', station='Nowe Batorowo', water_level_measurement_date=datetime.datetime(2024, 4, 22, 10, 0, tzinfo=datetime.timezone.utc), water_temperature_measurement_date=datetime.datetime(2024, 4, 22, 10, 10, tzinfo=datetime.timezone.utc), flood_alarm=False, flood_warning=False)
 # ---
 # name: test_hydrological_stations
   dict({

tests/test_init.py

@@ -9,7 +9,11 @@
 from syrupy import SnapshotAssertion
 
 from imgw_pib import ImgwPib
-from imgw_pib.const import API_HYDROLOGICAL_ENDPOINT, API_WEATHER_ENDPOINT
+from imgw_pib.const import (
+    API_HYDROLOGICAL_DETAILS_ENDPOINT,
+    API_HYDROLOGICAL_ENDPOINT,
+    API_WEATHER_ENDPOINT,
+)
 from imgw_pib.exceptions import ApiError
 from imgw_pib.model import ApiNames
 
@@ -100,6 +104,7 @@ async def test_hydrological_station(
     snapshot: SnapshotAssertion,
     hydrological_stations: list[dict[str, Any]],
     hydrological_station: dict[str, Any],
+    hydrological_details: dict[str, Any],
 ) -> None:
     """Test weather station."""
     session = aiohttp.ClientSession()
@@ -109,6 +114,12 @@ async def test_hydrological_station(
         session_mock.get(
             f"{API_HYDROLOGICAL_ENDPOINT}/id/154190050", payload=hydrological_station
         )
+        session_mock.get(
+            API_HYDROLOGICAL_DETAILS_ENDPOINT.format(
+                hydrological_station_id="154190050"
+            ),
+            payload=hydrological_details,
+        )
 
         imgwpib = await ImgwPib.create(session, hydrological_station_id="154190050")
         hydrological_data = await imgwpib.get_hydrological_data()
@@ -190,6 +201,7 @@ async def test_get_hydrological_data_without_station_id() -> None:
 async def test_invalid_water_level_value(
     hydrological_stations: list[dict[str, Any]],
     hydrological_station: dict[str, Any],
+    hydrological_details: dict[str, Any],
 ) -> None:
     """Test invalid water level value."""
     session = aiohttp.ClientSession()
@@ -201,6 +213,12 @@ async def test_invalid_water_level_value(
         session_mock.get(
             f"{API_HYDROLOGICAL_ENDPOINT}/id/154190050", payload=hydrological_station
         )
+        session_mock.get(
+            API_HYDROLOGICAL_DETAILS_ENDPOINT.format(
+                hydrological_station_id="154190050"
+            ),
+            payload=hydrological_details,
+        )
 
         imgwpib = await ImgwPib.create(session, hydrological_station_id="154190050")
 
@@ -216,6 +234,7 @@ async def test_invalid_water_level_value(
 async def test_invalid_date(
     hydrological_stations: list[dict[str, Any]],
     hydrological_station: dict[str, Any],
+    hydrological_details: dict[str, Any],
 ) -> None:
     """Test invalid water level value."""
     session = aiohttp.ClientSession()
@@ -227,6 +246,12 @@ async def test_invalid_date(
         session_mock.get(
             f"{API_HYDROLOGICAL_ENDPOINT}/id/154190050", payload=hydrological_station
         )
+        session_mock.get(
+            API_HYDROLOGICAL_DETAILS_ENDPOINT.format(
+                hydrological_station_id="154190050"
+            ),
+            payload=hydrological_details,
+        )
 
         imgwpib = await ImgwPib.create(session, hydrological_station_id="154190050")
         hydrological_data = await imgwpib.get_hydrological_data()