Add support for wash sub device with XML payload (issue #681)

custom_components/smartthinq_sensors/__init__.py

@@ -348,11 +348,14 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 class LGEDevice:
     """Generic class that represents a LGE device."""
 
-    def __init__(self, device: ThinQDevice, hass: HomeAssistant):
+    def __init__(
+        self, device: ThinQDevice, hass: HomeAssistant, root_dev_id: str | None = None
+    ):
         """initialize a LGE Device."""
 
         self._device = device
         self._hass = hass
+        self._root_dev_id = root_dev_id
         self._name = device.name
         self._device_id = device.unique_id
         self._type = device.device_info.type
@@ -425,8 +428,10 @@ def device_info(self) -> DeviceInfo:
         )
         if self._firmware:
             data["sw_version"] = self._firmware
-        if self._mac:
+        if self._mac and not self._root_dev_id:
             data["connections"] = {(dr.CONNECTION_NETWORK_MAC, self._mac)}
+        if self._root_dev_id:
+            data["via_device"] = (DOMAIN, self._root_dev_id)
 
         return data
 
@@ -558,50 +563,60 @@ async def lge_devices_setup(
     if hass.config.units.temperature_unit != UnitOfTemperature.CELSIUS:
         temp_unit = TemperatureUnit.FAHRENHEIT
 
+    async def init_device(
+        lge_dev: ThinQDevice, device_info: ThinQDeviceInfo, root_dev_id: str
+    ):
+        """Initialize a new device."""
+        root_dev = None if root_dev_id == lge_dev.unique_id else root_dev_id
+        dev = LGEDevice(lge_dev, hass, root_dev)
+        if not await dev.init_device():
+            _LOGGER.error(
+                "Error initializing LGE Device. Name: %s - Type: %s - InfoUrl: %s",
+                device_info.name,
+                device_info.type.name,
+                device_info.model_info_url,
+            )
+            return False
+
+        new_devices[device_info.device_id].append(dev.device_id)
+        wrapped_devices.setdefault(device_info.type, []).append(dev)
+        _LOGGER.info(
+            "LGE Device added. Name: %s - Type: %s - Model: %s - ID: %s",
+            dev.name,
+            device_info.type.name,
+            device_info.model_name,
+            dev.device_id,
+        )
+        return True
+
     for device_info in client_devices:
         device_id = device_info.device_id
         if device_id in discovered_devices:
             new_devices[device_id] = discovered_devices[device_id]
             continue
 
         new_devices[device_id] = []
-        device_name = device_info.name
-        device_type = device_info.type
-        network_type = device_info.network_type
-        model_name = device_info.model_name
         device_count += 1
 
         lge_devs = get_lge_device(client, device_info, temp_unit)
         if not lge_devs:
             _LOGGER.info(
                 "Found unsupported LGE Device. Name: %s - Type: %s - NetworkType: %s",
-                device_name,
-                device_type.name,
-                network_type.name,
+                device_info.name,
+                device_info.type.name,
+                device_info.network_type.name,
             )
-            unsupported_devices.setdefault(device_type, []).append(device_info)
+            unsupported_devices.setdefault(device_info.type, []).append(device_info)
             continue
 
-        for lge_dev in lge_devs:
-            dev = LGEDevice(lge_dev, hass)
-            if not await dev.init_device():
-                _LOGGER.error(
-                    "Error initializing LGE Device. Name: %s - Type: %s - InfoUrl: %s",
-                    device_name,
-                    device_type.name,
-                    device_info.model_info_url,
-                )
+        root_dev = None
+        for idx, lge_dev in enumerate(lge_devs):
+            if idx == 0:
+                root_dev = lge_dev.unique_id
+            if not await init_device(lge_dev, device_info, root_dev):
                 break
-
-            new_devices[device_id].append(dev.device_id)
-            wrapped_devices.setdefault(device_type, []).append(dev)
-            _LOGGER.info(
-                "LGE Device added. Name: %s - Type: %s - Model: %s - ID: %s",
-                dev.name,
-                device_type.name,
-                model_name,
-                dev.device_id,
-            )
+            if sub_dev := lge_dev.subkey_device:
+                await init_device(sub_dev, device_info, root_dev)
 
     if device_count > 0:
         _LOGGER.info("Founds %s LGE device(s)", device_count)

custom_components/smartthinq_sensors/wideq/device.py

@@ -437,6 +437,11 @@ def model_info(self) -> ModelInfo:
             raise DeviceNotInitialized()
         return self._model_info
 
+    @property
+    def subkey_device(self) -> Device | None:
+        """Return the available sub device."""
+        return None
+
     @property
     def available_features(self) -> dict:
         """Return available features."""

custom_components/smartthinq_sensors/wideq/devices/washerDryer.py

@@ -26,7 +26,7 @@
 ]
 
 WM_ROOT_DATA = "washerDryer"
-WM_SUB_KEYS = {"mini": "miniState"}
+WM_SUB_KEYS = {"mini": "miniState", "Sub": "SubState"}
 
 POWER_STATUS_KEY = ["State", "state"]
 
@@ -73,15 +73,6 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-def get_sub_keys(device_info: DeviceInfo, sub_device: str | None = None) -> list[str]:
-    """Search for valid sub devices and return related sub keys."""
-    if not (snapshot := device_info.snapshot):
-        return []
-    if not (payload := snapshot.get(sub_device or WM_ROOT_DATA)):
-        return []
-    return [k for k, s in WM_SUB_KEYS.items() if s in payload]
-
-
 class WMDevice(Device):
     """A higher-level interface for washer and dryer."""
 
