Add weather warning sensor to IPMA (#134054)

Co-authored-by: Abílio Costa <abmantis@users.noreply.github.com>
This commit is contained in:
Diogo Gomes 2025-01-07 22:11:24 +00:00 committed by GitHub
parent de9c05ad53
commit a1d43b9387
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 75 additions and 9 deletions

View File

@ -4,8 +4,9 @@ from __future__ import annotations
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
@ -28,23 +29,41 @@ _LOGGER = logging.getLogger(__name__)
class IPMASensorEntityDescription(SensorEntityDescription):
"""Describes a IPMA sensor entity."""
value_fn: Callable[[Location, IPMA_API], Coroutine[Location, IPMA_API, int | None]]
value_fn: Callable[
[Location, IPMA_API], Coroutine[Location, IPMA_API, tuple[Any, dict[str, Any]]]
]
async def async_retrieve_rcm(location: Location, api: IPMA_API) -> int | None:
async def async_retrieve_rcm(
location: Location, api: IPMA_API
) -> tuple[int, dict[str, Any]] | tuple[None, dict[str, Any]]:
"""Retrieve RCM."""
fire_risk: RCM = await location.fire_risk(api)
if fire_risk:
return fire_risk.rcm
return None
return fire_risk.rcm, {}
return None, {}
async def async_retrieve_uvi(location: Location, api: IPMA_API) -> int | None:
async def async_retrieve_uvi(
location: Location, api: IPMA_API
) -> tuple[int, dict[str, Any]] | tuple[None, dict[str, Any]]:
"""Retrieve UV."""
uv_risk: UV = await location.uv_risk(api)
if uv_risk:
return round(uv_risk.iUv)
return None
return round(uv_risk.iUv), {}
return None, {}
async def async_retrieve_warning(
location: Location, api: IPMA_API
) -> tuple[Any, dict[str, str]]:
"""Retrieve Warning."""
warnings = await location.warnings(api)
if len(warnings):
return warnings[0].awarenessLevelID, {
k: str(v) for k, v in asdict(warnings[0]).items()
}
return "green", {}
SENSOR_TYPES: tuple[IPMASensorEntityDescription, ...] = (
@ -58,6 +77,11 @@ SENSOR_TYPES: tuple[IPMASensorEntityDescription, ...] = (
translation_key="uv_index",
value_fn=async_retrieve_uvi,
),
IPMASensorEntityDescription(
key="alert",
translation_key="weather_alert",
value_fn=async_retrieve_warning,
),
)
@ -94,6 +118,8 @@ class IPMASensor(SensorEntity, IPMADevice):
async def async_update(self) -> None:
"""Update sensors."""
async with asyncio.timeout(10):
self._attr_native_value = await self.entity_description.value_fn(
state, attrs = await self.entity_description.value_fn(
self._location, self._api
)
self._attr_native_value = state
self._attr_extra_state_attributes = attrs

View File

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

View File

@ -6,6 +6,7 @@ from pyipma.forecast import Forecast, Forecast_Location, Weather_Type
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 @@ ENTRY_CONFIG = {
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))

View File

@ -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"