Repository: mill1000/midea-msmart
Pull request #118 (2024-03-15 20:52:27)

Accept PropertiesResponses with invalid CRCs. Redo of #114
Original attempt in #114 had some shortcomings due to the adhoc nature of responses. It was possible for a PropertiesResponse to appear to be received in response from another command which caused the CRC to fail.

This approach is a little more of a hammer. It simply ignores the CRC on all PropertiesResponse.


Patch:
diff --git a/msmart/device/AC/command.py b/msmart/device/AC/command.py
index 6e838e6..e136d31 100644
--- a/msmart/device/AC/command.py
+++ b/msmart/device/AC/command.py
@@ -335,47 +335,43 @@ class Response():
         return self._payload
 
     @classmethod
-    def validate(cls, frame: memoryview, skip_crc: bool = False) -> None:
+    def validate(cls, payload: memoryview) -> None:
         """Validate a response by checking the frame checksum and payload CRC."""
 
-        # Attempt to validate the frame
-        Frame.validate(frame)
-
-        # Exit early if not checking CRC
-        if skip_crc:
-            return
-
-        # Extract frame payload to validate CRC/checksum
-        payload = frame[10:-1]
-
         # Some devices use a CRC others seem to use a 2nd checksum
         payload_crc = crc8.calculate(payload[0:-1])
         payload_checksum = Frame.checksum(payload[0:-1])
-
         if payload_crc != payload[-1] and payload_checksum != payload[-1]:
             raise InvalidResponseException(
                 f"Payload '{payload.hex()}' failed CRC and checksum. Received: 0x{payload[-1]:X}, Expected: 0x{payload_crc:X} or 0x{payload_checksum:X}.")
 
     @classmethod
-    def construct(cls, frame: bytes, skip_crc: bool = False) -> Response:
+    def construct(cls, frame: bytes) -> Response:
         """Construct a response object from raw data."""
 
         # Build a memoryview of the frame for zero-copy slicing
         with memoryview(frame) as frame_mv:
-            # Ensure frame is valid before parsing
-            Response.validate(frame_mv, skip_crc)
+            # Validate the frame
+            Frame.validate(frame_mv)
 
-            # Parse frame depending on id
+            # Fetch the appropriate response class from the ID
             response_id = frame_mv[10]
-            payload = frame_mv[10:-2]
             if response_id == ResponseId.STATE:
-                return StateResponse(payload)
+                response_class = StateResponse
             elif response_id == ResponseId.CAPABILITIES:
-                return CapabilitiesResponse(payload)
+                response_class = CapabilitiesResponse
             elif response_id in [ResponseId.PROPERTIES, ResponseId.PROPERTIES_ACK]:
-                return PropertiesResponse(payload)
+                response_class = PropertiesResponse
             else:
-                return Response(payload)
+                response_class = Response
+
+            # Validate the payload CRC
+            # ...except for properties which certain devices send invalid CRCs
+            if response_class != PropertiesResponse:
+                Response.validate(frame_mv[10:-1])
+
+            # Build the response
+            return response_class(frame_mv[10:-2])
 
 
 class CapabilitiesResponse(Response):
diff --git a/msmart/device/AC/device.py b/msmart/device/AC/device.py
index c238251..f41358b 100644
--- a/msmart/device/AC/device.py
+++ b/msmart/device/AC/device.py
@@ -68,8 +68,6 @@ class AirConditioner(Device):
         super().__init__(ip=ip, port=port, device_id=device_id,
                          device_type=DeviceType.AIR_CONDITIONER, **kwargs)
 
-        self._disable_properties_crc_check = False
-
         self._beep_on = False
         self._power_state = False
         self._target_temperature = 17.0
@@ -244,32 +242,12 @@ class AirConditioner(Device):
         # Device is online if we received any response
         self._online = True
 
-        # Determine if CRCs will be checked
-        properties_command = isinstance(command,
-                                        (GetPropertiesCommand, SetPropertiesCommand))
-        skip_crc = self._disable_properties_crc_check and properties_command
-
         valid_responses = []
         for data in responses:
             try:
-                try:
-                    # Construct response from data
-                    response = Response.construct(data, skip_crc=skip_crc)
-                except InvalidResponseException as e:
-                    if not properties_command:
-                        _LOGGER.error(e)
-                        continue
-
-                    # Some devices return invalid CRCs on properties responses
-                    # Detect and disable CRC checks on these responses
-                    _LOGGER.info(
-                        "Disabling CRC check for properties commands.")
-                    self._disable_properties_crc_check = True
-
-                    # Construct the failing response again without CRC check
-                    response = Response.construct(data, skip_crc=True)
-
-            except InvalidFrameException as e:
+                # Construct response from data
+                response = Response.construct(data)
+            except (InvalidFrameException, InvalidResponseException) as e:
                 _LOGGER.error(e)
                 continue
 


Test patch:
diff --git a/msmart/device/AC/test_command.py b/msmart/device/AC/test_command.py
index 74807ec..14d4a9c 100644
--- a/msmart/device/AC/test_command.py
+++ b/msmart/device/AC/test_command.py
@@ -2,7 +2,7 @@ import unittest
 from typing import Union, cast
 
 from msmart.const import DeviceType, FrameType