@@ -98,6 +89,8 @@ def __init__(
         if sub_key:
             self._attr_unique_id += f"-{sub_key}"
             self._attr_name += f" {sub_key.capitalize()}"
+        self._subkey_device = None
+        self._internal_state = None
         self._stand_by = False
         self._remote_start_status = None
         self._remote_start_pressed = False
@@ -124,6 +117,42 @@ def sub_key(self) -> str | None:
         """Return device sub key."""
         return self._sub_key
 
+    @property
+    def subkey_device(self) -> Device | None:
+        """Return the available sub key device."""
+        return self._subkey_device
+
+    async def init_device_info(self) -> bool:
+        """Initialize the information for the device"""
+        if result := await super().init_device_info():
+            self._init_subkey_device()
+        return result
+
+    def _init_subkey_device(self) -> None:
+        """Initialize the available sub key device."""
+        if self._sub_key or self._subkey_device or not self.model_info:
+            return
+        for key, val in WM_SUB_KEYS.items():
+            if self.model_info.value_exist(val):
+                # we check for value in the snapshot if available
+                if snapshot := self.device_info.snapshot:
+                    if payload := snapshot.get(self._sub_device or WM_ROOT_DATA):
+                        if val not in payload:
+                            continue
+                self._subkey_device = WMDevice(
+                    self.client,
+                    self.device_info,
+                    sub_device=self._sub_device,
+                    sub_key=key,
+                )
+                return
+
+    def update_internal_state(self, state):
+        """Update internal state used by sub key device."""
+        if not self._sub_key:
+            return
+        self._internal_state = state
+
     def getkey(self, key: str | None) -> str | None:
         """Add subkey prefix to a key if required."""
         if not (key and self._sub_key):
@@ -414,7 +443,7 @@ def pause_enabled(self) -> bool:
     async def power_off(self):
         """Power off the device."""
         keys = self._get_cmd_keys(CMD_POWER_OFF)
-        await self.set(keys[0], keys[1], value=keys[2])
+        await self.set(keys[0], keys[1])
         self._remote_start_status = None
         self._update_status(POWER_STATUS_KEY, self._state_power_off)
 
@@ -424,7 +453,7 @@ async def wake_up(self):
             raise InvalidDeviceStatus()
 
         keys = self._get_cmd_keys(CMD_WAKE_UP)
-        await self.set(keys[0], keys[1], value=keys[2])
+        await self.set(keys[0], keys[1])
         self._stand_by = False
         self._update_status(POWER_STATUS_KEY, self._state_power_on_init)
 
@@ -443,7 +472,7 @@ async def pause(self):
             raise InvalidDeviceStatus()
 
         keys = self._get_cmd_keys(CMD_PAUSE)
-        await self.set(keys[0], keys[1], value=keys[2])
+        await self.set(keys[0], keys[1])
         self._update_status(POWER_STATUS_KEY, self._state_pause)
 
     async def set(
@@ -480,7 +509,7 @@ def _set_remote_start_opt(self, res):
             stand_by = self._status.device_features.get(WashDeviceFeatures.STANDBY)
             if stand_by is None:
                 standby_enable = self.model_info.config_value("standbyEnable")
-                if standby_enable and not self._should_poll:
+                if standby_enable and not self._should_poll and not self._sub_key:
                     self._stand_by = not self._status.is_on
                 else:
                     self._stand_by = False
@@ -497,7 +526,13 @@ def _set_remote_start_opt(self, res):
     async def poll(self) -> WMStatus | None:
         """Poll the device's current state."""
 
-        res = await self._device_poll(self._sub_device or WM_ROOT_DATA)
+        if not self._sub_key:
+            res = await self._device_poll(self._sub_device or WM_ROOT_DATA)
+            if self._subkey_device:
+                self._subkey_device.update_internal_state(res)
+        else:
+            res = self._internal_state
+
         if not res:
             self._stand_by = False
             return None

custom_components/smartthinq_sensors/wideq/factory.py

@@ -23,7 +23,7 @@
 from .devices.range import RangeDevice
 from .devices.refrigerator import RefrigeratorDevice
 from .devices.styler import StylerDevice
-from .devices.washerDryer import WMDevice, get_sub_keys
+from .devices.washerDryer import WMDevice
 from .devices.waterheater import WaterHeaterDevice
 
 
@@ -71,14 +71,8 @@ def get_lge_device(
     if device_type == DeviceType.WATER_HEATER:
         return [WaterHeaterDevice(client, device_info, temp_unit)]
     if device_type in WM_DEVICE_TYPES:
-        dev_list = []
-        for sub_device in _get_sub_devices(device_type):
-            dev_list.append(WMDevice(client, device_info, sub_device=sub_device))
-            dev_list.extend(
-                [
-                    WMDevice(client, device_info, sub_device=sub_device, sub_key=key)
-                    for key in get_sub_keys(device_info, sub_device)
-                ]
-            )
-        return dev_list
+        return [
+            WMDevice(client, device_info, sub_device=sub_device)
+            for sub_device in _get_sub_devices(device_type)
+        ]
     return None