Refactor bsblan schedule service helpers (#172340)

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(