Repository: mill1000/midea-msmart
Pull request #170 (2024-08-28 20:30:42)

Refactor property decode/encode
- Relocate property encoding from device.py to command.py
- Replace property parsers with decode() method
- Add testcases to verify property encoding/decoding


Patch:
diff --git a/msmart/device/AC/command.py b/msmart/device/AC/command.py
index de39d41..1e6c1fc 100644
--- a/msmart/device/AC/command.py
+++ b/msmart/device/AC/command.py
@@ -85,6 +85,51 @@ class PropertyId(IntEnum):
     IECO = 0x00E3
     ANION = 0x021E
 
+    @property
+    def _supported(self) -> bool:
+        """Check if a property ID is supported/tested."""
+        return self in [
+            PropertyId.BREEZE_AWAY,
+            PropertyId.BREEZE_CONTROL,
+            PropertyId.BREEZELESS,
+            PropertyId.BUZZER,
+            PropertyId.IECO,
+            PropertyId.RATE_SELECT,
+            PropertyId.SELF_CLEAN,
+            PropertyId.SWING_LR_ANGLE,
+            PropertyId.SWING_UD_ANGLE,
+        ]
+
+    def decode(self, data: bytes) -> Any:
+        """Decode raw property data into a convenient form."""
+        if not self._supported:
+            raise NotImplementedError(f"{repr(self)} decode is not supported.")
+
+        if self in [PropertyId.BREEZELESS, PropertyId.SELF_CLEAN]:
+            return bool(data[0])
+        elif self == PropertyId.BREEZE_AWAY:
+            return data[0] == 2
+        elif self == PropertyId.BUZZER:
+            return None  # Don't decode buzzer
+        elif self == PropertyId.IECO:
+            # data[0] - ieco_number, data[1] - ieco_switch
+            return bool(data[1])
+        else:
+            return data[0]
+
+    def encode(self, *args, **kwargs) -> bytes:
+        """Encode property into raw form."""
+        if not self._supported:
+            raise NotImplementedError(f"{repr(self)} encode is not supported.")
+
+        if self == PropertyId.BREEZE_AWAY:
+            return bytes([2 if args[0] else 1])
+        elif self == PropertyId.IECO:
+            # ieco_frame, ieco_number, ieco_switch, ...
+            return bytes([0, 1, args[0]]) + bytes(10)
+        else:
+            return bytes(args[0:1])
+
 
 class TemperatureType(IntEnum):
     UNKNOWN = 0
@@ -344,7 +389,7 @@ class GetPropertiesCommand(Command):
 class SetPropertiesCommand(Command):
     """Command to set specific properties of the device."""
 
-    def __init__(self, props: Mapping[PropertyId, Union[bytes, int]]) -> None:
+    def __init__(self, props: Mapping[PropertyId, Union[int, bool]]) -> None:
         super().__init__(frame_type=FrameType.CONTROL)
 
         self._properties = props
@@ -358,8 +403,8 @@ class SetPropertiesCommand(Command):
         for prop, value in self._properties.items():
             payload += struct.pack("<H", prop)
 
-            if isinstance(value, int):
-                value = bytes([value])
+            # Encode property value to bytes
+            value = prop.encode(value)
 
             payload += bytes([len(value)])
             payload += value
@@ -900,26 +945,6 @@ class PropertiesResponse(Response):
         # Clear existing properties
         self._properties.clear()
 
-        # Define parsing functions for supported properties
-        # TODO when a properties has multiple field .e.g fresh air
-        # should they be stored all under the fresh_air key or create different
-        # keys for each. e.g. capabilities
-        parsers = {
-            PropertyId.ANION: lambda v: v[0],
-            PropertyId.BREEZE_AWAY: lambda v: v[0],
-            PropertyId.BREEZE_CONTROL: lambda v: v[0],
-            PropertyId.BREEZELESS: lambda v: v[0],
-            PropertyId.BUZZER: lambda v: None,  # Don't bother parsing buzzer state
-            PropertyId.FRESH_AIR: lambda v: (v[0], v[1], v[2]),
-            # v[0] - ieco_number, v[1] - ieco_switch
-            PropertyId.IECO: lambda v: v[1],
-            PropertyId.INDOOR_HUMIDITY: lambda v: v[0],
-            PropertyId.RATE_SELECT: lambda v: v[0],
-            PropertyId.SELF_CLEAN: lambda v: v[0],
-            PropertyId.SWING_UD_ANGLE: lambda v: v[0],
-            PropertyId.SWING_LR_ANGLE: lambda v: v[0],
-        }
-
         count = payload[1]
         props = payload[2:]
 
