chore: typing db (#62)

midealocal/devices/db/__init__.py

@@ -1,5 +1,6 @@
 import logging
 import sys
+from typing import Any
 
 from .message import MessageDBResponse, MessagePower, MessageQuery, MessageStart
 
@@ -34,7 +35,7 @@ def __init__(
         model: str,
         subtype: int,
         customize: str,
-    ):
+    ) -> None:
         super().__init__(
             name=name,
             device_id=device_id,
@@ -55,10 +56,10 @@ def __init__(
             },
         )
 
-    def build_query(self):
+    def build_query(self) -> list[MessageQuery]:
         return [MessageQuery(self._protocol_version)]
 
-    def process_message(self, msg):
+    def process_message(self, msg: bytes) -> dict[str, Any]:
         message = MessageDBResponse(msg)
         _LOGGER.debug(f"[{self.device_id}] Received: {message}")
         new_status = {}
@@ -82,7 +83,8 @@ def process_message(self, msg):
                 new_status[str(status)] = self._attributes[status]
         return new_status
 
-    def set_attribute(self, attr, value):
+    def set_attribute(self, attr: str, value: Any) -> None:
+        message: MessagePower | MessageStart | None = None
         if attr == DeviceAttributes.power:
             message = MessagePower(self._protocol_version)
             message.power = value

midealocal/devices/db/message.py

@@ -7,7 +7,9 @@
 
 
 class MessageDBBase(MessageRequest):
-    def __init__(self, protocol_version, message_type, body_type):
+    def __init__(
+        self, protocol_version: int, message_type: int, body_type: int
+    ) -> None:
         super().__init__(
             device_type=0xDB,
             protocol_version=protocol_version,
@@ -16,25 +18,25 @@ def __init__(self, protocol_version, message_type, body_type):
         )
 
     @property
-    def _body(self):
+    def _body(self) -> bytearray:
         raise NotImplementedError
 
 
 class MessageQuery(MessageDBBase):
-    def __init__(self, protocol_version):
+    def __init__(self, protocol_version: int) -> None:
         super().__init__(
             protocol_version=protocol_version,
             message_type=MessageType.query,
             body_type=0x03,
         )
 
     @property
-    def _body(self):
+    def _body(self) -> bytearray:
         return bytearray([])
 
 
 class MessagePower(MessageDBBase):
-    def __init__(self, protocol_version):
+    def __init__(self, protocol_version: int) -> None:
         super().__init__(
             protocol_version=protocol_version,
             message_type=MessageType.set,
@@ -43,7 +45,7 @@ def __init__(self, protocol_version):
         self.power = False
 
     @property
-    def _body(self):
+    def _body(self) -> bytearray:
         power = 0x01 if self.power else 0x00
         return bytearray(
             [
@@ -73,7 +75,7 @@ def _body(self):
 
 
 class MessageStart(MessageDBBase):
-    def __init__(self, protocol_version):
+    def __init__(self, protocol_version: int) -> None:
         super().__init__(
             protocol_version=protocol_version,
             message_type=MessageType.set,
@@ -83,7 +85,7 @@ def __init__(self, protocol_version):
         self.washing_data = bytearray([])
 
     @property
-    def _body(self):
+    def _body(self) -> bytearray:
         if self.start:  # Pause
             return bytearray([0xFF, 0x01]) + self.washing_data
         else:
@@ -92,25 +94,24 @@ def _body(self):
 
 
 class DBGeneralMessageBody(MessageBody):
-    def __init__(self, body):
+    def __init__(self, body: bytearray) -> None:
         super().__init__(body)
         self.power = body[1] > 0
         self.start = True if body[2] in [2, 6] else False
         self.washing_data = body[3:16]
         self.progress = 0
+        self.time_remaining: float | None = None
         for i in range(7):
             if (body[16] & (1 << i)) > 0:
                 self.progress = i + 1
                 break
         if self.power:
             self.time_remaining = body[17] + (body[18] << 8)
-        else:
-            self.time_remaining = None
 
 
 class MessageDBResponse(MessageResponse):
-    def __init__(self, message):
-        super().__init__(message)
+    def __init__(self, message: bytes) -> None:
+        super().__init__(bytearray(message))
         if self.message_type in [MessageType.query, MessageType.set] or (
             self.message_type == MessageType.notify1 and self.body_type == 0x04
         ):