Add Risk of Fire sensor to IPMA (#80295)

homeassistant/components/ipma/__init__.py

@@ -18,7 +18,7 @@ from .const import DATA_API, DATA_LOCATION, DOMAIN
 
 DEFAULT_NAME = "ipma"
 
-PLATFORMS = [Platform.WEATHER]
+PLATFORMS = [Platform.WEATHER, Platform.SENSOR]
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/components/ipma/config_flow.py

@@ -5,8 +5,7 @@ from homeassistant import config_entries
 from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_MODE, CONF_NAME
 import homeassistant.helpers.config_validation as cv
 
-from .const import DOMAIN, HOME_LOCATION_NAME
-from .weather import FORECAST_MODE
+from .const import DOMAIN, FORECAST_MODE, HOME_LOCATION_NAME
 
 
 class IpmaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):

homeassistant/components/ipma/const.py

@@ -1,5 +1,24 @@
 """Constants for IPMA component."""
-from homeassistant.components.weather import DOMAIN as WEATHER_DOMAIN
+from datetime import timedelta
+
+from homeassistant.components.weather import (
+    ATTR_CONDITION_CLEAR_NIGHT,
+    ATTR_CONDITION_CLOUDY,
+    ATTR_CONDITION_EXCEPTIONAL,
+    ATTR_CONDITION_FOG,
+    ATTR_CONDITION_HAIL,
+    ATTR_CONDITION_LIGHTNING,
+    ATTR_CONDITION_LIGHTNING_RAINY,
+    ATTR_CONDITION_PARTLYCLOUDY,
+    ATTR_CONDITION_POURING,
+    ATTR_CONDITION_RAINY,
+    ATTR_CONDITION_SNOWY,
+    ATTR_CONDITION_SNOWY_RAINY,
+    ATTR_CONDITION_SUNNY,
+    ATTR_CONDITION_WINDY,
+    ATTR_CONDITION_WINDY_VARIANT,
+    DOMAIN as WEATHER_DOMAIN,
+)
 
 DOMAIN = "ipma"
 
@@ -9,3 +28,27 @@ DATA_API = "api"
 DATA_LOCATION = "location"
 
 ENTITY_ID_SENSOR_FORMAT_HOME = f"{WEATHER_DOMAIN}.ipma_{HOME_LOCATION_NAME}"
+
+MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=30)
+
+CONDITION_CLASSES = {
+    ATTR_CONDITION_CLOUDY: [4, 5, 24, 25, 27],
+    ATTR_CONDITION_FOG: [16, 17, 26],
+    ATTR_CONDITION_HAIL: [21, 22],
+    ATTR_CONDITION_LIGHTNING: [19],
+    ATTR_CONDITION_LIGHTNING_RAINY: [20, 23],
+    ATTR_CONDITION_PARTLYCLOUDY: [2, 3],
+    ATTR_CONDITION_POURING: [8, 11],
+    ATTR_CONDITION_RAINY: [6, 7, 9, 10, 12, 13, 14, 15],
+    ATTR_CONDITION_SNOWY: [18],
+    ATTR_CONDITION_SNOWY_RAINY: [],
+    ATTR_CONDITION_SUNNY: [1],
+    ATTR_CONDITION_WINDY: [],
+    ATTR_CONDITION_WINDY_VARIANT: [],
+    ATTR_CONDITION_EXCEPTIONAL: [],
+    ATTR_CONDITION_CLEAR_NIGHT: [-1],
+}
+
+FORECAST_MODE = ["hourly", "daily"]
+
+ATTRIBUTION = "Instituto Português do Mar e Atmosfera"

homeassistant/components/ipma/entity.py

@@ -0,0 +1,26 @@
+"""Base Entity for IPMA."""
+from __future__ import annotations
+
+from homeassistant.helpers.device_registry import DeviceEntryType
+from homeassistant.helpers.entity import DeviceInfo, Entity
+
+from .const import DOMAIN
+
+
+class IPMADevice(Entity):
+    """Common IPMA Device Information."""
+
+    def __init__(self, location) -> None:
+        """Initialize device information."""
+        self._location = location
+        self._attr_device_info = DeviceInfo(
+            entry_type=DeviceEntryType.SERVICE,
+            identifiers={
+                (
+                    DOMAIN,
+                    f"{self._location.station_latitude}, {self._location.station_longitude}",
+                )
+            },
+            manufacturer=DOMAIN,
+            name=self._location.name,
+        )

homeassistant/components/ipma/sensor.py

