1
0
mirror of https://github.com/home-assistant/core.git synced 2026-04-02 08:26:41 +01:00

Replace NINA attributes with sensors (#161882)

This commit is contained in:
DeerMaximum
2026-04-01 19:53:28 +00:00
committed by GitHub
parent f09602363c
commit 2881916c91
19 changed files with 2679 additions and 185 deletions

View File

@@ -18,7 +18,7 @@ from .const import (
)
from .coordinator import NinaConfigEntry, NINADataUpdateCoordinator
PLATFORMS: list[str] = [Platform.BINARY_SENSOR]
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: NinaConfigEntry) -> bool:

View File

@@ -1,4 +1,4 @@
"""NINA sensor platform."""
"""NINA binary sensor platform."""
from __future__ import annotations
@@ -88,15 +88,19 @@ class NINAMessage(NinaEntity, BinarySensorEntity):
data = self._get_warning_data()
return {
ATTR_HEADLINE: data.headline,
ATTR_DESCRIPTION: data.description,
ATTR_SENDER: data.sender,
ATTR_SEVERITY: data.severity,
ATTR_RECOMMENDED_ACTIONS: data.recommended_actions,
ATTR_AFFECTED_AREAS: data.affected_areas,
ATTR_WEB: data.web,
ATTR_HEADLINE: data.headline, # Deprecated, remove in 2026.11
ATTR_DESCRIPTION: data.description, # Deprecated, remove in 2026.11
ATTR_SENDER: data.sender, # Deprecated, remove in 2026.11
ATTR_SEVERITY: data.severity or "Unknown", # Deprecated, remove in 2026.11
ATTR_RECOMMENDED_ACTIONS: data.recommended_actions, # Deprecated, remove in 2026.11
ATTR_AFFECTED_AREAS: data.affected_areas, # Deprecated, remove in 2026.11
ATTR_WEB: data.more_info_url, # Deprecated, remove in 2026.11
ATTR_ID: data.id,
ATTR_SENT: data.sent,
ATTR_START: data.start,
ATTR_EXPIRES: data.expires,
ATTR_SENT: data.sent.isoformat(), # Deprecated, remove in 2026.11
ATTR_START: data.start.isoformat()
if data.start
else "", # Deprecated, remove in 2026.11
ATTR_EXPIRES: data.expires.isoformat()
if data.expires
else "", # Deprecated, remove in 2026.11
}

View File

@@ -31,6 +31,7 @@ from .const import (
CONST_REGIONS,
DOMAIN,
NO_MATCH_REGEX,
SENSOR_SUFFIXES,
)
@@ -243,32 +244,7 @@ class OptionsFlowHandler(OptionsFlowWithReload):
user_input, self._all_region_codes_sorted
)
entity_registry = er.async_get(self.hass)
entries = er.async_entries_for_config_entry(
entity_registry, self.config_entry.entry_id
)
removed_entities_slots = [
f"{region}-{slot_id}"
for region in self.data[CONF_REGIONS]
for slot_id in range(self.data[CONF_MESSAGE_SLOTS] + 1)
if slot_id > user_input[CONF_MESSAGE_SLOTS]
]
removed_entities_area = [
f"{cfg_region}-{slot_id}"
for slot_id in range(1, self.data[CONF_MESSAGE_SLOTS] + 1)
for cfg_region in self.data[CONF_REGIONS]
if cfg_region not in user_input[CONF_REGIONS]
]
for entry in entries:
for entity_uid in list(
set(removed_entities_slots + removed_entities_area)
):
if entry.unique_id == entity_uid:
entity_registry.async_remove(entry.entity_id)
await self.remove_unused_entities(user_input)
self.hass.config_entries.async_update_entry(
self.config_entry, data=user_input
@@ -287,3 +263,35 @@ class OptionsFlowHandler(OptionsFlowWithReload):
data_schema=schema_with_suggested,
errors=errors,
)
async def remove_unused_entities(self, user_input: dict[str, Any]) -> None:
"""Remove entities which are not used anymore."""
entity_registry = er.async_get(self.hass)
entries = er.async_entries_for_config_entry(
entity_registry, self.config_entry.entry_id
)
id_type_suffix = [f"-{sensor_id}" for sensor_id in SENSOR_SUFFIXES] + [""]
removed_entities_slots = [
f"{region}-{slot_id}{suffix}"
for region in self.data[CONF_REGIONS]
for slot_id in range(self.data[CONF_MESSAGE_SLOTS] + 1)
for suffix in id_type_suffix
if slot_id > user_input[CONF_MESSAGE_SLOTS]
]
removed_entities_area = [
f"{cfg_region}-{slot_id}{suffix}"
for slot_id in range(1, self.data[CONF_MESSAGE_SLOTS] + 1)
for cfg_region in self.data[CONF_REGIONS]
for suffix in id_type_suffix
if cfg_region not in user_input[CONF_REGIONS]
]
removed_uids = set(removed_entities_slots + removed_entities_area)
for entry in entries:
if entry.unique_id in removed_uids:
entity_registry.async_remove(entry.entity_id)

View File

