diff --git a/midealocal/devices/c2/__init__.py b/midealocal/devices/c2/__init__.py index 4eb5a94a..a155b518 100644 --- a/midealocal/devices/c2/__init__.py +++ b/midealocal/devices/c2/__init__.py @@ -1,6 +1,7 @@ import json import logging import sys +from typing import Any from .message import MessageC2Response, MessagePower, MessageQuery, MessageSet @@ -43,7 +44,7 @@ def __init__( model: str, subtype: int, customize: str, - ): + ) -> None: super().__init__( name=name, device_id=device_id, @@ -71,30 +72,30 @@ def __init__( DeviceAttributes.filter_life: None, }, ) - self._max_dry_level = None - self._max_water_temp_level = None - self._max_seat_temp_level = None + self._max_dry_level: int | None = None + self._max_water_temp_level: int | None = None + self._max_seat_temp_level: int | None = None self._default_max_dry_level = 3 self._default_max_water_temp_level = 5 self._default_max_seat_temp_level = 5 self.set_customize(customize) @property - def max_dry_level(self): + def max_dry_level(self) -> int | None: return self._max_dry_level @property - def max_water_temp_level(self): + def max_water_temp_level(self) -> int | None: return self._max_water_temp_level @property - def max_seat_temp_level(self): + def max_seat_temp_level(self) -> int | None: return self._max_seat_temp_level - 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 = MessageC2Response(msg) _LOGGER.debug(f"[{self.device_id}] Received: {message}") new_status = {} @@ -104,8 +105,8 @@ def process_message(self, msg): new_status[str(status)] = getattr(message, str(status)) return new_status - def set_attribute(self, attr, value): - message = None + def set_attribute(self, attr: str, value: Any) -> None: + message: MessagePower | MessageSet | None = None if attr == DeviceAttributes.power: message = MessagePower(self._protocol_version) message.power = value @@ -122,7 +123,7 @@ def set_attribute(self, attr, value): if message: self.build_send(message) - def set_customize(self, customize): + def set_customize(self, customize: str) -> None: self._max_dry_level = self._default_max_dry_level self._max_water_temp_level = self._default_max_water_temp_level self._max_seat_temp_level = self._default_max_seat_temp_level diff --git a/midealocal/devices/c2/message.py b/midealocal/devices/c2/message.py index b35f1e5f..6ba83999 100644 --- a/midealocal/devices/c2/message.py +++ b/midealocal/devices/c2/message.py @@ -1,4 +1,5 @@ from enum import IntEnum +from typing import cast from ...message import ( MessageBody, @@ -9,6 +10,7 @@ class C2MessageEnum(IntEnum): + none = 0x00 sensor_light = 0x01 child_lock = 0x10 foam_shield = 0x1F @@ -42,7 +44,9 @@ class C2MessageEnum(IntEnum): class MessageC2Base(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=0xC2, protocol_version=protocol_version, @@ -51,12 +55,12 @@ def __init__(self, protocol_version, message_type, body_type): ) @property - def _body(self): + def _body(self) -> bytearray: raise NotImplementedError class MessageQuery(MessageC2Base): - def __init__(self, protocol_version): + def __init__(self, protocol_version: int) -> None: super().__init__( protocol_version=protocol_version, message_type=MessageType.query, @@ -64,12 +68,12 @@ def __init__(self, protocol_version): ) @property - def _body(self): + def _body(self) -> bytearray: return bytearray([0x01]) class MessagePower(MessageC2Base): - def __init__(self, protocol_version): + def __init__(self, protocol_version: int) -> None: super().__init__( protocol_version=protocol_version, message_type=MessageType.set, @@ -78,7 +82,7 @@ def __init__(self, protocol_version): self.power = False @property - def _body(self): + def _body(self) -> bytearray: if self.power: self.body_type = 0x01 else: @@ -87,7 +91,7 @@ def _body(self): class MessagePowerOff(MessageC2Base): - def __init__(self, protocol_version): + def __init__(self, protocol_version: int) -> None: super().__init__( protocol_version=protocol_version, message_type=MessageType.set, @@ -95,30 +99,30 @@ def __init__(self, protocol_version): ) @property - def _body(self): + def _body(self) -> bytearray: return bytearray([0x01]) class MessageSet(MessageC2Base): - def __init__(self, protocol_version): + def __init__(self, protocol_version: int) -> None: super().__init__( protocol_version=protocol_version, message_type=MessageType.set, body_type=0x00, ) - self.child_lock = None - self.sensor_light = None - self.water_temp_level = None - self.seat_temp_level = None - self.dry_level = None - self.foam_shield = None + self.child_lock: bool | None = None + self.sensor_light: bool | None = None + self.water_temp_level: int | None = None + self.seat_temp_level: int | None = None + self.dry_level: int | None = None + self.foam_shield: bool | None = None @property - def _body(self): + def _body(self) -> bytearray: self.body_type = 0x14 - key = 0x00 - value = 0x00 + key = C2MessageEnum.none + value: int | bool = 0x00 if self.child_lock is not None: key = C2MessageEnum.child_lock value = self.child_lock @@ -137,12 +141,13 @@ def _body(self): elif self.foam_shield is not None: key = C2MessageEnum.foam_shield value = self.foam_shield - value = C2_MESSAGE_KEYS.get(key).get(value) + x = cast(dict, C2_MESSAGE_KEYS[key]) + value = cast(int | bool, x.get(value)) return bytearray([key, value]) class C2MessageBody(MessageBody): - def __init__(self, body): + def __init__(self, body: bytearray) -> None: super().__init__(body) self.power = (body[2] & 0x01) > 0 self.seat_status = (body[3] & 0x01) > 0 @@ -160,13 +165,13 @@ def __init__(self, body): class C2Notify1MessageBody(MessageBody): - def __init__(self, body): + def __init__(self, body: bytearray) -> None: super().__init__(body) class MessageC2Response(MessageResponse): - def __init__(self, message): - super().__init__(message) + def __init__(self, message: bytes) -> None: + super().__init__(bytearray(message)) if self.message_type in [ MessageType.notify1, MessageType.query,