@@ -0,0 +1,89 @@
+"""Support for IPMA sensors."""
+from __future__ import annotations
+
+from collections.abc import Callable, Coroutine
+from dataclasses import dataclass
+import logging
+
+import async_timeout
+from pyipma.api import IPMA_API
+from pyipma.location import Location
+
+from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.util import Throttle
+
+from .const import DATA_API, DATA_LOCATION, DOMAIN, MIN_TIME_BETWEEN_UPDATES
+from .entity import IPMADevice
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclass
+class IPMARequiredKeysMixin:
+    """Mixin for required keys."""
+
+    value_fn: Callable[[Location, IPMA_API], Coroutine[Location, IPMA_API, int | None]]
+
+
+@dataclass
+class IPMASensorEntityDescription(SensorEntityDescription, IPMARequiredKeysMixin):
+    """Describes IPMA sensor entity."""
+
+
+async def async_retrive_rcm(location: Location, api: IPMA_API) -> int | None:
+    """Retrieve RCM."""
+    fire_risk = await location.fire_risk(api)
+    if fire_risk:
+        return fire_risk.rcm
+    return None
+
+
+SENSOR_TYPES: tuple[IPMASensorEntityDescription, ...] = (
+    IPMASensorEntityDescription(
+        key="rcm",
+        translation_key="fire_risk",
+        value_fn=async_retrive_rcm,
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up the IPMA sensor platform."""
+    api = hass.data[DOMAIN][entry.entry_id][DATA_API]
+    location = hass.data[DOMAIN][entry.entry_id][DATA_LOCATION]
+
+    entities = [IPMASensor(api, location, description) for description in SENSOR_TYPES]
+
+    async_add_entities(entities, True)
+
+
+class IPMASensor(SensorEntity, IPMADevice):
+    """Representation of an IPMA sensor."""
+
+    entity_description: IPMASensorEntityDescription
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        api: IPMA_API,
+        location: Location,
+        description: IPMASensorEntityDescription,
+    ) -> None:
+        """Initialize the IPMA Sensor."""
+        IPMADevice.__init__(self, location)
+        self.entity_description = description
+        self._api = api
+        self._attr_unique_id = f"{self._location.station_latitude}, {self._location.station_longitude}, {self.entity_description.key}"
+
+    @Throttle(MIN_TIME_BETWEEN_UPDATES)
+    async def async_update(self) -> None:
+        """Update Fire risk."""
+        async with async_timeout.timeout(10):
+            self._attr_native_value = await self.entity_description.value_fn(
+                self._location, self._api
+            )

homeassistant/components/ipma/strings.json

@@ -18,5 +18,12 @@
     "info": {
       "api_endpoint_reachable": "IPMA API endpoint reachable"
     }
+  },
+  "entity": {
+    "sensor": {
+      "fire_risk": {
+        "name": "Fire risk"
+      }
+    }
   }
 }

homeassistant/components/ipma/weather.py

@@ -1,7 +1,6 @@
 """Support for IPMA weather service."""
 from __future__ import annotations
 
-from datetime import timedelta
 import logging
 
 import async_timeout
@@ -10,21 +9,6 @@ from pyipma.forecast import Forecast
 from pyipma.location import Location
 
 from homeassistant.components.weather import (
-    ATTR_CONDITION_CLEAR_NIGHT,
-    ATTR_CONDITION_CLOUDY,
-    ATTR_CONDITION_EXCEPTIONAL,
-    ATTR_CONDITION_FOG,
-    ATTR_CONDITION_HAIL,
-    ATTR_CONDITION_LIGHTNING,
-    ATTR_CONDITION_LIGHTNING_RAINY,
-    ATTR_CONDITION_PARTLYCLOUDY,
-    ATTR_CONDITION_POURING,
-    ATTR_CONDITION_RAINY,
-    ATTR_CONDITION_SNOWY,
-    ATTR_CONDITION_SNOWY_RAINY,
-    ATTR_CONDITION_SUNNY,
-    ATTR_CONDITION_WINDY,
-    ATTR_CONDITION_WINDY_VARIANT,
     ATTR_FORECAST_CONDITION,
     ATTR_FORECAST_NATIVE_TEMP,
     ATTR_FORECAST_NATIVE_TEMP_LOW,
@@ -48,34 +32,18 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.sun import is_up
 from homeassistant.util import Throttle
 
-from .const import DATA_API, DATA_LOCATION, DOMAIN
+from .const import (
+    ATTRIBUTION,
+    CONDITION_CLASSES,
+    DATA_API,
+    DATA_LOCATION,
+    DOMAIN,
+    MIN_TIME_BETWEEN_UPDATES,
+)
+from .entity import IPMADevice
 
 _LOGGER = logging.getLogger(__name__)
 
-ATTRIBUTION = "Instituto Português do Mar e Atmosfera"
-
-MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=30)
-
-CONDITION_CLASSES = {
-    ATTR_CONDITION_CLOUDY: [4, 5, 24, 25, 27],
-    ATTR_CONDITION_FOG: [16, 17, 26],
-    ATTR_CONDITION_HAIL: [21, 22],
-    ATTR_CONDITION_LIGHTNING: [19],
-    ATTR_CONDITION_LIGHTNING_RAINY: [20, 23],
-    ATTR_CONDITION_PARTLYCLOUDY: [2, 3],
-    ATTR_CONDITION_POURING: [8, 11],
-    ATTR_CONDITION_RAINY: [6, 7, 9, 10, 12, 13, 14, 15],
-    ATTR_CONDITION_SNOWY: [18],
-    ATTR_CONDITION_SNOWY_RAINY: [],
-    ATTR_CONDITION_SUNNY: [1],
-    ATTR_CONDITION_WINDY: [],
-    ATTR_CONDITION_WINDY_VARIANT: [],
-    ATTR_CONDITION_EXCEPTIONAL: [],
-    ATTR_CONDITION_CLEAR_NIGHT: [-1],
-}
-
-FORECAST_MODE = ["hourly", "daily"]
-
 
 async def async_setup_entry(
     hass: HomeAssistant,
@@ -110,7 +78,7 @@ async def async_setup_entry(
     async_add_entities([IPMAWeather(location, api, config_entry.data)], True)
 
 
-class IPMAWeather(WeatherEntity):
+class IPMAWeather(WeatherEntity, IPMADevice):
     """Representation of a weather condition."""
 
     _attr_native_pressure_unit = UnitOfPressure.HPA
@@ -121,13 +89,14 @@ class IPMAWeather(WeatherEntity):
 
     def __init__(self, location: Location, api: IPMA_API, config) -> None:
         """Initialise the platform with a data instance and station name."""
+        IPMADevice.__init__(self, location)
         self._api = api
-        self._location_name = config.get(CONF_NAME, location.name)
+        self._attr_name = config.get(CONF_NAME, location.name)
         self._mode = config.get(CONF_MODE)
         self._period = 1 if config.get(CONF_MODE) == "hourly" else 24
-        self._location = location
         self._observation = None
         self._forecast: list[Forecast] = []
+        self._attr_unique_id = f"{self._location.station_latitude}, {self._location.station_longitude}, {self._mode}"
 
     @Throttle(MIN_TIME_BETWEEN_UPDATES)
     async def async_update(self) -> None:
@@ -153,19 +122,6 @@ class IPMAWeather(WeatherEntity):
                 self._observation,
             )
 
-    @property
-    def unique_id(self) -> str:
-        """Return a unique id."""
-        return (
-            f"{self._location.station_latitude}, {self._location.station_longitude},"
-            f" {self._mode}"
-        )
-
-    @property
-    def name(self):
-        """Return the name of the station."""
-        return self._location_name
-
     def _condition_conversion(self, identifier, forecast_dt):
         """Convert from IPMA weather_type id to HA."""
         if identifier == 1 and not is_up(self.hass, forecast_dt):

tests/components/ipma/__init__.py

@@ -15,6 +15,18 @@ ENTRY_CONFIG = {
 class MockLocation:
     """Mock Location from pyipma."""
 
+    async def fire_risk(self, api):
+        """Mock Fire Risk."""
+        RCM = namedtuple(
+            "RCM",
+            [
+                "dico",
+                "rcm",
+                "coordinates",
+            ],
+        )
+        return RCM("some place", 3, (0, 0))
+
     async def observation(self, api):
         """Mock Observation."""
         Observation = namedtuple(

tests/components/ipma/test_sensor.py

@@ -0,0 +1,24 @@
+"""The sensor tests for the IPMA platform."""
+
+from unittest.mock import patch
+
+from . import ENTRY_CONFIG, MockLocation
+
+from tests.common import MockConfigEntry
+
+
+async def test_ipma_fire_risk_create_sensors(hass):
+    """Test creation of fire risk 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_fire_risk")
+
+    assert state.state == "3"