coordinator.py (16432B)
1 """DataUpdateCoordinator for in-lite integration. 2 3 Manages a persistent BLE connection with a connection lock to serialize 4 all hub communication. Connects once and queries all hubs before disconnecting. 5 Includes retry-with-reconnect for both commands and polling. 6 Receives OOB broadcast notifications for real-time state updates. 7 """ 8 9 from __future__ import annotations 10 11 import asyncio 12 import logging 13 from datetime import datetime, timedelta 14 from zoneinfo import ZoneInfo 15 16 from bleak_retry_connector import BleakClientWithServiceCache, establish_connection 17 from homeassistant.components import bluetooth 18 from homeassistant.config_entries import ConfigEntry 19 from homeassistant.core import HomeAssistant 20 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed 21 22 from inlite_ble.hub import InliteHub, ZoneState 23 24 from .const import ( 25 CONF_IDLE_DISCONNECT, 26 CONF_GARDEN_ID, 27 CONF_NETWORK_KEY, 28 CONF_PASSWORD, 29 CONF_SCAN_INTERVAL, 30 CONF_TRANSFORMERS, 31 DEFAULT_IDLE_DISCONNECT_SECONDS, 32 DEFAULT_SCAN_INTERVAL, 33 DOMAIN, 34 ) 35 36 _LOGGER = logging.getLogger(__name__) 37 38 MAX_COMMAND_ATTEMPTS = 3 39 MAX_POLL_ATTEMPTS = 2 40 RETRY_BACKOFF_SECONDS = 0.5 41 MAX_TIMEZONE_TRANSITIONS = 4 42 TIMEZONE_TRANSITION_LOOKAHEAD_DAYS = 365 * 3 43 44 45 class InliteCoordinator(DataUpdateCoordinator[dict[int, dict[int, ZoneState]]]): 46 """Coordinator that manages BLE communication with in-lite hubs. 47 48 Key reliability features: 49 - asyncio.Lock serializes all BLE operations (prevents race conditions) 50 - Persistent connection (connect once, reuse across polls and commands) 51 - Single BLE connection shared across all hubs (they share a gateway) 52 - Retry with disconnect-reconnect on command/poll failure 53 - Cached BLE device reference from advertisement callbacks 54 - OOB broadcast callback for real-time state push from the hub 55 - Proper cleanup on unload 56 """ 57 58 def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None: 59 """Initialize coordinator.""" 60 scan_interval = entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL) 61 super().__init__( 62 hass, 63 _LOGGER, 64 name=DOMAIN, 65 update_interval=timedelta(seconds=scan_interval), 66 ) 67 self.entry = entry 68 self._gateway_address: str = entry.data[CONF_GARDEN_ID] 69 self._hubs: dict[int, InliteHub] = {} 70 self._available = False 71 self._ble_lock = asyncio.Lock() 72 self._disconnect_timer: asyncio.TimerHandle | None = None 73 self._ble_service_info: bluetooth.BluetoothServiceInfoBleak | None = None 74 self._client: BleakClientWithServiceCache | None = None 75 self._active_hub: InliteHub | None = None 76 self._idle_disconnect_seconds = entry.options.get( 77 CONF_IDLE_DISCONNECT, DEFAULT_IDLE_DISCONNECT_SECONDS 78 ) 79 80 password = entry.data.get(CONF_PASSWORD) 81 network_key_hex = entry.data.get(CONF_NETWORK_KEY) 82 network_key = bytes.fromhex(network_key_hex) if network_key_hex else None 83 for tx_data in entry.data[CONF_TRANSFORMERS]: 84 device_id = tx_data["device_id"] 85 hub = InliteHub( 86 device_id=device_id, 87 passphrase=password, 88 network_key=network_key, 89 on_state_update=self._handle_oob_state_update, 90 on_teach_in_progress=self._handle_teach_in_progress, 91 ) 92 self._hubs[device_id] = hub 93 94 @property 95 def hubs(self) -> dict[int, InliteHub]: 96 return self._hubs 97 98 @property 99 def available(self) -> bool: 100 return self._available 101 102 def update_ble_service_info( 103 self, service_info: bluetooth.BluetoothServiceInfoBleak 104 ) -> None: 105 """Update the cached BLE device from an advertisement callback. 106 107 Called by the BLE callback registered in __init__.py. Keeps the device 108 reference fresh so _ensure_connected always uses the latest advertisement 109 (critical for ESPHome BLE proxy failover). 110 """ 111 self._ble_service_info = service_info 112 113 def _handle_oob_state_update(self) -> None: 114 """Handle an OOB broadcast notification from a hub. 115 116 Called (on the event loop via call_soon_threadsafe) when the hub receives 117 a state change broadcast (e.g., physical button press, timer trigger). 118 Builds the full state dict from all hubs and pushes it to HA entities. 119 """ 120 all_states: dict[int, dict[int, ZoneState]] = {} 121 for device_id, hub in self._hubs.items(): 122 if hub.zone_states: 123 all_states[device_id] = hub.zone_states 124 125 if all_states: 126 _LOGGER.debug("OOB state update received, pushing to HA entities") 127 self._available = True 128 self.async_set_updated_data(all_states) 129 130 def _handle_teach_in_progress(self, device_id: int, step: int) -> None: 131 """Publish the hub's local teach-in progress TLV as a HA event.""" 132 steps = { 133 0: "start_up", 134 10: "continue_to_next", 135 20: "teach_in_lines", 136 100: "wait", 137 250: "done", 138 } 139 self.hass.bus.async_fire( 140 f"{DOMAIN}_teach_in_progress", 141 { 142 "config_entry_id": self.entry.entry_id, 143 "device_id": device_id, 144 "step": steps.get(step, "unknown"), 145 "step_code": step, 146 }, 147 ) 148 149 def _find_ble_device(self) -> bluetooth.BluetoothServiceInfoBleak | None: 150 """Find this entry's gateway, preferring the freshest callback data.""" 151 if self._ble_service_info is not None: 152 return self._ble_service_info 153 return bluetooth.async_last_service_info( 154 self.hass, self._gateway_address, connectable=True 155 ) 156 157 async def _ensure_connected(self, hub: InliteHub) -> None: 158 """Ensure the hub has an active BLE connection, reconnecting if needed.""" 159 if self._client is not None and self._client.is_connected: 160 if not hub.is_connected: 161 await hub.connect(client=self._client, subscribe=False) 162 return 163 164 info = self._find_ble_device() 165 if info is None: 166 raise ConnectionError("in-lite hub not found in bluetooth scanner") 167 168 _LOGGER.debug("Connecting to %s via HA bluetooth", info.address) 169 client = await establish_connection( 170 BleakClientWithServiceCache, 171 info.device, 172 info.address, 173 max_attempts=3, 174 ) 175 176 self._client = client 177 connected = await hub.connect( 178 client=client, notification_handler=self._dispatch_notification 179 ) 180 if not connected: 181 self._client = None 182 raise ConnectionError("Hub notification setup failed") 183 184 def _dispatch_notification(self, sender: object, data: bytearray) -> None: 185 """Route notifications from the one shared BLE subscription.""" 186 if self._active_hub is not None: 187 self._active_hub.queue_notification(sender, data) 188 189 async def _disconnect_shared(self) -> None: 190 """Close the one gateway connection and detach every logical hub.""" 191 if self._client is not None: 192 try: 193 if self._client.is_connected: 194 await self._client.disconnect() 195 finally: 196 self._client = None 197 for hub in self._hubs.values(): 198 hub.detach_client() 199 200 def _schedule_idle_disconnect(self) -> None: 201 """Schedule a disconnect after the idle timeout.""" 202 self._cancel_idle_disconnect() 203 self._disconnect_timer = self.hass.loop.call_later( 204 self._idle_disconnect_seconds, 205 lambda: self.hass.async_create_task(self._idle_disconnect()), 206 ) 207 208 def _cancel_idle_disconnect(self) -> None: 209 """Cancel pending idle disconnect.""" 210 if self._disconnect_timer is not None: 211 self._disconnect_timer.cancel() 212 self._disconnect_timer = None 213 214 async def _idle_disconnect(self) -> None: 215 """Disconnect all hubs after idle timeout.""" 216 async with self._ble_lock: 217 await self._disconnect_shared() 218 _LOGGER.debug("Idle disconnect completed") 219 220 async def _async_update_data(self) -> dict[int, dict[int, ZoneState]]: 221 """Poll all hubs for zone states. 222 223 Connects once, queries all hubs, then schedules idle disconnect. 224 Retries once per hub on failure (disconnect-reconnect between attempts). 225 All operations are serialized under the BLE lock. 226 """ 227 async with self._ble_lock: 228 self._cancel_idle_disconnect() 229 all_states: dict[int, dict[int, ZoneState]] = {} 230 231 # A missing advertisement cannot be recovered by reconnecting the 232 # logical mesh hub. Fail the whole refresh once and let HA retry 233 # setup/polling when the address becomes visible again. 234 if ( 235 (self._client is None or not self._client.is_connected) 236 and self._find_ble_device() is None 237 ): 238 self._available = False 239 raise UpdateFailed( 240 f"in-lite gateway {self._gateway_address} is not available" 241 ) 242 243 for device_id, hub in self._hubs.items(): 244 for attempt in range(MAX_POLL_ATTEMPTS): 245 try: 246 await self._ensure_connected(hub) 247 self._active_hub = hub 248 states = await hub.query_zone_states() 249 all_states[device_id] = states 250 self._available = True 251 break 252 except Exception as err: 253 _LOGGER.warning( 254 "Poll attempt %d/%d for hub 0x%04X failed: %s", 255 attempt + 1, MAX_POLL_ATTEMPTS, device_id, err, 256 ) 257 await self._disconnect_shared() 258 if attempt < MAX_POLL_ATTEMPTS - 1: 259 await asyncio.sleep(RETRY_BACKOFF_SECONDS) 260 261 self._schedule_idle_disconnect() 262 263 if not all_states and self._hubs: 264 self._available = False 265 raise UpdateFailed("Could not connect to any hub") 266 267 return all_states 268 269 async def async_send_command( 270 self, device_id: int, output_id: int, on: bool 271 ) -> bool: 272 """Send an ON/OFF command to a specific zone. 273 274 Retries up to MAX_COMMAND_ATTEMPTS times with disconnect-reconnect 275 between attempts and increasing backoff. The BLE lock is released 276 between retries so polling can still proceed. 277 """ 278 hub = self._hubs.get(device_id) 279 if hub is None: 280 _LOGGER.error("No hub with device_id 0x%04X", device_id) 281 return False 282 283 last_error: Exception | None = None 284 for attempt in range(MAX_COMMAND_ATTEMPTS): 285 async with self._ble_lock: 286 self._cancel_idle_disconnect() 287 try: 288 await self._ensure_connected(hub) 289 self._active_hub = hub 290 result = await hub.set_outlet_mode(output_id, on) 291 if result: 292 self._schedule_idle_disconnect() 293 return True 294 # Command sent but hub didn't ACK — disconnect and retry 295 _LOGGER.debug( 296 "Command attempt %d/%d for hub 0x%04X zone %d: no ACK", 297 attempt + 1, MAX_COMMAND_ATTEMPTS, device_id, output_id, 298 ) 299 await self._disconnect_shared() 300 except Exception as err: 301 last_error = err 302 _LOGGER.debug( 303 "Command attempt %d/%d for hub 0x%04X zone %d failed: %s", 304 attempt + 1, MAX_COMMAND_ATTEMPTS, device_id, output_id, err, 305 ) 306 await self._disconnect_shared() 307 308 # Backoff between retries (lock released so other operations can proceed) 309 if attempt < MAX_COMMAND_ATTEMPTS - 1: 310 await asyncio.sleep(RETRY_BACKOFF_SECONDS * (attempt + 1)) 311 312 _LOGGER.error( 313 "Command to hub 0x%04X zone %d failed after %d attempts: %s", 314 device_id, output_id, MAX_COMMAND_ATTEMPTS, last_error, 315 ) 316 self._schedule_idle_disconnect() 317 return False 318 319 async def async_sync_clock(self) -> None: 320 """Synchronize all configured hubs to HA's current local time.""" 321 timezone = ZoneInfo(self.hass.config.time_zone) 322 now = datetime.now(timezone) 323 transitions = _find_timezone_transitions(now, timezone) 324 async with self._ble_lock: 325 self._cancel_idle_disconnect() 326 for device_id, hub in self._hubs.items(): 327 try: 328 await self._ensure_connected(hub) 329 self._active_hub = hub 330 if not await hub.sync_clock(now, transitions): 331 _LOGGER.warning("Clock sync was not acknowledged by hub 0x%04X", device_id) 332 except Exception as err: 333 _LOGGER.warning("Clock sync for hub 0x%04X failed: %s", device_id, err) 334 await self._disconnect_shared() 335 self._schedule_idle_disconnect() 336 337 async def async_start_teach_in(self, device_id: int) -> bool: 338 """Start teach-in on one hub after the caller has confirmed it.""" 339 hub = self._hubs.get(device_id) 340 if hub is None: 341 return False 342 async with self._ble_lock: 343 self._cancel_idle_disconnect() 344 try: 345 await self._ensure_connected(hub) 346 self._active_hub = hub 347 result = await hub.start_teach_in() 348 except Exception as err: 349 _LOGGER.warning("Teach-in for hub 0x%04X failed: %s", device_id, err) 350 await self._disconnect_shared() 351 return False 352 self._schedule_idle_disconnect() 353 return result 354 355 async def async_shutdown(self) -> None: 356 """Disconnect all hubs (called on unload).""" 357 self._cancel_idle_disconnect() 358 async with self._ble_lock: 359 await self._disconnect_shared() 360 _LOGGER.debug("All hubs disconnected") 361 362 363 def _find_timezone_transitions( 364 now: datetime, timezone: ZoneInfo 365 ) -> list[tuple[int, int]]: 366 """Return up to four exact future offset changes for the hub RTC protocol.""" 367 current_timestamp = int(now.timestamp()) 368 current_offset = int((now.utcoffset() or timedelta()).total_seconds()) 369 transitions: list[tuple[int, int]] = [] 370 previous_timestamp = current_timestamp 371 previous_offset = current_offset 372 373 # IANA timezone changes are sparse. Daily probes make this inexpensive and 374 # a binary search below finds the precise UTC second of each change. 375 for day in range(1, TIMEZONE_TRANSITION_LOOKAHEAD_DAYS + 1): 376 candidate_timestamp = current_timestamp + day * 24 * 60 * 60 377 candidate = datetime.fromtimestamp(candidate_timestamp, timezone) 378 candidate_offset = int((candidate.utcoffset() or timedelta()).total_seconds()) 379 if candidate_offset == previous_offset: 380 previous_timestamp = candidate_timestamp 381 continue 382 383 lower = previous_timestamp 384 upper = candidate_timestamp 385 while upper - lower > 1: 386 midpoint = (lower + upper) // 2 387 midpoint_offset = int( 388 (datetime.fromtimestamp(midpoint, timezone).utcoffset() or timedelta()).total_seconds() 389 ) 390 if midpoint_offset == previous_offset: 391 lower = midpoint 392 else: 393 upper = midpoint 394 395 transition_offset = int( 396 (datetime.fromtimestamp(upper, timezone).utcoffset() or timedelta()).total_seconds() 397 ) 398 transitions.append((upper, transition_offset)) 399 if len(transitions) == MAX_TIMEZONE_TRANSITIONS: 400 return transitions 401 previous_timestamp = candidate_timestamp 402 previous_offset = candidate_offset 403 404 return transitions