Move Netgear LTE services to their own file

homeassistant/components/netgear_lte/__init__.py

@@ -16,7 +16,7 @@
     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 . 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 @@
     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]
 

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)