@@ -15,6 +15,8 @@ DOMAIN: str = "nina"
NO_MATCH_REGEX: str = "/(?!)/"
ALL_MATCH_REGEX: str = ".*"
SEVERITY_VALUES: list[str] = ["extreme", "severe", "moderate", "minor", "unknown"]
CONF_REGIONS: str = "regions"
CONF_MESSAGE_SLOTS: str = "slots"
CONF_FILTERS: str = "filters"
@@ -34,6 +36,17 @@ ATTR_SENT: str = "sent"
ATTR_START: str = "start"
ATTR_EXPIRES: str = "expires"
SENSOR_SUFFIXES: list[str] = [
"headline",
"sender",
"severity",
"affected_areas",
"more_info_url",
"sent",
"start",
"expires",
]
CONST_LIST_A_TO_D: list[str] = ["A", "Ä", "B", "C", "D"]
CONST_LIST_E_TO_H: list[str] = ["E", "F", "G", "H"]
CONST_LIST_I_TO_L: list[str] = ["I", "J", "K", "L"]

View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio
from dataclasses import dataclass
from datetime import datetime
import re
from typing import Any
@@ -12,7 +13,6 @@ from pynina import ApiError, Nina
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
@@ -36,13 +36,14 @@ class NinaWarningData:
headline: str
description: str
sender: str
severity: str
severity: str | None
recommended_actions: str
affected_areas_short: str
affected_areas: str
web: str
sent: str
start: str
expires: str
more_info_url: str
sent: datetime
start: datetime | None
expires: datetime | None
is_valid: bool
@@ -65,12 +66,6 @@ class NINADataUpdateCoordinator(
]
self.area_filter: str = config_entry.data[CONF_FILTERS][CONF_AREA_FILTER]
self.device_info = DeviceInfo(
identifiers={(DOMAIN, config_entry.entry_id)},
manufacturer="NINA",
entry_type=DeviceEntryType.SERVICE,
)
regions: dict[str, str] = config_entry.data[CONF_REGIONS]
for region in regions:
self._nina.add_region(region)
@@ -146,18 +141,33 @@ class NINADataUpdateCoordinator(
)
continue
shortened_affected_areas: str = (
affected_areas_string[0:250] + "..."
if len(affected_areas_string) > 250
else affected_areas_string
)
severity = (
None
if raw_warn.severity.lower() == "unknown"
else raw_warn.severity
)
warning_data: NinaWarningData = NinaWarningData(
raw_warn.id,
raw_warn.headline,
raw_warn.description,
raw_warn.sender,
raw_warn.severity,
raw_warn.sender or "",
severity,
" ".join([str(action) for action in raw_warn.recommended_actions]),
shortened_affected_areas,
affected_areas_string,
raw_warn.web or "",
raw_warn.sent or "",
raw_warn.start or "",
raw_warn.expires or "",
datetime.fromisoformat(raw_warn.sent),
datetime.fromisoformat(raw_warn.start) if raw_warn.start else None,
datetime.fromisoformat(raw_warn.expires)
if raw_warn.expires
else None,
raw_warn.is_valid,
)
warnings_for_regions.append(warning_data)

View File

@@ -1,7 +1,9 @@
"""NINA common entity."""
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import NINADataUpdateCoordinator, NinaWarningData
@@ -20,12 +22,18 @@ class NinaEntity(CoordinatorEntity[NINADataUpdateCoordinator]):
self._region = region
self._warning_index = slot_id - 1
self._region_name = region_name
self._attr_translation_placeholders = {
"region_name": region_name,
"slot_id": str(slot_id),
}
self._attr_device_info = coordinator.device_info
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self._region)},
manufacturer="NINA",
name=self._region_name,
entry_type=DeviceEntryType.SERVICE,
)
def _get_active_warnings_count(self) -> int:
"""Return the number of active warnings for the region."""

View File

@@ -62,23 +62,17 @@ rules:
docs-supported-devices:
status: exempt
comment: |
This integration does not use devices.
docs-supported-functions: todo
This integration exposes Home Assistant devices only for logical grouping and does not integrate specific physical devices that need to be documented as supported hardware.
docs-supported-functions: done
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: done
entity-category: todo
entity-device-class:
status: todo
comment: |
Extract attributes into own entities.
entity-category: done
entity-device-class: done
entity-disabled-by-default: done
entity-translations: todo
entity-translations: done
exception-translations: todo
icon-translations:
status: exempt
comment: |
This integration does not custom icons.
icon-translations: todo
reconfiguration-flow: todo
repair-issues:
status: exempt

View File