@@ -948,17 +973,6 @@ class PropertiesResponse(Response):
                 props = props[4+size:]
                 continue
 
-            # Fetch parser for this property
-            parser = parsers.get(property, None)
-
-            # Check if parser exists
-            if parser is None:
-                _LOGGER.warning(
-                    "Unsupported property %r, Size: %d.", property, size)
-                # Advanced to next property
-                props = props[4+size:]
-                continue
-
             # Check execution result and log any errors
             error = props[2] & 0x10
             if error:
@@ -966,8 +980,12 @@ class PropertiesResponse(Response):
                     "Property %r failed, Result: 0x%02X.", property, props[2])
 
             # Parse the property
-            if (value := parser(props[4:])) is not None:
-                self._properties.update({property: value})
+            try:
+                if (value := property.decode(props[4:])) is not None:
+                    self._properties.update({property: value})
+            except NotImplementedError:
+                _LOGGER.warning(
+                    "Unsupported property %r, Size: %d.", property, size)
 
             # Advanced to next property
             props = props[4+size:]
diff --git a/msmart/device/AC/device.py b/msmart/device/AC/device.py
index a33d9e7..7327bbc 100644
--- a/msmart/device/AC/device.py
+++ b/msmart/device/AC/device.py
@@ -84,13 +84,11 @@ class AirConditioner(Device):
         DEFAULT = OFF
 
     # Create a dict to map attributes to property values
-    # TODO encoding of properties should be pushed into commands
     _PROPERTY_MAP = {
-        PropertyId.BREEZE_AWAY: lambda s: 2 if s._breeze_mode == AirConditioner.BreezeMode.BREEZE_AWAY else 1,
+        PropertyId.BREEZE_AWAY: lambda s: s._breeze_mode == AirConditioner.BreezeMode.BREEZE_AWAY,
         PropertyId.BREEZE_CONTROL: lambda s: s._breeze_mode,
         PropertyId.BREEZELESS: lambda s: s._breeze_mode == AirConditioner.BreezeMode.BREEZELESS,
-        # ieco_frame, ieco_number, ieco_switch, ...
-        PropertyId.IECO: lambda s: bytes([0, 1, s._ieco_mode]) + bytes(10),
+        PropertyId.IECO: lambda s: s._ieco_mode,
         PropertyId.RATE_SELECT: lambda s: s._rate_select,
         PropertyId.SWING_LR_ANGLE: lambda s: s._horizontal_swing_angle,
         PropertyId.SWING_UD_ANGLE: lambda s: s._vertical_swing_angle
@@ -220,7 +218,7 @@ class AirConditioner(Device):
                     AirConditioner.SwingAngle.get_from_value(angle))
 
             if (value := res.get_property(PropertyId.SELF_CLEAN)) is not None:
-                self._self_clean_active = bool(value)
+                self._self_clean_active = value
 
             if (rate := res.get_property(PropertyId.RATE_SELECT)) is not None:
                 self._rate_select = cast(
@@ -233,15 +231,15 @@ class AirConditioner(Device):
                                      else AirConditioner.BreezeMode.OFF)
             else:
                 if (value := res.get_property(PropertyId.BREEZE_AWAY)) is not None:
