Add support for Yeelink PTC Bath Heater v5/v6 (yeelink.bhf_light) with opt-in confirmed readback

custom_components/xiaomi_miot/climate.py

@@ -25,6 +25,7 @@
     SWING_OFF,
 )
 from homeassistant.components.sensor import SensorDeviceClass
+from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.restore_state import RestoreEntity
 
 from . import (
@@ -42,13 +43,18 @@
     MiotService,
     MiotProperty,
 )
+from .core.templates import YEELINK_BHF_FAN_GEAR_MODES
 
 _LOGGER = logging.getLogger(__name__)
 DATA_KEY = f'{ENTITY_DOMAIN}.{DOMAIN}'
 
 DEFAULT_MIN_TEMP = 16.0
 DEFAULT_MAX_TEMP = 31.0
 
+# Yeelight PTC bath heaters with composite supply+ventilation states; the
+# mode-aware gear codec lives in core/templates.py (YEELINK_BHF_FAN_MODE_GEARS).
+YEELINK_BATH_HEATER_COMPOSITE_MODELS = ('yeelink.bhf_light.v5', 'yeelink.bhf_light.v6')
+
 SERVICE_TO_METHOD = {}
 
 
@@ -277,8 +283,26 @@ def set_state(self, data: dict):
 
         if self._conv_speed:
             val = self._conv_speed.value_from_dict(data)
-            if val is not None:
+            if self._yeelink_bhf_composite():
+                # Composite output is physically High regardless of configured
+                # gears: the logical fan mode stays unconditionally unavailable.
+                self._attr_fan_mode = None
+                self._attr_extra_state_attributes['effective_fan_mode'] = 'high'
+                self._attr_extra_state_attributes.pop('raw_fan_gear', None)
+            elif val is not None:
                 self._attr_fan_mode = str(val).lower()
+                self._attr_extra_state_attributes.pop('effective_fan_mode', None)
+                self._attr_extra_state_attributes.pop('raw_fan_gear', None)
+            elif self._conv_speed.full_name in data and self._yeelink_bhf_tokens() is not None:
+                # The mode-aware codec reports Idle and out-of-domain gear codes
+                # as none: publish unavailable, never a stale mode. Out-of-domain
+                # singleton codes stay diagnosable via the raw gear digit.
+                self._attr_fan_mode = None
+                self._attr_extra_state_attributes.pop('effective_fan_mode', None)
+                if (raw := self._yeelink_bhf_unknown_raw_gear()) is not None:
+                    self._attr_extra_state_attributes['raw_fan_gear'] = raw
+                else:
+                    self._attr_extra_state_attributes.pop('raw_fan_gear', None)
         if self._conv_swing:
             val = self._conv_swing.value_from_dict(data)
             if val is not None:
@@ -380,9 +404,64 @@ async def async_set_humidity(self, humidity: int):
             return
         await self.device.async_write({self._conv_target_humidity.full_name: humidity})
 
+    def _yeelink_bhf_tokens(self):
+        """Active bh_mode tokens for yeelink v5/v6, None when unknown."""
+        if self.device.model not in YEELINK_BATH_HEATER_COMPOSITE_MODELS:
+            return None
+        m2m = getattr(self.device, 'miio2miot', None)
+        if not m2m:
+            return None
+        mode = (m2m.miio_props_values or {}).get('bh_mode')
+        if mode is None:
+            return None
+        return str(mode).split('|')
+
+    def _yeelink_bhf_composite(self):
+        """Composite state: supply channel (warm/cold) plus ventilation companion."""
+        tokens = self._yeelink_bhf_tokens()
+        if not tokens:
+            return False
+        return 'venting' in tokens and ('warmwind' in tokens or 'coolwind' in tokens)
+
+    def _yeelink_bhf_unknown_raw_gear(self):
+        """Raw gear digit of a singleton channel outside the mode-specific domain."""
+        tokens = self._yeelink_bhf_tokens()
+        if not tokens:
+            return None
+        active = [t for t in YEELINK_BHF_FAN_GEAR_MODES if t in tokens]
+        if len(active) != 1:
+            return None
+        token = active[0]
+        gear = (self.device.miio2miot.miio_props_values or {}).get(f'{token}_gear')
+        if gear is None or gear in YEELINK_BHF_FAN_GEAR_MODES[token]:
+            return None
+        return gear
+
     async def async_set_fan_mode(self, fan_mode: str):
         if not self._conv_speed:
             return
