starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-26 09:23:17 -04:00
Code Issues Packages Projects Releases Wiki Activity

Geosphere austria add advance warnings sensors (#178474)

Browse Source 
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
a7rk
2026-09-25 13:02:04 +02:00
committed by GitHub
co-authored by Copilot Autofix powered by AI
parent d59489473d
commit 02537a2f12
9 changed files with 785 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/geosphere_austria_warnings/coordinator.py
+7 -2
View File
@@ -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
),
)
homeassistant/components/geosphere_austria_warnings/icons.json
+6
View File
@@ -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"
}
homeassistant/components/geosphere_austria_warnings/sensor.py
+47 -15
View File
@@ -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)
homeassistant/components/geosphere_austria_warnings/strings.json
+13
View File
@@ -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": {
homeassistant/components/geosphere_austria_warnings/warnings.py
+118
View File
@@ -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)))
tests/components/geosphere_austria_warnings/fixtures/get_warnings_for_coords.json
+126 -6
View File
@@ -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"
}
}
}
tests/components/geosphere_austria_warnings/snapshots/test_sensor.ambr
+136 -6
View File
@@ -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': <ANY>,
@@ -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({
<EntityStateAttribute.ATTRIBUTION: 'attribution'>: 'Data provided by GeoSphere Austria',
@@ -81,10 +81,135 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'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({
<SensorEntityCapabilityAttribute.OPTIONS: 'options'>: list([
'none',
'yellow',
'orange',
'red',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'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': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Advance warning level',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'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({
<EntityStateAttribute.ATTRIBUTION: 'attribution'>: 'Data provided by GeoSphere Austria',
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'enum',
'end': '2023-03-28T14:00:00+00:00',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Schwechat Advance warning level',
'level': 'orange',
<SensorEntityCapabilityAttribute.OPTIONS: 'options'>: list([
'none',
'yellow',
'orange',
'red',
]),
'start': '2023-03-28T04:00:00+00:00',
'type': 'rain',
'warning_id': 4150,
}),
'context': <ANY>,
'entity_id': 'sensor.schwechat_advance_warning_level',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'orange',
})
# ---
# name: test_sensor_set_and_active_warning[sensor.schwechat_advance_warnings-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.schwechat_advance_warnings',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'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({
<EntityStateAttribute.ATTRIBUTION: 'attribution'>: 'Data provided by GeoSphere Austria',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Schwechat Advance warnings',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: 'warnings',
}),
'context': <ANY>,
'entity_id': 'sensor.schwechat_advance_warnings',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'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({
<EntityStateAttribute.ATTRIBUTION: 'attribution'>: 'Data provided by GeoSphere Austria',
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'enum',
'end': '2023-03-27T16:00:00+00:00',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Schwechat Warning level',
'level': 'orange',
<SensorEntityCapabilityAttribute.OPTIONS: 'options'>: list([
'none',
'yellow',
'orange',
'red',
]),
'start': '2023-03-27T06:00:00+00:00',
'type': 'storm',
'warning_id': 4149,
}),
'context': <ANY>,
'entity_id': 'sensor.schwechat_warning_level',
tests/components/geosphere_austria_warnings/test_sensor.py
+82 -12
View File
@@ -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
tests/components/geosphere_austria_warnings/test_warnings.py
+250
View File
@@ -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"}