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

Simplify screenlogic service actions (#146609)

Browse Source
This commit is contained in:
epenet 2025-06-12 14:02:06 +02:00 committed by GitHub
parent c34596e54d
commit 2991726d35
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 92 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
homeassistant/components/screenlogic/__init__.py
View File

@ -18,7 +18,7 @@ from homeassistant.util import slugify
from .const import DOMAIN
from .coordinator import ScreenlogicDataUpdateCoordinator, async_get_connect_info
from .data import ENTITY_MIGRATIONS
from .services import async_load_screenlogic_services
from .services import async_setup_services
from .util import generate_unique_id
type ScreenLogicConfigEntry = ConfigEntry[ScreenlogicDataUpdateCoordinator]
@ -48,7 +48,7 @@ CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up Screenlogic."""
async_load_screenlogic_services(hass)
async_setup_services(hass)
return True

175
homeassistant/components/screenlogic/services.py
View File

@ -54,105 +54,110 @@ TURN_ON_SUPER_CHLOR_SCHEMA = BASE_SERVICE_SCHEMA.extend(
)
async def _get_coordinators(
service_call: ServiceCall,
) -> list[ScreenlogicDataUpdateCoordinator]:
entry_ids = {service_call.data[ATTR_CONFIG_ENTRY]}
coordinators: list[ScreenlogicDataUpdateCoordinator] = []
for entry_id in entry_ids:
config_entry = cast(
ScreenLogicConfigEntry | None,
service_call.hass.config_entries.async_get_entry(entry_id),
)
if not config_entry:
raise ServiceValidationError(
f"Failed to call service '{service_call.service}'. Config entry "
f"'{entry_id}' not found"
)
if not config_entry.domain == DOMAIN:
raise ServiceValidationError(
f"Failed to call service '{service_call.service}'. Config entry "
f"'{entry_id}' is not a {DOMAIN} config"
)
if not config_entry.state == ConfigEntryState.LOADED:
raise ServiceValidationError(
f"Failed to call service '{service_call.service}'. Config entry "
f"'{entry_id}' not loaded"
)
coordinators.append(config_entry.runtime_data)
return coordinators
async def _async_set_color_mode(service_call: ServiceCall) -> None:
color_num = SUPPORTED_COLOR_MODES[service_call.data[ATTR_COLOR_MODE]]
coordinator: ScreenlogicDataUpdateCoordinator
for coordinator in await _get_coordinators(service_call):
_LOGGER.debug(
"Service %s called on %s with mode %s",
SERVICE_SET_COLOR_MODE,
coordinator.gateway.name,
color_num,
)
try:
await coordinator.gateway.async_set_color_lights(color_num)
# Debounced refresh to catch any secondary changes in the device
await coordinator.async_request_refresh()
except ScreenLogicError as error:
raise HomeAssistantError(error) from error
async def _async_set_super_chlor(
service_call: ServiceCall,
is_on: bool,
runtime: int | None = None,
) -> None:
coordinator: ScreenlogicDataUpdateCoordinator
for coordinator in await _get_coordinators(service_call):
if EQUIPMENT_FLAG.CHLORINATOR not in coordinator.gateway.equipment_flags:
raise ServiceValidationError(
f"Equipment configuration for {coordinator.gateway.name} does not"
f" support {service_call.service}"
)
rt_log = f" with runtime {runtime}" if runtime else ""
_LOGGER.debug(
"Service %s called on %s%s",
service_call.service,
coordinator.gateway.name,
rt_log,
)
try:
await coordinator.gateway.async_set_scg_config(
super_chlor_timer=runtime, super_chlorinate=is_on
)
# Debounced refresh to catch any secondary changes in the device
await coordinator.async_request_refresh()
except ScreenLogicError as error:
raise HomeAssistantError(error) from error
async def _async_start_super_chlor(service_call: ServiceCall) -> None:
runtime = service_call.data[ATTR_RUNTIME]
await _async_set_super_chlor(service_call, True, runtime)
async def _async_stop_super_chlor(service_call: ServiceCall) -> None:
await _async_set_super_chlor(service_call, False)
@callback
def async_load_screenlogic_services(hass: HomeAssistant):
def async_setup_services(hass: HomeAssistant):
"""Set up services for the ScreenLogic integration."""
async def get_coordinators(
service_call: ServiceCall,
) -> list[ScreenlogicDataUpdateCoordinator]:
entry_ids = {service_call.data[ATTR_CONFIG_ENTRY]}
coordinators: list[ScreenlogicDataUpdateCoordinator] = []
for entry_id in entry_ids:
config_entry = cast(
ScreenLogicConfigEntry | None,
hass.config_entries.async_get_entry(entry_id),
)
if not config_entry:
raise ServiceValidationError(
f"Failed to call service '{service_call.service}'. Config entry "
f"'{entry_id}' not found"
)
if not config_entry.domain == DOMAIN:
raise ServiceValidationError(
f"Failed to call service '{service_call.service}'. Config entry "
f"'{entry_id}' is not a {DOMAIN} config"
)
if not config_entry.state == ConfigEntryState.LOADED:
raise ServiceValidationError(
f"Failed to call service '{service_call.service}'. Config entry "
f"'{entry_id}' not loaded"
)
coordinators.append(config_entry.runtime_data)
return coordinators
async def async_set_color_mode(service_call: ServiceCall) -> None:
color_num = SUPPORTED_COLOR_MODES[service_call.data[ATTR_COLOR_MODE]]
coordinator: ScreenlogicDataUpdateCoordinator
for coordinator in await get_coordinators(service_call):
_LOGGER.debug(
"Service %s called on %s with mode %s",
SERVICE_SET_COLOR_MODE,
coordinator.gateway.name,
color_num,
)
try:
await coordinator.gateway.async_set_color_lights(color_num)
# Debounced refresh to catch any secondary changes in the device
await coordinator.async_request_refresh()
except ScreenLogicError as error:
raise HomeAssistantError(error) from error
async def async_set_super_chlor(
service_call: ServiceCall,
is_on: bool,
runtime: int | None = None,
) -> None:
coordinator: ScreenlogicDataUpdateCoordinator
for coordinator in await get_coordinators(service_call):
if EQUIPMENT_FLAG.CHLORINATOR not in coordinator.gateway.equipment_flags:
raise ServiceValidationError(
f"Equipment configuration for {coordinator.gateway.name} does not"
f" support {service_call.service}"
)
rt_log = f" with runtime {runtime}" if runtime else ""
_LOGGER.debug(
"Service %s called on %s%s",
service_call.service,
coordinator.gateway.name,
rt_log,
)
try:
await coordinator.gateway.async_set_scg_config(
super_chlor_timer=runtime, super_chlorinate=is_on
)
# Debounced refresh to catch any secondary changes in the device
await coordinator.async_request_refresh()
except ScreenLogicError as error:
raise HomeAssistantError(error) from error
async def async_start_super_chlor(service_call: ServiceCall) -> None:
runtime = service_call.data[ATTR_RUNTIME]
await async_set_super_chlor(service_call, True, runtime)
async def async_stop_super_chlor(service_call: ServiceCall) -> None:
await async_set_super_chlor(service_call, False)
hass.services.async_register(
DOMAIN, SERVICE_SET_COLOR_MODE, async_set_color_mode, SET_COLOR_MODE_SCHEMA
DOMAIN, SERVICE_SET_COLOR_MODE, _async_set_color_mode, SET_COLOR_MODE_SCHEMA
)
hass.services.async_register(
DOMAIN,
SERVICE_START_SUPER_CHLORINATION,
async_start_super_chlor,
_async_start_super_chlor,
TURN_ON_SUPER_CHLOR_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_STOP_SUPER_CHLORINATION,
async_stop_super_chlor,
_async_stop_super_chlor,
BASE_SERVICE_SCHEMA,
)