Repository: mill1000/midea-msmart
Pull request #24 (2023-08-06 20:15:54)

Reimplement LAN service with async support
Reimplement the LAN service with asyncio support.

Built around [async protocols](https://docs.python.org/3/library/asyncio-protocol.html).

The goal was to abstract the V2/V3 protocols properly such that the device doesn't know which protocol is being used, it simply just writes/reads commands and responses.


Patch:
diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml
index 4718189..9bbe0ec 100644
--- a/.github/workflows/checks.yml
+++ b/.github/workflows/checks.yml
@@ -12,7 +12,7 @@ jobs:
     - uses: actions/checkout@v3
     - uses: actions/setup-python@v4
       with:
-        python-version: "3.11"
+        python-version: "3.10"
     - uses: actions/cache@v3
       with:
         path: ~/.cache/pip
@@ -24,18 +24,12 @@ jobs:
     
   run-unittest:
     runs-on: ubuntu-latest
-    strategy:
-      max-parallel: 1 # Limit to one job due to cloud account reuse
-      matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11"]
-
     steps:
     - uses: actions/checkout@v3
     - uses: actions/setup-python@v4
       with:
-        python-version: ${{ matrix.python-version }}
+        python-version: "3.10"
         cache: "pip"
-    - run: python -c "import sys; print(sys.version)"
     - run: |
         python -m pip install --upgrade pip
         pip install --editable .
diff --git a/README.md b/README.md
index cc301ce..32e5745 100644
--- a/README.md
+++ b/README.md
@@ -1,5 +1,5 @@
 # midea-msmart
-A Python library for local control of Midea (and associated brands) smart air conditioners.
+A Python library to enable local communication with Midea (and associated brands) smart air conditioners.
 
 ![Code Quality Badge](https://github.com/mill1000/midea-msmart/actions/workflows/checks.yml/badge.svg)
 
diff --git a/example.py b/example.py
index 1efb211..7bc7a77 100644
--- a/example.py
+++ b/example.py
@@ -1,59 +1,85 @@
-
 import logging
-import time
+import asyncio
+
+from msmart.discover import Discover
 from msmart.device import air_conditioning as ac
 
-logging.basicConfig(level=logging.DEBUG)
-
-# Manually construct device. Async functions may use Discover.discover_single("YOUR_AC_IP")
-# See midea-discover to read ID, token and key
-# Prefer to use named args
-device = ac(ip='YOUR_AC_IP', port=6444, id=int('YOUR_AC_ID'))
-# But position args will work
-# device = ac('YOUR_AC_IP', int('YOUR_AC_ID'), 6444)
-
-# V3 devices require authentication
-device.authenticate('YOUR_AC_TOKEN', 'YOUR_AC_KEY')
-
-# Refresh the object with the actual state by querying it
-device.get_capabilities()
-device.refresh()
-print({
-    'id': device.id,
-    'name': device.ip,
-    'power_state': device.power_state,
-    'prompt_tone': device.prompt_tone,
-    'target_temperature': device.target_temperature,
-    'operational_mode': device.operational_mode,
-    'fan_speed': device.fan_speed,
-    'swing_mode': device.swing_mode,
-    'eco_mode': device.eco_mode,
-    'turbo_mode': device.turbo_mode,
-    'fahrenheit': device.fahrenheit,
-    'indoor_temperature': device.indoor_temperature,
-    'outdoor_temperature': device.outdoor_temperature
-})
-
-time.sleep(1)
-
-# Set the state of the device and
-device.power_state = True
-device.prompt_tone = False
-device.target_temperature = 25
-device.operational_mode = ac.operational_mode_enum.cool
-device.apply()
-
-print({
-    'id': device.id,
-    'name': device.ip,
-    'power_state': device.power_state,
-    'prompt_tone': device.prompt_tone,
-    'target_temperature': device.target_temperature,
-    'operational_mode': device.operational_mode,
-    'fan_speed': device.fan_speed,
-    'swing_mode': device.swing_mode,
-    'eco_mode': device.eco_mode,
-    'turbo_mode': device.turbo_mode,
-    'indoor_temperature': device.indoor_temperature,
-    'outdoor_temperature': device.outdoor_temperature
-})
+logging.basicConfig(level=logging.INFO)
+
+DEVICE_IP = "YOUR_DEVICE_IP"
+DEVICE_PORT = 6444
+DEVICE_ID = "YOUR_AC_ID"
+
+# For V3 devices
+DEVICE_TOKEN = None  # "YOUR_DEVICE_TOKEN"
+DEVICE_KEY = None  # "YOUR_DEVICE_KEY"
+
+
+async def main():
+
+    # There are 2 ways to connect
+
+    # Discover.discover_single can automatically construct a device from IP or hostname
+    #  - V3 devices will be automatically authenticated
+    #  - The Midea cloud will be accessed for V3 devices to fetch the token and key
+    # device = await Discover.discover_single(DEVICE_IP)
+
+    # Manually construct the device
+    #  - See midea-discover to read ID, token and key
+    device = ac(ip=DEVICE_IP, port=6444, id=int(DEVICE_ID))
+    if DEVICE_TOKEN and DEVICE_KEY:
+        await device.authenticate(DEVICE_TOKEN, DEVICE_KEY)
+
+    # Get device capabilities
+    await device.get_capabilities()
+
+    # Refresh the state
+    await device.refresh()
+
+    print({
+        'id': device.id,
+        'ip': device.ip,
+        "online": device.online,
+        "support": device._support,
+        'power_state': device.power_state,
+        'prompt_tone': device.prompt_tone,
+        'target_temperature': device.target_temperature,
+        'operational_mode': device.operational_mode,
+        'fan_speed': device.fan_speed,
+        'swing_mode': device.swing_mode,
+        'eco_mode': device.eco_mode,
+        'turbo_mode': device.turbo_mode,
+        'fahrenheit': device.fahrenheit,
+        'indoor_temperature': device.indoor_temperature,
+        'outdoor_temperature': device.outdoor_temperature
+    })
+
+    await asyncio.sleep(1)
+
+    # Change some device properties and apply them
+    device.power_state = True
+    device.prompt_tone = False
+    device.target_temperature = 25
+    device.operational_mode = ac.operational_mode_enum.cool
+    await device.apply()
+
+    print({
+        'id': device.id,
+        'ip': device.ip,
+        "online": device.online,
+        "support": device._support,
+        'power_state': device.power_state,
+        'prompt_tone': device.prompt_tone,
+        'target_temperature': device.target_temperature,
+        'operational_mode': device.operational_mode,
+        'fan_speed': device.fan_speed,
+        'swing_mode': device.swing_mode,
+        'eco_mode': device.eco_mode,
+        'turbo_mode': device.turbo_mode,
+        'fahrenheit': device.fahrenheit,
+        'indoor_temperature': device.indoor_temperature,
+        'outdoor_temperature': device.outdoor_temperature
+    })
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/msmart/__init__.py b/msmart/__init__.py
index 4f46a57..b4210f7 100644
--- a/msmart/__init__.py
+++ b/msmart/__init__.py
@@ -1,3 +1,8 @@
-from importlib import metadata
+import sys
+
+if sys.version_info >= (3, 8):
+    from importlib import metadata
+else:
+    import importlib_metadata as metadata
 
 __version__ = metadata.version("msmart")
diff --git a/msmart/cloud.py b/msmart/cloud.py
index 10fdf51..dcb8018 100644
--- a/msmart/cloud.py
+++ b/msmart/cloud.py
@@ -7,6 +7,7 @@ import json
 import logging
 import os
 from secrets import token_hex, token_urlsafe
+from typing import Optional
 import httpx
 
 
@@ -78,7 +79,7 @@ class Cloud:
 
         raise ApiError(body["msg"], code=response_code)
 
-    async def _post_request(self, url, headers, contents, retries=RETRIES) -> dict | None:
+    async def _post_request(self, url, headers, contents, retries=RETRIES) -> Optional[dict]:
         """Post a request to the API."""
 
         async with httpx.AsyncClient() as client:
@@ -98,7 +99,7 @@ class Cloud:
                     if retries == 0:
                         raise e
 
-    async def _api_request(self, endpoint, body) -> dict | None:
+    async def _api_request(self, endpoint, body) -> Optional[dict]:
         """Make a request to the Midea cloud return the results."""
 
         # Encode body as JSON
diff --git a/msmart/const.py b/msmart/const.py
index fb07b1d..9b1a57f 100644
--- a/msmart/const.py
+++ b/msmart/const.py
@@ -23,11 +23,6 @@ DEVICE_INFO_MSG = bytes([
     0xb7, 0xe4, 0x2d, 0x53, 0x49, 0x47, 0x62, 0xbe
 ])
 
-MSGTYPE_HANDSHAKE_REQUEST = 0x0
-MSGTYPE_HANDSHAKE_RESPONSE = 0x1
-MSGTYPE_ENCRYPTED_RESPONSE = 0x3
-MSGTYPE_ENCRYPTED_REQUEST = 0x6
-MSGTYPE_TRANSPARENT = 0xf
 
 OPEN_MIDEA_APP_ACCOUNT = 'midea_is_best@outlook.com'
 OPEN_MIDEA_APP_PASSWORD = 'lovemidea4ever'
diff --git a/msmart/device/AC/appliance.py b/msmart/device/AC/appliance.py
index 1b0ef29..575bc09 100644
--- a/msmart/device/AC/appliance.py
+++ b/msmart/device/AC/appliance.py
@@ -114,37 +114,38 @@ class air_conditioning(device):
     def __str__(self):
         return str(self.__dict__)
 
-    def get_capabilities(self):
+    async def get_capabilities(self):
         cmd = get_capabilities_command(self.type)
-        self.send_cmd(cmd)
+        await self.send_command(cmd)
 
-    def toggle_display(self):
+    async def toggle_display(self):
         if not self._supports_display_control:
             _LOGGER.warning("Device is not capable of display control.")
 
         cmd = toggle_display_command(self.type)
-        self.send_cmd(cmd, True)
+        await self.send_command(cmd, True)
 
         # Force a refresh to get the updated display state
-        self.refresh()
+        await self.refresh()
 
-    def refresh(self):
+    async def refresh(self):
         cmd = get_state_command(self.type)
-        self.send_cmd(cmd)
+        await self.send_command(cmd)
 
-    def send_cmd(self, cmd, ignore_response=False):
-        responses = super().send_cmd(cmd)
+    async def send_command(self, cmd, ignore_response=False):
+        responses = await super().send_command(cmd)
 
-        # Ignore responses if requested
-        if ignore_response:
+        # Ignore responses if requested, or nonexistent
+        if ignore_response or responses is None:
             return
 
-        # Process each response
         for response in responses:
             self.process_response(response)
 
     def process_response(self, data):
-        if super().process_response(data):
+        if data:
+            self._online = True
+
             # Construct response from data
             try:
                 response = base_response.construct(data)
@@ -158,13 +159,13 @@ class air_conditioning(device):
                 self.update(response)
             elif response.id == ResponseId.Capabilities:
                 self.update_capabilities(response)
-            elif response.id == 0xa1 or response.id == 0xa0:
-                _LOGGER.info("Ignored special response. %s:%d %s",
-                             self.ip, self.port, response.payload.hex())
+            else:
+                _LOGGER.debug("Ignored unknown response from %s:%d: %s",
+                              self.ip, self.port, response.payload.hex())
         elif not self._keep_last_known_online_state:
             self._online = False
 
-    def apply(self):
+    async def apply(self):
         self._updating = True
         try:
             # Warn if trying to apply unsupported modes
@@ -197,7 +198,7 @@ class air_conditioning(device):
             cmd.freeze_protection_mode = self._freeze_protection_mode
             cmd.sleep_mode = self._sleep_mode
             cmd.fahrenheit = self._fahrenheit_unit
-            self.send_cmd(cmd, self._defer_update)
+            await self.send_command(cmd, self._defer_update)
         finally:
             self._updating = False
             self._defer_update = False
diff --git a/msmart/device/base.py b/msmart/device/base.py
index 35225b9..2dcf3a9 100644
--- a/msmart/device/base.py
+++ b/msmart/device/base.py
@@ -2,9 +2,10 @@
 from abc import ABC, abstractmethod
 import logging
 import time
-from msmart.lan import lan
-from msmart.packet_builder import packet_builder
+from typing import Union, List
 
+from msmart.lan import LAN, ProtocolError, AuthenticationError
+from msmart.types import Token, Key
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -20,74 +21,53 @@ class device(ABC):
         self._name = kwargs.get("name", None)
         self._type = kwargs.get("type", None)
 
-        # For V3 devices
-        self._token = None
-        self._key = None
-
-        self._lan_service = lan(ip, id, port)
+        self._lan = LAN(ip, port, id)
         self._support = False
         self._online = False
 
-    def authenticate(self, token: str = None, key: str = None):
-        # Use existing token and key if none provided
-        if token is None or key is None:
-            token = self._token
-            key = self._key
-
-        def convert(x):
-            if isinstance(x, str):
-                return bytes.fromhex(x)
-
-        success = self._lan_service.authenticate(convert(token), convert(key))
-
-        # Update token and key if successful
-        if success:
-            self._token = token
-            self._key = key
-
-        return success
-
     @abstractmethod
-    def refresh(self):
-        pass
+    async def refresh(self):
+        raise NotImplementedError()
 
     @abstractmethod
-    def apply(self):
-        pass
-
-    def send_cmd(self, cmd):
-        pkt_builder = packet_builder(self.id)
-        pkt_builder.set_command(cmd)
-        data = pkt_builder.finalize()
-        _LOGGER.debug(
-            "pkt_builder: %s:%d len: %d data: %s", self.ip, self.port, len(data), data.hex())
-        send_time = time.time()
-
-        responses = self._lan_service.send(data)
-
-        request_time = round(time.time() - send_time, 2)
-        _LOGGER.debug(
-            "Got responses from %s:%d Version: %d Count: %d Spend time: %f", self.ip, self.port, self._lan_service.protocol_version, len(responses), request_time)
-        if len(responses) == 0:
-            _LOGGER.warning(
-                "Got Null from %s:%d Version: %d Count: %d Spend time: %f", self.ip, self.port, self._lan_service.protocol_version, len(responses), request_time)
-            self._support = False
-
-        # sort, put CMD_TYPE_QUERRY last, so we can get END(machine_status) from the last response
-        responses.sort()
+    async def apply(self):
+        raise NotImplementedError()
+
+    async def authenticate(self, token: Token, key: Key) -> bool:
+        """Authenticate with a V3 device."""
+        try:
+            await self._lan.authenticate(token, key)
+            return True
+        except (AuthenticationError, TimeoutError) as e:
+            _LOGGER.error("Authentication failed. Error: %s", e)
+            return False
+
+    async def send_command(self, command: bytes) -> List[object]:
+        """Send a command to the device and return any responses."""
+
+        data = command.pack()
+        _LOGGER.debug("Sending command to %s:%d: %s.",
+                      self.ip, self.port, data.hex())
+
+        start = time.time()
+        responses = None
+        try:
+            responses = await self._lan.send(data)
+        except (ProtocolError, TimeoutError) as e:
+            _LOGGER.error("Network error: %s", e)
+        finally:
+            response_time = round(time.time() - start, 2)
+
+        if responses is None:
+            _LOGGER.warning("No response from %s:%d in %f seconds. ",
+                            self.ip, self.port, response_time)
+            return None
+
+        _LOGGER.debug("Response from %s:%d in %f seconds.",
+                      self.ip, self.port, response_time)
 
         return responses
 
-    def process_response(self, data):
-        _LOGGER.debug("Update from %s:%d %s", self.ip, self.port, data.hex())
-        if len(data) > 0:
-            self._online = True
-            if data == b'ERROR':
-                self._support = False
-                _LOGGER.warning("Got ERROR from %s, %s", self.ip, self.id)
-                return
-            return data
-
     @property
     def ip(self) -> str:
         return self._ip
@@ -101,15 +81,15 @@ class device(ABC):
         return self._id
 
     @property
-    def type(self) -> str:
+    def type(self) -> Union[str, None]:
         return self._type
 
     @property
-    def name(self) -> str:
+    def name(self) -> Union[str, None]:
         return self._name
 
     @property
-    def sn(self) -> str:
+    def sn(self) -> Union[str, None]:
         return self._sn
 
     @property
diff --git a/msmart/discover.py b/msmart/discover.py
index 0148079..73fde85 100644
--- a/msmart/discover.py
+++ b/msmart/discover.py
@@ -41,10 +41,6 @@ class _V1DeviceInfoProtocol(asyncio.Protocol):
 
         self.response = data
 
-    def error_received(self, ex):
-        """Handle asyncio.Protocol errors."""
-        _LOGGER.error("Got error: %s", ex)
-
     def connection_lost(self, ex):
         """NOP implementation of connection lost."""
 
@@ -340,7 +336,7 @@ class Discover:
                 "Fetching token and key for udpid '%s' (%s).", udpid, endian)
             token, key = await cloud.get_token(udpid)
 
-            if dev.authenticate(token, key):
+            if await dev.authenticate(token, key):
                 return token, key
 
     @classmethod
@@ -363,6 +359,6 @@ class Discover:
         if version == 3:
             await Discover._authenticate_device(dev)
 
-        dev.refresh()
+        await dev.refresh()
 
         return dev
diff --git a/msmart/lan.py b/msmart/lan.py
index bfa04d0..3617212 100644
--- a/msmart/lan.py
+++ b/msmart/lan.py
@@ -1,312 +1,536 @@
 """Module for local network control of Midea AC devices."""
+import asyncio
 from Crypto.Cipher import AES
 from Crypto.Util import Padding
 from Crypto.Util.strxor import strxor
 from Crypto.Random import get_random_bytes
+from datetime import datetime
+from enum import IntEnum
 from hashlib import md5, sha256
 import logging
-import socket
-import time
-from msmart.const import (
-    MSGTYPE_ENCRYPTED_REQUEST,
-    MSGTYPE_HANDSHAKE_REQUEST,
-    MSGTYPE_ENCRYPTED_RESPONSE
-)
+import struct
+from typing import cast
 
+from msmart.types import Token, Key
 
 _LOGGER = logging.getLogger(__name__)
 
 
-class lan:
-    def __init__(self, device_ip, device_id, device_port=6444):
-        self.device_ip = device_ip
-        self.device_id = device_id
-        self.device_port = device_port
-        self.security = Security()
-        self._retries = 0
-        self._socket = None
+class ProtocolError(Exception):
+    pass
+
+
+class AuthenticationError(ProtocolError):
+    pass
+
+
+class _LanProtocol(asyncio.Protocol):
+    """Midea LAN protocol."""
+
+    # V2 Packet Overview
+    #
+    # Header: 40 bytes
+    #  2 byte start of packet: 0x5A5A
+    #  2 byte message type: 0x0111
+    #  2 byte packet length
+    #  2 byte magic bytes: Usually 0x2000, special responses differ
+    #  4 byte message ID
+    #  8 byte timestamp
+    #  8 byte device ID
+    #  12 byte ???
+    #
+    # Payload: N bytes
+    #  N byte data payload contains encrypted frame
+    #
+    # Sign: 16 bytes
+    #   16 byte MD5 of packet + fixed key
+    #
+
+    def __init__(self):
+        self._transport = None
+
+        self._peer = None
+        self._queue = asyncio.Queue()
+
+    @property
+    def peer(self):
+        return self._peer
+
+    @property
+    def alive(self):
+        if self._transport is None or self._transport.is_closing():
+            return False
+
+        return True
+
+    def _format_socket_name(self, sockname) -> str:
+        def _format(addr, port):
+            return f"{addr}:{port}"
+        return _format(*sockname)
+
+    def connection_made(self, transport) -> None:
+        """Handle connection events."""
+
+        # Save transport for later
+        self._transport = transport
+
+        # Save peer name for logging
+        peername = transport.get_extra_info('peername')
+        self._peer = self._format_socket_name(peername)
+
+        _LOGGER.debug("Connected to %s.", self._peer)
+
+    def data_received(self, data: bytes) -> None:
+        """Handle data received events."""
+
+        _LOGGER.debug("Received data from %s: %s", self.peer, data.hex())
+        self._queue.put_nowait(data)
+
+    def connection_lost(self, ex):
+        """Log connection lost."""
+        if ex:
+            _LOGGER.error("Connection to %s lost. Error: %s.", self.peer, ex)
+
+    def disconnect(self) -> None:
+        """Disconnect from the peer."""
+        if self._transport is None:
+            raise IOError()  # TODO better?
+
+        _LOGGER.debug("Disconnecting from %s.", self.peer)
+        self._transport.close()
+
+    def write(self, data: bytes) -> None:
+        """Send data to the peer."""
+
+        if self._transport is None:
+            raise IOError()  # TODO better
+
+        if not self.alive:
+            raise ProtocolError("Transport is closing or closed.")
+
+        _LOGGER.debug("Sending data to %s: %s", self.peer, data.hex())
+        self._transport.write(data)
+
+    async def _read_queue(self, timeout: int = 2) -> bytes:
+        """Read data from the receive queue."""
+
+        if timeout == 0:
+            return self._queue.get_nowait()
+
+        return await asyncio.wait_for(self._queue.get(), timeout=timeout)
+
+    async def read(self, timeout: int = 2) -> bytes:
+        """Asynchronously read data from the peer via the queue."""
+
+        # Fetch a packet from the queue
+        return await self._read_queue(timeout=timeout)
+
+
+class _LanProtocolV3(_LanProtocol):
+    """Midea LAN protocol V3."""
+
+    class PacketType(IntEnum):
+        HANDSHAKE_REQUEST = 0x0
+        HANDSHAKE_RESPONSE = 0x1
+        ENCRYPTED_RESPONSE = 0x3
+        ENCRYPTED_REQUEST = 0x6
+        ERROR = 0xF
+
+    # V3 Packet Overview
+    #
+    # Header: 6 bytes
+    #  2 byte start of packet: 0x8370
+    #  2 byte size of data payload, padding and sign
+    #  1 special byte: 0x20
+    #  1 padding and type byte: pad << 4 | type
+    #
+    # Payload: N + 2 bytes
+    #  2 byte request ID/count
+    #  N byte data payload
+    #
+    # Sign: 32 bytes
+    #   32 byte SHA256 of header + unencrypted payload
+    #
+    # Notes
+    #  - For padding purposes the 2 byte request ID is included in size,
+    #    but not in the size field
+    #  - When used for device command/response, the payload contains a V2 packet
+
+    def __init__(self):
+        super().__init__()
+
+        self._packet_id = 0
+        self._buffer = bytearray(0)
+        self._local_key = None
+
+    def data_received(self, data: bytes) -> None:
+        """Handle data received events."""
+
+        _LOGGER.debug("Received data from %s: %s", self.peer, data.hex())
+
+        # Add incoming data to buffer
+        self._buffer += data
+
+        # Process buffer until empty
+        while len(self._buffer) > 0:
+            # Find start of packet
+            start = self._buffer.find(b"\x83\x70")
+            if start == -1:
+                _LOGGER.warning(
+                    "No start of packet found. Buffer: %s", self._buffer.hex())
+                return
+
+            # Create a memoryview for zero copy slicing
+            with memoryview(self._buffer) as buf:
+                if start != 0:
+                    _LOGGER.warning(
+                        "Ignoring data before packet: %s", buf[:start].hex())
+
+                # Trim any leading data
+                buf = buf[start:]
+
+                # Check if the header has been received
+                if len(buf) < 6:
+                    _LOGGER.warning("Buffer too short. Buffer: %s", buf.hex())
+                    return
+
+                # 6 byte header + 2 packet id + padded encrypted payload
+                total_size = int.from_bytes(buf[2:4], "big") + 8
+
+                # Ensure entire packet is received
+                if len(buf) < total_size:
+                    _LOGGER.warning(
+                        "Partial packet received. Buffer: %s", buf.hex())
+                    return
+
+                # Extract the packet from the buffer
+                packet, self._buffer = buf[:total_size], bytearray(
+                    buf[total_size:])
+
+                # Queue the received packet
+                self._queue.put_nowait(packet.tobytes())
+
+    def _decode_encrypted_response(self, packet: memoryview):
+        """Decode an encrypted response packet."""
+
+        # Extract header, encrypted payload and hash
+        header = packet[:6]
+        payload = packet[6:-32]
+        hash = packet[-32:]
+
+        # Decrypt payload
+        decrypted_payload = Security.decrypt_aes_cbc(self._local_key, payload)
+
+        # TODO could padding module handle this padding?
+
+        # Verify hash
+        if sha256(bytes(header) + decrypted_payload).digest() != hash:
+            raise ProtocolError(
+                "Calculated and received SHA256 digest do not match.")
+
+        with memoryview(decrypted_payload) as payload:
+            # Decrypted payload consists of 2 byte packet ID + actual payload + padding
+
+            # Get pad count from header
+            pad = header[5] >> 4
+
+            # Get the frame from payload
+            return payload[2:-pad]
+
+    def _decode_handshake_response(self, packet: memoryview):
+        """Decode a handshake response packet."""
+
+        # Get payload from packet
+        payload = packet[6:]
+
+        # Return remaining raw payload
+        return payload[2:].tobytes()
+
+    def _process_packet(self, packet: memoryview):
+        """Process a received packet based on its type."""
+
+        if packet[:2] != b"\x83\x70":
+            raise ProtocolError(
+                f"Invalid start of packet: {packet[:2].hex()}")
+
+        if packet[4] != 0x20:
+            raise ProtocolError(
+                f"Invalid magic byte: 0x{packet[4]:X}")
+
+        # Handle packet based on type
+        type = packet[5] & 0xF
+        if type == self.PacketType.ENCRYPTED_RESPONSE:
+            return self._decode_encrypted_response(packet)
+        elif type == self.PacketType.HANDSHAKE_RESPONSE:
+            return self._decode_handshake_response(packet)
+        elif type == self.PacketType.ERROR:
+            raise ProtocolError("Error packet received.")
+        else:
+            raise ProtocolError(f"Unexpected type: {type}")
+
+    async def read(self, timeout: int = 2) -> bytes:
+        """Asynchronously read data from the peer via the queue."""
+
+        # Fetch a packet from the queue
+        packet = await self._read_queue(timeout=timeout)
+
+        with memoryview(packet) as packet_mv:
+            return self._process_packet(packet_mv)
+
+    def _build_header(self, length: int, extra: bytes):
+        # Build header
+        header = b"\x83\x70"
+        header += length.to_bytes(2, "big")
+        header += b"\x20"
+        header += extra
+
+        return header
+
+    def _encode_encrypted_request(self, packet_id: int, data: bytes):
+        """Encode an encrypted request packet."""
+
+        # Compute required padding for 16 byte alignment
+        # Include 2 bytes for packet ID in total length
+        remainder = (len(data) + 2) % 16
+        pad = 16 - remainder if remainder != 0 else 0
+
+        # Compute total length of payload, pad and hash
+        length = len(data) + pad + 32
+
+        # Build header
+        header = self._build_header(length, bytes(
+            [pad << 4 | self.PacketType.ENCRYPTED_REQUEST]))
+
+        # Build payload to encrypt
+        payload = packet_id.to_bytes(2, "big") + data + get_random_bytes(pad)
+
+        hash = sha256(header + payload).digest()
+        return header + Security.encrypt_aes_cbc(self._local_key, payload) + hash
+
+    def _encode_handshake_request(self, packet_id: int, data: bytes):
+        """Encode a handshake request packet."""
+
+        # Build header
+        header = self._build_header(len(data), bytes(
+            [self.PacketType.HANDSHAKE_REQUEST]))
+
+        # Build payload to encrypt
+        payload = packet_id.to_bytes(2, "big") + data
+
+        return header + payload
+
+    def write(self, data: bytes, *, type=PacketType.ENCRYPTED_REQUEST) -> None:
+        """Send a packet of the specified type to the peer."""
+
+        # Raise an error if attempting to send an encryptd request without authenticating
+        if type == self.PacketType.ENCRYPTED_REQUEST and self._local_key is None:
+            raise ProtocolError("Protocol has not been authenticated.")
+
+        # Encode the data according to the supplied type
+        if type == self.PacketType.ENCRYPTED_REQUEST:
+            packet = self._encode_encrypted_request(self._packet_id, data)
+        elif type == self.PacketType.HANDSHAKE_REQUEST:
+            packet = self._encode_handshake_request(self._packet_id, data)
+        else:
+            raise TypeError(f"Unknown type: {type}")
+
+        # Write to the peer
+        super().write(packet)
+
+        # Increment packet ID and handle rollover
+        self._packet_id += 1
+        self._packet_id &= 0xFFF  # Mask to 12 bits
+
+    def _get_local_key(self, key: Key, data: memoryview):
+
+        if len(data) != 64:
+            raise AuthenticationError(
+                "Invalid data length for key handshake.")
+
+        # Extract payload and hash
+        payload = data[:32]
+        hash = data[32:]
+
+        # Decrypt the payload with the provided key
+        decrypted_payload = Security.decrypt_aes_cbc(key, payload)
+
+        if sha256(decrypted_payload).digest() != hash:
+            raise AuthenticationError(
+                "Calculated and received SHA256 digest do not match.")
+
+        # Construct the local key
+        return strxor(decrypted_payload, key)
+
+    async def authenticate(self, token: Token, key: Key):
+
+        # Raise an exception if trying to auth without any token or key
+        if not token or not key:
+            raise AuthenticationError("Token and key must be supplied.")
+
+        try:
+            self.write(token, type=self.PacketType.HANDSHAKE_REQUEST)
+            response = await self.read()
+        except ProtocolError as e:
+            # Promote any protocol error to auth error
+            raise AuthenticationError(e)
+
+        # Generate local key from cloud key
+        with memoryview(response) as response_mv:
+            self._local_key = self._get_local_key(key, response_mv)
+
+        _LOGGER.info("Authentication successful. Local key: %s",
+                     self._local_key.hex())
+
+
+class LAN:
+    RETRIES = 3
+
+    def __init__(self, ip: str, port: int, device_id: int):
+        self._ip = ip
+        self._port = port
+        self._device_id = device_id
+
         self._token = None
         self._key = None
-        self._timestamp = time.time()
-        self._tcp_key = None
-        self._local = None
-        self._remote = device_ip + ":" + str(device_port)
         self._protocol_version = 2
+        self._protocol = None
 
-    def _connect(self):
-        if self._socket is None:
-            self._disconnect()
-            _LOGGER.debug("Attempting new connection to %s:%d",
-                          self.device_ip, self.device_port)
-            self._buffer = b''
-            self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-            # set timeout
-            self._socket.settimeout(2)
-            try:
-                self._socket.connect((self.device_ip, self.device_port))
-                self._timestamp = time.time()
-                self._local = ":".join(
-                    '%s' % i for i in self._socket.getsockname())
-            except Exception as error:
-                _LOGGER.error("Connect Error: %s:%d %s",
-                              self.device_ip, self.device_port, error)
-                self._disconnect()
+    async def _connect(self):
+        _LOGGER.info("Creating new connection to %s:%s", self._ip, self._port)
+
+        protocol_class = _LanProtocolV3 if self._protocol_version == 3 else _LanProtocol
+
+        # TODO throws OSError and ???
+        loop = asyncio.get_event_loop()
+        _transport, protocol = await loop.create_connection(
+            lambda: protocol_class(), self._ip, self._port)
+
+        if self._protocol_version == 3:
+            self._protocol = cast(_LanProtocolV3, protocol)
+        else:
+            self._protocol = cast(_LanProtocol, protocol)
 
     def _disconnect(self):
-        if self._socket:
-            self._socket.close()
-            self._socket = None
-            self._tcp_key = None
-
-    def get_socket_info(self):
-        socket_time = round(time.time() - self._timestamp, 2)
-        return "{} -> {} retries: {} time: {}".format(self._local, self._remote, self._retries, socket_time)
-
-    def request(self, message):
-        # Create a TCP/IP socket
-        self._connect()
-        if self._socket is None:
-            _LOGGER.error("Sokcet is None: %s", self._remote)
-            return bytearray(0), False
-        _LOGGER.debug("Socket %s tcp_key: %s",
-                      self.get_socket_info(), self._tcp_key)
-        # Send data
-        try:
-            _LOGGER.debug(
-                "Sending %s message: %s", self.get_socket_info(), message.hex())
-            self._socket.sendall(message)
-        except Exception as error:
-            _LOGGER.error("Send %s Error: %s", self.get_socket_info(), error)
+        if self._protocol:
+            self._protocol.disconnect()
+            self._protocol = None
+
+    async def authenticate(self, token: Token = None, key: Key = None, retries: int = RETRIES):
+        """Authenticate against a V3 device. Use cached token and key unless provided a new token and key."""
+
+        # Use existing token and key if none provided
+        if token is None or key is None:
+            token = self._token
+            key = self._key
+        else:
+            # Define a lambda to convert hex strings to bytes
+            def convert(x):
+                return bytes.fromhex(x) if isinstance(x, str) else x
+
+            # Ensure passed token and key are in byte form
+            token = convert(token)
+            key = convert(key)
+
+        # Disconnect any existing V2 protocol
+        if self._protocol_version == 2:
             self._disconnect()
-            self._retries += 1
-            return bytearray(0), True
 
-        # Received data
-        try:
-            response = self._socket.recv(1024)
-        except socket.timeout as error:
-            if error.args[0] == 'timed out':
-                _LOGGER.debug("Recv %s, timed out", self.get_socket_info())
-                self._retries += 1
-                return bytearray(0), True
-            else:
-                _LOGGER.debug("Recv %s TimeOut: %s",
-                              self.get_socket_info(), error)
-                self._disconnect()
-                self._retries += 1
-                return bytearray(0), True
-        except socket.error as error:
-            _LOGGER.debug("Recv %s Error: %s", self.get_socket_info(), error)
+        if self._protocol is None or not self._protocol.alive:
             self._disconnect()
-            self._retries += 1
-            return bytearray(0), True
-        else:
-            _LOGGER.debug("Recv %s Response: %s",
-                          self.get_socket_info(), response.hex())
-            if len(response) == 0:
-                _LOGGER.debug("Recv %s Server Closed Socket",
-                              self.get_socket_info())
-                self._disconnect()
-                self._retries += 1
-                return bytearray(0), True
-            else:
-                self._retries = 0
-                return response, True
-
-    def authenticate(self, token: bytearray, key: bytearray):
-        self._token, self._key = token, key
-        if not self._token or not self._key:
-            raise Exception('missing token key pair')
-        request = self.security.encode_8370(
-            self._token, MSGTYPE_HANDSHAKE_REQUEST)
-        response, _ = self.request(request)
-        response = response[8:72]
-
-        tcp_key, success = self.security.tcp_key(response, self._key)
-        if success:
-            self._tcp_key = tcp_key.hex()
-            _LOGGER.info('Got TCP key for %s tcp_key: %s',
-                         self.get_socket_info(), tcp_key.hex())
             self._protocol_version = 3
-            # After authentication, don’t send data immediately, so sleep 1s.
-            time.sleep(1)
-        else:
-            _LOGGER.error('Authentication failed for %s %s',
-                          self.get_socket_info(), tcp_key.hex())
-        return success
-
-    def _authenticate(self):
-        if not self._token or not self._key:
-            raise Exception('missing token key pair')
-        return self.authenticate(self._token, self._key)
-
-    def _send_v3(self, data, msgtype=MSGTYPE_ENCRYPTED_REQUEST):
-        # socket_time = time.time() - self._timestamp
-        # _LOGGER.debug("Data: {} msgtype: {} len: {} socket time: {}".format(data.hex(), msgtype, len(data), socket_time))
-        if self._socket is None or self._tcp_key is None:
-            _LOGGER.debug(
-                "Socket %s invalid, Create New Socket and Get New tcp_key %s", self.get_socket_info(), self._tcp_key)
-            self._disconnect()
-            if self._authenticate() == False:
-                return []
-        # copy from data in order to resend data
-        original_data = bytearray.copy(data)
-        data = self.security.encode_8370(data, msgtype)
-        # time sleep retries second befor send data, default is 0
-        time.sleep(self._retries)
-        responses, b = self.request(data)
-        _LOGGER.debug("Got responses len: %d", len(responses))
-        if responses[8:13] == b'ERROR':
+            await self._connect()
+
+        _LOGGER.info("Authenticating with %s.", self._protocol.peer)
+
+        # Attempt to authenticate
+        while retries > 0:
+            try:
+                await self._protocol.authenticate(token, key)
+                break
+            except (TimeoutError, asyncio.TimeoutError):
+                if retries > 1:
+                    _LOGGER.warning("Authentication timeout. Resending.")
+                    retries -= 1
+                else:
+                    raise TimeoutError("No response from host.")
+
+        # Update stored token and key if successful
+        self._token = token
+        self._key = key
+
+        # Sleep briefly before requesting more data
+        await asyncio.sleep(1)
+
+    async def _read(self, **kwargs):
+        """Read and decode a frame from the protocol."""
+
+        # Await a response
+        packet = await self._protocol.read(**kwargs)
+        _LOGGER.debug("Received packet from %s: %s",
+                      self._protocol.peer, packet.hex())
+
+        # Decode packet to frame
+        response = _Packet.decode(packet)
+        _LOGGER.debug("Received response from %s: %s",
+                      self._protocol.peer, response.hex())
+
+        return response
+
+    async def send(self, data: bytes, retries: int = RETRIES):
+        """Send data via the LAN protocol. Connecting to the peer if necessary."""
+
+        # Connect if protocol doesn't exist or is dead
+        if self._protocol is None or not self._protocol.alive:
             self._disconnect()
-            return [b'ERROR']
-        if responses == bytearray(0) and self._retries < 2 and b:
-            packets = self._send_v3(original_data, msgtype)
-            self._retries = 0
-            return packets
-        responses, self._buffer = self.security.decode_8370(
-            self._buffer + responses)
-        packets = []
-        for response in responses:
-            if len(response) > 40 + 16:
-                response = Security.decrypt_aes(response[40:-16])
-            # header lenght is 10
-            if len(response) > 10:
-                packets.append(response)
-        return packets
-
-    def _send(self, data):
-        # time sleep retries second befor send data, default is 0
-        time.sleep(self._retries)
-        responses, b = self.request(data)
-        _LOGGER.debug("Get responses len: %d", len(responses))
-        if responses == bytearray(0) and self._retries < 2 and b:
-            packets = self._send(data)
-            self._retries = 0
-            return packets
-        packets = []
-        if responses == bytearray(0):
-            return packets
-        dlen = len(responses)
-        if responses[:2].hex() == "5a5a" and dlen > 5:
-            i = 0
-            # maybe multiple response
-            while i < dlen:
-                size = responses[i+4]
-                data = Security.decrypt_aes(responses[i:i+size][40:-16])
-                # header lenght is 10
-                if len(data) > 10:
-                    packets.append(data)
-                i += size
-        elif responses[0] == 0xaa and dlen > 2:
-            i = 0
-            while i < dlen:
-                size = responses[i+1]
-                data = responses[i:i+size+1]
-                # header lenght is 10
-                if len(data) > 10:
-                    packets.append(data)
-                i += size + 1
-        else:
-            _LOGGER.error("Unknown responses %s", responses.hex())
-        return packets
+            await self._connect()
 
-    def send(self, data: bytes):
-        if self._protocol_version == 3:
-            return self._send_v3(data)
-        else:
-            return self._send(data)
+            # Reauthenticate if needed
+            if self._protocol_version == 3:
+                await self.authenticate()
 
-    @property
-    def protocol_version(self) -> int:
-        return self._protocol_version
+        # Encode frame to packet
+        packet = _Packet.encode(self._device_id, data)
+
+        responses = []
+        while retries > 0:
+            # Send the request
+            _LOGGER.debug("Sending packet to %s: %s",
+                          self._protocol.peer, packet.hex())
+            self._protocol.write(packet)
+
+            try:
+                # Await a response
+                responses.append(await self._read())
+                break
+            except (TimeoutError, asyncio.TimeoutError):
+                if retries > 1:
+                    _LOGGER.warning("Request timeout. Resending.")
+                    retries -= 1
+                else:
+                    self._disconnect()
+                    raise TimeoutError("No response from host.")
+
+        # Attempt to read any additional responses without blocking
+        while True:
+            try:
+                responses.append(await self._read(timeout=0))
+            except asyncio.QueueEmpty:
+                break
+
+        return responses
 
 
 class Security:
     SIGN_KEY = "xhdiwjnchekd4d512chdjx5d8e4c394D2D7S".encode()
     ENC_KEY = md5(SIGN_KEY).digest()
 
-    def __init__(self):
-        self._tcp_key = None
-        self._request_count = 0
-        self._response_count = 0
-
-    def _aes_cbc_decrypt(self, raw, key):
-        return AES.new(key, AES.MODE_CBC, iv=bytes(16)).decrypt(raw)
-
-    def _aes_cbc_encrypt(self, raw, key):
-        return AES.new(key, AES.MODE_CBC, iv=bytes(16)).encrypt(raw)
-
-    def tcp_key(self, response, key):
-        if response == b'ERROR':
-            _LOGGER.error('authentication failed')
-            return b'', False
-        if len(response) != 64:
-            _LOGGER.error('unexpected data length')
-            return b'', False
-        payload = response[:32]
-        sign = response[32:]
-        plain = self._aes_cbc_decrypt(payload, key)
-        if sha256(plain).digest() != sign:
-            _LOGGER.error("sign does not match")
-            return b'', False
-        self._tcp_key = strxor(plain, key)
-        self._request_count = 0
-        self._response_count = 0
-        return self._tcp_key, True
-
-    def encode_8370(self, data, msgtype):
-        header = bytes([0x83, 0x70])
-        size, padding = len(data), 0
-        if msgtype in (MSGTYPE_ENCRYPTED_RESPONSE, MSGTYPE_ENCRYPTED_REQUEST):
-            if (size + 2) % 16 != 0:
-                padding = 16 - (size + 2 & 0xf)
-                size += padding + 32
-                data += get_random_bytes(padding)
-        header += size.to_bytes(2, 'big')
-        header += bytes([0x20, padding << 4 | msgtype])
-        if self._request_count >= 0xfff:
-            _LOGGER.info("request_count is too big to convert: %d",
-                         self._request_count)
-            self._request_count = 0
-        data = self._request_count.to_bytes(2, 'big') + data
-        self._request_count += 1
-        if msgtype in (MSGTYPE_ENCRYPTED_RESPONSE, MSGTYPE_ENCRYPTED_REQUEST):
-            sign = sha256(header + data).digest()
-            data = self._aes_cbc_encrypt(data, self._tcp_key) + sign
-        return header + data
-
-    def decode_8370(self, data):
-        if len(data) < 6:
-            return [], data
-        header = data[:6]
-        if header[0] != 0x83 or header[1] != 0x70:
-            raise Exception('not an 8370 message')
-        size = int.from_bytes(header[2:4], 'big') + 8
-        leftover = None
-        if len(data) < size:
-            return [], data
-        elif len(data) > size:
-            leftover = data[size:]
-            data = data[:size]
-        if header[4] != 0x20:
-            raise Exception('missing byte 4')
-        padding = header[5] >> 4
-        msgtype = header[5] & 0xf
-        data = data[6:]
-        if msgtype in (MSGTYPE_ENCRYPTED_RESPONSE, MSGTYPE_ENCRYPTED_REQUEST):
-            sign = data[-32:]
-            data = data[:-32]
-            data = self._aes_cbc_decrypt(data, self._tcp_key)
-            if sha256(header + data).digest() != sign:
-                raise Exception('sign does not match')
-            if padding:
-                data = data[:-padding]
-        self._response_count = int.from_bytes(data[:2], 'big')
-        data = data[2:]
-        if leftover:
-            packets, incomplete = self.decode_8370(leftover)
-            return [data] + packets, incomplete
-        return [data], b''
+    @classmethod
+    def decrypt_aes_cbc(cls, key: bytes, data: bytes):
+        return AES.new(key, AES.MODE_CBC, iv=bytes(16)).decrypt(data)
+
+    @classmethod
+    def encrypt_aes_cbc(cls, key: bytes, data: bytes):
+        return AES.new(key, AES.MODE_CBC, iv=bytes(16)).encrypt(data)
 
     @classmethod
     def decrypt_aes(cls, data: bytes):
@@ -323,10 +547,84 @@ class Security:
         return cipher.encrypt(Padding.pad(data, 16))
 
     @classmethod
-    def encode32(cls, data: bytes):
+    def sign(cls, data: bytes):
         return md5(data + Security.SIGN_KEY).digest()
 
     @classmethod
     def udpid(cls, id: bytes):
         with memoryview(sha256(id).digest()) as hash:
             return strxor(hash[:16], hash[16:])
+
+
+class _Packet:
+    """Class to encode/decode command frames to packets."""
+
+    @classmethod
+    def encode(cls, device_id: int, command: bytes):
+        """Encode a command frame to a LAN packet."""
+        # Encrypt command
+        encrypted_payload = Security.encrypt_aes(command)
+
+        # Compute total length
+        length = 40 + len(encrypted_payload) + 16
+
+        header = b"\x5A\x5A"  # Start of packet
+        header += b"\x01\x11"  # Message type
+        header += length.to_bytes(2, 'little')  # Packet size
+        header += b"\x20\x00"  # Magic bytes
+        header += bytes(4)  # Message ID
+        header += cls._timestamp()  # Timestamp
+        header += device_id.to_bytes(8, 'little')  # Device ID
+        header += bytes(12)  # ???
+
+        packet = header + encrypted_payload
+
+        # Append hash
+        return packet + Security.sign(packet)
+
+    @classmethod
+    def decode(cls, data: bytes):
+        """Decode a LAN packet to a command frame."""
+
+        with memoryview(data) as packet:
+            if len(packet) < 6:
+                raise ProtocolError(f"Packet is too short: {packet.hex()}")
+
+            if packet[:2] != b"\x5a\x5a":
+                # TODO old code handled raw frames? e.g start = 0xAA
+                raise ProtocolError(f"Unsupported packet: %s", packet.hex())
+
+            length = int.from_bytes(packet[4:6], "little")
+
+            if len(packet) < length:
+                raise ProtocolError(
+                    f"Packet is truncated. Expected {length} bytes, only have {len(packet)} bytes: {packet.hex()}")
+
+            packet = packet[:length]
+            encrypted_frame = packet[40:-16]
+            hash = packet[-16:]
+
+            # Check that received hash matches
+            if Security.sign(bytes(packet[:-16])) != hash:
+                raise ProtocolError(
+                    f"Calculated and received MD5 digest do not match.")
+
+            # Decrypt frame
+            return Security.decrypt_aes(encrypted_frame)
+
+    @classmethod
+    def _timestamp(cls):
+        now = datetime.utcnow()
+
+        # Each byte is a 2 digit component of the timestamp
+        # YYYYMMDDHHMMSSmm
+        return struct.pack("BBBBBBBB",
+                           int(now.microsecond / 10000),
+                           now.second,
+                           now.minute,
+                           now.hour,
+                           now.day,
+                           now.month,
+                           now.year % 100,
+                           int(now.year / 100)
+                           )
diff --git a/msmart/packet_builder.py b/msmart/packet_builder.py
deleted file mode 100644
index 0e4d65c..0000000
--- a/msmart/packet_builder.py
+++ /dev/null
@@ -1,59 +0,0 @@
-import datetime
-from msmart.base_command import command as base_command
-from msmart.lan import Security
-
-
-class packet_builder:
-
-    def __init__(self, device_id: int):
-        self.command = None
-        # aa20ac00000000000003418100ff03ff000200000000000000000000000006f274
-        # Init the packet with the header data.
-        self.packet = bytearray([
-            # 2 bytes - StaicHeader
-            0x5a, 0x5a,
-            # 2 bytes - mMessageType
-            0x01, 0x11,
-            # 2 bytes - PacketLenght
-            0x00, 0x00,
-            # 2 bytes
-            0x20, 0x00,
-            # 4 bytes - MessageId
-            0x00, 0x00, 0x00, 0x00,
-            # 8 bytes - Date&Time
-            0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-            # 6 bytes - mDeviceID
-            0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-            # 12 bytes
-            0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00
-        ])
-        self.packet[12:20] = self.packet_time()
-        self.packet[20:28] = device_id.to_bytes(8, 'little')
-
-    def set_command(self, command: base_command, add_crc8: bool = True):
-        self.command = command.pack()
-
-    def finalize(self):
-        # Append the command data(48 bytes) to the packet
-        self.packet.extend(Security.encrypt_aes(self.command)[:48])
-        # PacketLenght
-        self.packet[4:6] = (len(self.packet) + 16).to_bytes(2, 'little')
-        # Append a basic checksum data(16 bytes) to the packet
-        self.packet.extend(self.encode32(self.packet))
-        return self.packet
-
-    def encode32(self, data: bytearray):
-        # 16 bytes encode32
-        return Security.encode32(data)
-
-    def checksum(self, data):
-        return (~ sum(data) + 1) & 0xff
-
-    def packet_time(self):
-        t = datetime.datetime.now().strftime('%Y%m%d%H%M%S%f')[
-            :16]
-        b = bytearray()
-        for i in range(0, len(t), 2):
-            d = int(t[i:i+2])
-            b.insert(0, d)
-        return b
diff --git a/msmart/types.py b/msmart/types.py
new file mode 100644
index 0000000..2daa2c1
--- /dev/null
+++ b/msmart/types.py
@@ -0,0 +1,4 @@
+from typing import Union
+
+Token = Union[str, bytes, None]
+Key = Union[str, bytes, None]
diff --git a/pyproject.toml b/pyproject.toml
index 4a053b3..802bb95 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,12 +4,13 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "msmart"
-description = "A Python library for local control of Midea (and associated brands) smart air conditioners."
+description = "A library to control Midea appliances via the local area network."
 readme = "README.md"
 authors = [
-  {name = "Tucker Kern", email = "tuckkern@gmail.com"},
+  {name = "mac_zhou"},
+  {email = "mac.zfl@gmail.com"},
 ]
-requires-python = ">=3.8"
+requires-python = ">=3.5"
 license = {text = "MIT"}
 classifiers = [
   "Programming Language :: Python :: 3",
@@ -18,13 +19,13 @@ classifiers = [
 ]
 dependencies = [
   "pycryptodome",
-  "httpx"
+  "httpx",
+  "importlib-metadata; python_version<'3.8'",
 ]
 dynamic = ["version"]
 
 [project.urls]
-Repository = "https://github.com/mill1000/midea-msmart"
-Issues = "https://github.com/mill1000/midea-msmart/issues"
+homepage = "https://github.com/mac-zhou/midea-msmart"
 
 [project.scripts]
 midea-discover = "msmart.cli:main"


Test patch:
diff --git a/msmart/tests/test_lan.py b/msmart/tests/test_lan.py
new file mode 100644
index 0000000..6a44918
--- /dev/null
+++ b/msmart/tests/test_lan.py
@@ -0,0 +1,54 @@
+import unittest
+from msmart.lan import _Packet, _LanProtocolV3
+
+
+class TestEncodeDecode(unittest.IsolatedAsyncioTestCase):
+
+    async def test_enocde_packet_roundtrip(self):
+        """Test that we can encode and decode a frame."""
+        FRAME = bytes.fromhex(
+            "aa21ac8d000000000003418100ff03ff000200000000000000000000000003016971")
+
+        packet = _Packet.encode(123456, FRAME)
+        self.assertIsNotNone(packet)
+
+        rx_frame = _Packet.decode(packet)
+        self.assertEqual(rx_frame, FRAME)
+
+    async def test_decode_packet(self):
+        """Test that we can decode a packet to a frame."""
+        PACKET = bytes.fromhex(
+            "5a5a01116800208000000000000000000000000060ca0000000e0000000000000000000001000000c6a90377a364cb55af337259514c6f96bf084e8c7a899b50b68920cdea36cecf11c882a88861d1f46cd87912f201218c66151f0c9fbe5941c5384e707c36ff76")
+        EXPTECTED_FRAME = bytes.fromhex(
+            "aa22ac00000000000303c0014566000000300010045cff2070000000000000008bed19")
+
+        frame = _Packet.decode(PACKET)
+        self.assertIsNotNone(frame)
+        self.assertEqual(frame, EXPTECTED_FRAME)
+
+    async def test_decode_v3_packet(self):
+        """Test that we can decode a V3 packet to payload to a frame."""
+        PACKET = bytes.fromhex("8370008e2063ec2b8aeb17d4e3aff77094dde7fa65cf22671adf807f490a97b927347943626e9b4f58362cf34b97a0d641f8bf0c8fcbf69ad8cca131d2d7baa70ef048c5e3f3dc78da8af4598ff47aee762a0345c18815d91b50a24dedcacde0663c4ec5e73a963dc8bbbea9a593859996eb79dcfcc6a29b96262fcaa8ea6346366efea214e4a2e48caf83489475246b6fef90192b00")
+        LOCAL_KEY = bytes.fromhex(
+            "55a0a178746a424bf1fc6bb74b9fb9e4515965048d24ce8dc72aca91597d05ab")
+
+        EXPECTED_PAYLOAD = bytes.fromhex(
+            "5a5a01116800208000000000eaa908020c0817143daa0000008600000000000000000180000000003e99f93bb0cf9ffa100cb24dbae7838641d6e63ccbcd366130cd74a372932526d98479ff1725dce7df687d32e1776bf68a3fa6fd6259d7eb25f32769fcffef78")
+        EXPTECTED_FRAME = bytes.fromhex(
+            "aa23ac00000000000303c00145660000003c0010045c6800000000000000000000018426")
+
+        # Setup the protocol
+        protocol = _LanProtocolV3()
+        protocol._local_key = LOCAL_KEY
+
+        payload = protocol._process_packet(PACKET)
+        self.assertIsNotNone(payload)
+        self.assertEqual(payload, EXPECTED_PAYLOAD)
+
+        frame = _Packet.decode(payload)
+        self.assertIsNotNone(frame)
+        self.assertEqual(frame, EXPTECTED_FRAME)
+
+
+if __name__ == '__main__':
+    unittest.main()