From 02537a2f1229fb29e4c789f6d02328b49e34aa56 Mon Sep 17 00:00:00 2001 From: a7rk Date: Fri, 25 Sep 2026 13:02:04 +0200 Subject: [PATCH] Geosphere austria add advance warnings sensors (#178474) Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../geosphere_austria_warnings/coordinator.py | 9 +- .../geosphere_austria_warnings/icons.json | 6 + .../geosphere_austria_warnings/sensor.py | 62 +++-- .../geosphere_austria_warnings/strings.json | 13 + .../geosphere_austria_warnings/warnings.py | 118 +++++++++ .../fixtures/get_warnings_for_coords.json | 132 ++++++++- .../snapshots/test_sensor.ambr | 142 +++++++++- .../geosphere_austria_warnings/test_sensor.py | 94 ++++++- .../test_warnings.py | 250 ++++++++++++++++++ 9 files changed, 785 insertions(+), 41 deletions(-) create mode 100644 homeassistant/components/geosphere_austria_warnings/warnings.py create mode 100644 tests/components/geosphere_austria_warnings/test_warnings.py diff --git a/homeassistant/components/geosphere_austria_warnings/coordinator.py b/homeassistant/components/geosphere_austria_warnings/coordinator.py index b2b89185bb6f..eae3fe595433 100644 --- a/homeassistant/components/geosphere_austria_warnings/coordinator.py +++ b/homeassistant/components/geosphere_austria_warnings/coordinator.py @@ -23,6 +23,7 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda from homeassistant.util import dt as dt_util from .const import DOMAIN, LOGGER, WARNINGS_URL +from .warnings import sort_warnings # Warnings are event driven and updated by GeoSphere Austria as needed. # The cheap HEAD precheck keeps the cost of a poll low, so a relatively @@ -38,6 +39,7 @@ class GeoSphereData: location_warnings: LocationWarnings active_warnings: list[WeatherWarning] + advance_warnings: list[WeatherWarning] class GeoSphereUpdateCoordinator(DataUpdateCoordinator[GeoSphereData]): @@ -106,9 +108,12 @@ class GeoSphereUpdateCoordinator(DataUpdateCoordinator[GeoSphereData]): now = dt_util.utcnow() return GeoSphereData( location_warnings=location_warnings, - active_warnings=[ + active_warnings=sort_warnings( warning for warning in location_warnings.warnings if warning.is_active(now) - ], + ), + advance_warnings=sort_warnings( + warning for warning in location_warnings.warnings if now < warning.start + ), ) diff --git a/homeassistant/components/geosphere_austria_warnings/icons.json b/homeassistant/components/geosphere_austria_warnings/icons.json index 66a2a9402f67..6b96042af229 100644 --- a/homeassistant/components/geosphere_austria_warnings/icons.json +++ b/homeassistant/components/geosphere_austria_warnings/icons.json @@ -4,6 +4,12 @@ "active_warnings": { "default": "mdi:alert-circle-outline" }, + "advance_warning_level": { + "default": "mdi:alert" + }, + "advance_warnings": { + "default": "mdi:alert-circle-outline" + }, "warning_level": { "default": "mdi:alert" } diff --git a/homeassistant/components/geosphere_austria_warnings/sensor.py b/homeassistant/components/geosphere_austria_warnings/sensor.py index 32080e30d3ce..de9da9e80113 100644 --- a/homeassistant/components/geosphere_austria_warnings/sensor.py +++ b/homeassistant/components/geosphere_austria_warnings/sensor.py @@ -1,8 +1,8 @@ """Sensors summarizing GeoSphere Austria weather warnings.""" -from collections.abc import Callable +from collections.abc import Callable, Mapping from dataclasses import dataclass -from typing import override +from typing import Any, override from pygeosphere_warnings import WeatherWarning @@ -12,30 +12,31 @@ from homeassistant.components.sensor import ( SensorEntityDescription, SensorStateClass, ) +from homeassistant.const import MATCH_ALL from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from homeassistant.helpers.typing import StateType -from .coordinator import GeoSphereConfigEntry +from .coordinator import GeoSphereConfigEntry, GeoSphereData from .entity import GeoSphereEntity +from .warnings import LEVEL_NONE, highest_warning_level, warning_sensor_attributes PARALLEL_UPDATES = 0 -LEVEL_NONE = "none" - - -def _max_level(active_warnings: list[WeatherWarning]) -> str: - """Return the highest level of the active warnings.""" - if not active_warnings: - return LEVEL_NONE - return max(warning.level for warning in active_warnings).name.lower() - @dataclass(frozen=True, kw_only=True) class GeoSphereSensorDescription(SensorEntityDescription): """Describes a GeoSphere Austria Warnings sensor.""" + warnings_fn: Callable[[GeoSphereData], list[WeatherWarning]] value_fn: Callable[[list[WeatherWarning]], StateType] + attributes_fn: ( + Callable[ + [list[WeatherWarning]], + Mapping[str, Any], + ] + | None + ) = None SENSORS: tuple[GeoSphereSensorDescription, ...] = ( @@ -44,12 +45,31 @@ SENSORS: tuple[GeoSphereSensorDescription, ...] = ( translation_key="warning_level", device_class=SensorDeviceClass.ENUM, options=[LEVEL_NONE, "yellow", "orange", "red"], - value_fn=_max_level, + warnings_fn=lambda data: data.active_warnings, + value_fn=highest_warning_level, + attributes_fn=warning_sensor_attributes, ), GeoSphereSensorDescription( key="active_warnings", translation_key="active_warnings", state_class=SensorStateClass.MEASUREMENT, + warnings_fn=lambda data: data.active_warnings, + value_fn=len, + ), + GeoSphereSensorDescription( + key="advance_warning_level", + translation_key="advance_warning_level", + device_class=SensorDeviceClass.ENUM, + options=[LEVEL_NONE, "yellow", "orange", "red"], + warnings_fn=lambda data: data.advance_warnings, + value_fn=highest_warning_level, + attributes_fn=warning_sensor_attributes, + ), + GeoSphereSensorDescription( + key="advance_warnings", + translation_key="advance_warnings", + state_class=SensorStateClass.MEASUREMENT, + warnings_fn=lambda data: data.advance_warnings, value_fn=len, ), ) @@ -68,12 +88,24 @@ async def async_setup_entry( class GeoSphereSensor(GeoSphereEntity, SensorEntity): - """Sensor summarizing the currently active warnings.""" + """Sensor summarizing GeoSphere Austria weather warnings.""" + _unrecorded_attributes = frozenset({MATCH_ALL}) entity_description: GeoSphereSensorDescription @property @override def native_value(self) -> StateType: """Return the state of the sensor.""" - return self.entity_description.value_fn(self.coordinator.data.active_warnings) + warnings = self.entity_description.warnings_fn(self.coordinator.data) + return self.entity_description.value_fn(warnings) + + @property + @override + def extra_state_attributes(self) -> Mapping[str, Any] | None: + """Return warning details as entity attributes.""" + if self.entity_description.attributes_fn is None: + return None + + warnings = self.entity_description.warnings_fn(self.coordinator.data) + return self.entity_description.attributes_fn(warnings) diff --git a/homeassistant/components/geosphere_austria_warnings/strings.json b/homeassistant/components/geosphere_austria_warnings/strings.json index 6f54357f96f2..0d1dd74945c6 100644 --- a/homeassistant/components/geosphere_austria_warnings/strings.json +++ b/homeassistant/components/geosphere_austria_warnings/strings.json @@ -26,6 +26,19 @@ "name": "Active warnings", "unit_of_measurement": "warnings" }, + "advance_warning_level": { + "name": "Advance warning level", + "state": { + "none": "No warning", + "orange": "Orange", + "red": "Red", + "yellow": "Yellow" + } + }, + "advance_warnings": { + "name": "Advance warnings", + "unit_of_measurement": "warnings" + }, "warning_level": { "name": "Warning level", "state": { diff --git a/homeassistant/components/geosphere_austria_warnings/warnings.py b/homeassistant/components/geosphere_austria_warnings/warnings.py new file mode 100644 index 000000000000..b864b8d05cd0 --- /dev/null +++ b/homeassistant/components/geosphere_austria_warnings/warnings.py @@ -0,0 +1,118 @@ +"""Shared helpers for GeoSphere Austria weather warnings.""" + +from collections.abc import Iterable, Mapping +from datetime import datetime +from typing import Any + +from pygeosphere_warnings import WarningLevel, WarningType, WeatherWarning + +LEVEL_NONE = "none" + +WARNING_TYPE_SLUGS: Mapping[WarningType, str] = { + WarningType.STORM: "storm", + WarningType.RAIN: "rain", + WarningType.SNOW: "snow", + WarningType.BLACK_ICE: "black_ice", + WarningType.THUNDERSTORM: "thunderstorm", + WarningType.HEAT: "heat", + WarningType.COLD: "cold", +} + +WARNING_LEVEL_SLUGS: Mapping[WarningLevel, str] = { + WarningLevel.YELLOW: "yellow", + WarningLevel.ORANGE: "orange", + WarningLevel.RED: "red", +} + +SUSTAINED_WARNING_TYPES: frozenset[WarningType] = frozenset( + {WarningType.HEAT, WarningType.COLD} +) +RANKING_DEMOTION = 1 + + +def warning_type_slug(warning_type: WarningType) -> str: + """Return the stable slug for a warning type.""" + return WARNING_TYPE_SLUGS[warning_type] + + +def warning_level_slug(level: WarningLevel) -> str: + """Return the stable slug for a warning level.""" + return WARNING_LEVEL_SLUGS[level] + + +def _ranking_level(warning: WeatherWarning) -> int: + """Return the level value used for ordering, adjusted for warning type. + + Sustained/ambient types (heat, cold) are demoted by one tier relative to + acute/event types when ranking warnings against each other, since they + describe a background condition rather than something to react to right + now. This never affects the warning's actual, reported level. + """ + demotion = ( + RANKING_DEMOTION if warning.warning_type in SUSTAINED_WARNING_TYPES else 0 + ) + return warning.level.value - demotion + + +def warning_sort_key( + warning: WeatherWarning, +) -> tuple[int, datetime, datetime, int, int, int, str]: + """Return the deterministic ordering key for a warning. + + Warnings with a higher ranking level sort first. For equal ranking levels, the + warning with the earliest end time sorts first, followed by its start time. + The remaining fields provide stable tie-breakers. + """ + return ( + -_ranking_level(warning), + warning.end, + warning.start, + warning.warning_id, + warning.change_id, + warning.course_id, + warning_type_slug(warning.warning_type), + ) + + +def sort_warnings(warnings: Iterable[WeatherWarning]) -> list[WeatherWarning]: + """Return warnings in deterministic, actionability-ranked order. + + This ordering is used for display and for selecting the "featured" + warning; it is intentionally not pure severity order. For the true + worst-case severity, use ``highest_warning_level`` instead. + """ + return sorted(warnings, key=warning_sort_key) + + +def warning_sensor_attributes( + warnings: Iterable[WeatherWarning], +) -> dict[str, Any]: + """Return the agreed attributes for the selected warning. + + ``warnings`` are expected in descending priority order. The full warning + payload is intentionally not exposed on sensor entities. + """ + warning = next(iter(warnings), None) + if warning is None: + return {} + + return { + "type": warning_type_slug(warning.warning_type), + "level": warning_level_slug(warning.level), + "start": warning.start.isoformat(), + "end": warning.end.isoformat(), + "warning_id": warning.warning_id, + } + + +def highest_warning_level(warnings: Iterable[WeatherWarning]) -> str: + """Return the highest *actual* warning level, or ``none`` for an empty bucket. + + Deliberately independent of ``sort_warnings``' type-adjusted ranking: the + reported level must reflect true worst-case severity, e.g. a red heat + warning must never be masked by a concurrent yellow thunderstorm. + """ + levels = [warning.level.value for warning in warnings] + if not levels: + return LEVEL_NONE + return warning_level_slug(WarningLevel(max(levels))) diff --git a/tests/components/geosphere_austria_warnings/fixtures/get_warnings_for_coords.json b/tests/components/geosphere_austria_warnings/fixtures/get_warnings_for_coords.json index 5df61a5e7ce7..8f0c1d01ec5d 100644 --- a/tests/components/geosphere_austria_warnings/fixtures/get_warnings_for_coords.json +++ b/tests/components/geosphere_austria_warnings/fixtures/get_warnings_for_coords.json @@ -14,17 +14,41 @@ } }, "warnings": [ + { + "type": "Warning", + "properties": { + "warnid": 10, + "chgid": 1, + "verlaufid": 11, + "warntypid": 6, + "begin": "27.03.2023 00:00", + "end": "27.03.2023 23:59", + "create": "2023-03-26 12:00:00+00", + "text": "Enhanced heat stress can be expected.", + "auswirkungen": "", + "empfehlungen": "", + "meteotext": "", + "updategrund": "", + "warnstufeid": 1, + "rawinfo": { + "wtype": 6, + "wlevel": 1, + "start": "1679868000", + "end": "1679954340" + } + } + }, { "type": "Warning", "properties": { "warnid": 4149, "chgid": 6, - "verlaufid": 2, + "verlaufid": 12, "warntypid": 1, "begin": "27.03.2023 08:00", "end": "27.03.2023 18:00", "create": "2023-03-27 06:00:00+00", - "text": "Orange wind warning from Mon, 27.03.2023 08:00 until Mon, 27.03.2023 18:00", + "text": "Orange storm warning from Mon, 27.03.2023 08:00 until Mon, 27.03.2023 18:00", "auswirkungen": "* Branches may fall and objects may be thrown around.", "empfehlungen": "* Be careful in forests, parks and avenues!", "meteotext": "Strong northwest winds with gusts between 60 and 80 km/h.", @@ -43,12 +67,84 @@ "properties": { "warnid": 4150, "chgid": 2, - "verlaufid": 1, + "verlaufid": 52, "warntypid": 2, + "begin": "28.03.2023 06:00", + "end": "28.03.2023 16:00", + "create": "2023-03-27 08:00:00+00", + "text": "Orange rain warning from Tue, 28.03.2023 06:00 until Tue, 28.03.2023 16:00", + "auswirkungen": "", + "empfehlungen": "", + "meteotext": "", + "updategrund": "", + "warnstufeid": 2, + "rawinfo": { + "wtype": 2, + "wlevel": 2, + "start": "1679976000", + "end": "1680012000" + } + } + }, + { + "type": "Warning", + "properties": { + "warnid": 4149, + "chgid": 6, + "verlaufid": 31, + "warntypid": 1, "begin": "28.03.2023 08:00", "end": "28.03.2023 18:00", + "create": "2023-03-27 06:00:00+00", + "text": "Orange storm warning from Tue, 28.03.2023 08:00 until Tue, 28.03.2023 18:00", + "auswirkungen": "", + "empfehlungen": "", + "meteotext": "", + "updategrund": "", + "warnstufeid": 2, + "rawinfo": { + "wtype": 1, + "wlevel": 2, + "start": "1679983200", + "end": "1680019200" + } + } + }, + { + "type": "Warning", + "properties": { + "warnid": 10, + "chgid": 1, + "verlaufid": 51, + "warntypid": 6, + "begin": "29.03.2023 00:00", + "end": "29.03.2023 23:59", + "create": "2023-03-28 12:00:00+00", + "text": "Enhanced heat stress can be expected.", + "auswirkungen": "", + "empfehlungen": "", + "meteotext": "", + "updategrund": "", + "warnstufeid": 2, + "rawinfo": { + "wtype": 6, + "wlevel": 2, + "start": "1680040800", + "end": "1680127140" + } + } + }, + { + "type": "Warning", + "properties": { + "warnid": 4150, + "chgid": 2, + "verlaufid": 61, + "warntypid": 2, + "begin": "29.03.2023 08:00", + "end": "29.03.2023 18:00", "create": "2023-03-27 08:00:00+00", - "text": "Yellow rain warning from Tue, 28.03.2023 08:00 until Tue, 28.03.2023 18:00", + "text": "Yellow rain warning from Wed, 29.03.2023 08:00 until Wed, 29.03.2023 18:00", "auswirkungen": "* Local flooding is possible.", "empfehlungen": "* Avoid underpasses and flooded roads!", "meteotext": "Persistent rain with amounts between 40 and 60 mm.", @@ -57,8 +153,32 @@ "rawinfo": { "wtype": 2, "wlevel": 1, - "start": "1679983200", - "end": "1680019200" + "start": "1680069600", + "end": "1680105600" + } + } + }, + { + "type": "Warning", + "properties": { + "warnid": 4837, + "chgid": 1, + "verlaufid": 2, + "warntypid": 5, + "begin": "29.03.2023 14:00", + "end": "29.03.2023 21:00", + "create": "2023-03-29 12:00:00+00", + "text": "Yellow thunderstorm warning from Wed, 29.03.2023 14:00 until Wed, 29.03.2023 21:00", + "auswirkungen": "* Caution! Heightened likelihood of thunderstorms. From region to region, heavy rainfall and storm-strength winds can be expected. The major danger stems from strokes of lightning, falling tree branches and loosed flying objects. From place to place, inundations are possible.", + "empfehlungen": "", + "meteotext": "", + "updategrund": "", + "warnstufeid": 1, + "rawinfo": { + "wtype": 5, + "wlevel": 1, + "start": "1680091200", + "end": "1680116400" } } } diff --git a/tests/components/geosphere_austria_warnings/snapshots/test_sensor.ambr b/tests/components/geosphere_austria_warnings/snapshots/test_sensor.ambr index 5efaa53eaaae..9c5eb04cd83a 100644 --- a/tests/components/geosphere_austria_warnings/snapshots/test_sensor.ambr +++ b/tests/components/geosphere_austria_warnings/snapshots/test_sensor.ambr @@ -1,5 +1,5 @@ # serializer version: 1 -# name: test_sensors.4 +# name: test_sensor_set_and_active_warning.8 DeviceRegistryEntrySnapshot({ 'area_id': None, 'config_entry_id': , @@ -29,7 +29,7 @@ 'via_device_id': None, }) # --- -# name: test_sensors[sensor.schwechat_active_warnings-entry] +# name: test_sensor_set_and_active_warning[sensor.schwechat_active_warnings-entry] EntityRegistryEntrySnapshot({ 'aliases': list([ None, @@ -68,7 +68,7 @@ 'unit_of_measurement': 'warnings', }) # --- -# name: test_sensors[sensor.schwechat_active_warnings-state] +# name: test_sensor_set_and_active_warning[sensor.schwechat_active_warnings-state] StateSnapshot({ 'attributes': ReadOnlyDict({ : 'Data provided by GeoSphere Austria', @@ -81,10 +81,135 @@ 'last_changed': , 'last_reported': , 'last_updated': , - 'state': '1', + 'state': '2', }) # --- -# name: test_sensors[sensor.schwechat_warning_level-entry] +# name: test_sensor_set_and_active_warning[sensor.schwechat_advance_warning_level-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : list([ + 'none', + 'yellow', + 'orange', + 'red', + ]), + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.schwechat_advance_warning_level', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Advance warning level', + 'options': dict({ + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Advance warning level', + 'platform': 'geosphere_austria_warnings', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'advance_warning_level', + 'unique_id': '30740-advance_warning_level', + 'unit_of_measurement': None, + }) +# --- +# name: test_sensor_set_and_active_warning[sensor.schwechat_advance_warning_level-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'Data provided by GeoSphere Austria', + : 'enum', + 'end': '2023-03-28T14:00:00+00:00', + : 'Schwechat Advance warning level', + 'level': 'orange', + : list([ + 'none', + 'yellow', + 'orange', + 'red', + ]), + 'start': '2023-03-28T04:00:00+00:00', + 'type': 'rain', + 'warning_id': 4150, + }), + 'context': , + 'entity_id': 'sensor.schwechat_advance_warning_level', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'orange', + }) +# --- +# name: test_sensor_set_and_active_warning[sensor.schwechat_advance_warnings-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.schwechat_advance_warnings', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Advance warnings', + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': 'Advance warnings', + 'platform': 'geosphere_austria_warnings', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'advance_warnings', + 'unique_id': '30740-advance_warnings', + 'unit_of_measurement': 'warnings', + }) +# --- +# name: test_sensor_set_and_active_warning[sensor.schwechat_advance_warnings-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'Data provided by GeoSphere Austria', + : 'Schwechat Advance warnings', + : , + : 'warnings', + }), + 'context': , + 'entity_id': 'sensor.schwechat_advance_warnings', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '5', + }) +# --- +# name: test_sensor_set_and_active_warning[sensor.schwechat_warning_level-entry] EntityRegistryEntrySnapshot({ 'aliases': list([ None, @@ -128,18 +253,23 @@ 'unit_of_measurement': None, }) # --- -# name: test_sensors[sensor.schwechat_warning_level-state] +# name: test_sensor_set_and_active_warning[sensor.schwechat_warning_level-state] StateSnapshot({ 'attributes': ReadOnlyDict({ : 'Data provided by GeoSphere Austria', : 'enum', + 'end': '2023-03-27T16:00:00+00:00', : 'Schwechat Warning level', + 'level': 'orange', : list([ 'none', 'yellow', 'orange', 'red', ]), + 'start': '2023-03-27T06:00:00+00:00', + 'type': 'storm', + 'warning_id': 4149, }), 'context': , 'entity_id': 'sensor.schwechat_warning_level', diff --git a/tests/components/geosphere_austria_warnings/test_sensor.py b/tests/components/geosphere_austria_warnings/test_sensor.py index 236e9c59fdda..fc44361784cc 100644 --- a/tests/components/geosphere_austria_warnings/test_sensor.py +++ b/tests/components/geosphere_austria_warnings/test_sensor.py @@ -1,5 +1,6 @@ """Tests for the GeoSphere Austria Warnings sensors.""" +from typing import Any from unittest.mock import AsyncMock from freezegun.api import FrozenDateTimeFactory @@ -12,7 +13,7 @@ from homeassistant.components.geosphere_austria_warnings.coordinator import ( UPDATE_INTERVAL, ) from homeassistant.const import STATE_UNAVAILABLE -from homeassistant.core import HomeAssistant +from homeassistant.core import HomeAssistant, State from homeassistant.helpers import device_registry as dr, entity_registry as er from . import setup_integration @@ -21,20 +22,72 @@ from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_plat pytestmark = pytest.mark.usefixtures("mock_client") +WARNING_LEVEL_ENTITY_ID = "sensor.schwechat_warning_level" ACTIVE_WARNINGS_ENTITY_ID = "sensor.schwechat_active_warnings" +ADVANCE_WARNING_LEVEL_ENTITY_ID = "sensor.schwechat_advance_warning_level" +ADVANCE_WARNINGS_ENTITY_ID = "sensor.schwechat_advance_warnings" + +EXPECTED_ENTITY_IDS = { + WARNING_LEVEL_ENTITY_ID, + ACTIVE_WARNINGS_ENTITY_ID, + ADVANCE_WARNING_LEVEL_ENTITY_ID, + ADVANCE_WARNINGS_ENTITY_ID, +} + +WARNING_DETAIL_ATTRIBUTE_KEYS = {"type", "start", "end", "warning_id"} -@pytest.mark.freeze_time("2023-03-27 12:00:00+00:00") -async def test_sensors( +def warning_details(state: State) -> dict[str, Any]: + """Return integration-specific warning attributes only.""" + return { + key: value + for key, value in state.attributes.items() + if key in WARNING_DETAIL_ATTRIBUTE_KEYS + } + + +@pytest.mark.freeze_time("2023-03-27 12:00:00+02:00") +async def test_sensor_set_and_active_warning( hass: HomeAssistant, mock_config_entry: MockConfigEntry, entity_registry: er.EntityRegistry, device_registry: dr.DeviceRegistry, snapshot: SnapshotAssertion, ) -> None: - """Test the state of the sensors while a warning is active.""" + """Test the complete sensor set while a warning is active.""" await setup_integration(hass, mock_config_entry) + entity_ids = { + entity_id + for entity_id in hass.states.async_entity_ids("sensor") + if entity_id.startswith("sensor.schwechat_") + } + assert entity_ids == EXPECTED_ENTITY_IDS + + assert (state := hass.states.get(WARNING_LEVEL_ENTITY_ID)) + assert state.state == "orange" + assert warning_details(state) == { + "type": "storm", + "start": "2023-03-27T06:00:00+00:00", + "end": "2023-03-27T16:00:00+00:00", + "warning_id": 4149, + } + + assert (state := hass.states.get(ACTIVE_WARNINGS_ENTITY_ID)) + assert state.state == "2" + + assert (state := hass.states.get(ADVANCE_WARNING_LEVEL_ENTITY_ID)) + assert state.state == "orange" + assert warning_details(state) == { + "type": "rain", + "start": "2023-03-28T04:00:00+00:00", + "end": "2023-03-28T14:00:00+00:00", + "warning_id": 4150, + } + + assert (state := hass.states.get(ADVANCE_WARNINGS_ENTITY_ID)) + assert state.state == "5" + await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id) device_entry = device_registry.async_get_device_by_identifier( @@ -44,21 +97,35 @@ async def test_sensors( assert device_entry == snapshot -@pytest.mark.freeze_time("2023-03-27 20:00:00+00:00") +@pytest.mark.freeze_time("2023-03-28 00:00:00+02:00") async def test_sensors_without_active_warning( hass: HomeAssistant, mock_config_entry: MockConfigEntry, ) -> None: - """Test the state of the sensors when no warning is active.""" + """Test the sensor states and empty attributes when no warning is active.""" await setup_integration(hass, mock_config_entry) - assert (state := hass.states.get("sensor.schwechat_warning_level")) + assert (state := hass.states.get(WARNING_LEVEL_ENTITY_ID)) assert state.state == "none" + assert warning_details(state) == {} + assert (state := hass.states.get(ACTIVE_WARNINGS_ENTITY_ID)) assert state.state == "0" + assert (state := hass.states.get(ADVANCE_WARNING_LEVEL_ENTITY_ID)) + assert state.state == "orange" + assert warning_details(state) == { + "type": "rain", + "start": "2023-03-28T04:00:00+00:00", + "end": "2023-03-28T14:00:00+00:00", + "warning_id": 4150, + } -@pytest.mark.freeze_time("2023-03-27 12:00:00+00:00") + assert (state := hass.states.get(ADVANCE_WARNINGS_ENTITY_ID)) + assert state.state == "5" + + +@pytest.mark.freeze_time("2023-03-27 12:00:00+02:00") async def test_entities_unavailable_on_error( hass: HomeAssistant, mock_client: AsyncMock, @@ -67,12 +134,15 @@ async def test_entities_unavailable_on_error( ) -> None: """Test that entities become unavailable when the update fails.""" await setup_integration(hass, mock_config_entry) - assert (state := hass.states.get(ACTIVE_WARNINGS_ENTITY_ID)) - assert state.state != STATE_UNAVAILABLE + for entity_id in EXPECTED_ENTITY_IDS: + assert (state := hass.states.get(entity_id)) + assert state.state != STATE_UNAVAILABLE mock_client.get_last_modified.side_effect = GeoSphereConnectionError freezer.tick(UPDATE_INTERVAL) async_fire_time_changed(hass) await hass.async_block_till_done() - assert (state := hass.states.get(ACTIVE_WARNINGS_ENTITY_ID)) - assert state.state == STATE_UNAVAILABLE + + for entity_id in EXPECTED_ENTITY_IDS: + assert (state := hass.states.get(entity_id)) + assert state.state == STATE_UNAVAILABLE diff --git a/tests/components/geosphere_austria_warnings/test_warnings.py b/tests/components/geosphere_austria_warnings/test_warnings.py new file mode 100644 index 000000000000..82a959e85b13 --- /dev/null +++ b/tests/components/geosphere_austria_warnings/test_warnings.py @@ -0,0 +1,250 @@ +"""Tests for shared GeoSphere Austria warning helpers.""" + +from datetime import UTC, datetime + +from pygeosphere_warnings import ( + LocationWarnings, + WarningLevel, + WarningType, + WeatherWarning, +) +import pytest + +from homeassistant.components.geosphere_austria_warnings.const import DOMAIN +from homeassistant.components.geosphere_austria_warnings.warnings import ( + LEVEL_NONE, + highest_warning_level, + sort_warnings, + warning_sensor_attributes, +) + +from tests.common import load_json_object_fixture + + +@pytest.fixture +def warnings() -> list[WeatherWarning]: + """Return warnings parsed from the deterministic API fixture.""" + location_warnings = LocationWarnings.from_api( + load_json_object_fixture("get_warnings_for_coords.json", DOMAIN) + ) + return location_warnings.warnings + + +def test_sort_warnings_is_deterministic(warnings: list[WeatherWarning]) -> None: + """Test actionability ranking is independent of source order. + + The fixture contains no overlapping warnings of the same type and level. + Equal-ranked warnings are ordered by their end time: orange rain before + orange storm, and yellow thunderstorm before demoted orange heat. + """ + sorted_warnings = sort_warnings(warnings) + sorted_again = sort_warnings(reversed(warnings)) + + expected = [ + (4149, 12), + (4150, 52), + (4149, 31), + (4150, 61), + (4837, 2), + (10, 51), + (10, 11), + ] + assert [ + (warning.warning_id, warning.course_id) for warning in sorted_warnings + ] == expected + assert [ + (warning.warning_id, warning.course_id) for warning in sorted_again + ] == expected + + +def test_sort_warnings_uses_severity_then_end_time( + warnings: list[WeatherWarning], +) -> None: + """Test orange warnings outrank yellow and earlier end wins within orange.""" + selected = sort_warnings(warnings)[0] + + assert selected.level == WarningLevel.ORANGE + assert selected.warning_type == WarningType.STORM + assert selected.end.isoformat() == "2023-03-27T16:00:00+00:00" + + +def test_highest_warning_level(warnings: list[WeatherWarning]) -> None: + """Test highest-level values and the empty-bucket value.""" + assert highest_warning_level(warnings) == "orange" + assert highest_warning_level([]) == LEVEL_NONE + + +def test_sort_warnings_prefers_storm_over_concurrent_heat( + warnings: list[WeatherWarning], +) -> None: + """Test acute orange storm wins over concurrent demoted yellow heat.""" + at = datetime(2023, 3, 27, 10, 0, tzinfo=UTC) + concurrent_warnings = [ + warning for warning in warnings if warning.start <= at < warning.end + ] + + assert {warning.warning_type for warning in concurrent_warnings} == { + WarningType.HEAT, + WarningType.STORM, + } + + selected = sort_warnings(concurrent_warnings)[0] + + assert selected.course_id == 12 + assert selected.warning_type == WarningType.STORM + + +def test_sort_warnings_demotes_sustained_heat_below_acute_thunderstorm( + warnings: list[WeatherWarning], +) -> None: + """Test all-day orange heat ranks below concurrent yellow thunderstorm.""" + heat_and_thunderstorm = [ + warning + for warning in warnings + if warning.warning_type in {WarningType.HEAT, WarningType.THUNDERSTORM} + ] + + sorted_warnings = sort_warnings(heat_and_thunderstorm) + + assert [warning.course_id for warning in sorted_warnings] == [2, 51, 11] + + +def test_highest_warning_level_ignores_type_demotion( + warnings: list[WeatherWarning], +) -> None: + """Test severity reflects orange heat despite its display demotion.""" + heat_and_thunderstorm = [ + warning + for warning in warnings + if warning.warning_type in {WarningType.HEAT, WarningType.THUNDERSTORM} + ] + + assert highest_warning_level(heat_and_thunderstorm) == "orange" + + +def test_sort_warnings_prefers_acute_over_sustained( + warnings: list[WeatherWarning], +) -> None: + """Test a concurrent yellow thunderstorm wins over orange all-day heat.""" + at = datetime(2023, 3, 29, 12, 0, tzinfo=UTC) + concurrent_warnings = [ + warning + for warning in warnings + if warning.start <= at < warning.end + and warning.warning_type in {WarningType.HEAT, WarningType.THUNDERSTORM} + ] + + assert {warning.warning_type for warning in concurrent_warnings} == { + WarningType.HEAT, + WarningType.THUNDERSTORM, + } + + selected = sort_warnings(concurrent_warnings)[0] + + assert selected.course_id == 2 + assert selected.warning_type == WarningType.THUNDERSTORM + + +def test_ranking_tie_between_equal_levels_prefers_soonest_end() -> None: + """Test that the earliest end time breaks ties between equally ranked warnings.""" + all_day_heat = WeatherWarning( + warning_id=100, + change_id=1, + course_id=1, + warning_type=WarningType.HEAT, + level=WarningLevel.ORANGE, + start=datetime(2023, 3, 26, 22, 0, tzinfo=UTC), + end=datetime(2023, 3, 27, 21, 59, tzinfo=UTC), + text="", + impacts="", + recommendations="", + meteo_text="", + update_reason="", + ) + afternoon_thunderstorm = WeatherWarning( + warning_id=200, + change_id=1, + course_id=1, + warning_type=WarningType.THUNDERSTORM, + level=WarningLevel.YELLOW, + start=datetime(2023, 3, 27, 11, 0, tzinfo=UTC), + end=datetime(2023, 3, 27, 13, 0, tzinfo=UTC), + text="", + impacts="", + recommendations="", + meteo_text="", + update_reason="", + ) + + selected = sort_warnings([all_day_heat, afternoon_thunderstorm])[0] + + assert selected.warning_id == 200 + assert highest_warning_level([all_day_heat, afternoon_thunderstorm]) == "orange" + + +def test_warning_sensor_attributes_are_flat_and_minimal( + warnings: list[WeatherWarning], +) -> None: + """Test that sensor attributes expose only the selected warning details.""" + sorted_warnings = sort_warnings(warnings) + + assert warning_sensor_attributes(sorted_warnings) == { + "type": "storm", + "level": "orange", + "start": "2023-03-27T06:00:00+00:00", + "end": "2023-03-27T16:00:00+00:00", + "warning_id": 4149, + } + + attributes = warning_sensor_attributes(sorted_warnings) + assert set(attributes) == {"type", "level", "start", "end", "warning_id"} + assert warning_sensor_attributes([]) == {} + + +def test_warning_sensor_attributes_include_diverging_warning_level() -> None: + """Test that sensor attributes expose only the selected warning details.""" + all_day_heat = WeatherWarning( + warning_id=100, + change_id=1, + course_id=1, + warning_type=WarningType.HEAT, + level=WarningLevel.ORANGE, + start=datetime(2023, 3, 26, 22, 0, tzinfo=UTC), + end=datetime(2023, 3, 27, 21, 59, tzinfo=UTC), + text="", + impacts="", + recommendations="", + meteo_text="", + update_reason="", + ) + afternoon_thunderstorm = WeatherWarning( + warning_id=200, + change_id=1, + course_id=1, + warning_type=WarningType.THUNDERSTORM, + level=WarningLevel.YELLOW, + start=datetime(2023, 3, 27, 11, 0, tzinfo=UTC), + end=datetime(2023, 3, 27, 13, 0, tzinfo=UTC), + text="", + impacts="", + recommendations="", + meteo_text="", + update_reason="", + ) + + all_warnings = [all_day_heat, afternoon_thunderstorm] + sorted_warnings = sort_warnings(all_warnings) + + selected = sorted_warnings[0] + assert selected.warning_id == 200 # thunderstorm wins the tie + + assert warning_sensor_attributes(sorted_warnings) == { + "type": "thunderstorm", + "start": "2023-03-27T11:00:00+00:00", + "end": "2023-03-27T13:00:00+00:00", + "warning_id": 200, + "level": "yellow", + } + + attributes = warning_sensor_attributes(sorted_warnings) + assert set(attributes) == {"type", "start", "end", "warning_id", "level"}