Skip to content

Commit

Permalink
chore: typing c3
Browse files Browse the repository at this point in the history
  • Loading branch information
chemelli74 committed Jun 4, 2024
1 parent 7e66870 commit 592848a
Show file tree
Hide file tree
Showing 2 changed files with 27 additions and 22 deletions.
19 changes: 11 additions & 8 deletions midealocal/devices/c3/__init__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import logging
import sys
from typing import Any

from .message import (
MessageC3Response,
Expand Down Expand Up @@ -75,7 +76,7 @@ def __init__(
model: str,
subtype: int,
customize: str,
):
) -> None:
super().__init__(
name=name,
device_id=device_id,
Expand Down Expand Up @@ -131,10 +132,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 = MessageC3Response(msg)
_LOGGER.debug(f"[{self.device_id}] Received: {message}")
new_status = {}
Expand Down Expand Up @@ -219,7 +220,7 @@ def process_message(self, msg):

return new_status

def make_message_set(self):
def make_message_set(self) -> MessageSet:
message = MessageSet(self._protocol_version)
message.zone1_power = self._attributes[DeviceAttributes.zone1_power]
message.zone2_power = self._attributes[DeviceAttributes.zone2_power]
Expand All @@ -235,8 +236,8 @@ def make_message_set(self):
message.fast_dhw = self._attributes[DeviceAttributes.fast_dhw]
return message

def set_attribute(self, attr, value):
message = None
def set_attribute(self, attr: str, value: Any) -> None:
message: MessageSet | MessageSetECO | MessageSetSilent | None = None
if attr in [
DeviceAttributes.zone1_power,
DeviceAttributes.zone2_power,
Expand All @@ -259,7 +260,7 @@ def set_attribute(self, attr, value):
if message is not None:
self.build_send(message)

def set_mode(self, zone, mode):
def set_mode(self, zone: int, mode: int) -> None:
message = self.make_message_set()
if zone == 0:
message.zone1_power = True
Expand All @@ -268,7 +269,9 @@ def set_mode(self, zone, mode):
message.mode = mode
self.build_send(message)

def set_target_temperature(self, zone, target_temperature, mode):
def set_target_temperature(
self, zone: int, target_temperature: int, mode: int
) -> None:
message = self.make_message_set()
if self._attributes[DeviceAttributes.zone_temp_type][zone]:
message.zone_target_temp[zone] = target_temperature
Expand Down
30 changes: 16 additions & 14 deletions midealocal/devices/c3/message.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,9 @@


class MessageC3Base(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=0xC3,
protocol_version=protocol_version,
Expand All @@ -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(MessageC3Base):
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([])


class MessageSet(MessageC3Base):
def __init__(self, protocol_version):
def __init__(self, protocol_version: int) -> None:
super().__init__(
protocol_version=protocol_version,
message_type=MessageType.set,
Expand All @@ -54,7 +56,7 @@ def __init__(self, protocol_version):
self.tbh = False

@property
def _body(self):
def _body(self) -> bytearray:
# Byte 1
zone1_power = 0x01 if self.zone1_power else 0x00
zone2_power = 0x02 if self.zone2_power else 0x00
Expand Down Expand Up @@ -82,7 +84,7 @@ def _body(self):


class MessageSetSilent(MessageC3Base):
def __init__(self, protocol_version):
def __init__(self, protocol_version: int) -> None:
super().__init__(
protocol_version=protocol_version,
message_type=MessageType.set,
Expand All @@ -92,7 +94,7 @@ def __init__(self, protocol_version):
self.super_silent = False

@property
def _body(self):
def _body(self) -> bytearray:
silent_mode = 0x01 if self.silent_mode else 0
super_silent = 0x02 if self.super_silent else 0

Expand All @@ -112,7 +114,7 @@ def _body(self):


class MessageSetECO(MessageC3Base):
def __init__(self, protocol_version):
def __init__(self, protocol_version: int) -> None:
super().__init__(
protocol_version=protocol_version,
message_type=MessageType.set,
Expand All @@ -121,14 +123,14 @@ def __init__(self, protocol_version):
self.eco_mode = False

@property
def _body(self):
def _body(self) -> bytearray:
eco_mode = 0x01 if self.eco_mode else 0

return bytearray([eco_mode, 0x00, 0x00, 0x00, 0x00, 0x00])


class C3MessageBody(MessageBody):
def __init__(self, body, data_offset=0):
def __init__(self, body: bytearray, data_offset: int = 0) -> None:
super().__init__(body)
self.zone1_power = body[data_offset + 0] & 0x01 > 0
self.zone2_power = body[data_offset + 0] & 0x02 > 0
Expand Down Expand Up @@ -162,7 +164,7 @@ def __init__(self, body, data_offset=0):


class C3Notify1MessageBody(MessageBody):
def __init__(self, body, data_offset=0):
def __init__(self, body: bytearray, data_offset: int = 0) -> None:
super().__init__(body)
status_byte = body[data_offset]
self.status_tbh = (status_byte & 0x08) > 0
Expand Down Expand Up @@ -190,8 +192,8 @@ def __init__(self, body, data_offset=0):


class MessageC3Response(MessageResponse):
def __init__(self, message):
super().__init__(message)
def __init__(self, message: bytes) -> None:
super().__init__(bytearray(message))
if (
self.message_type
in [MessageType.set, MessageType.notify1, MessageType.query]
Expand Down

0 comments on commit 592848a

Please sign in to comment.