Skip to content

Commit

Permalink
chore: typing c2 (#52)
Browse files Browse the repository at this point in the history
<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

- **Refactor**
- Introduced type annotations across various methods to enhance code
readability and maintainability.
- Updated method signatures to include type hints for parameters and
return types.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->
  • Loading branch information
chemelli74 authored Jun 4, 2024
1 parent 95adbf4 commit 75493c3
Show file tree
Hide file tree
Showing 2 changed files with 41 additions and 35 deletions.
25 changes: 13 additions & 12 deletions midealocal/devices/c2/__init__.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import json
import logging
import sys
from typing import Any

from .message import MessageC2Response, MessagePower, MessageQuery, MessageSet

Expand Down Expand Up @@ -43,7 +44,7 @@ def __init__(
model: str,
subtype: int,
customize: str,
):
) -> None:
super().__init__(
name=name,
device_id=device_id,
Expand Down Expand Up @@ -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 = {}
Expand All @@ -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
Expand All @@ -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
Expand Down
51 changes: 28 additions & 23 deletions midealocal/devices/c2/message.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from enum import IntEnum
from typing import cast

from ...message import (
MessageBody,
Expand All @@ -9,6 +10,7 @@


class C2MessageEnum(IntEnum):
none = 0x00
sensor_light = 0x01
child_lock = 0x10
foam_shield = 0x1F
Expand Down Expand Up @@ -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,
Expand All @@ -51,25 +55,25 @@ 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,
body_type=0x01,
)

@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,
Expand All @@ -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:
Expand All @@ -87,38 +91,38 @@ 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,
body_type=0x02,
)

@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
Expand All @@ -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
Expand All @@ -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,
Expand Down

0 comments on commit 75493c3

Please sign in to comment.