-                    self._breeze_mode = (AirConditioner.BreezeMode.BREEZE_AWAY if (value == 2)
+                    self._breeze_mode = (AirConditioner.BreezeMode.BREEZE_AWAY if value
                                          else AirConditioner.BreezeMode.OFF)
 
                 if (value := res.get_property(PropertyId.BREEZELESS)) is not None:
-                    self._breeze_mode = (AirConditioner.BreezeMode.BREEZELESS if bool(value)
+                    self._breeze_mode = (AirConditioner.BreezeMode.BREEZELESS if value
                                          else AirConditioner.BreezeMode.OFF)
 
             if (value := res.get_property(PropertyId.IECO)) is not None:
-                self._ieco_mode = bool(value)
+                self._ieco_mode = value
 
         elif isinstance(res, EnergyUsageResponse):
             self._total_energy_usage = res.total_energy
@@ -482,7 +480,7 @@ class AirConditioner(Device):
             for response in await self._send_command_get_responses(cmd):
                 self._update_state(response)
 
-    async def _apply_properties(self, properties: dict[PropertyId, Union[bytes, int]]) -> None:
+    async def _apply_properties(self, properties: dict[PropertyId, Union[int, bool]]) -> None:
         """Apply the provided properties to the device."""
 
         # Warn if attempting to update a property that isn't supported


Test patch:
diff --git a/msmart/device/AC/test_command.py b/msmart/device/AC/test_command.py
index 8514a99..4acfca5 100644
--- a/msmart/device/AC/test_command.py
+++ b/msmart/device/AC/test_command.py
@@ -590,11 +590,38 @@ class TestGetPropertiesCommand(unittest.TestCase):
 
 class TestSetPropertiesCommand(unittest.TestCase):
 
+    def test_encode(self) -> None:
+        """Test encoding of property values to bytes objects."""
+        TEST_ENCODES = {
+            # Breeze away: 0x02 - On, 0x01 - Off
+            (PropertyId.BREEZE_AWAY, True): bytes([0x02]),
+            (PropertyId.BREEZE_AWAY, False): bytes([0x01]),
+
+            # Breezeless: Boolean
+            (PropertyId.BREEZELESS, True): bytes([0x01]),
+            (PropertyId.BREEZELESS, False): bytes([0x00]),
+
+            # Breeze control: Passthru
+            (PropertyId.BREEZE_CONTROL, 0x04): bytes([0x04]),
+            (PropertyId.BREEZE_CONTROL, 0x00): bytes([0x00]),
+
+            # IECO: 13 bytes ieco_frame, ieco_number, ieco_switch, ...
+            (PropertyId.IECO, True): bytes([0, 1, 1]) + bytes(10),
+            (PropertyId.IECO, False): bytes([0, 1, 0]) + bytes(10),
+        }
+
+        for (prop, value), expected_data in TEST_ENCODES.items():
+            self.assertEqual(prop.encode(value), expected_data, msg=f"""Encode {
+                             repr(prop)}, Value: {value}, Expected: {expected_data}""")
+
+        # Validate "unsupported" properties raise exceptions
+        with self.assertRaisesRegex(NotImplementedError, ".* encode is not supported."):
+            PropertyId.ANION.encode(True)
+
     def test_payload(self) -> None:
         """Test that we encode set properties payloads correctly."""
         # TODO this test is not based on a real world sample
-        PROPS = {PropertyId.SWING_UD_ANGLE: bytes(
-            [25]), PropertyId.SWING_LR_ANGLE: bytes([75])}
+        PROPS = {PropertyId.SWING_UD_ANGLE: 25, PropertyId.SWING_LR_ANGLE: 75}
 
         # Build command
         command = SetPropertiesCommand(PROPS)
@@ -610,30 +637,64 @@ class TestSetPropertiesCommand(unittest.TestCase):
         self.assertEqual(payload[2], PropertyId.SWING_UD_ANGLE & 0xFF)
         self.assertEqual(payload[3], PropertyId.SWING_UD_ANGLE >> 8 & 0xFF)
 
-        # Assert length is correct
-        self.assertEqual(payload[4], len(PROPS[PropertyId.SWING_UD_ANGLE]))
-
-        # Assert data is correct
-        self.assertEqual(payload[5], PROPS[PropertyId.SWING_UD_ANGLE][0])
+        # Assert length is correct and data is correct
+        self.assertEqual(payload[4], 1)
+        self.assertEqual(payload[5], PROPS[PropertyId.SWING_UD_ANGLE])
 
 
 class TestPropertiesResponse(_TestResponseBase):
     """Test properties response messages."""
 
+    def test_decode(self) -> None:
+        """Test decoding of bytes objects to property values."""
+        TEST_DECODES = {
+            # Breeze away 0x02 - On, 0x01 - Off
+            (PropertyId.BREEZE_AWAY, bytes([0x02])): True,
+            (PropertyId.BREEZE_AWAY, bytes([0x01])): False,
+
+            # Breezeless: Boolean
+            (PropertyId.BREEZELESS, bytes([0x01])): True,
+            (PropertyId.BREEZELESS, bytes([0x00])): False,
+            (PropertyId.BREEZELESS, bytes([0x02])): True,
+
+            # Breeze control: Passthru
+            (PropertyId.BREEZE_CONTROL, bytes([0x04])): 0x04,
+            (PropertyId.BREEZE_CONTROL, bytes([0x00])): 0x00,
+
+            # Buzzer: Don't decode
+            (PropertyId.BUZZER, bytes([0x00])): None,
+
+            # IECO: 2 bytes
+            (PropertyId.IECO, bytes([0x00, 0x00])): False,
+            (PropertyId.IECO, bytes([0x00, 0x01])): True,
+        }
+
+        for (prop, data), expected_value in TEST_DECODES.items():
+            self.assertEqual(prop.decode(data), expected_value, msg=f"""Decode {
+                             repr(prop)}, Data: {data}, Expected: {expected_value}""")
+
+        # Validate "unsupported" properties raise exceptions
+        with self.assertRaisesRegex(NotImplementedError, ".* decode is not supported."):
+            PropertyId.INDOOR_HUMIDITY.decode(bytes([1]))
+
     def test_properties_parsing(self) -> None:
-        """Test we decode properties correctly."""
+        """Test we decode properties responses correctly."""
         # https://github.com/mill1000/midea-ac-py/issues/60#issuecomment-1936976587
         TEST_RESPONSE = bytes.fromhex(
             "aa21ac00000000000303b10409000001000a00000100150000012b1e020000005fa3")
 
-        resp = self._test_build_response(TEST_RESPONSE)
+        # Response contains an unsupported property so check the log for warnings
+        with self.assertLogs("msmart", logging.WARNING) as log:
+            resp = self._test_build_response(TEST_RESPONSE)
+
+            self.assertRegex("\n".join(log.output),
+                             "Unsupported property .*INDOOR_HUMIDITY.*")
 
         # Assert response is a correct type
         self.assertEqual(type(resp), PropertiesResponse)
         resp = cast(PropertiesResponse, resp)
 
         EXPECTED_RAW_PROPERTIES = {
-            PropertyId.INDOOR_HUMIDITY: 43,
             PropertyId.SWING_LR_ANGLE: 0,
             PropertyId.SWING_UD_ANGLE: 0,
         }
@@ -641,7 +702,6 @@ class TestPropertiesResponse(_TestResponseBase):
         self.assertEqual(resp._properties, EXPECTED_RAW_PROPERTIES)
 
         # Check state
-        self.assertEqual(resp.get_property(PropertyId.INDOOR_HUMIDITY), 43)
         self.assertEqual(resp.get_property(PropertyId.SWING_LR_ANGLE), 0)
         self.assertEqual(resp.get_property(PropertyId.SWING_UD_ANGLE), 0)
 
diff --git a/msmart/device/AC/test_device.py b/msmart/device/AC/test_device.py
index d9a2bd7..2580a6b 100644
--- a/msmart/device/AC/test_device.py
+++ b/msmart/device/AC/test_device.py
@@ -151,10 +151,15 @@ class TestUpdateStateFromResponse(unittest.TestCase):
         device.horizontal_swing_angle = AC.SwingAngle.POS_5
         device.vertical_swing_angle = AC.SwingAngle.POS_5
 
-        resp = Response.construct(TEST_RESPONSE)
-        self.assertIsNotNone(resp)
+        # Response contains an unsupported property so check the log for warnings
+        with self.assertLogs("msmart", logging.WARNING) as log:
+            resp = Response.construct(TEST_RESPONSE)
+
+            self.assertRegex("\n".join(log.output),
+                             "Unsupported property .*INDOOR_HUMIDITY.*")
 
         # Assert response is a state response
+        self.assertIsNotNone(resp)
         self.assertEqual(type(resp), PropertiesResponse)
 
         # Process the response