ha-inlite

__init__.py (5857B)

---

 """The in-lite integration."""
 
 from __future__ import annotations
 
 import logging
 import sys
 from pathlib import Path
 
 import voluptuous as vol
 
 # Make the bundled inlite_ble package importable as a top-level module.
 # This allows `from inlite_ble.hub import ...` to work without pip-installing.
 _LIB_DIR = str(Path(__file__).parent / "lib")
 if _LIB_DIR not in sys.path:
     sys.path.insert(0, _LIB_DIR)
 
 from homeassistant.components import bluetooth
 from homeassistant.components.bluetooth import BluetoothScanningMode
 from homeassistant.components.bluetooth.match import BluetoothCallbackMatcher
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import ATTR_DEVICE_ID, Platform
 from homeassistant.core import HomeAssistant, ServiceCall, callback
 from homeassistant.exceptions import ConfigEntryNotReady, ServiceValidationError
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.event import async_track_time_change
 from homeassistant.helpers.update_coordinator import UpdateFailed
 
 from .const import CONF_CONFIRM, CONF_GARDEN_ID, DOMAIN
 from .coordinator import InliteCoordinator
 # Preload the config flow with the integration package. Home Assistant imports
 # package-level modules in its import executor; loading this later for options
 # or setup would otherwise perform disk I/O from the event loop.
 from . import config_flow as _config_flow  # noqa: F401
 
 _LOGGER = logging.getLogger(__name__)
 
 PLATFORMS = [Platform.LIGHT]
 
 SERVICE_START_TEACH_IN = "start_teach_in"
 EVENT_TEACH_IN_PROGRESS = f"{DOMAIN}_teach_in_progress"
 DATA_COORDINATORS = "coordinators"
 
 type InliteConfigEntry = ConfigEntry[InliteCoordinator]
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: InliteConfigEntry) -> bool:
     """Set up in-lite from a config entry."""
     coordinator = InliteCoordinator(hass, entry)
 
     @callback
     def _async_update_ble(
         service_info: bluetooth.BluetoothServiceInfoBleak,
         change: bluetooth.BluetoothChange,
     ) -> None:
         """Update the cached BLE device from advertisement data."""
         coordinator.update_ble_service_info(service_info)
 
     entry.async_on_unload(
         bluetooth.async_register_callback(
             hass,
             _async_update_ble,
             BluetoothCallbackMatcher(address=entry.data[CONF_GARDEN_ID]),
             BluetoothScanningMode.ACTIVE,
         )
     )
 
     try:
         await coordinator.async_config_entry_first_refresh()
     except UpdateFailed as err:
         raise ConfigEntryNotReady("Hub not reachable") from err
 
     entry.runtime_data = coordinator
 
     coordinators: dict[str, InliteCoordinator] = hass.data.setdefault(
         DOMAIN, {}
     ).setdefault(DATA_COORDINATORS, {})
     coordinators[entry.entry_id] = coordinator
     _async_register_services(hass)
 
     @callback
     def _async_sync_clock_daily(_: datetime) -> None:
         """Start the daily local clock synchronization without blocking HA."""
         hass.async_create_task(coordinator.async_sync_clock())
 
     # Synchronize immediately after a successful local setup and then at
     # 03:00 in Home Assistant's configured local timezone every day.
     hass.async_create_task(coordinator.async_sync_clock())
     entry.async_on_unload(
         async_track_time_change(hass, _async_sync_clock_daily, hour=3, minute=0, second=0)
     )
 
     @callback
     def _async_remove_coordinator() -> None:
         """Remove this entry's coordinator and its last shared service."""
         coordinators.pop(entry.entry_id, None)
         if not coordinators:
             hass.services.async_remove(DOMAIN, SERVICE_START_TEACH_IN)
 
     entry.async_on_unload(_async_remove_coordinator)
 
     # Disconnect hubs when the config entry is unloaded
     entry.async_on_unload(coordinator.async_shutdown)
 
     # Reload integration when options change (scan interval, idle disconnect)
     entry.async_on_unload(entry.add_update_listener(_async_options_updated))
 
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
     return True
 
 
 async def _async_options_updated(
     hass: HomeAssistant, entry: InliteConfigEntry
 ) -> None:
     """Reload when options are updated."""
     await hass.config_entries.async_reload(entry.entry_id)
 
 
 async def async_unload_entry(hass: HomeAssistant, entry: InliteConfigEntry) -> bool:
     """Unload a config entry."""
     return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
 
 
 @callback
 def _async_register_services(hass: HomeAssistant) -> None:
     """Register the explicit, confirmation-gated teach-in service once."""
     if hass.services.has_service(DOMAIN, SERVICE_START_TEACH_IN):
         return
 
     schema = vol.Schema(
         {
             vol.Required(ATTR_DEVICE_ID): vol.Coerce(int),
             vol.Required(CONF_CONFIRM): cv.boolean,
         }
     )
 
     async def _async_start_teach_in(call: ServiceCall) -> None:
         if not call.data[CONF_CONFIRM]:
             raise ServiceValidationError(
                 "Set confirm: true to start teach-in; it changes mesh group assignments."
             )
 
         device_id = call.data[ATTR_DEVICE_ID]
         coordinators: dict[str, InliteCoordinator] = hass.data[DOMAIN][
             DATA_COORDINATORS
         ]
         matches = [
             coordinator
             for coordinator in coordinators.values()
             if device_id in coordinator.hubs
         ]
         if len(matches) != 1:
             raise ServiceValidationError(
                 f"Expected exactly one configured in-lite hub with device_id {device_id}."
             )
         if not await matches[0].async_start_teach_in(device_id):
             raise ServiceValidationError("The hub did not acknowledge the teach-in request.")
 
     hass.services.async_register(
         DOMAIN, SERVICE_START_TEACH_IN, _async_start_teach_in, schema=schema
     )