Move Netgear LTE constants to their own file (#92924)

* Move Netgear LTE parts to their own files

* Revert to only move constants

* uno mas
This commit is contained in:
Robert Hillis 2023-05-11 11:49:11 -04:00 committed by GitHub
parent 52c773a776
commit aee3f115d6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 62 additions and 45 deletions

View File

@ -1,8 +1,6 @@
"""Support for Netgear LTE modems.""" """Support for Netgear LTE modems."""
import asyncio import asyncio
from datetime import timedelta from datetime import timedelta
import logging
from typing import Final
import aiohttp import aiohttp
import attr import attr
@ -30,18 +28,24 @@ from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.typing import ConfigType from homeassistant.helpers.typing import ConfigType
from . import sensor_types from . import sensor_types
from .const import (
_LOGGER = logging.getLogger(__name__) ATTR_AUTOCONNECT,
ATTR_FAILOVER,
ATTR_FROM,
ATTR_HOST,
ATTR_MESSAGE,
ATTR_SMS_ID,
AUTOCONNECT_MODES,
CONF_BINARY_SENSOR,
CONF_NOTIFY,
CONF_SENSOR,
DISPATCHER_NETGEAR_LTE,
DOMAIN,
FAILOVER_MODES,
LOGGER,
)
SCAN_INTERVAL = timedelta(seconds=10) SCAN_INTERVAL = timedelta(seconds=10)
DISPATCHER_NETGEAR_LTE = "netgear_lte_update"
CONF_NOTIFY: Final = "notify"
CONF_BINARY_SENSOR: Final = "binary_sensor"
CONF_SENSOR: Final = "sensor"
DOMAIN = "netgear_lte"
DATA_KEY = "netgear_lte"
EVENT_SMS = "netgear_lte_sms" EVENT_SMS = "netgear_lte_sms"
@ -50,16 +54,6 @@ SERVICE_SET_OPTION = "set_option"
SERVICE_CONNECT_LTE = "connect_lte" SERVICE_CONNECT_LTE = "connect_lte"
SERVICE_DISCONNECT_LTE = "disconnect_lte" SERVICE_DISCONNECT_LTE = "disconnect_lte"
ATTR_HOST = "host"
ATTR_SMS_ID = "sms_id"
ATTR_FROM = "from"
ATTR_MESSAGE = "message"
ATTR_FAILOVER = "failover"
ATTR_AUTOCONNECT = "autoconnect"
FAILOVER_MODES = ["auto", "wire", "mobile"]
AUTOCONNECT_MODES = ["never", "home", "always"]
NOTIFY_SCHEMA = vol.Schema( NOTIFY_SCHEMA = vol.Schema(
{ {
@ -148,11 +142,11 @@ class ModemData:
try: try:
self.data = await self.modem.information() self.data = await self.modem.information()
if not self.connected: if not self.connected:
_LOGGER.warning("Connected to %s", self.host) LOGGER.warning("Connected to %s", self.host)
self.connected = True self.connected = True
except eternalegypt.Error: except eternalegypt.Error:
if self.connected: if self.connected:
_LOGGER.warning("Lost connection to %s", self.host) LOGGER.warning("Lost connection to %s", self.host)
self.connected = False self.connected = False
self.data = None self.data = None
@ -177,20 +171,20 @@ class LTEData:
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up Netgear LTE component.""" """Set up Netgear LTE component."""
if DATA_KEY not in hass.data: if DOMAIN not in hass.data:
websession = async_create_clientsession( websession = async_create_clientsession(
hass, cookie_jar=aiohttp.CookieJar(unsafe=True) hass, cookie_jar=aiohttp.CookieJar(unsafe=True)
) )
hass.data[DATA_KEY] = LTEData(websession) hass.data[DOMAIN] = LTEData(websession)
async def service_handler(service: ServiceCall) -> None: async def service_handler(service: ServiceCall) -> None:
"""Apply a service.""" """Apply a service."""
host = service.data.get(ATTR_HOST) host = service.data.get(ATTR_HOST)
conf = {CONF_HOST: host} conf = {CONF_HOST: host}
modem_data = hass.data[DATA_KEY].get_modem_data(conf) modem_data = hass.data[DOMAIN].get_modem_data(conf)
if not modem_data: if not modem_data:
_LOGGER.error("%s: host %s unavailable", service.service, host) LOGGER.error("%s: host %s unavailable", service.service, host)
return return
if service.service == SERVICE_DELETE_SMS: if service.service == SERVICE_DELETE_SMS:
@ -272,7 +266,7 @@ async def _setup_lte(hass, lte_config):
host = lte_config[CONF_HOST] host = lte_config[CONF_HOST]
password = lte_config[CONF_PASSWORD] password = lte_config[CONF_PASSWORD]
websession = hass.data[DATA_KEY].websession websession = hass.data[DOMAIN].websession
modem = eternalegypt.Modem(hostname=host, websession=websession) modem = eternalegypt.Modem(hostname=host, websession=websession)
modem_data = ModemData(hass, host, modem) modem_data = ModemData(hass, host, modem)
@ -308,7 +302,7 @@ async def _login(hass, modem_data, password):
await modem_data.modem.add_sms_listener(fire_sms_event) await modem_data.modem.add_sms_listener(fire_sms_event)
await modem_data.async_update() await modem_data.async_update()
hass.data[DATA_KEY].modem_data[modem_data.host] = modem_data hass.data[DOMAIN].modem_data[modem_data.host] = modem_data
async def _update(now): async def _update(now):
"""Periodic update.""" """Periodic update."""
@ -320,7 +314,7 @@ async def _login(hass, modem_data, password):
"""Clean up resources.""" """Clean up resources."""
update_unsub() update_unsub()
await modem_data.modem.logout() await modem_data.modem.logout()
del hass.data[DATA_KEY].modem_data[modem_data.host] del hass.data[DOMAIN].modem_data[modem_data.host]
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, cleanup) hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, cleanup)
@ -328,7 +322,7 @@ async def _login(hass, modem_data, password):
async def _retry_login(hass, modem_data, password): async def _retry_login(hass, modem_data, password):
"""Sleep and retry setup.""" """Sleep and retry setup."""
_LOGGER.warning("Could not connect to %s. Will keep trying", modem_data.host) LOGGER.warning("Could not connect to %s. Will keep trying", modem_data.host)
modem_data.connected = False modem_data.connected = False
delay = 15 delay = 15

View File

@ -2,12 +2,14 @@
from __future__ import annotations from __future__ import annotations
from homeassistant.components.binary_sensor import BinarySensorEntity from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.const import CONF_MONITORED_CONDITIONS
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import CONF_BINARY_SENSOR, CONF_MONITORED_CONDITIONS, DATA_KEY, LTEEntity from . import LTEEntity
from .const import CONF_BINARY_SENSOR, DOMAIN
from .sensor_types import BINARY_SENSOR_CLASSES from .sensor_types import BINARY_SENSOR_CLASSES
@ -21,7 +23,7 @@ async def async_setup_platform(
if discovery_info is None: if discovery_info is None:
return return
modem_data = hass.data[DATA_KEY].get_modem_data(discovery_info) modem_data = hass.data[DOMAIN].get_modem_data(discovery_info)
if not modem_data or not modem_data.data: if not modem_data or not modem_data.data:
raise PlatformNotReady raise PlatformNotReady

View File

@ -0,0 +1,22 @@
"""Constants for the Netgear LTE integration."""
import logging
from typing import Final
ATTR_HOST = "host"
ATTR_SMS_ID = "sms_id"
ATTR_FROM = "from"
ATTR_MESSAGE = "message"
ATTR_FAILOVER = "failover"
ATTR_AUTOCONNECT = "autoconnect"
AUTOCONNECT_MODES = ["never", "home", "always"]
CONF_BINARY_SENSOR: Final = "binary_sensor"
CONF_NOTIFY: Final = "notify"
CONF_SENSOR: Final = "sensor"
DISPATCHER_NETGEAR_LTE = "netgear_lte_update"
DOMAIN: Final = "netgear_lte"
FAILOVER_MODES = ["auto", "wire", "mobile"]
LOGGER = logging.getLogger(__package__)

View File

@ -1,18 +1,15 @@
"""Support for Netgear LTE notifications.""" """Support for Netgear LTE notifications."""
from __future__ import annotations from __future__ import annotations
import logging
import attr import attr
import eternalegypt import eternalegypt
from homeassistant.components.notify import ATTR_TARGET, BaseNotificationService from homeassistant.components.notify import ATTR_TARGET, BaseNotificationService
from homeassistant.const import CONF_RECIPIENT
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import CONF_NOTIFY, CONF_RECIPIENT, DATA_KEY from .const import CONF_NOTIFY, DOMAIN, LOGGER
_LOGGER = logging.getLogger(__name__)
async def async_get_service( async def async_get_service(
@ -37,14 +34,14 @@ class NetgearNotifyService(BaseNotificationService):
async def async_send_message(self, message="", **kwargs): async def async_send_message(self, message="", **kwargs):
"""Send a message to a user.""" """Send a message to a user."""
modem_data = self.hass.data[DATA_KEY].get_modem_data(self.config) modem_data = self.hass.data[DOMAIN].get_modem_data(self.config)
if not modem_data: if not modem_data:
_LOGGER.error("Modem not ready") LOGGER.error("Modem not ready")
return return
targets = kwargs.get(ATTR_TARGET, self.config[CONF_NOTIFY][CONF_RECIPIENT]) targets = kwargs.get(ATTR_TARGET, self.config[CONF_NOTIFY][CONF_RECIPIENT])
if not targets: if not targets:
_LOGGER.warning("No recipients") LOGGER.warning("No recipients")
return return
if not message: if not message:
@ -54,4 +51,4 @@ class NetgearNotifyService(BaseNotificationService):
try: try:
await modem_data.modem.sms(target, message) await modem_data.modem.sms(target, message)
except eternalegypt.Error: except eternalegypt.Error:
_LOGGER.error("Unable to send to %s", target) LOGGER.error("Unable to send to %s", target)

View File

@ -2,12 +2,14 @@
from __future__ import annotations from __future__ import annotations
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
from homeassistant.const import CONF_MONITORED_CONDITIONS
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import CONF_MONITORED_CONDITIONS, CONF_SENSOR, DATA_KEY, LTEEntity from . import LTEEntity
from .const import CONF_SENSOR, DOMAIN
from .sensor_types import SENSOR_SMS, SENSOR_SMS_TOTAL, SENSOR_UNITS, SENSOR_USAGE from .sensor_types import SENSOR_SMS, SENSOR_SMS_TOTAL, SENSOR_UNITS, SENSOR_USAGE
@ -21,7 +23,7 @@ async def async_setup_platform(
if discovery_info is None: if discovery_info is None:
return return
modem_data = hass.data[DATA_KEY].get_modem_data(discovery_info) modem_data = hass.data[DOMAIN].get_modem_data(discovery_info)
if not modem_data or not modem_data.data: if not modem_data or not modem_data.data:
raise PlatformNotReady raise PlatformNotReady