From 5b1e2ecdd5eabb1bbe6f552bdc2baa4b07528ea6 Mon Sep 17 00:00:00 2001 From: yuzhuohuang Date: Mon, 3 Aug 2026 17:06:08 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8Ddocker=E7=8E=AF=E5=A2=83?= =?UTF-8?q?=E4=B8=8B=E8=AE=BE=E5=A4=87=E9=87=8D=E5=90=AF=E5=AF=BC=E8=87=B4?= =?UTF-8?q?=E6=97=A0=E6=B3=95=E6=90=9C=E7=B4=A2=E5=88=B0=E8=AE=BE=E5=A4=87?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- custom_components/em1003/__init__.py | 59 ++++- custom_components/em1003/const.py | 14 ++ custom_components/em1003/device.py | 286 +++++++++++++++++++------ custom_components/em1003/manifest.json | 2 +- custom_components/em1003/switch.py | 20 +- 5 files changed, 301 insertions(+), 80 deletions(-) diff --git a/custom_components/em1003/__init__.py b/custom_components/em1003/__init__.py index 32a784a..23368e4 100644 --- a/custom_components/em1003/__init__.py +++ b/custom_components/em1003/__init__.py @@ -9,8 +9,8 @@ from homeassistant.components import bluetooth from homeassistant.config_entries import ConfigEntry -from homeassistant.const import Platform -from homeassistant.core import HomeAssistant, ServiceCall +from homeassistant.const import EVENT_HOMEASSISTANT_STOP, Platform +from homeassistant.core import Event, HomeAssistant, ServiceCall, callback from homeassistant.helpers import device_registry as dr import voluptuous as vol @@ -36,6 +36,25 @@ PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.SWITCH] +@callback +def async_advertised_name(hass: HomeAssistant, mac_address: str) -> str | None: + """Return the name carried in the BLE advertisement, without connecting. + + Opening a link here would race with the platform setups, and BlueZ then rejects + both connects with "Operation already in progress". + """ + device = bluetooth.async_ble_device_from_address(hass, mac_address, connectable=True) + + if device is None or not device.name: + return None + + name = device.name.strip() + if not name or name == mac_address: + return None + + return name + + async def async_read_device_name(hass: HomeAssistant, mac_address: str) -> str | None: """Read device name from BLE device using Device Name characteristic. @@ -65,6 +84,14 @@ async def async_read_device_name(hass: HomeAssistant, mac_address: str) -> str | try: _LOGGER.debug("Connected to device %s to read name", mac_address) + # Not every device exposes Generic Access, which is not an error. + if client.services.get_characteristic(DEVICE_NAME_UUID) is None: + _LOGGER.debug( + "Device %s does not expose the Device Name characteristic", + mac_address, + ) + return None + # Read the Device Name characteristic (0x2A00) value = await client.read_gatt_char(DEVICE_NAME_UUID) device_name = value.decode('utf-8').strip() @@ -91,17 +118,29 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: _LOGGER.info("Setting up EM1003 device with MAC: %s", mac_address) - # Try to read device name from BLE device - device_name = await async_read_device_name(hass, mac_address) + # Create the device first so a link left open by a previous run can be released + # before anything tries to connect. BlueZ lives outside the container, so an + # unclean restart leaves the device connected and therefore not advertising. + em1003_device = EM1003Device(hass, mac_address) + await em1003_device.async_release_stale_connection() + + device_name = async_advertised_name(hass, mac_address) if device_name: - _LOGGER.info("Successfully read device name: %s", device_name) + _LOGGER.info("Using advertised device name: %s", device_name) else: - _LOGGER.warning("Could not read device name, using default: %s", entry.title) + _LOGGER.debug("No advertised name for %s, using %s", mac_address, entry.title) device_name = entry.title - # Create EM1003 device instance with device name - em1003_device = EM1003Device(hass, mac_address, device_name) + em1003_device.device_name = device_name + + async def _async_release_link_on_stop(_event: Event) -> None: + """Drop the BLE link so the device advertises again after a restart.""" + await em1003_device.async_shutdown() + + entry.async_on_unload( + hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _async_release_link_on_stop) + ) # Register device in device registry before creating entities device_registry = dr.async_get(hass) @@ -135,7 +174,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Unload a config entry.""" if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS): - hass.data[DOMAIN].pop(entry.entry_id) + data = hass.data[DOMAIN].pop(entry.entry_id) + # Release the link, otherwise the device stays silent for the next setup. + await data["device"].async_shutdown() return unload_ok diff --git a/custom_components/em1003/const.py b/custom_components/em1003/const.py index 96dab21..7b61172 100644 --- a/custom_components/em1003/const.py +++ b/custom_components/em1003/const.py @@ -129,5 +129,19 @@ DEFAULT_SCAN_INTERVAL = 60 # Default polling interval in seconds DEVICE_TIMEOUT = 30.0 +# Stale connection recovery. +# BlueZ runs on the host, so an unclean container/HA restart leaves the ACL link +# open. The device then stops advertising and no scan can find it again. +STALE_SETTLE_TIME = 3.0 # Seconds to wait after tearing the link down +STALE_REDISCOVER_TIMEOUT = 12.0 # How long to wait for the device to advertise again +STALE_REDISCOVER_INTERVAL = 0.5 # Poll interval while waiting for re-discovery + +# BlueZ answers "Operation already in progress" while it is still finishing a +# previous connect or disconnect, which is common right after a restart. It rejects +# the call without starting a connection, so no BLE slot is consumed and retrying is +# safe. +BLUEZ_BUSY_RETRY_DELAY = 5.0 +BLUEZ_BUSY_MAX_ATTEMPTS = 4 + # Version VERSION = "1.0.3" diff --git a/custom_components/em1003/device.py b/custom_components/em1003/device.py index 3b69765..6fdfe9d 100644 --- a/custom_components/em1003/device.py +++ b/custom_components/em1003/device.py @@ -9,7 +9,12 @@ from bleak import BleakClient from bleak.exc import BleakError -from bleak_retry_connector import establish_connection +from bleak_retry_connector import ( + IS_LINUX, + close_stale_connections_by_address, + establish_connection, +) +from bleak_retry_connector import get_device as get_bluez_device from homeassistant.components import bluetooth from homeassistant.core import HomeAssistant @@ -23,6 +28,11 @@ EM1003_NOTIFY_CHAR_UUID, EM1003_WRITE_CHAR_UUID, SENSOR_TYPES, + BLUEZ_BUSY_MAX_ATTEMPTS, + BLUEZ_BUSY_RETRY_DELAY, + STALE_REDISCOVER_INTERVAL, + STALE_REDISCOVER_TIMEOUT, + STALE_SETTLE_TIME, ) _LOGGER = logging.getLogger(__name__) @@ -159,6 +169,9 @@ def __init__(self, hass: HomeAssistant, mac_address: str, device_name: str | Non self.mac_address = mac_address self.device_name = device_name or mac_address # Use MAC as fallback self._client: BleakClient | None = None + # BlueZ rejects a second connect while one is still in flight, so only one + # caller may establish the link; the rest reuse the client it produced. + self._connect_lock = asyncio.Lock() self.sensor_data: dict[int, float | None] = {} self.buzzer_state: bool | None = None # Buzzer state (True=on, False=off, None=unknown) self._last_disconnect_time: float | None = None @@ -287,6 +300,121 @@ async def _ensure_connection_delay(self) -> None: _LOGGER.info("Waiting %.1fs before retry (connection abort backoff)", abort_backoff) await asyncio.sleep(abort_backoff) + async def _async_bluez_device(self): + """Query BlueZ directly for the device object. + + Unlike Home Assistant's advertisement cache, BlueZ keeps the device object + around even while the device is connected and therefore not advertising. + """ + if not IS_LINUX: + return None + try: + return await get_bluez_device(self.mac_address) + except Exception as err: + _LOGGER.debug("[STALE] Could not query BlueZ for %s: %s", self.mac_address, err) + return None + + async def async_release_stale_connection(self) -> bool: + """Tear down a connection left open in the host Bluetooth stack. + + BlueZ runs on the host, so an unclean container or Home Assistant restart + leaves the ACL link established. A connected BLE peripheral stops + advertising, which makes it undiscoverable until that link is dropped - + previously this could only be resolved by power cycling the device. + + Returns: + True if a stale connection was found and released. + """ + device = await self._async_bluez_device() + if device is None: + return False + + props = (device.details or {}).get("props") or {} + if not props.get("Connected"): + return False + + _LOGGER.warning( + "[STALE] %s is still connected in the host Bluetooth stack but is not " + "advertising. Releasing the stale link to make it discoverable again", + self._device_id(), + ) + try: + await close_stale_connections_by_address( + self.mac_address, only_other_adapters=False + ) + except Exception as err: + _LOGGER.error("[STALE] ✗ Failed to release stale connection: %s", err) + return False + + self._client = None + self._last_disconnect_time = time.time() + + # Give BlueZ time to finish the teardown, otherwise the next connect gets + # rejected with "Operation already in progress". + await asyncio.sleep(STALE_SETTLE_TIME) + + _LOGGER.info("[STALE] ✓ Released stale connection to %s", self.mac_address) + return True + + async def _async_rediscover(self): + """Wait for the device to advertise again after its link was torn down.""" + deadline = time.time() + STALE_REDISCOVER_TIMEOUT + while time.time() < deadline: + device = bluetooth.async_ble_device_from_address( + self.hass, self.mac_address, connectable=True + ) + if device: + _LOGGER.info("[STALE] ✓ %s is advertising again", self.mac_address) + return device + await asyncio.sleep(STALE_REDISCOVER_INTERVAL) + + # Advertising can take longer than we are willing to wait here, but BlueZ + # still knows the device and bleak can connect using that object directly. + device = await self._async_bluez_device() + if device: + _LOGGER.info( + "[STALE] %s is not advertising yet, connecting via the BlueZ device object", + self.mac_address, + ) + return device + + async def _connect_retrying_while_busy(self, device) -> BleakClient: + """Connect, retrying once while the Bluetooth stack is still busy. + + Right after a restart BlueZ is often still finishing a disconnect and rejects + the next connect with "Operation already in progress". It does so without + starting a connection, so no BLE slot is consumed and waiting it out is safe. + + establish_connection keeps max_attempts=1 on purpose: a genuinely failed + attempt can hold a connection slot that is not released immediately, so + retrying inside bleak-retry-connector can exhaust every slot. + """ + for attempt in range(1, BLUEZ_BUSY_MAX_ATTEMPTS + 1): + try: + return await establish_connection( + BleakClient, + device, + self.mac_address, + disconnected_callback=lambda _: None, + max_attempts=1, + timeout=30.0, + ) + except Exception as err: + is_busy = "already in progress" in str(err).lower() + if not is_busy or attempt == BLUEZ_BUSY_MAX_ATTEMPTS: + raise + _LOGGER.info( + "[CONN] Bluetooth stack still busy for %s (attempt %d/%d), " + "retrying in %.1fs", + self.mac_address, + attempt, + BLUEZ_BUSY_MAX_ATTEMPTS, + BLUEZ_BUSY_RETRY_DELAY, + ) + await asyncio.sleep(BLUEZ_BUSY_RETRY_DELAY) + + raise BleakError(f"Could not connect to {self.mac_address}") + async def _establish_connection(self) -> BleakClient: """Establish a connection to the device with proper error handling. @@ -316,6 +444,12 @@ async def _establish_connection(self) -> BleakClient: connectable=True ) + if not device: + # A link left open by a previous run stops the device from advertising, + # so release it and look again instead of failing outright. + if await self.async_release_stale_connection(): + device = await self._async_rediscover() + if not device: _LOGGER.error( "[DIAG] ✗ Device %s not found", @@ -351,14 +485,7 @@ async def _establish_connection(self) -> BleakClient: self.mac_address, getattr(device, 'rssi', 'N/A') ) - client = await establish_connection( - BleakClient, - device, - self.mac_address, - disconnected_callback=lambda _: None, - max_attempts=1, # CRITICAL: Reduced to 1 to prevent slot exhaustion - timeout=30.0, # 30 second timeout per attempt - ) + client = await self._connect_retrying_while_busy(device) connection_duration = time.time() - connection_start_time _LOGGER.info( @@ -538,75 +665,84 @@ async def _ensure_connected(self) -> BleakClient: ) return self._client - _LOGGER.debug( - "[CONN] No active connection to %s, need to establish new connection", - self.mac_address - ) - - # Fast-fail if we recently failed to connect (unless circuit breaker is testing) - if self._last_connection_failure_time is not None: - time_since_failure = time.time() - self._last_connection_failure_time - - # Only fast-fail if we're not in HALF_OPEN state (testing phase) - if time_since_failure < self._fast_fail_window and self._circuit_breaker.state != "HALF_OPEN": - remaining = self._fast_fail_window - time_since_failure + async with self._connect_lock: + # Another caller may have finished connecting while we waited. + if self._client and self._client.is_connected: _LOGGER.debug( - "[CONN] Fast-fail: Recent connection failure (%.0fs ago), " - "skipping connection attempt for %.0fs more", - time_since_failure, remaining - ) - raise BleakError( - f"Fast-fail: Connection failed {time_since_failure:.0f}s ago, " - f"will retry after {remaining:.0f}s" + "[CONN] ✓ Reusing connection to %s opened by a concurrent caller", + self.mac_address ) + return self._client - # PRIORITY 2: Need to establish a new connection via active BLE scanning - _LOGGER.info( - "[CONN] Establishing new connection to %s via active BLE scan", - self.mac_address - ) + _LOGGER.debug( + "[CONN] No active connection to %s, need to establish new connection", + self.mac_address + ) - try: - self._client = await self._establish_connection() + # Fast-fail if we recently failed to connect (unless circuit breaker is testing) + if self._last_connection_failure_time is not None: + time_since_failure = time.time() - self._last_connection_failure_time + + # Only fast-fail if we're not in HALF_OPEN state (testing phase) + if time_since_failure < self._fast_fail_window and self._circuit_breaker.state != "HALF_OPEN": + remaining = self._fast_fail_window - time_since_failure + _LOGGER.debug( + "[CONN] Fast-fail: Recent connection failure (%.0fs ago), " + "skipping connection attempt for %.0fs more", + time_since_failure, remaining + ) + raise BleakError( + f"Fast-fail: Connection failed {time_since_failure:.0f}s ago, " + f"will retry after {remaining:.0f}s" + ) + + # PRIORITY 2: Need to establish a new connection via active BLE scanning + _LOGGER.info( + "[CONN] Establishing new connection to %s via active BLE scan", + self.mac_address + ) - # Subscribe to notifications (only need to do this once per connection) try: - await self._client.start_notify(EM1003_NOTIFY_CHAR_UUID, self._notification_handler) - _LOGGER.debug("[CONN] ✓ Connected and subscribed to %s", self.mac_address) + self._client = await self._establish_connection() - # Connection successful - clear failure timestamp - self._last_connection_failure_time = None + # Subscribe to notifications (only need to do this once per connection) + try: + await self._client.start_notify(EM1003_NOTIFY_CHAR_UUID, self._notification_handler) + _LOGGER.debug("[CONN] ✓ Connected and subscribed to %s", self.mac_address) + + # Connection successful - clear failure timestamp + self._last_connection_failure_time = None + + except Exception as err: + # Failed to subscribe, disconnect and re-raise + _LOGGER.error("[CONN] Failed to subscribe to notifications: %s", err) + # CRITICAL: Ensure connection is properly cleaned up to free slot + if self._client is not None: + try: + await self._client.disconnect() + _LOGGER.debug("[CONN] Disconnected after subscription failure to free slot") + except Exception as disconnect_err: + _LOGGER.debug("[CONN] Error during cleanup disconnect: %s", disconnect_err) + self._client = None + # Record failure timestamp + self._last_connection_failure_time = time.time() + raise + + return self._client except Exception as err: - # Failed to subscribe, disconnect and re-raise - _LOGGER.error("[CONN] Failed to subscribe to notifications: %s", err) - # CRITICAL: Ensure connection is properly cleaned up to free slot + # Record failure timestamp for fast-fail + self._last_connection_failure_time = time.time() + # CRITICAL: Ensure client is cleared so it doesn't hold a stale connection if self._client is not None: try: await self._client.disconnect() - _LOGGER.debug("[CONN] Disconnected after subscription failure to free slot") - except Exception as disconnect_err: - _LOGGER.debug("[CONN] Error during cleanup disconnect: %s", disconnect_err) - self._client = None - # Record failure timestamp - self._last_connection_failure_time = time.time() + _LOGGER.debug("[CONN] Disconnected after connection error to free slot") + except Exception: + pass + self._client = None raise - return self._client - - except Exception as err: - # Record failure timestamp for fast-fail - self._last_connection_failure_time = time.time() - # CRITICAL: Ensure client is cleared so it doesn't hold a stale connection - if self._client is not None: - try: - await self._client.disconnect() - _LOGGER.debug("[CONN] Disconnected after connection error to free slot") - except Exception: - pass - self._client = None - raise - async def disconnect(self) -> None: """Explicitly disconnect from the device.""" if self._client and self._client.is_connected: @@ -630,6 +766,28 @@ async def disconnect(self) -> None: _LOGGER.debug("[CONN] Already disconnected from %s", self.mac_address) self._client = None + async def async_shutdown(self) -> None: + """Disconnect and make sure the host Bluetooth stack drops the link. + + bleak's disconnect does not reliably tear down the ACL link in BlueZ, and a + link left open keeps the device from advertising on the next start, so force + the teardown here. + """ + await self.disconnect() + + if not IS_LINUX: + return + + try: + await close_stale_connections_by_address( + self.mac_address, only_other_adapters=False + ) + _LOGGER.debug( + "[STALE] Forced link teardown for %s during shutdown", self.mac_address + ) + except Exception as err: + _LOGGER.debug("[STALE] Could not force link teardown: %s", err) + def _notification_handler(self, sender, data: bytearray) -> None: """Handle notification from device. diff --git a/custom_components/em1003/manifest.json b/custom_components/em1003/manifest.json index 8d1b488..5a7916f 100644 --- a/custom_components/em1003/manifest.json +++ b/custom_components/em1003/manifest.json @@ -3,7 +3,7 @@ "name": "EM1003 BLE Sensor (720环境宝3)", "codeowners": [], "config_flow": true, - "dependencies": [], + "dependencies": ["bluetooth"], "documentation": "https://github.com/yourusername/em1003", "integration_type": "device", "iot_class": "local_polling", diff --git a/custom_components/em1003/switch.py b/custom_components/em1003/switch.py index 3d610de..5d50393 100644 --- a/custom_components/em1003/switch.py +++ b/custom_components/em1003/switch.py @@ -70,9 +70,7 @@ def __init__( self._attr_is_on = None self._attr_available = True - # Subscribe to coordinator updates if available - if self._coordinator: - self._coordinator_listener = None + self._coordinator_listener = None @property def device_info(self) -> DeviceInfo: @@ -95,7 +93,16 @@ async def async_added_to_hass(self) -> None: self._handle_coordinator_update ) - # Try to read initial buzzer state + # A BLE read takes tens of seconds when the device is asleep, so keep it off + # the setup path; the coordinator also refreshes this state on every cycle. + self._config_entry.async_create_background_task( + self.hass, + self._async_read_initial_state(), + f"em1003 initial buzzer state {self._mac_address}", + ) + + async def _async_read_initial_state(self) -> None: + """Read the buzzer state once, in the background.""" try: _LOGGER.debug("Reading initial buzzer state for %s", self._mac_address) state = await self._em1003_device.read_buzzer_state() @@ -106,14 +113,15 @@ async def async_added_to_hass(self) -> None: self._mac_address, "ON" if state else "OFF" ) + self.async_write_ha_state() else: - _LOGGER.warning( + _LOGGER.debug( "Could not read initial buzzer state for %s, will retry on first interaction", self._mac_address ) self._attr_is_on = None except Exception as err: - _LOGGER.warning( + _LOGGER.debug( "Error reading initial buzzer state for %s: %s", self._mac_address, err