mirror of
https://github.com/home-assistant/core.git
synced 2026-09-24 23:41:48 -05:00
92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
"""Services for the TP-Link Omada integration."""
|
|
|
|
from typing import cast
|
|
|
|
from tplink_omada_client.exceptions import OmadaClientException
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
|
|
from homeassistant.const import ATTR_CONFIG_ENTRY_ID
|
|
from homeassistant.core import HomeAssistant, ServiceCall, callback
|
|
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
|
|
from homeassistant.helpers import config_validation as cv, selector
|
|
|
|
from .const import DOMAIN
|
|
from .controller import OmadaSiteController
|
|
|
|
SERVICE_RECONNECT_CLIENT = "reconnect_client"
|
|
|
|
ATTR_MAC = "mac"
|
|
|
|
|
|
def _get_controller(call: ServiceCall) -> OmadaSiteController:
|
|
if call.data.get(ATTR_CONFIG_ENTRY_ID):
|
|
entry = call.hass.config_entries.async_get_entry(
|
|
call.data[ATTR_CONFIG_ENTRY_ID]
|
|
)
|
|
if not entry:
|
|
raise ServiceValidationError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="controller_not_found",
|
|
)
|
|
else:
|
|
# Assume first loaded entry if none specified
|
|
# (for backward compatibility/99% use case)
|
|
entries = call.hass.config_entries.async_entries(DOMAIN)
|
|
if len(entries) == 0:
|
|
raise ServiceValidationError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="no_controllers",
|
|
)
|
|
entry = entries[0]
|
|
|
|
entry = cast(ConfigEntry[OmadaSiteController], entry)
|
|
|
|
if entry.state is not ConfigEntryState.LOADED:
|
|
raise ServiceValidationError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="controller_unavailable",
|
|
)
|
|
return entry.runtime_data
|
|
|
|
|
|
SCHEMA_RECONNECT_CLIENT = vol.Schema(
|
|
{
|
|
vol.Optional(ATTR_CONFIG_ENTRY_ID): selector.ConfigEntrySelector(
|
|
{
|
|
"integration": DOMAIN,
|
|
}
|
|
),
|
|
vol.Required(ATTR_MAC): cv.string,
|
|
}
|
|
)
|
|
|
|
|
|
async def _handle_reconnect_client(call: ServiceCall) -> None:
|
|
"""Handle the service action to force reconnection of a network client."""
|
|
controller = _get_controller(call)
|
|
|
|
mac: str = call.data[ATTR_MAC]
|
|
|
|
try:
|
|
await controller.omada_client.reconnect_client(mac)
|
|
except OmadaClientException as ex:
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="reconnect_failed",
|
|
translation_placeholders={"mac": mac},
|
|
) from ex
|
|
|
|
|
|
SERVICES = [
|
|
(SERVICE_RECONNECT_CLIENT, SCHEMA_RECONNECT_CLIENT, _handle_reconnect_client)
|
|
]
|
|
|
|
|
|
@callback
|
|
def async_setup_services(hass: HomeAssistant) -> None:
|
|
"""Set up the services for the TP-Link Omada integration."""
|
|
|
|
for service_name, schema, handler in SERVICES:
|
|
hass.services.async_register(DOMAIN, service_name, handler, schema=schema)
|