jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-16 17:57:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add update coordinator to Netgear LTE (#115474)

Browse Source
This commit is contained in:
Robert Hillis 2024-06-18 03:12:02 -04:00 committed by GitHub
parent 2555827030
commit d5d906e148
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 166 additions and 194 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

142
homeassistant/components/netgear_lte/__init__.py
View File

@ -1,25 +1,17 @@
"""Support for Netgear LTE modems."""
from datetime import timedelta
from typing import Any
from aiohttp.cookiejar import CookieJar
import attr
import eternalegypt
from eternalegypt.eternalegypt import SMS
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
CONF_PASSWORD,
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.core import Event, HomeAssistant
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv, discovery
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.typing import ConfigType
from .const import (
@ -28,14 +20,12 @@ from .const import (
ATTR_MESSAGE,
ATTR_SMS_ID,
DATA_HASS_CONFIG,
DISPATCHER_NETGEAR_LTE,
DATA_SESSION,
DOMAIN,
LOGGER,
)
from .coordinator import NetgearLTEDataUpdateCoordinator
from .services import async_setup_services
SCAN_INTERVAL = timedelta(seconds=10)
EVENT_SMS = "netgear_lte_sms"
ALL_SENSORS = [
@ -65,54 +55,11 @@ PLATFORMS = [
Platform.NOTIFY,
Platform.SENSOR,
]
type NetgearLTEConfigEntry = ConfigEntry[NetgearLTEDataUpdateCoordinator]
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
@attr.s
class ModemData:
"""Class for modem state."""
hass = attr.ib()
host = attr.ib()
modem = attr.ib()
data = attr.ib(init=False, default=None)
connected = attr.ib(init=False, default=True)
async def async_update(self):
"""Call the API to update the data."""
try:
self.data = await self.modem.information()
if not self.connected:
LOGGER.warning("Connected to %s", self.host)
self.connected = True
except eternalegypt.Error:
if self.connected:
LOGGER.warning("Lost connection to %s", self.host)
self.connected = False
self.data = None
async_dispatcher_send(self.hass, DISPATCHER_NETGEAR_LTE)
@attr.s
class LTEData:
"""Shared state."""
websession = attr.ib()
modem_data: dict[str, ModemData] = attr.ib(init=False, factory=dict)
def get_modem_data(self, config):
"""Get modem_data for the host in config."""
if config[CONF_HOST] is not None:
return self.modem_data.get(config[CONF_HOST])
if len(self.modem_data) != 1:
return None
return next(iter(self.modem_data.values()))
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up Netgear LTE component."""
hass.data[DATA_HASS_CONFIG] = config
@ -120,44 +67,44 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: NetgearLTEConfigEntry) -> bool:
"""Set up Netgear LTE from a config entry."""
host = entry.data[CONF_HOST]
password = entry.data[CONF_PASSWORD]
if not (data := hass.data.get(DOMAIN)) or data.websession.closed:
websession = async_create_clientsession(hass, cookie_jar=CookieJar(unsafe=True))
data: dict[str, Any] = hass.data.setdefault(DOMAIN, {})
if not (session := data.get(DATA_SESSION)) or session.closed:
session = async_create_clientsession(hass, cookie_jar=CookieJar(unsafe=True))
modem = eternalegypt.Modem(hostname=host, websession=session)
hass.data[DOMAIN] = LTEData(websession)
try:
await modem.login(password=password)
except eternalegypt.Error as ex:
raise ConfigEntryNotReady("Cannot connect/authenticate") from ex
modem = eternalegypt.Modem(hostname=host, websession=hass.data[DOMAIN].websession)
modem_data = ModemData(hass, host, modem)
def fire_sms_event(sms: SMS) -> None:
"""Send an SMS event."""
data = {
ATTR_HOST: modem.hostname,
ATTR_SMS_ID: sms.id,
ATTR_FROM: sms.sender,
ATTR_MESSAGE: sms.message,
}
hass.bus.async_fire(EVENT_SMS, data)
await _login(hass, modem_data, password)
await modem.add_sms_listener(fire_sms_event)
async def _update(now):
"""Periodic update."""
await modem_data.async_update()
coordinator = NetgearLTEDataUpdateCoordinator(hass, modem)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
update_unsub = async_track_time_interval(hass, _update, SCAN_INTERVAL)
async def cleanup(event: Event | None = None) -> None:
"""Clean up resources."""
update_unsub()
await modem.logout()
if DOMAIN in hass.data:
del hass.data[DOMAIN].modem_data[modem_data.host]
entry.async_on_unload(cleanup)
entry.async_on_unload(hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, cleanup))
await async_setup_services(hass)
await async_setup_services(hass, modem)
await discovery.async_load_platform(
hass,
Platform.NOTIFY,
DOMAIN,
{CONF_HOST: entry.data[CONF_HOST], CONF_NAME: entry.title},
{CONF_NAME: entry.title, "modem": modem},
hass.data[DATA_HASS_CONFIG],
)
@ -168,7 +115,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: NetgearLTEConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
loaded_entries = [
@ -178,28 +125,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
]
if len(loaded_entries) == 1:
hass.data.pop(DOMAIN, None)
for service_name in hass.services.async_services()[DOMAIN]:
hass.services.async_remove(DOMAIN, service_name)
return unload_ok
async def _login(hass: HomeAssistant, modem_data: ModemData, password: str) -> None:
"""Log in and complete setup."""
try:
await modem_data.modem.login(password=password)
except eternalegypt.Error as ex:
raise ConfigEntryNotReady("Cannot connect/authenticate") from ex
def fire_sms_event(sms):
"""Send an SMS event."""
data = {
ATTR_HOST: modem_data.host,
ATTR_SMS_ID: sms.id,
ATTR_FROM: sms.sender,
ATTR_MESSAGE: sms.message,
}
hass.bus.async_fire(EVENT_SMS, data)
await modem_data.modem.add_sms_listener(fire_sms_event)
await modem_data.async_update()
hass.data[DOMAIN].modem_data[modem_data.host] = modem_data

13
homeassistant/components/netgear_lte/binary_sensor.py
View File

@ -7,12 +7,11 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from . import NetgearLTEConfigEntry
from .entity import LTEEntity
BINARY_SENSORS: tuple[BinarySensorEntityDescription, ...] = (
@ -38,13 +37,13 @@ BINARY_SENSORS: tuple[BinarySensorEntityDescription, ...] = (
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
hass: HomeAssistant,
entry: NetgearLTEConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Netgear LTE binary sensor."""
modem_data = hass.data[DOMAIN].get_modem_data(entry.data)
async_add_entities(
NetgearLTEBinarySensor(entry, modem_data, sensor) for sensor in BINARY_SENSORS
NetgearLTEBinarySensor(entry, description) for description in BINARY_SENSORS
)
@ -54,4 +53,4 @@ class NetgearLTEBinarySensor(LTEEntity, BinarySensorEntity):
@property
def is_on(self):
"""Return true if the binary sensor is on."""
return getattr(self.modem_data.data, self.entity_description.key)
return getattr(self.coordinator.data, self.entity_description.key)

2
homeassistant/components/netgear_lte/const.py
View File

@ -16,9 +16,9 @@ CONF_NOTIFY: Final = "notify"
CONF_SENSOR: Final = "sensor"
DATA_HASS_CONFIG = "netgear_lte_hass_config"
DATA_SESSION = "session"
# https://kb.netgear.com/31160/How-do-I-change-my-4G-LTE-Modem-s-IP-address-range
DEFAULT_HOST = "192.168.5.1"
DISPATCHER_NETGEAR_LTE = "netgear_lte_update"
DOMAIN: Final = "netgear_lte"
FAILOVER_MODES = ["auto", "wire", "mobile"]

43
homeassistant/components/netgear_lte/coordinator.py Normal file
View File

@ -0,0 +1,43 @@
"""Data update coordinator for the Netgear LTE integration."""
from __future__ import annotations
from datetime import timedelta
from typing import TYPE_CHECKING
from eternalegypt.eternalegypt import Error, Information, Modem
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, LOGGER
if TYPE_CHECKING:
from . import NetgearLTEConfigEntry
class NetgearLTEDataUpdateCoordinator(DataUpdateCoordinator[Information]):
"""Data update coordinator for the Netgear LTE integration."""
config_entry: NetgearLTEConfigEntry
def __init__(
self,
hass: HomeAssistant,
modem: Modem,
) -> None:
"""Initialize the coordinator."""
super().__init__(
hass=hass,
logger=LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=10),
)
self.modem = modem
async def _async_update_data(self) -> Information:
"""Get the latest data."""
try:
return await self.modem.information()
except Error as ex:
raise UpdateFailed(ex) from ex

50
homeassistant/components/netgear_lte/entity.py
View File

@ -1,54 +1,36 @@
"""Entity representing a Netgear LTE entity."""
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import Entity, EntityDescription
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import ModemData
from .const import DISPATCHER_NETGEAR_LTE, DOMAIN, MANUFACTURER
from . import NetgearLTEConfigEntry
from .const import DOMAIN, MANUFACTURER
from .coordinator import NetgearLTEDataUpdateCoordinator
class LTEEntity(Entity):
class LTEEntity(CoordinatorEntity[NetgearLTEDataUpdateCoordinator]):
"""Base LTE entity."""
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(
self,
config_entry: ConfigEntry,
modem_data: ModemData,
entry: NetgearLTEConfigEntry,
description: EntityDescription,
) -> None:
"""Initialize a Netgear LTE entity."""
super().__init__(entry.runtime_data)
self.entity_description = description
self.modem_data = modem_data
self._attr_unique_id = f"{description.key}_{modem_data.data.serial_number}"
data = entry.runtime_data.data
self._attr_unique_id = f"{description.key}_{data.serial_number}"
self._attr_device_info = DeviceInfo(
configuration_url=f"http://{config_entry.data[CONF_HOST]}",
identifiers={(DOMAIN, modem_data.data.serial_number)},
configuration_url=f"http://{entry.data[CONF_HOST]}",
identifiers={(DOMAIN, data.serial_number)},
manufacturer=MANUFACTURER,
model=modem_data.data.items["general.model"],
serial_number=modem_data.data.serial_number,
sw_version=modem_data.data.items["general.fwversion"],
hw_version=modem_data.data.items["general.hwversion"],
model=data.items["general.model"],
serial_number=data.serial_number,
sw_version=data.items["general.fwversion"],
hw_version=data.items["general.hwversion"],
)
async def async_added_to_hass(self) -> None:
"""Register callback."""
self.async_on_remove(
async_dispatcher_connect(
self.hass, DISPATCHER_NETGEAR_LTE, self.async_write_ha_state
)
)
async def async_update(self) -> None:
"""Force update of state."""
await self.modem_data.async_update()
@property
def available(self) -> bool:
"""Return the availability of the sensor."""
return self.modem_data.data is not None

24
homeassistant/components/netgear_lte/notify.py
View File

@ -2,15 +2,17 @@
from __future__ import annotations
import attr
from typing import Any
import eternalegypt
from eternalegypt.eternalegypt import Modem
from homeassistant.components.notify import ATTR_TARGET, BaseNotificationService
from homeassistant.const import CONF_RECIPIENT
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import CONF_NOTIFY, DOMAIN, LOGGER
from .const import CONF_NOTIFY, LOGGER
async def async_get_service(
@ -22,21 +24,25 @@ async def async_get_service(
if discovery_info is None:
return None
return NetgearNotifyService(hass, discovery_info)
return NetgearNotifyService(config, discovery_info)
@attr.s
class NetgearNotifyService(BaseNotificationService):
"""Implementation of a notification service."""
hass = attr.ib()
config = attr.ib()
def __init__(
self,
config: ConfigType,
discovery_info: dict[str, Any],
) -> None:
"""Initialize the service."""
self.config = config
self.modem: Modem = discovery_info["modem"]
async def async_send_message(self, message="", **kwargs):
"""Send a message to a user."""
modem_data = self.hass.data[DOMAIN].get_modem_data(self.config)
if not modem_data:
if not self.modem.token:
LOGGER.error("Modem not ready")
return
if not (targets := kwargs.get(ATTR_TARGET)):
@ -50,6 +56,6 @@ class NetgearNotifyService(BaseNotificationService):
for target in targets:
try:
await modem_data.modem.sms(target, message)
await self.modem.sms(target, message)
except eternalegypt.Error:
LOGGER.error("Unable to send to %s", target)

28
homeassistant/components/netgear_lte/sensor.py
View File

@ -5,12 +5,13 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from eternalegypt.eternalegypt import Information
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
@ -21,8 +22,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from . import ModemData
from .const import DOMAIN
from . import NetgearLTEConfigEntry
from .entity import LTEEntity
@ -30,7 +30,7 @@ from .entity import LTEEntity
class NetgearLTESensorEntityDescription(SensorEntityDescription):
"""Class describing Netgear LTE entities."""
value_fn: Callable[[ModemData], StateType] | None = None
value_fn: Callable[[Information], StateType] | None = None
SENSORS: tuple[NetgearLTESensorEntityDescription, ...] = (
@ -38,13 +38,13 @@ SENSORS: tuple[NetgearLTESensorEntityDescription, ...] = (
key="sms",
translation_key="sms",
native_unit_of_measurement="unread",
value_fn=lambda modem_data: sum(1 for x in modem_data.data.sms if x.unread),
value_fn=lambda data: sum(1 for x in data.sms if x.unread),
),
NetgearLTESensorEntityDescription(
key="sms_total",
translation_key="sms_total",
native_unit_of_measurement="messages",
value_fn=lambda modem_data: len(modem_data.data.sms),
value_fn=lambda data: len(data.sms),
),
NetgearLTESensorEntityDescription(
key="usage",
@ -54,7 +54,7 @@ SENSORS: tuple[NetgearLTESensorEntityDescription, ...] = (
native_unit_of_measurement=UnitOfInformation.BYTES,
suggested_unit_of_measurement=UnitOfInformation.MEBIBYTES,
suggested_display_precision=1,
value_fn=lambda modem_data: modem_data.data.usage,
value_fn=lambda data: data.usage,
),
NetgearLTESensorEntityDescription(
key="radio_quality",
@ -125,14 +125,12 @@ SENSORS: tuple[NetgearLTESensorEntityDescription, ...] = (
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
hass: HomeAssistant,
entry: NetgearLTEConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Netgear LTE sensor."""
modem_data = hass.data[DOMAIN].get_modem_data(entry.data)
async_add_entities(
NetgearLTESensor(entry, modem_data, sensor) for sensor in SENSORS
)
async_add_entities(NetgearLTESensor(entry, description) for description in SENSORS)
class NetgearLTESensor(LTEEntity, SensorEntity):
@ -144,5 +142,5 @@ class NetgearLTESensor(LTEEntity, SensorEntity):
def native_value(self) -> StateType:
"""Return the state of the sensor."""
if self.entity_description.value_fn is not None:
return self.entity_description.value_fn(self.modem_data)
return getattr(self.modem_data.data, self.entity_description.key)
return self.entity_description.value_fn(self.coordinator.data)
return getattr(self.coordinator.data, self.entity_description.key)

23
homeassistant/components/netgear_lte/services.py
View File

@ -1,10 +1,8 @@
"""Services for the Netgear LTE integration."""
from typing import TYPE_CHECKING
from eternalegypt.eternalegypt import Modem
import voluptuous as vol
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import config_validation as cv
@ -19,9 +17,6 @@ from .const import (
LOGGER,
)
if TYPE_CHECKING:
from . import LTEData, ModemData
SERVICE_DELETE_SMS = "delete_sms"
SERVICE_SET_OPTION = "set_option"
SERVICE_CONNECT_LTE = "connect_lte"
@ -50,31 +45,29 @@ CONNECT_LTE_SCHEMA = vol.Schema({vol.Optional(ATTR_HOST): cv.string})
DISCONNECT_LTE_SCHEMA = vol.Schema({vol.Optional(ATTR_HOST): cv.string})
async def async_setup_services(hass: HomeAssistant) -> None:
async def async_setup_services(hass: HomeAssistant, modem: Modem) -> None:
"""Set up services for Netgear LTE integration."""
async def service_handler(call: ServiceCall) -> None:
"""Apply a service."""
host = call.data.get(ATTR_HOST)
data: LTEData = hass.data[DOMAIN]
modem_data: ModemData = data.get_modem_data({CONF_HOST: host})
if not modem_data:
if not modem.token:
LOGGER.error("%s: host %s unavailable", call.service, host)
return
if call.service == SERVICE_DELETE_SMS:
for sms_id in call.data[ATTR_SMS_ID]:
await modem_data.modem.delete_sms(sms_id)
await modem.delete_sms(sms_id)
elif call.service == SERVICE_SET_OPTION:
if failover := call.data.get(ATTR_FAILOVER):
await modem_data.modem.set_failover_mode(failover)
await modem.set_failover_mode(failover)
if autoconnect := call.data.get(ATTR_AUTOCONNECT):
await modem_data.modem.set_autoconnect_mode(autoconnect)
await modem.set_autoconnect_mode(autoconnect)
elif call.service == SERVICE_CONNECT_LTE:
await modem_data.modem.connect_lte()
await modem.connect_lte()
elif call.service == SERVICE_DISCONNECT_LTE:
await modem_data.modem.disconnect_lte()
await modem.disconnect_lte()
service_schemas = {
SERVICE_DELETE_SMS: DELETE_SMS_SCHEMA,

9
tests/components/netgear_lte/test_config_flow.py
View File

@ -2,7 +2,6 @@
from unittest.mock import patch
from homeassistant import data_entry_flow
from homeassistant.components.netgear_lte.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_SOURCE
@ -25,7 +24,7 @@ async def test_flow_user_form(hass: HomeAssistant, connection: None) -> None:
context={CONF_SOURCE: SOURCE_USER},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
with _patch_setup():
@ -33,7 +32,7 @@ async def test_flow_user_form(hass: HomeAssistant, connection: None) -> None:
result["flow_id"],
user_input=CONF_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Netgear LM1200"
assert result["data"] == CONF_DATA
assert result["context"]["unique_id"] == "FFFFFFFFFFFFF"
@ -63,7 +62,7 @@ async def test_flow_user_cannot_connect(
data=CONF_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"]["base"] == "cannot_connect"
@ -78,6 +77,6 @@ async def test_flow_user_unknown_error(hass: HomeAssistant, unknown: None) -> No
result["flow_id"],
user_input=CONF_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"]["base"] == "unknown"

26
tests/components/netgear_lte/test_init.py
View File

@ -1,14 +1,22 @@
"""Test Netgear LTE integration."""
from datetime import timedelta
from unittest.mock import patch
from eternalegypt.eternalegypt import Error
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.netgear_lte.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
import homeassistant.util.dt as dt_util
from .conftest import CONF_DATA
from tests.common import async_fire_time_changed
async def test_setup_unload(hass: HomeAssistant, setup_integration: None) -> None:
"""Test setup and unload."""
@ -43,3 +51,21 @@ async def test_device(
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={(DOMAIN, entry.unique_id)})
assert device == snapshot
async def test_update_failed(
hass: HomeAssistant,
entity_registry_enabled_by_default: None,
setup_integration: None,
) -> None:
"""Test coordinator throws UpdateFailed after failed update."""
with patch(
"homeassistant.components.netgear_lte.eternalegypt.Modem.information",
side_effect=Error,
) as updater:
next_update = dt_util.utcnow() + timedelta(seconds=10)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
updater.assert_called_once()
state = hass.states.get("sensor.netgear_lm1200_radio_quality")
assert state.state == STATE_UNAVAILABLE