+        if self._yeelink_bhf_composite():
+            raise HomeAssistantError(
+                f'{self.device.model} runs a composite mode: fan mode is not settable, '
+                'effective fan mode is high'
+            )
+        tokens = self._yeelink_bhf_tokens()
+        if self.device.model in YEELINK_BATH_HEATER_COMPOSITE_MODELS and tokens is None:
+            # bh_mode is not read yet: the active warm/cold/vent channel is
+            # unknown, so the mode-specific gear domain can't be validated.
+            # Writing anyway would let the encoder fall back to the coolwind
+            # codec and form a wrong set_bh_mode.
+            raise HomeAssistantError(
+                f'{self.device.model} has no read warm/cold/vent channel: '
+                'fan mode is not settable before the device state is read'
+            )
+        if tokens is not None and not any(t in YEELINK_BHF_FAN_GEAR_MODES for t in tokens):
+            # Idle/drying have no active warm/cold/vent channel; a gear payload
+            # for 'bh_off'/'drying' is undefined on the device, so reject.
+            raise HomeAssistantError(
+                f'{self.device.model} has no active warm/cold/vent channel: '
+                'fan mode is not settable in the current state'
+            )
         dat = {
             ATTR_FAN_MODE: fan_mode,
         }

custom_components/xiaomi_miot/core/converters.py

@@ -110,7 +110,7 @@ def __post_init__(self):
                 self.desc = self.prop.use_desc(self.domain)
 
     def decode(self, device: 'Device', payload: dict, value):
-        if self.desc and self.prop:
+        if self.desc and self.prop and value is not None:
             value = self.prop.list_description(value)
             if self.domain == 'sensor' and isinstance(value, str):
                 value = value.lower()

custom_components/xiaomi_miot/core/coordinator.py

@@ -1,4 +1,7 @@
+import asyncio
+import inspect
 import logging
+import time
 from typing import TYPE_CHECKING
 
 from homeassistant.core import HassJob, HassJobType
@@ -10,6 +13,13 @@
 
 _LOGGER = logging.getLogger(__name__)
 
+# Minimum gap in seconds between physical device polls of confirmed-readback
+# (non_optimistic) devices, including write-initiated refreshes that bypass
+# the HA debouncer. The lock and timestamp live on the shared Device so that
+# every coordinator of one physical device obeys the same gap. Legacy
+# optimistic devices are debounced by HA and poll immediately.
+MIN_DEVICE_POLL_GAP_SECONDS = 3.0
+
 class DataCoordinator(DataUpdateCoordinator):
     def __init__(self, device: 'Device', update_method, **kwargs):
         kwargs.setdefault('always_update', True)
@@ -23,11 +33,17 @@ def __init__(self, device: 'Device', update_method, **kwargs):
             raise ValueError('Invalid update method')
         name = kwargs.pop('name', name)
 
+        config_entry = getattr(device.entry, 'entry', getattr(device, 'entry', None)) if hasattr(device, 'entry') else None
+        if config_entry and 'config_entry' in inspect.signature(DataUpdateCoordinator.__init__).parameters:
+            kwargs.setdefault('config_entry', config_entry)
+        self._device_update_method = update_method
+        self._device_update_tasks = set()
+
         super().__init__(
             device.hass,
             logger=device.log,
             name=f'{device.unique_id}-{name}',
-            update_method=update_method,
+            update_method=self._async_update if callable(update_method) else None,
             **kwargs,
         )
         self.device = device
