Replace NINA corona filter with regex (#83181)
* Change headline filter to regex * Add config migration * Update config flowpull/95441/head
parent
36b0fc17df
commit
0bfb81ecf3
|
@ -2,6 +2,7 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from async_timeout import timeout
|
||||
|
@ -13,7 +14,15 @@ from homeassistant.core import HomeAssistant
|
|||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import _LOGGER, CONF_FILTER_CORONA, CONF_REGIONS, DOMAIN, SCAN_INTERVAL
|
||||
from .const import (
|
||||
_LOGGER,
|
||||
CONF_FILTER_CORONA,
|
||||
CONF_HEADLINE_FILTER,
|
||||
CONF_REGIONS,
|
||||
DOMAIN,
|
||||
NO_MATCH_REGEX,
|
||||
SCAN_INTERVAL,
|
||||
)
|
||||
|
||||
PLATFORMS: list[str] = [Platform.BINARY_SENSOR]
|
||||
|
||||
|
@ -23,8 +32,18 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||
|
||||
regions: dict[str, str] = entry.data[CONF_REGIONS]
|
||||
|
||||
if CONF_HEADLINE_FILTER not in entry.data:
|
||||
filter_regex = NO_MATCH_REGEX
|
||||
|
||||
if entry.data[CONF_FILTER_CORONA]:
|
||||
filter_regex = ".*corona.*"
|
||||
|
||||
new_data = {**entry.data, CONF_HEADLINE_FILTER: filter_regex}
|
||||
new_data.pop(CONF_FILTER_CORONA, None)
|
||||
hass.config_entries.async_update_entry(entry, data=new_data)
|
||||
|
||||
coordinator = NINADataUpdateCoordinator(
|
||||
hass, regions, entry.data[CONF_FILTER_CORONA]
|
||||
hass, regions, entry.data[CONF_HEADLINE_FILTER]
|
||||
)
|
||||
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
@ -70,12 +89,12 @@ class NINADataUpdateCoordinator(
|
|||
"""Class to manage fetching NINA data API."""
|
||||
|
||||
def __init__(
|
||||
self, hass: HomeAssistant, regions: dict[str, str], corona_filter: bool
|
||||
self, hass: HomeAssistant, regions: dict[str, str], headline_filter: str
|
||||
) -> None:
|
||||
"""Initialize."""
|
||||
self._regions: dict[str, str] = regions
|
||||
self._nina: Nina = Nina(async_get_clientsession(hass))
|
||||
self.corona_filter: bool = corona_filter
|
||||
self.headline_filter: str = headline_filter
|
||||
|
||||
for region in regions:
|
||||
self._nina.addRegion(region)
|
||||
|
@ -125,7 +144,9 @@ class NINADataUpdateCoordinator(
|
|||
warnings_for_regions: list[NinaWarningData] = []
|
||||
|
||||
for raw_warn in raw_warnings:
|
||||
if "corona" in raw_warn.headline.lower() and self.corona_filter:
|
||||
if re.search(
|
||||
self.headline_filter, raw_warn.headline, flags=re.IGNORECASE
|
||||
):
|
||||
continue
|
||||
|
||||
warning_data: NinaWarningData = NinaWarningData(
|
||||
|
|
|
@ -18,12 +18,13 @@ from homeassistant.helpers.entity_registry import (
|
|||
|
||||
from .const import (
|
||||
_LOGGER,
|
||||
CONF_FILTER_CORONA,
|
||||
CONF_HEADLINE_FILTER,
|
||||
CONF_MESSAGE_SLOTS,
|
||||
CONF_REGIONS,
|
||||
CONST_REGION_MAPPING,
|
||||
CONST_REGIONS,
|
||||
DOMAIN,
|
||||
NO_MATCH_REGEX,
|
||||
)
|
||||
|
||||
|
||||
|
@ -125,6 +126,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
if group_input := user_input.get(group):
|
||||
user_input[CONF_REGIONS] += group_input
|
||||
|
||||
if not user_input[CONF_HEADLINE_FILTER]:
|
||||
user_input[CONF_HEADLINE_FILTER] = NO_MATCH_REGEX
|
||||
|
||||
if user_input[CONF_REGIONS]:
|
||||
return self.async_create_entry(
|
||||
title="NINA",
|
||||
|
@ -144,7 +148,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
vol.Required(CONF_MESSAGE_SLOTS, default=5): vol.All(
|
||||
int, vol.Range(min=1, max=20)
|
||||
),
|
||||
vol.Required(CONF_FILTER_CORONA, default=True): cv.boolean,
|
||||
vol.Optional(CONF_HEADLINE_FILTER, default=""): cv.string,
|
||||
}
|
||||
),
|
||||
errors=errors,
|
||||
|
@ -255,10 +259,10 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
|
|||
CONF_MESSAGE_SLOTS,
|
||||
default=self.data[CONF_MESSAGE_SLOTS],
|
||||
): vol.All(int, vol.Range(min=1, max=20)),
|
||||
vol.Required(
|
||||
CONF_FILTER_CORONA,
|
||||
default=self.data[CONF_FILTER_CORONA],
|
||||
): cv.boolean,
|
||||
vol.Optional(
|
||||
CONF_HEADLINE_FILTER,
|
||||
default=self.data[CONF_HEADLINE_FILTER],
|
||||
): cv.string,
|
||||
}
|
||||
),
|
||||
errors=errors,
|
||||
|
|
|
@ -11,9 +11,12 @@ SCAN_INTERVAL: timedelta = timedelta(minutes=5)
|
|||
|
||||
DOMAIN: str = "nina"
|
||||
|
||||
NO_MATCH_REGEX: str = "/(?!)/"
|
||||
|
||||
CONF_REGIONS: str = "regions"
|
||||
CONF_MESSAGE_SLOTS: str = "slots"
|
||||
CONF_FILTER_CORONA: str = "corona_filter"
|
||||
CONF_FILTER_CORONA: str = "corona_filter" # deprecated
|
||||
CONF_HEADLINE_FILTER: str = "headline_filter"
|
||||
|
||||
ATTR_HEADLINE: str = "headline"
|
||||
ATTR_DESCRIPTION: str = "description"
|
||||
|
|
|
@ -11,7 +11,7 @@
|
|||
"_r_to_u": "City/county (R-U)",
|
||||
"_v_to_z": "City/county (V-Z)",
|
||||
"slots": "Maximum warnings per city/county",
|
||||
"corona_filter": "Remove Corona Warnings"
|
||||
"headline_filter": "Blacklist regex to filter warning headlines"
|
||||
}
|
||||
}
|
||||
},
|
||||
|
@ -36,7 +36,7 @@
|
|||
"_r_to_u": "City/county (R-U)",
|
||||
"_v_to_z": "City/county (V-Z)",
|
||||
"slots": "Maximum warnings per city/county",
|
||||
"corona_filter": "Remove Corona Warnings"
|
||||
"headline_filter": "Blacklist regex to filter warning headlines"
|
||||
}
|
||||
}
|
||||
},
|
||||
|
|
|
@ -10,7 +10,7 @@ from pynina import ApiError
|
|||
|
||||
from homeassistant import data_entry_flow
|
||||
from homeassistant.components.nina.const import (
|
||||
CONF_FILTER_CORONA,
|
||||
CONF_HEADLINE_FILTER,
|
||||
CONF_MESSAGE_SLOTS,
|
||||
CONF_REGIONS,
|
||||
CONST_REGION_A_TO_D,
|
||||
|
@ -37,7 +37,7 @@ DUMMY_DATA: dict[str, Any] = {
|
|||
CONST_REGION_M_TO_Q: ["071380000000_0", "071380000000_1"],
|
||||
CONST_REGION_R_TO_U: ["072320000000_0", "072320000000_1"],
|
||||
CONST_REGION_V_TO_Z: ["081270000000_0", "081270000000_1"],
|
||||
CONF_FILTER_CORONA: True,
|
||||
CONF_HEADLINE_FILTER: ".*corona.*",
|
||||
}
|
||||
|
||||
DUMMY_RESPONSE_REGIONS: dict[str, Any] = json.loads(
|
||||
|
@ -113,7 +113,7 @@ async def test_step_user_no_selection(hass: HomeAssistant) -> None:
|
|||
wraps=mocked_request_function,
|
||||
):
|
||||
result: dict[str, Any] = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}, data={}
|
||||
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HEADLINE_FILTER: ""}
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.FlowResultType.FORM
|
||||
|
@ -145,7 +145,7 @@ async def test_options_flow_init(hass: HomeAssistant) -> None:
|
|||
domain=DOMAIN,
|
||||
title="NINA",
|
||||
data={
|
||||
CONF_FILTER_CORONA: deepcopy(DUMMY_DATA[CONF_FILTER_CORONA]),
|
||||
CONF_HEADLINE_FILTER: deepcopy(DUMMY_DATA[CONF_HEADLINE_FILTER]),
|
||||
CONF_MESSAGE_SLOTS: deepcopy(DUMMY_DATA[CONF_MESSAGE_SLOTS]),
|
||||
CONST_REGION_A_TO_D: deepcopy(DUMMY_DATA[CONST_REGION_A_TO_D]),
|
||||
CONF_REGIONS: {"095760000000": "Aach"},
|
||||
|
@ -183,7 +183,7 @@ async def test_options_flow_init(hass: HomeAssistant) -> None:
|
|||
assert result["data"] is None
|
||||
|
||||
assert dict(config_entry.data) == {
|
||||
CONF_FILTER_CORONA: deepcopy(DUMMY_DATA[CONF_FILTER_CORONA]),
|
||||
CONF_HEADLINE_FILTER: deepcopy(DUMMY_DATA[CONF_HEADLINE_FILTER]),
|
||||
CONF_MESSAGE_SLOTS: deepcopy(DUMMY_DATA[CONF_MESSAGE_SLOTS]),
|
||||
CONST_REGION_A_TO_D: ["072350000000_1"],
|
||||
CONST_REGION_E_TO_H: [],
|
||||
|
@ -229,6 +229,7 @@ async def test_options_flow_with_no_selection(hass: HomeAssistant) -> None:
|
|||
CONST_REGION_M_TO_Q: [],
|
||||
CONST_REGION_R_TO_U: [],
|
||||
CONST_REGION_V_TO_Z: [],
|
||||
CONF_HEADLINE_FILTER: "",
|
||||
},
|
||||
)
|
||||
|
||||
|
|
|
@ -15,7 +15,7 @@ from tests.common import MockConfigEntry
|
|||
|
||||
ENTRY_DATA: dict[str, Any] = {
|
||||
"slots": 5,
|
||||
"corona_filter": True,
|
||||
"headline_filter": ".*corona.*",
|
||||
"regions": {"083350000000": "Aach, Stadt"},
|
||||
}
|
||||
|
||||
|
@ -37,6 +37,27 @@ async def init_integration(hass) -> MockConfigEntry:
|
|||
return entry
|
||||
|
||||
|
||||
async def test_config_migration(hass: HomeAssistant) -> None:
|
||||
"""Test the migration to a new configuration layout."""
|
||||
|
||||
old_entry_data: dict[str, Any] = {
|
||||
"slots": 5,
|
||||
"corona_filter": True,
|
||||
"regions": {"083350000000": "Aach, Stadt"},
|
||||
}
|
||||
|
||||
old_conf_entry: MockConfigEntry = MockConfigEntry(
|
||||
domain=DOMAIN, title="NINA", data=old_entry_data
|
||||
)
|
||||
|
||||
old_conf_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(old_conf_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert dict(old_conf_entry.data) == ENTRY_DATA
|
||||
|
||||
|
||||
async def test_config_entry_not_ready(hass: HomeAssistant) -> None:
|
||||
"""Test the configuration entry."""
|
||||
entry: MockConfigEntry = await init_integration(hass)
|
||||
|
|
Loading…
Reference in New Issue