coordinator.py (15885B)
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 BLE_LOCAL_NAME, 26 CONF_IDLE_DISCONNECT, 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._hubs: dict[int, InliteHub] = {} 69 self._available = False 70 self._ble_lock = asyncio.Lock() 71 self._disconnect_timer: asyncio.TimerHandle | None = None 72 self._ble_service_info: bluetooth.BluetoothServiceInfoBleak | None = None 73 self._client: BleakClientWithServiceCache | None = None 74 self._active_hub: InliteHub | None = None 75 self._idle_disconnect_seconds = entry.options.get( 76 CONF_IDLE_DISCONNECT, DEFAULT_IDLE_DISCONNECT_SECONDS 77 ) 78 79 password = entry.data.get(CONF_PASSWORD) 80 network_key_hex = entry.data.get(CONF_NETWORK_KEY) 81 network_key = bytes.fromhex(network_key_hex) if network_key_hex else None 82 for tx_data in entry.data[CONF_TRANSFORMERS]: 83 device_id = tx_data["device_id"] 84 hub = InliteHub( 85 device_id=device_id, 86 passphrase=password, 87 network_key=network_key, 88 on_state_update=self._handle_oob_state_update, 89 on_teach_in_progress=self._handle_teach_in_progress, 90 ) 91 self._hubs[device_id] = hub 92 93 @property 94 def hubs(self) -> dict[int, InliteHub]: 95 return self._hubs 96 97 @property 98 def available(self) -> bool: 99 return self._available 100 101 def update_ble_service_info( 102 self, service_info: bluetooth.BluetoothServiceInfoBleak 103 ) -> None: 104 """Update the cached BLE device from an advertisement callback. 105 106 Called by the BLE callback registered in __init__.py. Keeps the device 107 reference fresh so _ensure_connected always uses the latest advertisement 108 (critical for ESPHome BLE proxy failover). 109 """ 110 self._ble_service_info = service_info 111 112 def _handle_oob_state_update(self) -> None: 113 """Handle an OOB broadcast notification from a hub. 114 115 Called (on the event loop via call_soon_threadsafe) when the hub receives 116 a state change broadcast (e.g., physical button press, timer trigger). 117 Builds the full state dict from all hubs and pushes it to HA entities. 118 """ 119 all_states: dict[int, dict[int, ZoneState]] = {} 120 for device_id, hub in self._hubs.items(): 121 if hub.zone_states: 122 all_states[device_id] = hub.zone_states 123 124 if all_states: 125 _LOGGER.debug("OOB state update received, pushing to HA entities") 126 self._available = True 127 self.async_set_updated_data(all_states) 128 129 def _handle_teach_in_progress(self, device_id: int, step: int) -> None: 130 """Publish the hub's local teach-in progress TLV as a HA event.""" 131 steps = { 132 0: "start_up", 133 10: "continue_to_next", 134 20: "teach_in_lines", 135 100: "wait", 136 250: "done", 137 } 138 self.hass.bus.async_fire( 139 f"{DOMAIN}_teach_in_progress", 140 { 141 "config_entry_id": self.entry.entry_id, 142 "device_id": device_id, 143 "step": steps.get(step, "unknown"), 144 "step_code": step, 145 }, 146 ) 147 148 def _find_ble_device(self) -> bluetooth.BluetoothServiceInfoBleak | None: 149 """Find the in-lite hub, preferring the cached reference.""" 150 if self._ble_service_info is not None: 151 return self._ble_service_info 152 for info in bluetooth.async_discovered_service_info(self.hass, connectable=True): 153 if info.name and info.name.lower() == BLE_LOCAL_NAME: 154 return info 155 return None 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.handle_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 for device_id, hub in self._hubs.items(): 232 for attempt in range(MAX_POLL_ATTEMPTS): 233 try: 234 await self._ensure_connected(hub) 235 self._active_hub = hub 236 states = await hub.query_zone_states() 237 all_states[device_id] = states 238 self._available = True 239 break 240 except Exception as err: 241 _LOGGER.warning( 242 "Poll attempt %d/%d for hub 0x%04X failed: %s", 243 attempt + 1, MAX_POLL_ATTEMPTS, device_id, err, 244 ) 245 await self._disconnect_shared() 246 if attempt < MAX_POLL_ATTEMPTS - 1: 247 await asyncio.sleep(RETRY_BACKOFF_SECONDS) 248 249 self._schedule_idle_disconnect() 250 251 if not all_states and self._hubs: 252 self._available = False 253 raise UpdateFailed("Could not connect to any hub") 254 255 return all_states 256 257 async def async_send_command( 258 self, device_id: int, output_id: int, on: bool 259 ) -> bool: 260 """Send an ON/OFF command to a specific zone. 261 262 Retries up to MAX_COMMAND_ATTEMPTS times with disconnect-reconnect 263 between attempts and increasing backoff. The BLE lock is released 264 between retries so polling can still proceed. 265 """ 266 hub = self._hubs.get(device_id) 267 if hub is None: 268 _LOGGER.error("No hub with device_id 0x%04X", device_id) 269 return False 270 271 last_error: Exception | None = None 272 for attempt in range(MAX_COMMAND_ATTEMPTS): 273 async with self._ble_lock: 274 self._cancel_idle_disconnect() 275 try: 276 await self._ensure_connected(hub) 277 self._active_hub = hub 278 result = await hub.set_outlet_mode(output_id, on) 279 if result: 280 self._schedule_idle_disconnect() 281 return True 282 # Command sent but hub didn't ACK — disconnect and retry 283 _LOGGER.debug( 284 "Command attempt %d/%d for hub 0x%04X zone %d: no ACK", 285 attempt + 1, MAX_COMMAND_ATTEMPTS, device_id, output_id, 286 ) 287 await self._disconnect_shared() 288 except Exception as err: 289 last_error = err 290 _LOGGER.debug( 291 "Command attempt %d/%d for hub 0x%04X zone %d failed: %s", 292 attempt + 1, MAX_COMMAND_ATTEMPTS, device_id, output_id, err, 293 ) 294 await self._disconnect_shared() 295 296 # Backoff between retries (lock released so other operations can proceed) 297 if attempt < MAX_COMMAND_ATTEMPTS - 1: 298 await asyncio.sleep(RETRY_BACKOFF_SECONDS * (attempt + 1)) 299 300 _LOGGER.error( 301 "Command to hub 0x%04X zone %d failed after %d attempts: %s", 302 device_id, output_id, MAX_COMMAND_ATTEMPTS, last_error, 303 ) 304 self._schedule_idle_disconnect() 305 return False 306 307 async def async_sync_clock(self) -> None: 308 """Synchronize all configured hubs to HA's current local time.""" 309 timezone = ZoneInfo(self.hass.config.time_zone) 310 now = datetime.now(timezone) 311 transitions = _find_timezone_transitions(now, timezone) 312 async with self._ble_lock: 313 self._cancel_idle_disconnect() 314 for device_id, hub in self._hubs.items(): 315 try: 316 await self._ensure_connected(hub) 317 self._active_hub = hub 318 if not await hub.sync_clock(now, transitions): 319 _LOGGER.warning("Clock sync was not acknowledged by hub 0x%04X", device_id) 320 except Exception as err: 321 _LOGGER.warning("Clock sync for hub 0x%04X failed: %s", device_id, err) 322 await self._disconnect_shared() 323 self._schedule_idle_disconnect() 324 325 async def async_start_teach_in(self, device_id: int) -> bool: 326 """Start teach-in on one hub after the caller has confirmed it.""" 327 hub = self._hubs.get(device_id) 328 if hub is None: 329 return False 330 async with self._ble_lock: 331 self._cancel_idle_disconnect() 332 try: 333 await self._ensure_connected(hub) 334 self._active_hub = hub 335 result = await hub.start_teach_in() 336 except Exception as err: 337 _LOGGER.warning("Teach-in for hub 0x%04X failed: %s", device_id, err) 338 await self._disconnect_shared() 339 return False 340 self._schedule_idle_disconnect() 341 return result 342 343 async def async_shutdown(self) -> None: 344 """Disconnect all hubs (called on unload).""" 345 self._cancel_idle_disconnect() 346 async with self._ble_lock: 347 await self._disconnect_shared() 348 _LOGGER.debug("All hubs disconnected") 349 350 351 def _find_timezone_transitions( 352 now: datetime, timezone: ZoneInfo 353 ) -> list[tuple[int, int]]: 354 """Return up to four exact future offset changes for the hub RTC protocol.""" 355 current_timestamp = int(now.timestamp()) 356 current_offset = int((now.utcoffset() or timedelta()).total_seconds()) 357 transitions: list[tuple[int, int]] = [] 358 previous_timestamp = current_timestamp 359 previous_offset = current_offset 360 361 # IANA timezone changes are sparse. Daily probes make this inexpensive and 362 # a binary search below finds the precise UTC second of each change. 363 for day in range(1, TIMEZONE_TRANSITION_LOOKAHEAD_DAYS + 1): 364 candidate_timestamp = current_timestamp + day * 24 * 60 * 60 365 candidate = datetime.fromtimestamp(candidate_timestamp, timezone) 366 candidate_offset = int((candidate.utcoffset() or timedelta()).total_seconds()) 367 if candidate_offset == previous_offset: 368 previous_timestamp = candidate_timestamp 369 continue 370 371 lower = previous_timestamp 372 upper = candidate_timestamp 373 while upper - lower > 1: 374 midpoint = (lower + upper) // 2 375 midpoint_offset = int( 376 (datetime.fromtimestamp(midpoint, timezone).utcoffset() or timedelta()).total_seconds() 377 ) 378 if midpoint_offset == previous_offset: 379 lower = midpoint 380 else: 381 upper = midpoint 382 383 transition_offset = int( 384 (datetime.fromtimestamp(upper, timezone).utcoffset() or timedelta()).total_seconds() 385 ) 386 transitions.append((upper, transition_offset)) 387 if len(transitions) == MAX_TIMEZONE_TRANSITIONS: 388 return transitions 389 previous_timestamp = candidate_timestamp 390 previous_offset = candidate_offset 391 392 return transitions