From 3cd3082eba32cb2a5ee5d348596b02cc3e67d674 Mon Sep 17 00:00:00 2001 From: Willem-Jan van Rootselaar Date: Wed, 8 Jul 2026 16:54:03 +0200 Subject: [PATCH] Refactor bsblan schedule service helpers (#172340) --- homeassistant/components/bsblan/services.py | 145 +++++++------------- 1 file changed, 50 insertions(+), 95 deletions(-) diff --git a/homeassistant/components/bsblan/services.py b/homeassistant/components/bsblan/services.py index 315320fd5b83..f48ea2552213 100644 --- a/homeassistant/components/bsblan/services.py +++ b/homeassistant/components/bsblan/services.py @@ -2,7 +2,7 @@ from datetime import time import logging -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any, Final from bsblan import BSBLANError, DaySchedule, DHWSchedule, TimeSlot import voluptuous as vol @@ -29,6 +29,16 @@ ATTR_FRIDAY_SLOTS = "friday_slots" ATTR_SATURDAY_SLOTS = "saturday_slots" ATTR_SUNDAY_SLOTS = "sunday_slots" +_DAY_NAME_SLOT_ATTR_PAIRS: tuple[tuple[str, str], ...] = ( + ("monday", ATTR_MONDAY_SLOTS), + ("tuesday", ATTR_TUESDAY_SLOTS), + ("wednesday", ATTR_WEDNESDAY_SLOTS), + ("thursday", ATTR_THURSDAY_SLOTS), + ("friday", ATTR_FRIDAY_SLOTS), + ("saturday", ATTR_SATURDAY_SLOTS), + ("sunday", ATTR_SUNDAY_SLOTS), +) + # Schema for a single time slot _SLOT_SCHEMA = vol.Schema( @@ -39,16 +49,16 @@ _SLOT_SCHEMA = vol.Schema( ) +_WEEKLY_SCHEDULE_FIELDS: Final[dict[vol.Marker, Any]] = { + vol.Optional(slot_attr): vol.All(cv.ensure_list, [_SLOT_SCHEMA]) + for _, slot_attr in _DAY_NAME_SLOT_ATTR_PAIRS +} + + SERVICE_SET_HOT_WATER_SCHEDULE_SCHEMA = vol.Schema( { vol.Required(ATTR_DEVICE_ID): cv.string, - vol.Optional(ATTR_MONDAY_SLOTS): vol.All(cv.ensure_list, [_SLOT_SCHEMA]), - vol.Optional(ATTR_TUESDAY_SLOTS): vol.All(cv.ensure_list, [_SLOT_SCHEMA]), - vol.Optional(ATTR_WEDNESDAY_SLOTS): vol.All(cv.ensure_list, [_SLOT_SCHEMA]), - vol.Optional(ATTR_THURSDAY_SLOTS): vol.All(cv.ensure_list, [_SLOT_SCHEMA]), - vol.Optional(ATTR_FRIDAY_SLOTS): vol.All(cv.ensure_list, [_SLOT_SCHEMA]), - vol.Optional(ATTR_SATURDAY_SLOTS): vol.All(cv.ensure_list, [_SLOT_SCHEMA]), - vol.Optional(ATTR_SUNDAY_SLOTS): vol.All(cv.ensure_list, [_SLOT_SCHEMA]), + **_WEEKLY_SCHEDULE_FIELDS, } ) @@ -98,11 +108,26 @@ def _convert_time_slots_to_day_schedule( return DaySchedule(slots=time_slots) -async def set_hot_water_schedule(service_call: ServiceCall) -> None: - """Set hot water heating schedule.""" - device_id = service_call.data[ATTR_DEVICE_ID] +def _build_weekly_schedule_days( + service_call: ServiceCall, +) -> dict[str, DaySchedule | None]: + """Build day-name -> schedule values from the service call data. + + Days omitted from the service call map to None, which tells python-bsblan not to + modify that day. + """ + return { + day_name: _convert_time_slots_to_day_schedule(service_call.data.get(attr_name)) + for day_name, attr_name in _DAY_NAME_SLOT_ATTR_PAIRS + } + + +def _resolve_config_entry( + service_call: ServiceCall, +) -> tuple[BSBLanConfigEntry, dr.DeviceEntry]: + """Resolve device_id from a service call into a loaded BSBLAN config entry.""" + device_id: str = service_call.data[ATTR_DEVICE_ID] - # Get the device and config entry device_registry = dr.async_get(service_call.hass) device_entry = device_registry.async_get(device_id) @@ -137,56 +162,20 @@ async def set_hot_water_schedule(service_call: ServiceCall) -> None: translation_placeholders={"device_name": device_entry.name or device_id}, ) + return entry, device_entry + + +async def set_hot_water_schedule(service_call: ServiceCall) -> None: + """Set hot water heating schedule.""" + entry, _ = _resolve_config_entry(service_call) client = entry.runtime_data.client - # Convert time slots to DaySchedule objects - monday = _convert_time_slots_to_day_schedule( - service_call.data.get(ATTR_MONDAY_SLOTS) - ) - tuesday = _convert_time_slots_to_day_schedule( - service_call.data.get(ATTR_TUESDAY_SLOTS) - ) - wednesday = _convert_time_slots_to_day_schedule( - service_call.data.get(ATTR_WEDNESDAY_SLOTS) - ) - thursday = _convert_time_slots_to_day_schedule( - service_call.data.get(ATTR_THURSDAY_SLOTS) - ) - friday = _convert_time_slots_to_day_schedule( - service_call.data.get(ATTR_FRIDAY_SLOTS) - ) - saturday = _convert_time_slots_to_day_schedule( - service_call.data.get(ATTR_SATURDAY_SLOTS) - ) - sunday = _convert_time_slots_to_day_schedule( - service_call.data.get(ATTR_SUNDAY_SLOTS) - ) + days = _build_weekly_schedule_days(service_call) + dhw_schedule = DHWSchedule(**days) - # Create the DHWSchedule object - dhw_schedule = DHWSchedule( - monday=monday, - tuesday=tuesday, - wednesday=wednesday, - thursday=thursday, - friday=friday, - saturday=saturday, - sunday=sunday, - ) - - LOGGER.debug( - "Setting hot water schedule - Monday: %s, Tuesday: %s, Wednesday: %s, " - "Thursday: %s, Friday: %s, Saturday: %s, Sunday: %s", - monday, - tuesday, - wednesday, - thursday, - friday, - saturday, - sunday, - ) + LOGGER.debug("Setting hot water schedule: %s", dhw_schedule) try: - # Call the BSB-LAN API to set the schedule await client.set_hot_water_schedule(dhw_schedule) except BSBLANError as err: raise HomeAssistantError( @@ -201,45 +190,11 @@ async def set_hot_water_schedule(service_call: ServiceCall) -> None: async def async_sync_time(service_call: ServiceCall) -> None: """Synchronize BSB-LAN device time with Home Assistant.""" - device_id: str = service_call.data[ATTR_DEVICE_ID] - - # Get the device and config entry - device_registry = dr.async_get(service_call.hass) - device_entry = device_registry.async_get(device_id) - - if device_entry is None: - raise ServiceValidationError( - translation_domain=DOMAIN, - translation_key="invalid_device_id", - translation_placeholders={"device_id": device_id}, - ) - - # Find the config entry for this device - matching_entries: list[BSBLanConfigEntry] = [ - entry - for entry in service_call.hass.config_entries.async_entries(DOMAIN) - if entry.entry_id in device_entry.config_entries - ] - - if not matching_entries: - raise ServiceValidationError( - translation_domain=DOMAIN, - translation_key="no_config_entry_for_device", - translation_placeholders={"device_id": device_entry.name or device_id}, - ) - - entry = matching_entries[0] - - # Verify the config entry is loaded - if entry.state is not ConfigEntryState.LOADED: - raise ServiceValidationError( - translation_domain=DOMAIN, - translation_key="config_entry_not_loaded", - translation_placeholders={"device_name": device_entry.name or device_id}, - ) - + entry, device_entry = _resolve_config_entry(service_call) client = entry.runtime_data.client - await async_sync_device_time(client, device_entry.name or device_id) + await async_sync_device_time( + client, device_entry.name or service_call.data[ATTR_DEVICE_ID] + ) SYNC_TIME_SCHEMA = vol.Schema(