Add weather warning sensor to IPMA

homeassistant/components/ipma/sensor.py

@@ -4,8 +4,9 @@
 
 import asyncio
 from collections.abc import Callable, Coroutine
-from dataclasses import dataclass
+from dataclasses import asdict, dataclass
 import logging
+from typing import Any
 
 from pyipma.api import IPMA_API
 from pyipma.location import Location
@@ -29,6 +30,8 @@
     """Describes a IPMA sensor entity."""
 
     value_fn: Callable[[Location, IPMA_API], Coroutine[Location, IPMA_API, int | None]]
+    value_extractor: Callable[[Any], Any] | None = None
+    extra_attr_fn: Callable[[Any], dict[str, Any]] | None = None
 
 
 async def async_retrieve_rcm(location: Location, api: IPMA_API) -> int | None:
@@ -47,6 +50,19 @@
     return None
 
 
+async def async_retrieve_warning(location: Location, api: IPMA_API) -> int | None:
+    """Retrieve Warning."""
+    warning = await location.warnings(api)
+    if len(warning):
+        return warning[0]
+    return None
+
+
+def get_extra_attr(data: Any) -> dict[str, Any]:
+    """Return the extra attributes."""
+    return {k: str(v) for k, v in asdict(data).items()}
+
+
 SENSOR_TYPES: tuple[IPMASensorEntityDescription, ...] = (
     IPMASensorEntityDescription(
         key="rcm",
@@ -58,6 +74,13 @@
         translation_key="uv_index",
         value_fn=async_retrieve_uvi,
     ),
+    IPMASensorEntityDescription(
+        key="alert",
+        translation_key="weather_alert",
+        value_fn=async_retrieve_warning,
+        value_extractor=lambda data: data.awarenessLevelID if data else "green",
+        extra_attr_fn=get_extra_attr,
+    ),
 )
 
 
@@ -89,11 +112,29 @@
         IPMADevice.__init__(self, api, location)
         self.entity_description = description
         self._attr_unique_id = f"{self._location.station_latitude}, {self._location.station_longitude}, {self.entity_description.key}"
+        self._ipma_data: Any | None = None
 
     @Throttle(MIN_TIME_BETWEEN_UPDATES)
     async def async_update(self) -> None:
         """Update sensors."""
         async with asyncio.timeout(10):
-            self._attr_native_value = await self.entity_description.value_fn(
+            self._ipma_data = await self.entity_description.value_fn(
                 self._location, self._api
             )
+
+    @property
+    def native_value(self) -> str | None:
+        """Return the state of the sensor."""
+        if self.entity_description.value_extractor is not None:
+            return self.entity_description.value_extractor(self._ipma_data)
+        return self._ipma_data
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any] | None:
+        """Return the state attributes."""
+        if (
+            self.entity_description.extra_attr_fn is not None
+            and self._ipma_data is not None
+        ):
+            return self.entity_description.extra_attr_fn(self._ipma_data)
+        return None

homeassistant/components/ipma/strings.json

@@ -31,6 +31,15 @@
       },
       "uv_index": {
         "name": "UV index"
+      },
+      "weather_alert": {
+        "name": "Weather Alert",
+        "state": {
+          "red": "Red",
+          "yellow": "Yellow",
+          "orange": "Orange",
+          "green": "Green"
+        }
       }
     }
   }

tests/components/ipma/__init__.py

@@ -6,6 +6,7 @@
 from pyipma.observation import Observation
 from pyipma.rcm import RCM
 from pyipma.uv import UV
+from pyipma.warnings import Warning
 
 from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_MODE, CONF_NAME
 
@@ -20,6 +21,20 @@
 class MockLocation:
     """Mock Location from pyipma."""
 
+    async def warnings(self, api):
+        """Mock Warnings."""
+        return [
+            Warning(
+                text="Na costa Sul, ondas de sueste com 2 a 2,5 metros, em especial "
+                "no barlavento.",
+                awarenessTypeName="Agitação Marítima",
+                idAreaAviso="FAR",
+                startTime=datetime(2024, 12, 26, 12, 24),
+                awarenessLevelID="yellow",
+                endTime=datetime(2024, 12, 28, 6, 0),
+            )
+        ]
+
     async def fire_risk(self, api):
         """Mock Fire Risk."""
         return RCM("some place", 3, (0, 0))

tests/components/ipma/test_sensor.py

@@ -35,3 +35,19 @@ async def test_ipma_uv_index_create_sensors(hass: HomeAssistant) -> None:
     state = hass.states.get("sensor.hometown_uv_index")
 
     assert state.state == "6"
+
+
+async def test_ipma_warning_create_sensors(hass: HomeAssistant) -> None:
+    """Test creation of warning sensors."""
+
+    with patch("pyipma.location.Location.get", return_value=MockLocation()):
+        entry = MockConfigEntry(domain="ipma", data=ENTRY_CONFIG)
+        entry.add_to_hass(hass)
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.hometown_weather_alert")
+
+    assert state.state == "yellow"
+
+    assert state.attributes["awarenessTypeName"] == "Agitação Marítima"