@@ -0,0 +1,159 @@
"""NINA sensor platform."""
from __future__ import annotations
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from datetime import datetime
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import CONF_MESSAGE_SLOTS, CONF_REGIONS, SENSOR_SUFFIXES, SEVERITY_VALUES
from .coordinator import NinaConfigEntry, NINADataUpdateCoordinator, NinaWarningData
from .entity import NinaEntity
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class NinaSensorEntityDescription(SensorEntityDescription):
"""Describes NINA sensor entity."""
value_fn: Callable[[NinaWarningData], str | datetime | None]
SENSOR_TYPES: tuple[NinaSensorEntityDescription, ...] = (
NinaSensorEntityDescription(
key=SENSOR_SUFFIXES[0],
translation_key="headline",
value_fn=lambda data: data.headline,
),
NinaSensorEntityDescription(
key=SENSOR_SUFFIXES[1],
translation_key="sender",
value_fn=lambda data: data.sender,
),
NinaSensorEntityDescription(
key=SENSOR_SUFFIXES[2],
options=SEVERITY_VALUES,
device_class=SensorDeviceClass.ENUM,
translation_key="severity",
value_fn=lambda data: (
data.severity.lower() if data.severity is not None else None
),
),
NinaSensorEntityDescription(
key=SENSOR_SUFFIXES[3],
translation_key="affected_areas",
value_fn=lambda data: data.affected_areas_short,
),
NinaSensorEntityDescription(
key=SENSOR_SUFFIXES[4],
translation_key="more_info_url",
value_fn=lambda data: data.more_info_url,
),
NinaSensorEntityDescription(
key=SENSOR_SUFFIXES[5],
translation_key="sent",
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=lambda data: data.sent,
),
NinaSensorEntityDescription(
key=SENSOR_SUFFIXES[6],
translation_key="start",
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=lambda data: data.start,
),
NinaSensorEntityDescription(
key=SENSOR_SUFFIXES[7],
translation_key="expires",
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=lambda data: data.expires,
),
)
def create_sensors_for_warning(
coordinator: NINADataUpdateCoordinator, region: str, region_name: str, slot_id: int
) -> Sequence[NinaSensor]:
"""Create sensors for a warning."""
return [
NinaSensor(
coordinator,
region,
region_name,
slot_id,
description,
)
for description in SENSOR_TYPES
]
async def async_setup_entry(
_: HomeAssistant,
config_entry: NinaConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the NINA sensor platform."""
coordinator = config_entry.runtime_data
regions: dict[str, str] = config_entry.data[CONF_REGIONS]
message_slots: int = config_entry.data[CONF_MESSAGE_SLOTS]
entities = [
create_sensors_for_warning(coordinator, ent, regions[ent], i + 1)
for ent in coordinator.data
for i in range(message_slots)
]
async_add_entities(
[entity for slot_entities in entities for entity in slot_entities]
)
class NinaSensor(NinaEntity, SensorEntity):
"""Representation of a NINA sensor."""
_attr_has_entity_name = True
_attr_entity_category = EntityCategory.DIAGNOSTIC
entity_description: NinaSensorEntityDescription
def __init__(
self,
coordinator: NINADataUpdateCoordinator,
region: str,
region_name: str,
slot_id: int,
description: NinaSensorEntityDescription,
) -> None:
"""Initialize."""
super().__init__(coordinator, region, region_name, slot_id)
self.entity_description = description
self._attr_unique_id = f"{region}-{slot_id}-{self.entity_description.key}"
@property
def available(self) -> bool:
"""Return if entity is available."""
if self._get_active_warnings_count() <= self._warning_index:
return False
return self._get_warning_data().is_valid and super().available
@property
def native_value(self) -> str | datetime | None:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self._get_warning_data())

View File

@@ -48,7 +48,39 @@
"entity": {
"binary_sensor": {
"warning": {
"name": "Warning: {region_name} {slot_id}"
"name": "Warning {slot_id}"
}
},
"sensor": {
"affected_areas": {
"name": "Affected areas {slot_id}"
},
"expires": {
"name": "Expires {slot_id}"
},
"headline": {
"name": "Headline {slot_id}"
},
"more_info_url": {
"name": "More information URL {slot_id}"
},
"sender": {
"name": "Sender {slot_id}"
},
"sent": {
"name": "Sent {slot_id}"
},
"severity": {
"name": "Severity {slot_id}",
"state": {
"extreme": "Extreme",
"minor": "Minor",
"moderate": "Moderate",
"severe": "Severe"
}
},
"start": {
"name": "Start {slot_id}"
}
}
},

View File

@@ -1,12 +1,13 @@
"""Tests for the Nina integration."""
from copy import deepcopy
from unittest.mock import AsyncMock
from unittest.mock import AsyncMock, patch
from pynina import Warning
from homeassistant.components.nina.const import CONF_REGIONS
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@@ -18,7 +19,7 @@ async def setup_platform(
mock_nina_class: AsyncMock,
nina_warnings: list[Warning],
) -> None:
"""Set up the NINA platform."""
"""Set up the NINA platforms."""
mock_nina_class.warnings = {
region: deepcopy(nina_warnings)
for region in config_entry.data.get(CONF_REGIONS, {})
@@ -28,3 +29,25 @@ async def setup_platform(
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
async def setup_single_platform(
hass: HomeAssistant,
config_entry: MockConfigEntry,
platform: Platform | None,
mock_nina_class: AsyncMock,
nina_warnings: list[Warning],
) -> None:
"""Set up a single NINA platform."""
mock_nina_class.warnings = {
region: deepcopy(nina_warnings)
for region in config_entry.data.get(CONF_REGIONS, {})
}
platforms = [platform] if platform else []
with patch("homeassistant.components.nina.PLATFORMS", platforms):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED

View File

@@ -10,7 +10,11 @@ import pytest
from homeassistant.components.nina.const import DOMAIN
from homeassistant.core import HomeAssistant
from .const import DUMMY_CONFIG_ENTRY
from .const import (
DUMMY_CONFIG_ENTRY,
DUMMY_CONFIG_ENTRY_AREA_FILTERS,
DUMMY_CONFIG_ENTRY_DEFAULT_FILTERS,
)
from tests.common import (
MockConfigEntry,
@@ -44,6 +48,38 @@ def mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
return config_entry
@pytest.fixture
def mock_config_entry_default_filter(hass: HomeAssistant) -> MockConfigEntry:
"""Provide a common mock config entry with no filters."""
config_entry = MockConfigEntry(
domain=DOMAIN,
title="NINA",
data=deepcopy(DUMMY_CONFIG_ENTRY_DEFAULT_FILTERS),
version=1,
minor_version=3,
)
config_entry.add_to_hass(hass)
return config_entry
@pytest.fixture
def mock_config_entry_area_filter(hass: HomeAssistant) -> MockConfigEntry:
"""Provide a common mock config entry with an area filter."""
config_entry = MockConfigEntry(
domain=DOMAIN,
title="NINA",
data=deepcopy(DUMMY_CONFIG_ENTRY_AREA_FILTERS),
version=1,
minor_version=3,
)
config_entry.add_to_hass(hass)
return config_entry
@pytest.fixture
def mock_nina_class(nina_region_codes: dict[str, str]) -> Generator[AsyncMock]:
"""Fixture to mock the NINA class."""

View File

@@ -37,3 +37,21 @@ DUMMY_CONFIG_ENTRY: dict[str, Any] = {
CONST_REGION_A_TO_D: deepcopy(DUMMY_USER_INPUT[CONST_REGION_A_TO_D]),
CONF_REGIONS: {"095760000000": "Aach"},
}
DUMMY_CONFIG_ENTRY_DEFAULT_FILTERS: dict[str, Any] = {
CONF_MESSAGE_SLOTS: 5,
CONF_REGIONS: {"083350000000": "Aach, Stadt"},
CONF_FILTERS: {
CONF_HEADLINE_FILTER: "/(?!)/",
CONF_AREA_FILTER: ".*",
},
}
DUMMY_CONFIG_ENTRY_AREA_FILTERS: dict[str, Any] = {
CONF_MESSAGE_SLOTS: 5,
CONF_REGIONS: {"083350000000": "Aach, Stadt"},
CONF_FILTERS: {
CONF_HEADLINE_FILTER: "/(?!)/",
CONF_AREA_FILTER: ".*nagold.*",
},
}

View File

@@ -23,7 +23,9 @@
"affected_areas": [
"Gemeinde Oberreichenbach, Gemeinde Neuweiler, Stadt Nagold, Stadt Neubulach, Gemeinde Schömberg, Gemeinde Simmersfeld, Gemeinde Simmozheim, Gemeinde Rohrdorf, Gemeinde Ostelsheim, Gemeinde Ebhausen, Gemeinde Egenhausen, Gemeinde Dobel, Stadt Bad Liebenzell, Stadt Solingen, Stadt Haiterbach, Stadt Bad Herrenalb, Gemeinde Höfen an der Enz, Gemeinde Gechingen, Gemeinde Enzklösterle, Gemeinde Gutach (Schwarzwaldbahn) und 3392 weitere."
],
"recommended_actions": [],
"recommended_actions": [
"ACHTUNG! Hinweis auf mögliche Gefahren: Es können zum Beispiel einzelne Äste herabstürzen. Achte besonders auf herabfallende Gegenstände."
],
"web": "https://www.wettergefahren.de",
"sent": "2021-10-11T05:20:00+01:00",
"start": "2021-11-01T05:20:00+01:00",

View File

@@ -1,5 +1,5 @@
# serializer version: 1
# name: test_sensors[binary_sensor.nina_warning_aach_1-entry]
# name: test_binary_sensors[binary_sensor.aach_warning_1-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
@@ -13,7 +13,7 @@
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.nina_warning_aach_1',
'entity_id': 'binary_sensor.aach_warning_1',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
@@ -21,12 +21,12 @@
'labels': set({
}),
'name': None,
'object_id_base': 'Warning: Aach 1',
'object_id_base': 'Warning 1',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.SAFETY: 'safety'>,
'original_icon': None,
'original_name': 'Warning: Aach 1',
'original_name': 'Warning 1',
'platform': 'nina',
'previous_unique_id': None,
'suggested_object_id': None,
@@ -36,17 +36,17 @@
'unit_of_measurement': None,
})
# ---
# name: test_sensors[binary_sensor.nina_warning_aach_1-state]
# name: test_binary_sensors[binary_sensor.aach_warning_1-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'affected_areas': 'Gemeinde Oberreichenbach, Gemeinde Neuweiler, Stadt Nagold, Stadt Neubulach, Gemeinde Schömberg, Gemeinde Simmersfeld, Gemeinde Simmozheim, Gemeinde Rohrdorf, Gemeinde Ostelsheim, Gemeinde Ebhausen, Gemeinde Egenhausen, Gemeinde Dobel, Stadt Bad Liebenzell, Stadt Solingen, Stadt Haiterbach, Stadt Bad Herrenalb, Gemeinde Höfen an der Enz, Gemeinde Gechingen, Gemeinde Enzklösterle, Gemeinde Gutach (Schwarzwaldbahn) und 3392 weitere.',
'description': 'Es treten Sturmböen mit Geschwindigkeiten zwischen 70 km/h (20m/s, 38kn, Bft 8) und 85 km/h (24m/s, 47kn, Bft 9) aus westlicher Richtung auf. In Schauernähe sowie in exponierten Lagen muss mit schweren Sturmböen bis 90 km/h (25m/s, 48kn, Bft 10) gerechnet werden.',
'device_class': 'safety',
'expires': '3021-11-22T05:19:00+01:00',
'friendly_name': 'NINA Warning: Aach 1',
'friendly_name': 'Aach Warning 1',
'headline': 'Ausfall Notruf 112',
'id': 'mow.DE-NW-BN-SE030-20201014-30-000',
'recommended_actions': '',
'recommended_actions': 'ACHTUNG! Hinweis auf mögliche Gefahren: Es können zum Beispiel einzelne Äste herabstürzen. Achte besonders auf herabfallende Gegenstände.',
'sender': 'Deutscher Wetterdienst',
'sent': '2021-10-11T05:20:00+01:00',
'severity': 'Minor',
@@ -54,14 +54,14 @@
'web': 'https://www.wettergefahren.de',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.nina_warning_aach_1',
'entity_id': 'binary_sensor.aach_warning_1',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_sensors[binary_sensor.nina_warning_aach_2-entry]
# name: test_binary_sensors[binary_sensor.aach_warning_2-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
@@ -75,7 +75,7 @@
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.nina_warning_aach_2',
'entity_id': 'binary_sensor.aach_warning_2',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
@@ -83,12 +83,12 @@
'labels': set({
}),
'name': None,
'object_id_base': 'Warning: Aach 2',
'object_id_base': 'Warning 2',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.SAFETY: 'safety'>,
'original_icon': None,
'original_name': 'Warning: Aach 2',
'original_name': 'Warning 2',
'platform': 'nina',
'previous_unique_id': None,
'suggested_object_id': None,
@@ -98,21 +98,21 @@
'unit_of_measurement': None,
})
# ---
# name: test_sensors[binary_sensor.nina_warning_aach_2-state]
# name: test_binary_sensors[binary_sensor.aach_warning_2-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'safety',
'friendly_name': 'NINA Warning: Aach 2',
'friendly_name': 'Aach Warning 2',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.nina_warning_aach_2',
'entity_id': 'binary_sensor.aach_warning_2',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_sensors[binary_sensor.nina_warning_aach_3-entry]
# name: test_binary_sensors[binary_sensor.aach_warning_3-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
@@ -126,7 +126,7 @@
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.nina_warning_aach_3',
'entity_id': 'binary_sensor.aach_warning_3',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
@@ -134,12 +134,12 @@
'labels': set({
}),
'name': None,
'object_id_base': 'Warning: Aach 3',
'object_id_base': 'Warning 3',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.SAFETY: 'safety'>,
'original_icon': None,
'original_name': 'Warning: Aach 3',
'original_name': 'Warning 3',
'platform': 'nina',
'previous_unique_id': None,
'suggested_object_id': None,
@@ -149,21 +149,21 @@
'unit_of_measurement': None,
})
# ---
# name: test_sensors[binary_sensor.nina_warning_aach_3-state]
# name: test_binary_sensors[binary_sensor.aach_warning_3-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'safety',
'friendly_name': 'NINA Warning: Aach 3',
'friendly_name': 'Aach Warning 3',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.nina_warning_aach_3',
'entity_id': 'binary_sensor.aach_warning_3',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_sensors[binary_sensor.nina_warning_aach_4-entry]
# name: test_binary_sensors[binary_sensor.aach_warning_4-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
@@ -177,7 +177,7 @@
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.nina_warning_aach_4',
'entity_id': 'binary_sensor.aach_warning_4',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
@@ -185,12 +185,12 @@
'labels': set({
}),
'name': None,
'object_id_base': 'Warning: Aach 4',
'object_id_base': 'Warning 4',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.SAFETY: 'safety'>,
'original_icon': None,
'original_name': 'Warning: Aach 4',
'original_name': 'Warning 4',
'platform': 'nina',
'previous_unique_id': None,
'suggested_object_id': None,
@@ -200,21 +200,21 @@
'unit_of_measurement': None,
})
# ---
# name: test_sensors[binary_sensor.nina_warning_aach_4-state]
# name: test_binary_sensors[binary_sensor.aach_warning_4-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'safety',
'friendly_name': 'NINA Warning: Aach 4',
'friendly_name': 'Aach Warning 4',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.nina_warning_aach_4',
'entity_id': 'binary_sensor.aach_warning_4',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_sensors[binary_sensor.nina_warning_aach_5-entry]
# name: test_binary_sensors[binary_sensor.aach_warning_5-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
@@ -228,7 +228,7 @@
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.nina_warning_aach_5',
'entity_id': 'binary_sensor.aach_warning_5',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
@@ -236,12 +236,12 @@
'labels': set({
}),
'name': None,
'object_id_base': 'Warning: Aach 5',
'object_id_base': 'Warning 5',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.SAFETY: 'safety'>,
'original_icon': None,
'original_name': 'Warning: Aach 5',
'original_name': 'Warning 5',
'platform': 'nina',
'previous_unique_id': None,
'suggested_object_id': None,
@@ -251,14 +251,14 @@
'unit_of_measurement': None,
})
# ---
# name: test_sensors[binary_sensor.nina_warning_aach_5-state]
# name: test_binary_sensors[binary_sensor.aach_warning_5-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'safety',
'friendly_name': 'NINA Warning: Aach 5',
'friendly_name': 'Aach Warning 5',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.nina_warning_aach_5',
'entity_id': 'binary_sensor.aach_warning_5',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,

View File

@@ -5,31 +5,33 @@
'095760000000': list([
dict({
'affected_areas': 'Gemeinde Oberreichenbach, Gemeinde Neuweiler, Stadt Nagold, Stadt Neubulach, Gemeinde Schömberg, Gemeinde Simmersfeld, Gemeinde Simmozheim, Gemeinde Rohrdorf, Gemeinde Ostelsheim, Gemeinde Ebhausen, Gemeinde Egenhausen, Gemeinde Dobel, Stadt Bad Liebenzell, Stadt Solingen, Stadt Haiterbach, Stadt Bad Herrenalb, Gemeinde Höfen an der Enz, Gemeinde Gechingen, Gemeinde Enzklösterle, Gemeinde Gutach (Schwarzwaldbahn) und 3392 weitere.',
'affected_areas_short': 'Gemeinde Oberreichenbach, Gemeinde Neuweiler, Stadt Nagold, Stadt Neubulach, Gemeinde Schömberg, Gemeinde Simmersfeld, Gemeinde Simmozheim, Gemeinde Rohrdorf, Gemeinde Ostelsheim, Gemeinde Ebhausen, Gemeinde Egenhausen, Gemeinde Dobel, Stadt Bad Lieb...',
'description': 'Es treten Sturmböen mit Geschwindigkeiten zwischen 70 km/h (20m/s, 38kn, Bft 8) und 85 km/h (24m/s, 47kn, Bft 9) aus westlicher Richtung auf. In Schauernähe sowie in exponierten Lagen muss mit schweren Sturmböen bis 90 km/h (25m/s, 48kn, Bft 10) gerechnet werden.',
'expires': '3021-11-22T05:19:00+01:00',
'headline': 'Ausfall Notruf 112',
'id': 'mow.DE-NW-BN-SE030-20201014-30-000',
'is_valid': True,
'recommended_actions': '',
'more_info_url': 'https://www.wettergefahren.de',
'recommended_actions': 'ACHTUNG! Hinweis auf mögliche Gefahren: Es können zum Beispiel einzelne Äste herabstürzen. Achte besonders auf herabfallende Gegenstände.',
'sender': 'Deutscher Wetterdienst',
'sent': '2021-10-11T05:20:00+01:00',
'severity': 'Minor',
'start': '2021-11-01T05:20:00+01:00',
'web': 'https://www.wettergefahren.de',
}),
dict({
'affected_areas': 'Axstedt, Gnarrenburg, Grasberg, Hagen im Bremischen, Hambergen, Hepstedt, Holste, Lilienthal, Lübberstedt, Osterholz-Scharmbeck, Ritterhude, Schwanewede, Vollersode, Worpswede',
'affected_areas_short': 'Axstedt, Gnarrenburg, Grasberg, Hagen im Bremischen, Hambergen, Hepstedt, Holste, Lilienthal, Lübberstedt, Osterholz-Scharmbeck, Ritterhude, Schwanewede, Vollersode, Worpswede',
'description': 'In Beverstedt im Landkreis Cuxhaven ist am 20. Juli 2022 in einer Geflügelhaltung der Ausbruch der Geflügelpest (Vogelgrippe, Aviäre Influenza) amtlich festgestellt worden. Durch die geografische Nähe des Ausbruchsbetriebes zum Gebiet des Landkreises Osterholz musste das Veterinäramt des Landkreises zum Schutz vor einer Ausbreitung der Geflügelpest auch für sein Gebiet ein Restriktionsgebiet festlegen. Rund um den Ausbruchsort wurde eine Überwachungszone ausgewiesen. Eine entsprechende Tierseuchenbehördliche Allgemeinverfügung wurde vom Landkreis Osterholz erlassen und tritt am 23.07.2022 in Kraft.<br>\xa0<br>Die Überwachungszone mit einem Radius von mindestens zehn Kilometern um den Ausbruchsbetrieb erstreckt sich im Landkreis Osterholz innerhalb der Samtgemeinde Hambergen auf die Mitgliedsgemeinden Axstedt, Holste und Lübberstedt. Die vorgenannten Gemeinden sind vollständig zur Überwachungszone erklärt worden. Der genaue Grenzverlauf des Gebietes kann auch der interaktiven Karte im Internet entnommen werden.<br>\xa0<br>In der Überwachungszone liegen im Landkreis Osterholz rund 70 Geflügelhaltungen mit einem Gesamtbestand von rund 1.800 Tieren. Sie alle unterliegen mit der Allgemeinverfügung der sogenannten amtlichen Beobachtung. Für die Betriebe sind die Biosicherheitsmaßnahmen einzuhalten. Dazu zählen insbesondere Hygienemaßnahmen im laufenden Betrieb und eine ordnungsgemäße Schadnagerbekämpfung.<br>\xa0<br>Das Verbringen von Vögeln, Fleisch von Geflügel, Eiern und sonstige Nebenprodukte von Geflügel in und aus Betrieben in der Überwachungszone ist verboten. Auch Geflügeltransporte sind in der Überwachungszone verboten. Jeder Verdacht der Erkrankung auf Geflügelpest ist zudem dem Veterinäramt des Landkreises Osterholz unter der E-Mail-Adresse veterinaeramt@landkreis-osterholz.de sofort zu melden. Alle Hinweise, die innerhalb der Überwachungszone zu beachten sind, sind unter www.landkreis-osterholz.de/gefluegelpest zusammengefasst dargestellt.<br>\xa0<br>Die Veterinärbehörde weist zudem darauf hin, dass sämtliche Geflügelhaltungen Hühner, Enten, Gänse, Fasane, Perlhühner, Rebhühner, Truthühner, Wachteln oder Laufvögel der zuständigen Behörde angezeigt werden müssen. Wer dies bisher noch nicht gemacht hat und über keine Registriernummer für seinen Geflügelbestand verfügt, sollte die Meldung über das Veterinäramt umgehend nachholen.<br>\xa0<br>Das Beobachtungsgebiet kann frühestens 30 Tage nach der Grobreinigung des Ausbruchsbetriebes wieder aufgehoben werden. Hierüber wird der Landkreis Osterholz informieren.<br>\xa0<br>Die Allgemeinverfügung, eine Übersicht zur Überwachungszone und weitere Hinweise sind auf der Internetseite unter www.landkreis-osterholz.de/gefluegelpest zu finden.',
'expires': '2002-08-07T10:59:00+02:00',
'headline': 'Geflügelpest im Landkreis Cuxhaven - Teile des Landkreises Osterholz zur Überwachungszone erklärt',
'id': 'biw.BIWAPP-69634',
'is_valid': False,
'more_info_url': '',
'recommended_actions': '',
'sender': None,
'sender': '',
'sent': '1999-08-07T10:59:00+02:00',
'severity': 'Minor',
'start': '',
'web': '',
'start': None,
}),
]),
}),

File diff suppressed because it is too large Load Diff

View File

@@ -2,48 +2,22 @@
from __future__ import annotations
from typing import Any
from unittest.mock import AsyncMock
from pynina import Warning
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.nina.const import (
ATTR_HEADLINE,
CONF_AREA_FILTER,
CONF_FILTERS,
CONF_HEADLINE_FILTER,
CONF_MESSAGE_SLOTS,
CONF_REGIONS,
DOMAIN,
)
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.components.nina.const import ATTR_HEADLINE
from homeassistant.const import STATE_OFF, STATE_ON, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_platform
from . import setup_single_platform
from tests.common import MockConfigEntry, snapshot_platform
ENTRY_DATA_NO_CORONA: dict[str, Any] = {
CONF_MESSAGE_SLOTS: 5,
CONF_REGIONS: {"083350000000": "Aach, Stadt"},
CONF_FILTERS: {
CONF_HEADLINE_FILTER: "/(?!)/",
CONF_AREA_FILTER: ".*",
},
}
ENTRY_DATA_SPECIFIC_AREA: dict[str, Any] = {
CONF_MESSAGE_SLOTS: 5,
CONF_REGIONS: {"083350000000": "Aach, Stadt"},
CONF_FILTERS: {
CONF_HEADLINE_FILTER: "/(?!)/",
CONF_AREA_FILTER: ".*nagold.*",
},
}
async def test_sensors(
async def test_binary_sensors(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
@@ -51,32 +25,31 @@ async def test_sensors(
mock_nina_class: AsyncMock,
nina_warnings: list[Warning],
) -> None:
"""Test the creation and values of the NINA sensors."""
await setup_platform(hass, mock_config_entry, mock_nina_class, nina_warnings)
"""Test the creation and values of the NINA binary sensors."""
await setup_single_platform(
hass, mock_config_entry, Platform.BINARY_SENSOR, mock_nina_class, nina_warnings
)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_sensors_without_corona_filter(
async def test_binary_sensors_without_corona_filter(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
mock_config_entry_default_filter: MockConfigEntry,
mock_nina_class: AsyncMock,
nina_warnings: list[Warning],
) -> None:
"""Test the creation and values of the NINA sensors without the corona filter."""
"""Test the creation and values of the NINA binary sensors without the corona filter."""
conf_entry: MockConfigEntry = MockConfigEntry(
domain=DOMAIN,
title="NINA",
data=ENTRY_DATA_NO_CORONA,
version=1,
minor_version=3,
await setup_single_platform(
hass,
mock_config_entry_default_filter,
Platform.BINARY_SENSOR,
mock_nina_class,
nina_warnings,
)
conf_entry.add_to_hass(hass)
await setup_platform(hass, conf_entry, mock_nina_class, nina_warnings)
state_w1 = hass.states.get("binary_sensor.nina_warning_aach_stadt_1")
state_w1 = hass.states.get("binary_sensor.aach_stadt_warning_1")
assert state_w1.state == STATE_ON
assert (
@@ -84,61 +57,58 @@ async def test_sensors_without_corona_filter(
== "Corona-Verordnung des Landes: Warnstufe durch Landesgesundheitsamt ausgerufen"
)
state_w2 = hass.states.get("binary_sensor.nina_warning_aach_stadt_2")
state_w2 = hass.states.get("binary_sensor.aach_stadt_warning_2")
assert state_w2.state == STATE_ON
assert state_w2.attributes.get(ATTR_HEADLINE) == "Ausfall Notruf 112"
state_w3 = hass.states.get("binary_sensor.nina_warning_aach_stadt_3")
state_w3 = hass.states.get("binary_sensor.aach_stadt_warning_3")
assert state_w3.state == STATE_OFF
assert state_w3.state == STATE_OFF # Warning expired
state_w4 = hass.states.get("binary_sensor.nina_warning_aach_stadt_4")
state_w4 = hass.states.get("binary_sensor.aach_stadt_warning_4")
assert state_w4.state == STATE_OFF
state_w5 = hass.states.get("binary_sensor.nina_warning_aach_stadt_5")
state_w5 = hass.states.get("binary_sensor.aach_stadt_warning_5")
assert state_w5.state == STATE_OFF
async def test_sensors_with_area_filter(
async def test_binary_sensors_with_area_filter(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
mock_config_entry_area_filter: MockConfigEntry,
mock_nina_class: AsyncMock,
nina_warnings: list[Warning],
) -> None:
"""Test the creation and values of the NINA sensors with a restrictive area filter."""
"""Test the creation and values of the NINA binary sensors with a restrictive area filter."""
conf_entry: MockConfigEntry = MockConfigEntry(
domain=DOMAIN,
title="NINA",
data=ENTRY_DATA_SPECIFIC_AREA,
version=1,
minor_version=3,
await setup_single_platform(
hass,
mock_config_entry_area_filter,
Platform.BINARY_SENSOR,
mock_nina_class,
nina_warnings,
)
conf_entry.add_to_hass(hass)
await setup_platform(hass, conf_entry, mock_nina_class, nina_warnings)
state_w1 = hass.states.get("binary_sensor.nina_warning_aach_stadt_1")
state_w1 = hass.states.get("binary_sensor.aach_stadt_warning_1")
assert state_w1.state == STATE_ON
assert state_w1.attributes.get(ATTR_HEADLINE) == "Ausfall Notruf 112"
state_w2 = hass.states.get("binary_sensor.nina_warning_aach_stadt_2")
state_w2 = hass.states.get("binary_sensor.aach_stadt_warning_2")
assert state_w2.state == STATE_OFF
state_w3 = hass.states.get("binary_sensor.nina_warning_aach_stadt_3")
state_w3 = hass.states.get("binary_sensor.aach_stadt_warning_3")
assert state_w3.state == STATE_OFF
state_w4 = hass.states.get("binary_sensor.nina_warning_aach_stadt_4")
state_w4 = hass.states.get("binary_sensor.aach_stadt_warning_4")
assert state_w4.state == STATE_OFF
state_w5 = hass.states.get("binary_sensor.nina_warning_aach_stadt_5")
state_w5 = hass.states.get("binary_sensor.aach_stadt_warning_5")
assert state_w5.state == STATE_OFF

View File

@@ -6,7 +6,7 @@ from copy import deepcopy
from typing import Any
from unittest.mock import AsyncMock
from pynina import ApiError
from pynina import ApiError, Warning
from homeassistant.components.nina.const import (
CONF_AREA_FILTER,
@@ -21,6 +21,7 @@ from homeassistant.components.nina.const import (
CONST_REGION_R_TO_U,
CONST_REGION_V_TO_Z,
DOMAIN,
SENSOR_SUFFIXES,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.core import HomeAssistant
@@ -285,6 +286,17 @@ async def test_options_flow_entity_removal(
"""Test if old entities are removed."""
await setup_platform(hass, mock_config_entry, mock_nina_class, nina_warnings)
entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
entities_per_slot = len(SENSOR_SUFFIXES) + 1
assert (
len(entries)
== mock_config_entry.data.get(CONF_MESSAGE_SLOTS) * entities_per_slot
)
result = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
new_slot_count = 2
@@ -309,4 +321,4 @@ async def test_options_flow_entity_removal(
entity_registry, mock_config_entry.entry_id
)
assert len(entries) == new_slot_count
assert len(entries) == new_slot_count * entities_per_slot

View File

@@ -0,0 +1,107 @@
"""Test the Nina sensor."""
from __future__ import annotations
from unittest.mock import AsyncMock
from pynina import Warning
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_single_platform
from tests.common import MockConfigEntry, snapshot_platform
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_sensors(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
mock_nina_class: AsyncMock,
nina_warnings: list[Warning],
) -> None:
"""Test the creation and values of the NINA sensors."""
await setup_single_platform(
hass, mock_config_entry, Platform.SENSOR, mock_nina_class, nina_warnings
)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_sensors_without_corona_filter(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry_default_filter: MockConfigEntry,
mock_nina_class: AsyncMock,
nina_warnings: list[Warning],
) -> None:
"""Test the creation and values of the NINA sensors without the corona filter."""
await setup_single_platform(
hass,
mock_config_entry_default_filter,
Platform.SENSOR,
mock_nina_class,
nina_warnings,
)
state_w1 = hass.states.get("sensor.aach_stadt_severity_1")
assert state_w1.state == "minor"
state_w2 = hass.states.get("sensor.aach_stadt_severity_2")
assert state_w2.state == "minor"
state_w3 = hass.states.get("sensor.aach_stadt_severity_3")
assert state_w3.state == STATE_UNAVAILABLE # Warning expired
state_w4 = hass.states.get("sensor.aach_stadt_severity_4")
assert state_w4.state == STATE_UNAVAILABLE
state_w5 = hass.states.get("sensor.aach_stadt_severity_5")
assert state_w5.state == STATE_UNAVAILABLE
async def test_sensors_with_area_filter(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry_area_filter: MockConfigEntry,
mock_nina_class: AsyncMock,
nina_warnings: list[Warning],
) -> None:
"""Test the creation and values of the NINA sensors with a restrictive area filter."""
await setup_single_platform(
hass,
mock_config_entry_area_filter,
Platform.SENSOR,
mock_nina_class,
nina_warnings,
)
state_w1 = hass.states.get("sensor.aach_stadt_severity_1")
assert state_w1.state == "minor"
state_w2 = hass.states.get("sensor.aach_stadt_severity_2")
assert state_w2.state == STATE_UNAVAILABLE
state_w3 = hass.states.get("sensor.aach_stadt_severity_3")
assert state_w3.state == STATE_UNAVAILABLE
state_w4 = hass.states.get("sensor.aach_stadt_severity_4")
assert state_w4.state == STATE_UNAVAILABLE
state_w5 = hass.states.get("sensor.aach_stadt_severity_5")
assert state_w5.state == STATE_UNAVAILABLE