Skip to content

Commit

Permalink
Move Netgear LTE services to their own file (#92945)
Browse files Browse the repository at this point in the history
* 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>
  • Loading branch information
tkdrob and epenet committed May 12, 2023
1 parent 30f52b8 commit bd7e943
Show file tree
Hide file tree
Showing 2 changed files with 89 additions and 66 deletions.
69 changes: 3 additions & 66 deletions homeassistant/components/netgear_lte/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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(
{
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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]

Expand Down
86 changes: 86 additions & 0 deletions homeassistant/components/netgear_lte/services.py
Original file line number Diff line number Diff line change
@@ -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)

0 comments on commit bd7e943

Please sign in to comment.