-from msmart.frame import Frame
+from msmart.frame import Frame, InvalidFrameException
 
 from .command import (CapabilitiesResponse, CapabilityId, Command,
                       GetPropertiesCommand, GetStateCommand,
@@ -657,32 +657,37 @@ class TestPropertiesResponse(_TestResponseBase):
         # Check state
         self.assertEqual(resp.swing_horizontal_angle, 50)
 
+
+class TestResponseConstruct(_TestResponseBase):
+    """Test construction of responses from raw data."""
+
+    def test_invalid_checksum(self) -> None:
+        """Test that invalid checksums raise exceptions."""
+        TEST_RESPONSE_BAD_CHECKSUM = bytes.fromhex(
+            "aa14ac00000000000303b10109000001003c0000FF")
+
+        with self.assertRaises(InvalidFrameException):
+            Response.construct(TEST_RESPONSE_BAD_CHECKSUM)
+
     def test_properties_response_invalid_crc(self) -> None:
-        """Test we decode a properties response sent with an invalid CRC correctly."""
+        """Test that PropertiesResponses with invalid CRCs are accepted."""
+        # PropertiesResponse with invalid CRC
         # https://github.com/mill1000/midea-ac-py/issues/101#issuecomment-1994824924
-        TEST_RESPONSE = bytes.fromhex(
+        TEST_RESPONSE_PROPERTIES_BAD_CRC = bytes.fromhex(
             "aa14ac00000000000303b10109000001003c000042")
+        # Mocked up StateResponse with invalid CRC
+        TEST_RESPONSE_STATE_BAD_CRC = bytes.fromhex(
+            "aa22ac00000000000303c0014566000000300010045eff00000000000000000069aa0c")
 
-        # Assert that constructing with CRC checks results in an exception
+        # Assert that constructing a StateResponse with invalid CRC raises an exception
         with self.assertRaises(InvalidResponseException):
-            resp = Response.construct(TEST_RESPONSE, skip_crc=False)
+            resp = Response.construct(TEST_RESPONSE_STATE_BAD_CRC)
 
-        # Create response without checking CRC and assert it's expected type
-        resp = Response.construct(TEST_RESPONSE, skip_crc=True)
+        # Now construct a PropertiesResponse with an invalid CRC
+        resp = Response.construct(TEST_RESPONSE_PROPERTIES_BAD_CRC)
 
         self.assertIsNotNone(resp)
         self.assertEqual(type(resp), PropertiesResponse)
-        resp = cast(PropertiesResponse, resp)
-
-        EXPECTED_RAW_PROPERTIES = {
-            PropertyId.SWING_UD_ANGLE: 0,
-        }
-        # Ensure raw decoded properties match
-        self.assertEqual(resp._properties, EXPECTED_RAW_PROPERTIES)
-
-        # Check state
-        self.assertEqual(resp.swing_vertical_angle, 0)
-        self.assertIsNone(resp.swing_horizontal_angle)
 
 
 if __name__ == "__main__":
diff --git a/msmart/device/AC/test_device.py b/msmart/device/AC/test_device.py
index 22606af..6b8712b 100644
--- a/msmart/device/AC/test_device.py
+++ b/msmart/device/AC/test_device.py
@@ -1,8 +1,6 @@
 import unittest
-from unittest.mock import patch
 
-from .command import (GetPropertiesCommand, GetStateCommand,
-                      PropertiesResponse, PropertyId, Response, StateResponse)
+from .command import PropertiesResponse, Response, StateResponse
 from .device import AirConditioner as AC
 
 
@@ -212,68 +210,5 @@ class TestUpdateStateFromResponse(unittest.TestCase):
         self.assertEqual(device.vertical_swing_angle, AC.SwingAngle.POS_5)
 
 
-class TestSendCommand(unittest.IsolatedAsyncioTestCase):
-
-    @patch("msmart.device.Device._send_command")
-    async def test_response_ignore_bad_crc(self, mock_send_command) -> None:
-        """Test that bad CRCs are ignored when in response to properties commands."""
-        TEST_RESPONSE_BAD_CRC = bytes.fromhex(
-            "aa14ac00000000000303b10109000001003c000042")
-        TEST_RESPONSE_BAD_CHECKSUM = bytes.fromhex(
-            "aa14ac00000000000303b10109000001003c0000FF")
-
-        # Set return value of mocked method
-        mock_send_command.return_value = [TEST_RESPONSE_BAD_CRC]
-
-        # Create a dummy device
-        device = AC(0, 0, 0)
-
-        # Assert we start not ignoring CRCs
-        self.assertEqual(device._disable_properties_crc_check, False)
-
-        # Send a properties command
-        command = GetPropertiesCommand([PropertyId.SWING_UD_ANGLE])
-        responses = await device._send_command_get_responses(command)
-
-        # Assert that responses are received
-        self.assertTrue(responses)
-
-        # Assert we start are now ignoring CRCs
-        self.assertEqual(device._disable_properties_crc_check, True)
-
-        # Check that we still check CRCs on non-properties commands
-        with self.assertLogs("msmart") as log:
-            command = GetStateCommand()
-            responses = await device._send_command_get_responses(command)
-
-            # Assert that no responses received
-            self.assertFalse(responses)
-
-        # Assert that an error was generated
-        self.assertRegex(log.output[0], "^ERROR:.*Payload.*failed CRC")
-
-        # Change mocked return to an invalid frame
-        mock_send_command.return_value = [TEST_RESPONSE_BAD_CHECKSUM]
-
-        # Assert that checksums are always checked regardless of command
-        with self.assertLogs("msmart") as log:
-            command = GetStateCommand()
-            responses = await device._send_command_get_responses(command)
-
-            # Assert that no responses received
-            self.assertFalse(responses)
-
-            # Send a properties command
-            command = GetPropertiesCommand([PropertyId.SWING_UD_ANGLE])
-            responses = await device._send_command_get_responses(command)
-
-            # Assert that no responses received
-            self.assertFalse(responses)
-
-        # Assert that an error was generated
-        self.assertRegex(log.output[0], "^ERROR:.*Frame.*failed checksum")
-        self.assertRegex(log.output[1], "^ERROR:.*Frame.*failed checksum")
-
-
 if __name__ == "__main__":
     unittest.main()