Simplify netgear_lte service actions (#146606)

This commit is contained in:
epenet 2025-06-12 14:02:17 +02:00 committed by GitHub
parent 2991726d35
commit 78ed1097c4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 34 additions and 26 deletions

View File

@ -96,7 +96,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: NetgearLTEConfigEntry) -
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await async_setup_services(hass, modem)
async_setup_services(hass)
await discovery.async_load_platform(
hass,

View File

@ -1,9 +1,9 @@
"""Services for the Netgear LTE integration."""
from eternalegypt.eternalegypt import Modem
import voluptuous as vol
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.helpers import config_validation as cv
from .const import (
@ -16,6 +16,7 @@ from .const import (
FAILOVER_MODES,
LOGGER,
)
from .coordinator import NetgearLTEConfigEntry
SERVICE_DELETE_SMS = "delete_sms"
SERVICE_SET_OPTION = "set_option"
@ -45,30 +46,37 @@ 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, modem: Modem) -> None:
async def _service_handler(call: ServiceCall) -> None:
"""Apply a service."""
host = call.data.get(ATTR_HOST)
entry: NetgearLTEConfigEntry | None = None
for entry in call.hass.config_entries.async_loaded_entries(DOMAIN):
if entry.data.get(CONF_HOST) == host:
break
if not entry or not (modem := entry.runtime_data.modem).token:
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.delete_sms(sms_id)
elif call.service == SERVICE_SET_OPTION:
if failover := call.data.get(ATTR_FAILOVER):
await modem.set_failover_mode(failover)
if autoconnect := call.data.get(ATTR_AUTOCONNECT):
await modem.set_autoconnect_mode(autoconnect)
elif call.service == SERVICE_CONNECT_LTE:
await modem.connect_lte()
elif call.service == SERVICE_DISCONNECT_LTE:
await modem.disconnect_lte()
@callback
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)
if not modem.token:
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.delete_sms(sms_id)
elif call.service == SERVICE_SET_OPTION:
if failover := call.data.get(ATTR_FAILOVER):
await modem.set_failover_mode(failover)
if autoconnect := call.data.get(ATTR_AUTOCONNECT):
await modem.set_autoconnect_mode(autoconnect)
elif call.service == SERVICE_CONNECT_LTE:
await modem.connect_lte()
elif call.service == SERVICE_DISCONNECT_LTE:
await modem.disconnect_lte()
service_schemas = {
SERVICE_DELETE_SMS: DELETE_SMS_SCHEMA,
SERVICE_SET_OPTION: SET_OPTION_SCHEMA,
@ -77,4 +85,4 @@ async def async_setup_services(hass: HomeAssistant, modem: Modem) -> None:
}
for service, schema in service_schemas.items():
hass.services.async_register(DOMAIN, service, service_handler, schema=schema)
hass.services.async_register(DOMAIN, service, _service_handler, schema=schema)