From bd7e943efe7b85e7b1208867a2027d605daa0c27 Mon Sep 17 00:00:00 2001 From: Robert Hillis Date: Fri, 12 May 2023 09:18:34 -0400 Subject: [PATCH] Move Netgear LTE services to their own file (#92945) * Move Netgear LTE services to their own file * Update homeassistant/components/netgear_lte/services.py Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> --------- Co-authored-by: epenet <6771947+epenet@users.noreply.github.com> --- .../components/netgear_lte/__init__.py | 69 +-------------- .../components/netgear_lte/services.py | 86 +++++++++++++++++++ 2 files changed, 89 insertions(+), 66 deletions(-) create mode 100644 homeassistant/components/netgear_lte/services.py diff --git a/homeassistant/components/netgear_lte/__init__.py b/homeassistant/components/netgear_lte/__init__.py index 14aee229617..ed9ee49a0de 100644 --- a/homeassistant/components/netgear_lte/__init__.py +++ b/homeassistant/components/netgear_lte/__init__.py @@ -16,7 +16,7 @@ from homeassistant.const import ( EVENT_HOMEASSISTANT_STOP, Platform, ) -from homeassistant.core import HomeAssistant, ServiceCall, callback +from homeassistant.core import HomeAssistant, callback 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 @@ -25,31 +25,23 @@ from homeassistant.helpers.typing import ConfigType from . import sensor_types from .const import ( - 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, ) +from .services import async_setup_services SCAN_INTERVAL = timedelta(seconds=10) EVENT_SMS = "netgear_lte_sms" -SERVICE_DELETE_SMS = "delete_sms" -SERVICE_SET_OPTION = "set_option" -SERVICE_CONNECT_LTE = "connect_lte" -SERVICE_DISCONNECT_LTE = "disconnect_lte" - NOTIFY_SCHEMA = vol.Schema( { @@ -98,28 +90,6 @@ CONFIG_SCHEMA = vol.Schema( extra=vol.ALLOW_EXTRA, ) -DELETE_SMS_SCHEMA = vol.Schema( - { - vol.Optional(ATTR_HOST): cv.string, - vol.Required(ATTR_SMS_ID): vol.All(cv.ensure_list, [cv.positive_int]), - } -) - -SET_OPTION_SCHEMA = vol.Schema( - vol.All( - cv.has_at_least_one_key(ATTR_FAILOVER, ATTR_AUTOCONNECT), - { - vol.Optional(ATTR_HOST): cv.string, - vol.Optional(ATTR_FAILOVER): vol.In(FAILOVER_MODES), - vol.Optional(ATTR_AUTOCONNECT): vol.In(AUTOCONNECT_MODES), - }, - ) -) - -CONNECT_LTE_SCHEMA = vol.Schema({vol.Optional(ATTR_HOST): cv.string}) - -DISCONNECT_LTE_SCHEMA = vol.Schema({vol.Optional(ATTR_HOST): cv.string}) - @attr.s class ModemData: @@ -173,40 +143,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: ) hass.data[DOMAIN] = LTEData(websession) - async def service_handler(service: ServiceCall) -> None: - """Apply a service.""" - host = service.data.get(ATTR_HOST) - conf = {CONF_HOST: host} - modem_data = hass.data[DOMAIN].get_modem_data(conf) - - if not modem_data: - LOGGER.error("%s: host %s unavailable", service.service, host) - return - - if service.service == SERVICE_DELETE_SMS: - for sms_id in service.data[ATTR_SMS_ID]: - await modem_data.modem.delete_sms(sms_id) - elif service.service == SERVICE_SET_OPTION: - if failover := service.data.get(ATTR_FAILOVER): - await modem_data.modem.set_failover_mode(failover) - if autoconnect := service.data.get(ATTR_AUTOCONNECT): - await modem_data.modem.set_autoconnect_mode(autoconnect) - elif service.service == SERVICE_CONNECT_LTE: - await modem_data.modem.connect_lte() - elif service.service == SERVICE_DISCONNECT_LTE: - await modem_data.modem.disconnect_lte() - - service_schemas = { - SERVICE_DELETE_SMS: DELETE_SMS_SCHEMA, - SERVICE_SET_OPTION: SET_OPTION_SCHEMA, - SERVICE_CONNECT_LTE: CONNECT_LTE_SCHEMA, - SERVICE_DISCONNECT_LTE: DISCONNECT_LTE_SCHEMA, - } - - for service, schema in service_schemas.items(): - hass.services.async_register( - DOMAIN, service, service_handler, schema=schema - ) + await async_setup_services(hass) netgear_lte_config = config[DOMAIN] diff --git a/homeassistant/components/netgear_lte/services.py b/homeassistant/components/netgear_lte/services.py new file mode 100644 index 00000000000..2ea98896791 --- /dev/null +++ b/homeassistant/components/netgear_lte/services.py @@ -0,0 +1,86 @@ +"""Services for the Netgear LTE integration.""" +from typing import TYPE_CHECKING + +import voluptuous as vol + +from homeassistant.const import CONF_HOST +from homeassistant.core import HomeAssistant, ServiceCall +from homeassistant.helpers import config_validation as cv + +from .const import ( + ATTR_AUTOCONNECT, + ATTR_FAILOVER, + ATTR_HOST, + ATTR_SMS_ID, + AUTOCONNECT_MODES, + DOMAIN, + FAILOVER_MODES, + LOGGER, +) + +if TYPE_CHECKING: + from . import LTEData, ModemData + +SERVICE_DELETE_SMS = "delete_sms" +SERVICE_SET_OPTION = "set_option" +SERVICE_CONNECT_LTE = "connect_lte" +SERVICE_DISCONNECT_LTE = "disconnect_lte" + +DELETE_SMS_SCHEMA = vol.Schema( + { + vol.Optional(ATTR_HOST): cv.string, + vol.Required(ATTR_SMS_ID): vol.All(cv.ensure_list, [cv.positive_int]), + } +) + +SET_OPTION_SCHEMA = vol.Schema( + vol.All( + cv.has_at_least_one_key(ATTR_FAILOVER, ATTR_AUTOCONNECT), + { + vol.Optional(ATTR_HOST): cv.string, + vol.Optional(ATTR_FAILOVER): vol.In(FAILOVER_MODES), + vol.Optional(ATTR_AUTOCONNECT): vol.In(AUTOCONNECT_MODES), + }, + ) +) + +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: + """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: + 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) + elif call.service == SERVICE_SET_OPTION: + if failover := call.data.get(ATTR_FAILOVER): + await modem_data.modem.set_failover_mode(failover) + if autoconnect := call.data.get(ATTR_AUTOCONNECT): + await modem_data.modem.set_autoconnect_mode(autoconnect) + elif call.service == SERVICE_CONNECT_LTE: + await modem_data.modem.connect_lte() + elif call.service == SERVICE_DISCONNECT_LTE: + await modem_data.modem.disconnect_lte() + + service_schemas = { + SERVICE_DELETE_SMS: DELETE_SMS_SCHEMA, + SERVICE_SET_OPTION: SET_OPTION_SCHEMA, + SERVICE_CONNECT_LTE: CONNECT_LTE_SCHEMA, + SERVICE_DISCONNECT_LTE: DISCONNECT_LTE_SCHEMA, + } + + for service, schema in service_schemas.items(): + hass.services.async_register(DOMAIN, service, service_handler, schema=schema)