jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-21 20:27:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move Netgear LTE services to their own file (#92945)

Browse Source 
* 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>
This commit is contained in:
Robert Hillis 2023-05-12 09:18:34 -04:00 committed by GitHub
parent 30f52b8d96
commit bd7e943efe
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 89 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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]

86
homeassistant/components/netgear_lte/services.py Normal file
View File

@ -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)