diff --git a/midealocal/devices/c3/__init__.py b/midealocal/devices/c3/__init__.py index 0c9dd9b0..17c49d0e 100644 --- a/midealocal/devices/c3/__init__.py +++ b/midealocal/devices/c3/__init__.py @@ -2,10 +2,10 @@ import json import logging -from enum import IntEnum, StrEnum from typing import Any from midealocal.device import MideaDevice +from midealocal.devices.c3.const import C3DeviceMode, C3SilentLevel, DeviceAttributes from .message import ( MessageC3Response, @@ -18,58 +18,6 @@ _LOGGER = logging.getLogger(__name__) -class DeviceAttributes(StrEnum): - """Midea C3 device attributes.""" - - zone1_power = "zone1_power" - zone2_power = "zone2_power" - dhw_power = "dhw_power" - zone1_curve = "zone1_curve" - zone2_curve = "zone2_curve" - disinfect = "disinfect" - fast_dhw = "fast_dhw" - zone_temp_type = "zone_temp_type" - zone1_room_temp_mode = "zone1_room_temp_mode" - zone2_room_temp_mode = "zone2_room_temp_mode" - zone1_water_temp_mode = "zone1_water_temp_mode" - zone2_water_temp_mode = "zone2_water_temp_mode" - mode = "mode" - mode_auto = "mode_auto" - zone_target_temp = "zone_target_temp" - dhw_target_temp = "dhw_target_temp" - room_target_temp = "room_target_temp" - zone_heating_temp_max = "zone_heating_temp_max" - zone_heating_temp_min = "zone_heating_temp_min" - zone_cooling_temp_max = "zone_cooling_temp_max" - zone_cooling_temp_min = "zone_cooling_temp_min" - tank_actual_temperature = "tank_actual_temperature" - room_temp_max = "room_temp_max" - room_temp_min = "room_temp_min" - dhw_temp_max = "dhw_temp_max" - dhw_temp_min = "dhw_temp_min" - target_temperature = "target_temperature" - temperature_max = "temperature_max" - temperature_min = "temperature_min" - status_heating = "status_heating" - status_dhw = "status_dhw" - status_tbh = "status_tbh" - status_ibh = "status_ibh" - total_energy_consumption = "total_energy_consumption" - total_produced_energy = "total_produced_energy" - outdoor_temperature = "outdoor_temperature" - silent_mode = "silent_mode" - eco_mode = "eco_mode" - tbh = "tbh" - error_code = "error_code" - - -class C3DeviceMode(IntEnum): - """C3 Device Mode.""" - - COOL = 2 - HEAT = 3 - - class MideaC3Device(MideaDevice): """Midea C3 device.""" @@ -112,6 +60,7 @@ def __init__( DeviceAttributes.zone1_water_temp_mode: False, DeviceAttributes.zone2_water_temp_mode: False, DeviceAttributes.silent_mode: False, + DeviceAttributes.SILENT_LEVEL: C3SilentLevel.SILENT, DeviceAttributes.eco_mode: False, DeviceAttributes.tbh: False, DeviceAttributes.mode: 1, diff --git a/midealocal/devices/c3/const.py b/midealocal/devices/c3/const.py new file mode 100644 index 00000000..2d429849 --- /dev/null +++ b/midealocal/devices/c3/const.py @@ -0,0 +1,64 @@ +"""Midea local C3 device const.""" + +from enum import IntEnum, StrEnum + + +class DeviceAttributes(StrEnum): + """Midea C3 device attributes.""" + + zone1_power = "zone1_power" + zone2_power = "zone2_power" + dhw_power = "dhw_power" + zone1_curve = "zone1_curve" + zone2_curve = "zone2_curve" + disinfect = "disinfect" + fast_dhw = "fast_dhw" + zone_temp_type = "zone_temp_type" + zone1_room_temp_mode = "zone1_room_temp_mode" + zone2_room_temp_mode = "zone2_room_temp_mode" + zone1_water_temp_mode = "zone1_water_temp_mode" + zone2_water_temp_mode = "zone2_water_temp_mode" + mode = "mode" + mode_auto = "mode_auto" + zone_target_temp = "zone_target_temp" + dhw_target_temp = "dhw_target_temp" + room_target_temp = "room_target_temp" + zone_heating_temp_max = "zone_heating_temp_max" + zone_heating_temp_min = "zone_heating_temp_min" + zone_cooling_temp_max = "zone_cooling_temp_max" + zone_cooling_temp_min = "zone_cooling_temp_min" + tank_actual_temperature = "tank_actual_temperature" + room_temp_max = "room_temp_max" + room_temp_min = "room_temp_min" + dhw_temp_max = "dhw_temp_max" + dhw_temp_min = "dhw_temp_min" + target_temperature = "target_temperature" + temperature_max = "temperature_max" + temperature_min = "temperature_min" + status_heating = "status_heating" + status_dhw = "status_dhw" + status_tbh = "status_tbh" + status_ibh = "status_ibh" + total_energy_consumption = "total_energy_consumption" + total_produced_energy = "total_produced_energy" + outdoor_temperature = "outdoor_temperature" + silent_mode = "silent_mode" + SILENT_LEVEL = "silent_level" + eco_mode = "eco_mode" + tbh = "tbh" + error_code = "error_code" + + +class C3SilentLevel(IntEnum): + """C3 Silent Level.""" + + OFF = 0x0 + SILENT = 0x1 + SUPER_SILENT = 0x3 + + +class C3DeviceMode(IntEnum): + """C3 Device Mode.""" + + COOL = 2 + HEAT = 3 diff --git a/midealocal/devices/c3/message.py b/midealocal/devices/c3/message.py index 49066d7d..18a79bc3 100644 --- a/midealocal/devices/c3/message.py +++ b/midealocal/devices/c3/message.py @@ -1,5 +1,6 @@ """Midea local C3 message.""" +from midealocal.devices.c3.const import C3SilentLevel from midealocal.message import ( BodyType, MessageBody, @@ -111,16 +112,13 @@ def __init__(self, protocol_version: int) -> None: body_type=0x05, ) self.silent_mode = False - self.super_silent = False + self.silent_level = C3SilentLevel.OFF @property def _body(self) -> bytearray: - silent_mode = 0x01 if self.silent_mode else 0 - super_silent = 0x02 if self.super_silent else 0 - return bytearray( [ - silent_mode | super_silent, + self.silent_level if self.silent_mode else C3SilentLevel.OFF, 0x00, 0x00, 0x00, @@ -220,6 +218,33 @@ def __init__(self, body: bytearray, data_offset: int = 0) -> None: ) +class C3QuerySilenceMessageBody(MessageBody): + """C3 Query silence message body.""" + + def __init__(self, body: bytearray, data_offset: int = 0) -> None: + """Initialize C3 notify1 message body.""" + super().__init__(body) + self.silence_mode = body[data_offset] & 0x1 > 0 + self.silence_level = ( + (body[data_offset] & 0x1) + ((body[data_offset] & 0x8) >> 2) + if self.silence_mode + else C3SilentLevel.OFF + ) + # Message protocol information: + # silence_function_state: Byte 1, BIT 0 + # silence_timer1_state: Byte 1, BIT 1 + # silence_timer2_state: Byte 1, BIT 2 + # silence_function_level: Byte 1, BIT 3 + # silence_timer1_starthour: Byte 2 + # silence_timer1_startmin: Byte 3 + # silence_timer1_endhour: Byte 4 + # silence_timer1_endmin: Byte 5 + # silence_timer2_starthour: Byte 6 + # silence_timer2_startmin: Byte 7 + # silence_timer2_endhour: Byte 8 + # silence_timer2_endmin: Byte 9 + + class MessageC3Response(MessageResponse): """C3 message response.""" @@ -236,4 +261,6 @@ def __init__(self, message: bytes) -> None: self.message_type == MessageType.notify1 and self.body_type == BodyType.X04 ): self.set_body(C3Notify1MessageBody(super().body, data_offset=1)) + elif self.message_type == MessageType.query and self.body_type == BodyType.X05: + self.set_body(C3QuerySilenceMessageBody(super().body, data_offset=1)) self.set_attr() diff --git a/tests/devices/c3/device_c3_test.py b/tests/devices/c3/device_c3_test.py index 5bf0dcc9..31ca8a30 100644 --- a/tests/devices/c3/device_c3_test.py +++ b/tests/devices/c3/device_c3_test.py @@ -4,7 +4,10 @@ import pytest -from midealocal.devices.c3 import C3DeviceMode, DeviceAttributes, MideaC3Device +from midealocal.devices.c3 import ( + MideaC3Device, +) +from midealocal.devices.c3.const import C3DeviceMode, C3SilentLevel, DeviceAttributes from midealocal.devices.c3.message import ( MessageQuery, ) @@ -46,6 +49,10 @@ def test_initial_attributes(self) -> None: assert self.device.attributes[DeviceAttributes.zone1_water_temp_mode] is False assert self.device.attributes[DeviceAttributes.zone2_water_temp_mode] is False assert self.device.attributes[DeviceAttributes.silent_mode] is False + assert ( + self.device.attributes[DeviceAttributes.SILENT_LEVEL] + == C3SilentLevel.SILENT + ) assert self.device.attributes[DeviceAttributes.eco_mode] is False assert self.device.attributes[DeviceAttributes.tbh] is False assert self.device.attributes[DeviceAttributes.mode] == 1 diff --git a/tests/devices/c3/message_c3_test.py b/tests/devices/c3/message_c3_test.py new file mode 100644 index 00000000..9f1f11e5 --- /dev/null +++ b/tests/devices/c3/message_c3_test.py @@ -0,0 +1,312 @@ +"""Test c3 message.""" + +import pytest + +from midealocal.devices.c3.const import C3DeviceMode, C3SilentLevel +from midealocal.devices.c3.message import ( + MessageC3Base, + MessageC3Response, + MessageQuery, + MessageSet, + MessageSetECO, + MessageSetSilent, +) +from midealocal.message import BodyType, MessageType + + +class TestMessageC3Base: + """Test C3 Message Base.""" + + def test_body_not_implemented(self) -> None: + """Test body not implemented.""" + msg = MessageC3Base(protocol_version=1, message_type=1, body_type=1) + with pytest.raises(NotImplementedError): + _ = msg.body + + +class TestC3MessageQuery: + """Test C3 message query.""" + + def test_query_body(self) -> None: + """Test query body.""" + msg = MessageQuery(protocol_version=1) + expected_body = bytearray([0x1]) + assert msg.body == expected_body + + +class TestC3MessageSet: + """Test C3 message set.""" + + def test_set_body(self) -> None: + """Test set body.""" + msg = MessageSet(protocol_version=1) + msg.zone1_power = True + msg.zone2_power = True + msg.dhw_power = True + msg.mode = C3DeviceMode.COOL + msg.zone_target_temp = [23.0, 22.0] + msg.dhw_target_temp = 45 + msg.room_target_temp = 24.0 + msg.zone1_curve = True + msg.zone2_curve = True + msg.disinfect = True + msg.fast_dhw = True + msg.tbh = True + + expected_body = bytearray( + [ + msg.body_type, + 0x1 | 0x2 | 0x4, + 0x2, + 23, + 22, + 45, + 24 * 2, + 0x1 | 0x2 | 0x4 | 0x8, + ], + ) + assert msg.body == expected_body + + +class TestC3MessageSetSilent: + """Test C3 message set silent.""" + + def test_set_silent_body(self) -> None: + """Test set silent body.""" + msg = MessageSetSilent(protocol_version=1) + expected_body_off = bytearray([0x5] + [0x0] * 9) + expected_body_silent = bytearray([0x5, 0x1] + [0x0] * 8) + expected_body_super_silent = bytearray([0x5, 0x3] + [0x0] * 8) + assert msg.body == expected_body_off + msg.silent_mode = True + assert msg.body == expected_body_off # mode true and level unset + + msg.silent_level = C3SilentLevel.SILENT + assert msg.body == expected_body_silent + msg.silent_mode = False + assert msg.body == expected_body_off # mode false and level silent + + msg.silent_level = C3SilentLevel.SUPER_SILENT + assert msg.body == expected_body_off # mode false and level super silent + msg.silent_mode = True + assert msg.body == expected_body_super_silent + + +class TestC3MessageSetECO: + """Test C3 message set ECO.""" + + def test_set_eco_body(self) -> None: + """Test set ECO body.""" + msg = MessageSetECO(protocol_version=1) + expected_body_off = bytearray([0x7] + [0x0] * 6) + expected_body_eco = bytearray([0x7, 0x1] + [0x0] * 5) + + assert msg.body == expected_body_off + msg.eco_mode = True + assert msg.body == expected_body_eco + + +class TestMessageC3Response: + """Test Message C3 Response.""" + + @pytest.fixture(autouse=True) + def _setup_header(self) -> None: + """Do setup header.""" + self.header = bytearray( + [ + 0xAA, + 0x00, + 0xC3, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x01, + 0x00, # message type + ], + ) + + def test_message_generic_response(self) -> None: + """Test message generic response.""" + body = bytearray( + [ + BodyType.X01, + 0x01 + | 0x04 + | 0x08 + | 0x20, # BYTE 1: zone_power1 + dhw_power + zone1_curve + tbh + 0x30, # BYTE 2: temp_type [True, True] + 0x2 | 0x8, # BYTE 3: silent on, eco on + 0x3, # BYTE 4: Mode HEAT + 0x2, # BYTE 5: Mode Auto COOL + 21, # BYTE 6: Zone1 Target Temp + 22, # BYTE 7: Zone2 Target Temp + 42, # BYTE 8: DHW Target Temp + 45, # BYTE 9: Room Target Temp * 2 + 30, # BYTE 10: zone1_heating_temp_max + 20, # BYTE 11: zone1_heating_temp_min + 25, # BYTE 12: zone1_cooling_temp_max + 16, # BYTE 13: zone1_cooling_temp_min + 35, # BYTE 14: zone2_heating_temp_max + 20, # BYTE 15: zone2_heating_temp_min + 30, # BYTE 16: zone2_cooling_temp_max + 18, # BYTE 17: zone2_cooling_temp_min + 61, # BYTE 18: room_temp_max / 2 + 32, # BYTE 19: room_temp_min / 2 + 50, # BYTE 20: dhw_temp_max + 34, # BYTE 21: dhw_temp_min + 44, # BYTE 22: tank_actual_temperature + 0x0, # BYTE 23; error_code + 0x0, # CRC + ], + ) + + for message_type in ( + MessageType.set, + MessageType.query, + MessageType.notify1, + MessageType.notify2, + ): + self.header[-1] = message_type + response = MessageC3Response(self.header + body) + + assert response.body_type == BodyType.X01 + assert hasattr(response, "zone1_power") + assert response.zone1_power is True + assert hasattr(response, "zone2_power") + assert response.zone2_power is False + assert hasattr(response, "dhw_power") + assert response.dhw_power is True + assert hasattr(response, "zone1_curve") + assert response.zone1_curve is True + assert hasattr(response, "zone2_curve") + assert response.zone2_curve is False + assert hasattr(response, "disinfect") + assert response.disinfect is True + assert hasattr(response, "tbh") + assert response.tbh is True + assert hasattr(response, "fast_dhw") + assert response.fast_dhw is False + assert hasattr(response, "zone_temp_type") + assert response.zone_temp_type == [True, True] + assert hasattr(response, "silent_mode") + assert response.silent_mode is True + assert hasattr(response, "eco_mode") + assert response.eco_mode is True + assert hasattr(response, "mode") + assert response.mode == C3DeviceMode.HEAT + assert hasattr(response, "mode_auto") + assert response.mode_auto == C3DeviceMode.COOL + assert hasattr(response, "zone_target_temp") + assert response.zone_target_temp == [21.0, 22.0] + assert hasattr(response, "dhw_target_temp") + assert response.dhw_target_temp == 42.0 + assert hasattr(response, "room_target_temp") + assert response.room_target_temp == 22.5 + assert hasattr(response, "zone_heating_temp_max") + assert response.zone_heating_temp_max == [30.0, 35.0] + assert hasattr(response, "zone_heating_temp_min") + assert response.zone_heating_temp_min == [20.0, 20.0] + assert hasattr(response, "zone_cooling_temp_max") + assert response.zone_cooling_temp_max == [25.0, 30.0] + assert hasattr(response, "zone_cooling_temp_min") + assert response.zone_cooling_temp_min == [16.0, 18.0] + assert hasattr(response, "room_temp_max") + assert response.room_temp_max == 30.5 + assert hasattr(response, "room_temp_min") + assert response.room_temp_min == 16.0 + assert hasattr(response, "dhw_temp_max") + assert response.dhw_temp_max == 50 + assert hasattr(response, "dhw_temp_min") + assert response.dhw_temp_min == 34 + assert hasattr(response, "tank_actual_temperature") + assert response.tank_actual_temperature == 44 + assert hasattr(response, "error_code") + assert response.error_code == 0x0 + + def test_message_notify1_x04_response(self) -> None: + """Test message notify1 x04 response.""" + self.header[-1] = MessageType.notify1 + body = bytearray( + [ + BodyType.X04, + 0x01 | 0x04, # BYTE 1: status_dhw + status_heating + 0x32, # BYTE 2: total_energy_consumption + 0x1A, # BYTE 3: total_energy_consumption + 0xB3, # BYTE 4: total_energy_consumption + 0xC2, # BYTE 5: total_energy_consumption + 21, # BYTE 6: total_produced_energy + 22, # BYTE 7: total_produced_energy + 42, # BYTE 8: total_produced_energy + 45, # BYTE 9: total_produced_energy + 30, # BYTE 10: outdoor_temperature + 0x0, # CRC + ], + ) + response = MessageC3Response(self.header + body) + assert response.body_type == BodyType.X04 + assert hasattr(response, "status_tbh") + assert response.status_tbh is False + assert hasattr(response, "status_dhw") + assert response.status_dhw is True + assert hasattr(response, "status_ibh") + assert response.status_ibh is False + assert hasattr(response, "status_heating") + assert response.status_heating is True + assert hasattr(response, "total_energy_consumption") + assert response.total_energy_consumption == 214750114754 + assert hasattr(response, "total_produced_energy") + assert response.total_produced_energy == 90195765805 + assert hasattr(response, "outdoor_temperature") + assert response.outdoor_temperature == 30 + + body[10] = 253 + response = MessageC3Response(self.header + body) + assert hasattr(response, "outdoor_temperature") + assert response.outdoor_temperature == -3 + + def test_message_silence_response(self) -> None: + """Test message silence response.""" + self.header[-1] = MessageType.query + body = bytearray( + [ + BodyType.X05, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + ], + ) + response = MessageC3Response(self.header + body) + assert hasattr(response, "silence_mode") + assert response.silence_mode is False + assert hasattr(response, "silence_level") + assert response.silence_level == C3SilentLevel.OFF + + body[1] = 0x1 + response = MessageC3Response(self.header + body) + assert hasattr(response, "silence_mode") + assert response.silence_mode is True + assert hasattr(response, "silence_level") + assert response.silence_level == C3SilentLevel.SILENT + + body[1] = 0x8 + response = MessageC3Response(self.header + body) + assert hasattr(response, "silence_mode") + assert response.silence_mode is False + assert hasattr(response, "silence_level") + assert response.silence_level == C3SilentLevel.OFF + + body[1] = 0x9 + response = MessageC3Response(self.header + body) + assert hasattr(response, "silence_mode") + assert response.silence_mode is True + assert hasattr(response, "silence_level") + assert response.silence_level == C3SilentLevel.SUPER_SILENT