@@ -36,6 +52,39 @@ def __init__(self, device: 'Device', update_method, **kwargs):
             # hass v2024.7-
             self.async_add_listener(self.coordinator_updated)
 
+    async def _async_update(self):
+        task = asyncio.current_task()
+        if self._shutdown_requested:
+            if task:
+                task.cancel()
+            raise asyncio.CancelledError
+        if task:
+            self._device_update_tasks.add(task)
+        try:
+            # One lock per physical device: chunk coordinators and
+            # write-initiated refreshes must never poll in parallel.
+            async with self.device.poll_lock:
+                if self._shutdown_requested:
+                    if task:
+                        task.cancel()
+                    raise asyncio.CancelledError
+                if self._device_update_method is None:
+                    raise NotImplementedError('Update method not implemented')
+                non_optimistic = self.device.custom_config_bool('non_optimistic')
+                if non_optimistic and self.device.last_poll_monotonic is not None:
+                    delay = MIN_DEVICE_POLL_GAP_SECONDS - (time.monotonic() - self.device.last_poll_monotonic)
+                    if delay > 0:
+                        _LOGGER.debug('%s: Coalesce device poll for %.1fs', self.device.name_model, delay)
+                        await asyncio.sleep(delay)
+                try:
+                    return await self._device_update_method()
+                finally:
+                    if non_optimistic:
+                        self.device.last_poll_monotonic = time.monotonic()
+        finally:
+            if task:
+                self._device_update_tasks.discard(task)
+
     async def async_setup(self, index=0):
         await self._async_setup()
 
@@ -47,6 +96,11 @@ async def async_shutdown(self):
             self._unsub_setup_refresh()
             self._unsub_setup_refresh = None
         await super().async_shutdown()
+        tasks = self._device_update_tasks - {asyncio.current_task()}
+        for task in tasks:
+            task.cancel()
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
 
     async def _async_setup(self):
         """Set up coordinator."""

custom_components/xiaomi_miot/core/device.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 import copy
 import re
@@ -192,6 +193,16 @@ def __init__(self, info: DeviceInfo, entry: HassEntry):
         self.converters: list[BaseConv] = []
         self.coordinators: list[DataCoordinator] = []
         self.main_coordinators: list[DataCoordinator] = []
+        # One poll lock per physical device: every DataCoordinator obeys it.
+        self.poll_lock = asyncio.Lock()
+        self.last_poll_monotonic: float | None = None
+        # Trailing coalescing of confirmed readbacks after non-optimistic
+        # writes: one poll confirms every write that finished before the poll
+        # started; a write landing during an active poll schedules at most one
+        # trailing poll.
+        self.write_poll_lock = asyncio.Lock()
+        self.write_poll_generation = 0
+        self.write_poll_done_generation = 0
         self.log = logging.getLogger(f'{__name__}.{self.model}')
 
     async def async_init(self):
@@ -617,9 +628,37 @@ async def update_status(self):
         for coo in self.coordinators:
             await coo.async_request_refresh()
 
-    async def update_main_status(self):
-        for coo in self.main_coordinators:
-            await coo.async_request_refresh()
+    async def update_main_status(self, immediate=False):
+        coos = self.main_coordinators
+        if not coos and immediate and self.custom_config_bool('non_optimistic'):
+            # miio2miot devices always get the miot_status main coordinator;
+            # this fallback only covers foreign YAML customizes that leave
+            # main_coordinators empty. Deliberate guard, not dead code.
+            coos = self.coordinators
+        for coo in coos:
+            if immediate:
+                await coo.async_refresh()
+            else:
+                await coo.async_request_refresh()
+
+    async def async_write_refresh(self):
+        """Trailing-edge coalesced confirmed refresh after a non-optimistic write."""
+        self.write_poll_generation += 1
+        generation = self.write_poll_generation
+        async with self.write_poll_lock:
+            if generation <= self.write_poll_done_generation:
+                # A poll that started after this write finished confirmed it.
+                return
+            # Yield one loop tick so writes issued in the same batch join
+            # this poll instead of each scheduling a trailing one.
+            await asyncio.sleep(0)
+            covered = self.write_poll_generation
+            try:
+                await self.update_main_status(immediate=True)
+            except Exception as exc:
+                self.log.warning('Failed to refresh status after write: %s', exc)
+            finally:
+                self.write_poll_done_generation = covered
 
     async def update_all_status(self, _=None):
         all = []
