Add support for device type 0xC3 (heat pumps)

msmart/base_device.py

@@ -2,8 +2,8 @@
 import time
 from typing import List, Optional
 
-from msmart.base_command import Command
 from msmart.const import DeviceType
+from msmart.frame import Frame
 from msmart.lan import LAN, AuthenticationError, Key, ProtocolError, Token
 
 _LOGGER = logging.getLogger(__name__)
@@ -25,7 +25,7 @@ def __init__(self, *, ip: str, port: int, device_id: int, device_type: DeviceTyp
         self._supported = False
         self._online = False
 
-    async def _send_command(self, command: Command) -> Optional[List[bytes]]:
+    async def _send_command(self, command: Frame) -> Optional[List[bytes]]:
         """Send a command to the device and return any responses."""
 
         data = command.tobytes()

msmart/const.py

@@ -29,11 +29,12 @@
 
 class DeviceType(IntEnum):
     AIR_CONDITIONER = 0xAC
+    HEAT_PUMP = 0xC3
 
 
 class FrameType(IntEnum):
     UNKNOWN = 0
-    SET = 0x02
-    REQUEST = 0x03
+    CONTROL = 0x02
+    QUERY = 0x03
     REPORT = 0x04
     ABNORMAL_REPORT = 0x06

msmart/crc8.py

@@ -37,10 +37,5 @@
 def calculate(data: bytes) -> int:
     crc_value = 0
     for m in data:
-        k = crc_value ^ m
-        if k > 256:
-            k -= 256
-        if k < 0:
-            k += 256
-        crc_value = _CRC8_854_TABLE[k]
+        crc_value = _CRC8_854_TABLE[(crc_value ^ m) & 0xFF]
     return crc_value

msmart/device/AC/command.py

@@ -8,8 +8,8 @@
 from typing import Callable, Collection, Mapping, Optional, Union
 
 import msmart.crc8 as crc8
-from msmart.base_command import Command
 from msmart.const import DeviceType, FrameType
+from msmart.frame import Frame, InvalidFrameException
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -85,31 +85,47 @@ class TemperatureType(IntEnum):
     OUTDOOR = 0x3
 
 
+class Command(Frame):
+    """Base class for AC commands."""
+
+    _message_id = 0
+
+    def __init__(self, device_type: DeviceType, frame_type: FrameType) -> None:
+        super().__init__(device_type, frame_type)
+
+    def tobytes(self, data: Union[bytes, bytearray] = bytes()) -> bytes:
+        # Append message ID to payload
+        # TODO Message ID in reference is just a random value
+        payload = data + bytes([self._next_message_id()])
+
+        # Append CRC
+        return super().tobytes(payload + bytes([crc8.calculate(payload)]))
+
+    def _next_message_id(self) -> int:
+        Command._message_id += 1
+        return Command._message_id & 0xFF
+
+
 class GetCapabilitiesCommand(Command):
     def __init__(self, additional: bool = False) -> None:
-        super().__init__(DeviceType.AIR_CONDITIONER, frame_type=FrameType.REQUEST)
+        super().__init__(DeviceType.AIR_CONDITIONER, frame_type=FrameType.QUERY)
 
         self._additional = additional
 
-    @property
-    def payload(self) -> bytes:
-        if not self._additional:
-            # Get capabilities
-            return bytes([0xB5, 0x01, 0x00])
-        else:
-            # Get more capabilities
-            return bytes([0xB5, 0x01, 0x01, 0x1])
+    def tobytes(self) -> bytes:
+        paylaod = bytes([0xB5, 0x01, 0x00] if not self._additional
+                        else [0xB5, 0x01, 0x01, 0x1])
+        return super().tobytes(paylaod)
 
 
 class GetStateCommand(Command):
     def __init__(self) -> None:
-        super().__init__(DeviceType.AIR_CONDITIONER, frame_type=FrameType.REQUEST)
+        super().__init__(DeviceType.AIR_CONDITIONER, frame_type=FrameType.QUERY)
 
         self.temperature_type = TemperatureType.INDOOR
 
-    @property
-    def payload(self) -> bytes:
-        return bytes([
+    def tobytes(self) -> bytes:
+        return super().tobytes(bytes([
             # Get state
             0x41,
             # Unknown
@@ -122,12 +138,12 @@ def payload(self) -> bytes:
             0x00, 0x00, 0x00, 0x00,
             # Unknown
             0x03,
-        ])
+        ]))
 
 
 class SetStateCommand(Command):
     def __init__(self) -> None:
-        super().__init__(DeviceType.AIR_CONDITIONER, frame_type=FrameType.SET)
+        super().__init__(DeviceType.AIR_CONDITIONER, frame_type=FrameType.CONTROL)
 
         self.beep_on = True
         self.power_on = False
@@ -142,8 +158,7 @@ def __init__(self) -> None:
         self.freeze_protection_mode = False
         self.follow_me = False
 
-    @property
-    def payload(self) -> bytes:
+    def tobytes(self) -> bytes:
         # Build beep and power status bytes
         beep = 0x40 if self.beep_on else 0
         power = 0x1 if self.power_on else 0
@@ -185,7 +200,7 @@ def payload(self) -> bytes:
         # Build alternate turbo byte
         freeze_protect = 0x80 if self.freeze_protection_mode else 0
 
-        return bytes([
+        return super().tobytes(bytes([
             # Set state
             0x40,
             # Beep and power state
@@ -215,22 +230,21 @@ def payload(self) -> bytes:
             freeze_protect,
             # Unknown
             0x00, 0x00,
-        ])
+        ]))
 
 
 class ToggleDisplayCommand(Command):
     def __init__(self) -> None:
         # For whatever reason, toggle display uses a request type...
-        super().__init__(DeviceType.AIR_CONDITIONER, frame_type=FrameType.REQUEST)
+        super().__init__(DeviceType.AIR_CONDITIONER, frame_type=FrameType.QUERY)
 
         self.beep_on = True
 
-    @property
-    def payload(self) -> bytes:
+    def tobytes(self) -> bytes:
         # Set beep bit
         beep = 0x40 if self.beep_on else 0
 
-        return bytes([
+        return super().tobytes(bytes([
             # Get state
             0x41,
             # Beep and other flags
@@ -241,14 +255,14 @@ def payload(self) -> bytes:
             0x00, 0x00, 0x00, 0x00,
             0x00, 0x00, 0x00, 0x00,
             0x00, 0x00, 0x00, 0x00,
-        ])
+        ]))
 
 
 class GetPropertiesCommand(Command):
     """Command to query specific properties from the device."""
 
     def __init__(self, props: Collection[PropertyId]) -> None:
-        super().__init__(DeviceType.AIR_CONDITIONER, frame_type=FrameType.REQUEST)
+        super().__init__(DeviceType.AIR_CONDITIONER, frame_type=FrameType.QUERY)
 
         self._properties = props
 
@@ -269,7 +283,7 @@ class SetPropertiesCommand(Command):
     """Command to set specific properties of the device."""
 
     def __init__(self, props: Mapping[PropertyId, Union[bytes, int]]) -> None:
-        super().__init__(DeviceType.AIR_CONDITIONER, frame_type=FrameType.SET)
+        super().__init__(DeviceType.AIR_CONDITIONER, frame_type=FrameType.CONTROL)
 
         self._properties = props
 
@@ -308,18 +322,17 @@ def payload(self) -> bytes:
 
     @classmethod
     def validate(cls, frame: memoryview) -> None:
-        # Validate frame checksum
-        frame_checksum = Command.checksum(frame[1:-1])
-        if frame_checksum != frame[-1]:
-            raise InvalidResponseException(
-                f"Frame '{frame.hex()}' failed checksum. Received: 0x{frame[-1]:X}, Expected: 0x{frame_checksum:X}.")
+        try:
+            Frame.validate(frame)
+        except InvalidFrameException as e:
+            raise InvalidResponseException(e) from e
 
         # 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 = Command.checksum(payload[0:-1])
+        payload_checksum = Frame.checksum(payload[0:-1])
 
         if payload_crc != payload[-1] and payload_checksum != payload[-1]:
             raise InvalidResponseException(