@@ -728,8 +767,9 @@ async def async_write(self, payload: dict):
         self.log.info('Device write data: %s', [payload, data])
         result = None
         method = data.get('method')
-        success = None
-
+        non_optimistic = self.custom_config_bool('non_optimistic')
+        success = False
+        write_exc = None
         try:
             if method == 'update_status':
                 result = await self.update_main_status()
@@ -740,6 +780,7 @@ async def async_write(self, payload: dict):
                 success = True if result else False
                 if err := MiotResults(result).has_error:
                     success = False
+                    write_exc = DeviceException(f'Device write error: {err.spec_error}')
                     self.log.warning('Device write error: %s', [payload, data, err])
 
             if method == 'action':
@@ -749,14 +790,24 @@ async def async_write(self, payload: dict):
                 ins = param.get('in') or []
                 result = await self.async_call_action(siid, aiid, ins)
                 success = result.is_success
+                if not success and non_optimistic:
+                    write_exc = DeviceException(
+                        f'Device action error: {result.error or result.code}'
+                    )
 
         except (DeviceException, MiCloudException) as exc:
             success = False
+            write_exc = exc
             self.log.exception('Device write failed: %s', [exc, payload, data])
+        finally:
+            if method in ['set_properties', 'action'] and non_optimistic:
+                await self.async_write_refresh()
 
         self.log.info('Device write result: %s', [payload, result])
-        if success:
+        if success and not non_optimistic:
             self.dispatch(payload)
+        if write_exc and non_optimistic:
+            raise write_exc
         return result
 
     @property

custom_components/xiaomi_miot/core/device_customizes.py

@@ -2788,11 +2788,13 @@
     'yeelink.bhf_light.v5': {
         'interval_seconds': 30,
         'select_properties': 'heat_mode,cold_mode,vent_mode',
+        'non_optimistic': True,
         'chunk_coordinators': [],
     },
     'yeelink.bhf_light.v6': {
         'interval_seconds': 30,
         'select_properties': 'heat_mode,cold_mode,vent_mode',
+        'non_optimistic': True,
         'chunk_coordinators': [],
     },
     'yeelink.bhf_light.v10': {

custom_components/xiaomi_miot/core/miio2miot.py

@@ -217,10 +217,14 @@ async def async_set_property(self, device, siid, piid, value):
         if iok and cbk:
             cbk(prop=cfg.get('prop'), config=cfg, setter=setter, params=pms, props=self.miio_props_values)
         return {
-            'code': 0 if iok else 1,
+            # Never 1 here: MIoT code 1 means "operation not completed" and is
+            # treated as success by MiotResult.is_success; a failed miio setter
+            # must surface as an error so non-optimistic writes can raise.
+            'code': 0 if iok else -1,
             'siid': siid,
             'piid': piid,
             'result': ret,
+            **({} if iok else {'error': f'miio command failed: {setter} {pms} -> {ret}'}),
         }
 
     async def async_call_action(self, device, siid, aiid, params):
@@ -249,10 +253,12 @@ async def async_call_action(self, device, siid, aiid, params):
         if self.config.get('ignore_result'):
             iok = ret or isinstance(ret, list)
         return {
-            'code': 0 if iok else 1,
+            # See async_set_property: code 1 would be swallowed as success.
+            'code': 0 if iok else -1,
             'siid': siid,
             'aiid': aiid,
             'result': ret,
+            **({} if iok else {'error': f'miio action failed: {setter} {pms} -> {ret}'}),
         }
 
     def entity_attrs(self):