from enum import IntEnum
from ...core.message import (
MessageType,
MessageRequest,
MessageResponse,
MessageBody,
NewProtocolMessageBody
)
from ...core.crc8 import calculate
BB_AC_MODES = [0, 3, 1, 2, 4, 5]
class NewProtocolTags(IntEnum):
indoor_humidity = 0x0015
screen_display = 0x0017
breezeless = 0x0018
prompt_tone = 0x001A
indirect_wind = 0x0042
fresh_air_1 = 0x0233
fresh_air_2 = 0x004b
class MessageACBase(MessageRequest):
_message_serial = 0
def __init__(self, device_protocol_version, message_type, body_type):
super().__init__(
device_protocol_version=device_protocol_version,
device_type=0xAC,
message_type=message_type,
body_type=body_type
)
MessageACBase._message_serial += 1
if MessageACBase._message_serial >= 254:
MessageACBase._message_serial = 1
self._message_id = MessageACBase._message_serial
@property
def _body(self):
raise NotImplementedError
@property
def body(self):
body = bytearray([self.body_type]) + self._body + bytearray([self._message_id])
body.append(calculate(body))
return body
class MessageQuery(MessageACBase):
def __init__(self, device_protocol_version):
super().__init__(
device_protocol_version=device_protocol_version,
message_type=MessageType.query,
body_type=0x41)
@property
def _body(self):
return bytearray([
0x81, 0x00, 0xFF, 0x03,
0xFF, 0x00, 0x02,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00
])
class MessagePowerQuery(MessageACBase):
def __init__(self, device_protocol_version):
super().__init__(
device_protocol_version=device_protocol_version,
message_type=MessageType.query,
body_type=0x41)
@property
def _body(self):
return bytearray([
0x21, 0x01, 0x44, 0x00, 0x01
])
@property
def body(self):
body = bytearray([self.body_type]) + self._body
body.append(calculate(body))
return body
class MessageSwitchDisplay(MessageACBase):
def __init__(self, device_protocol_version):
super().__init__(
device_protocol_version=device_protocol_version,
message_type=MessageType.query,
body_type=0x41)
@property
def _body(self):
return bytearray([
0x81, 0x00, 0xFF, 0x02,
0xFF, 0x02, 0x02,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00
])
class MessageNewProtocolQuery(MessageACBase):
def __init__(self, device_protocol_version):
super().__init__(
device_protocol_version=device_protocol_version,
message_type=MessageType.query,
body_type=0xB1)
@property
def _body(self):
query_params = [
NewProtocolTags.indirect_wind,
NewProtocolTags.breezeless,
NewProtocolTags.indoor_humidity,
NewProtocolTags.screen_display,
NewProtocolTags.fresh_air_1,
NewProtocolTags.fresh_air_2
]
_body = bytearray([len(query_params)])
for param in query_params:
_body.extend([param & 0xFF, param >> 8])
return _body
class MessageSubProtocol(MessageACBase):
def __init__(self, device_protocol_version, message_type, subprotocol_query_type):
super().__init__(
device_protocol_version=device_protocol_version,
message_type=message_type,
body_type=0xAA)
self._subprotocol_query_type = subprotocol_query_type
@property
def _subprotocol_body(self):
return bytes([])
@property
def body(self):
body = bytearray([self.body_type]) + self._body
body.append(calculate(body))
body.append(self.checksum(body))
return body
@property
def _body(self):
_subprotocol_body = self._subprotocol_body
_body = bytearray([
6 + 2 + (len(_subprotocol_body) if _subprotocol_body is not None else 0),
0x00, 0xFF, 0xFF, self._subprotocol_query_type
])
if _subprotocol_body is not None:
_body.extend(_subprotocol_body)
return _body
class MessageSubProtocolQuery(MessageSubProtocol):
def __init__(self, device_protocol_version, subprotocol_query_type):
super().__init__(
device_protocol_version=device_protocol_version,
message_type=MessageType.query,
subprotocol_query_type=subprotocol_query_type)
class MessageSubProtocolSet(MessageSubProtocol):
def __init__(self, device_protocol_version):
super().__init__(
device_protocol_version=device_protocol_version,
message_type=MessageType.set,
subprotocol_query_type=0x20)
self.power = False
self.mode = 0
self.target_temperature = 20.0
self.fan_speed = 102
self.boost_mode = False
self.aux_heating = False
self.dry = False
self.eco_mode = False
self.sleep_mode = False
self.sn8_flag = False
self.timer = False
self.prompt_tone = False
@property
def _subprotocol_body(self):
power = 0x01 if self.power else 0
dry = 0x10 if self.power and self.dry else 0
boost_mode = 0x20 if self.boost_mode else 0
aux_heating = 0x40 if self.aux_heating else 0x80
sleep_mode = 0x80 if self.sleep_mode else 0
try:
mode = 0 if self.mode == 0 else BB_AC_MODES[self.mode] - 1
except IndexError:
mode = 2 # set Auto if invalid mode
target_temperature = int(self.target_temperature * 2 + 30)
water_model_temperature_set = int((self.target_temperature - 1) * 2 + 50)
fan_speed = self.fan_speed
eco = 0x40 if self.eco_mode else 0
prompt_tone = 0x01 if self.prompt_tone else 0
timer = 0x04 if (self.sn8_flag and self.timer) else 0
return bytearray([
boost_mode | power | dry, aux_heating, sleep_mode, 0x00,
0x00, mode, target_temperature, fan_speed,
0x32, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x01,
0x01, 0x00, 0x01, water_model_temperature_set,
prompt_tone, target_temperature, 0x32, 0x66,
0x00, eco | timer, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
0x08
])
class MessageGeneralSet(MessageACBase):
def __init__(self, device_protocol_version):
super().__init__(
device_protocol_version=device_protocol_version,
message_type=MessageType.set,
body_type=0x40)
self.power = False
self.prompt_tone = True
self.mode = 0
self.target_temperature = 20.0
self.fan_speed = 102
self.swing_vertical = False
self.swing_horizontal = False
self.boost_mode = False
self.smart_eye = False
self.dry = False
self.aux_heating = False
self.eco_mode = False
self.temp_fahrenheit = False
self.sleep_mode = False
self.natural_wind = False
self.frost_protect = False
self.comfort_mode = False
@property
def _body(self):
# Byte1, Power, prompt_tone
power = 0x01 if self.power else 0
prompt_tone = 0x40 if self.prompt_tone else 0
# Byte2, mode target_temperature
mode = (self.mode << 5) & 0xe0
target_temperature = (int(self.target_temperature) & 0xf) | \
(0x10 if int(round(self.target_temperature * 2)) % 2 != 0 else 0)
# Byte 3, fan_speed
fan_speed = self.fan_speed & 0x7f
# Byte 7, swing_mode
swing_mode = 0x30 | \
(0x0c if self.swing_vertical else 0) | \
(0x03 if self.swing_horizontal else 0)
# Byte 8, turbo
boost_mode = 0x20 if self.boost_mode else 0
# Byte 9 aux_heating eco_mode
smart_eye = 0x01 if self.smart_eye else 0
dry = 0x04 if self.dry else 0
aux_heating = 0x08 if self.aux_heating else 0
eco_mode = 0x80 if self.eco_mode else 0
# Byte 10 temp_fahrenheit
temp_fahrenheit = 0x04 if self.temp_fahrenheit else 0
sleep_mode = 0x01 if self.sleep_mode else 0
boost_mode_1 = 0x02 if self.boost_mode else 0
# Byte 17 natural_wind
natural_wind = 0x40 if self.natural_wind else 0
# Byte 21 frost_protect
frost_protect = 0x80 if self.frost_protect else 0
# Byte 22 comfort_mode
comfort_mode = 0x01 if self.comfort_mode else 0
return bytearray([
power | prompt_tone,
mode | target_temperature,
fan_speed,
0x00, 0x00, 0x00,
swing_mode,
boost_mode,
smart_eye | dry | aux_heating | eco_mode,
temp_fahrenheit | sleep_mode | boost_mode_1,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00,
natural_wind,
0x00, 0x00, 0x00,
frost_protect,
comfort_mode
])
class MessageNewProtocolSet(MessageACBase):
def __init__(self, device_protocol_version):
super().__init__(
device_protocol_version=device_protocol_version,
message_type=MessageType.set,
body_type=0xB0)
self.indirect_wind = None
self.prompt_tone = None
self.breezeless = None
self.screen_display = None
self.fresh_air_1 = None
self.fresh_air_2 = None
@property
def _body(self):
pack_count = 0
payload = bytearray([0x00])
if self.breezeless is not None:
pack_count += 1
payload.extend(
NewProtocolMessageBody.pack(
param=NewProtocolTags.breezeless,
value=bytearray([0x01 if self.breezeless else 0x00])
))
if self.indirect_wind is not None:
pack_count += 1
payload.extend(
NewProtocolMessageBody.pack(
param=NewProtocolTags.indirect_wind,
value=bytearray([0x02 if self.indirect_wind else 0x01])
))
if self.prompt_tone is not None:
pack_count += 1
payload.extend(
NewProtocolMessageBody.pack(
param=NewProtocolTags.prompt_tone,
value=bytearray([0x01 if self.prompt_tone else 0x00])
))
if self.screen_display is not None:
pack_count += 1
payload.extend(
NewProtocolMessageBody.pack(
param=NewProtocolTags.screen_display,
value=bytearray([0x64 if self.screen_display else 0x00])
))
if self.fresh_air_1 is not None and len(self.fresh_air_1) == 2:
pack_count += 1
fresh_air_power = 2 if self.fresh_air_1[0] > 0 else 1
fresh_air_fan_speed = self.fresh_air_1[1]
payload.extend(
NewProtocolMessageBody.pack(
param=NewProtocolTags.fresh_air_1,
value=bytearray([
fresh_air_power,
fresh_air_fan_speed,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00
])
))
if self.fresh_air_2 is not None and len(self.fresh_air_2) == 2:
pack_count += 1
fresh_air_power = 1 if self.fresh_air_2[0] > 0 else 0
fresh_air_fan_speed = self.fresh_air_2[1]
payload.extend(
NewProtocolMessageBody.pack(
param=NewProtocolTags.fresh_air_2,
value=bytearray([
fresh_air_power,
fresh_air_fan_speed,
0xFF
])
))
payload[0] = pack_count
return payload
class XA0MessageBody(MessageBody):
def __init__(self, body):
super().__init__(body)
self.power = (body[1] & 0x1) > 0
self.target_temperature = ((body[1] & 0x3E) >> 1) - 4 + 16.0 + (0.5 if body[1] & 0x40 > 0 else 0.0)
self.mode = (body[2] & 0xe0) >> 5
self.fan_speed = body[3] & 0x7f
self.swing_vertical = (body[7] & 0xC) > 0
self.swing_horizontal = (body[7] & 0x3) > 0
self.boost_mode = ((body[8] & 0x20) > 0) or ((body[10] & 0x2) > 0)
self.smart_eye = (body[9] & 0x01) > 0
self.dry = (body[9] & 0x04) > 0
self.aux_heating = (body[9] & 0x08) > 0
self.eco_mode = (body[9] & 0x10) > 0
self.sleep_mode = (body[10] & 0x01) > 0
self.natural_wind = (body[10] & 0x40) > 0
self.full_dust = (body[13] & 0x20) > 0
self.comfort_mode = (body[14] & 0x1) > 0 if len(body) > 16 else False
class XA1MessageBody(MessageBody):
def __init__(self, body):
super().__init__(body)
if body[13] != 0xFF:
temp_integer = int((body[13] - 50) / 2)
temp_decimal = ((body[18] & 0xF) * 0.1) if len(body) > 20 else 0
if body[13] > 49:
self.indoor_temperature = temp_integer + temp_decimal
else:
self.indoor_temperature = temp_integer - temp_decimal
if body[14] == 0xFF:
self.outdoor_temperature = None
else:
temp_integer = int((body[14] - 50) / 2)
temp_decimal = (((body[18] & 0xF0) >> 4) * 0.1) if len(body) > 20 else 0
if body[14] > 49:
self.outdoor_temperature = temp_integer + temp_decimal
else:
self.outdoor_temperature = temp_integer - temp_decimal
self.indoor_humidity = body[17]
class XBXMessageBody(NewProtocolMessageBody):
def __init__(self, body, bt):
super().__init__(body, bt)
params = self.parse()
if NewProtocolTags.indirect_wind in params:
self.indirect_wind = (params[NewProtocolTags.indirect_wind][0] == 0x02)
if NewProtocolTags.indoor_humidity in params:
self.indoor_humidity = params[NewProtocolTags.indoor_humidity][0]
if NewProtocolTags.breezeless in params:
self.breezeless = (params[NewProtocolTags.breezeless][0] == 1)
if NewProtocolTags.screen_display in params:
self.screen_display = (params[NewProtocolTags.screen_display][0] > 0)
self.screen_display_new = True
if NewProtocolTags.fresh_air_1 in params:
self.fresh_air_1 = True
data = params[NewProtocolTags.fresh_air_1]
self.fresh_air_power = data[0] == 0x02
self.fresh_air_fan_speed = data[1]
if NewProtocolTags.fresh_air_2 in params:
self.fresh_air_2 = True
data = params[NewProtocolTags.fresh_air_2]
self.fresh_air_power = data[0] > 0
self.fresh_air_fan_speed = data[1]
class XC0MessageBody(MessageBody):
def __init__(self, body):
super().__init__(body)
self.power = (body[1] & 0x1) > 0
self.mode = (body[2] & 0xe0) >> 5
self.target_temperature = (body[2] & 0x0F) + 16.0 + (0.5 if body[0x02] & 0x10 > 0 else 0.0)
self.fan_speed = body[3] & 0x7F
self.swing_vertical = (body[7] & 0x0C) > 0
self.swing_horizontal = (body[7] & 0x03) > 0
self.boost_mode = ((body[8] & 0x20) > 0) or ((body[10] & 0x2) > 0)
self.smart_eye = (body[8] & 0x40) > 0
self.natural_wind = (body[9] & 0x2) > 0
self.dry = (body[9] & 0x4) > 0
self.eco_mode = (body[9] & 0x10) > 0
self.aux_heating = (body[9] & 0x08) > 0
self.temp_fahrenheit = (body[10] & 0x04) > 0
self.sleep_mode = (body[10] & 0x01) > 0
if body[11] != 0xFF:
temp_integer = int((body[11] - 50) / 2)
temp_decimal = (body[15] & 0x0F) * 0.1
if body[11] > 49:
self.indoor_temperature = temp_integer + temp_decimal
else:
self.indoor_temperature = temp_integer - temp_decimal
if body[12] == 0xFF:
self.outdoor_temperature = None
else:
temp_integer = int((body[12] - 50) / 2)
temp_decimal = ((body[15] & 0xF0) >> 4) * 0.1
if body[12] > 49:
self.outdoor_temperature = temp_integer + temp_decimal
else:
self.outdoor_temperature = temp_integer - temp_decimal
self.full_dust = (body[13] & 0x20) > 0
self.screen_display = ((body[14] >> 4 & 0x7) != 0x07) and self.power
self.frost_protect = (body[21] & 0x80) > 0 if len(body) > 23 else False
self.comfort_mode = (body[22] & 0x1) > 0 if len(body) > 24 else False
class XC1MessageBody(MessageBody):
def __init__(self, body, analysis_method=3):
super().__init__(body)
if body[3] == 0x44:
self.total_energy_consumption = XC1MessageBody.parse_consumption(
analysis_method,
body[4], body[5], body[6], body[7]
)
self.current_energy_consumption = XC1MessageBody.parse_consumption(
analysis_method,
body[12], body[13], body[14], body[15]
)
self.realtime_power = XC1MessageBody.parse_power(
analysis_method,
body[16], body[17], body[18]
)
elif body[3] == 0x40:
pass
@staticmethod
def parse_value(byte):
return (byte >> 4) * 10 + (byte & 0x0F)
@staticmethod
def parse_power(analysis_method, byte1, byte2, byte3):
if analysis_method == 1:
return float(XC1MessageBody.parse_value(byte1) * 10000 +
XC1MessageBody.parse_value(byte2) * 100 +
XC1MessageBody.parse_value(byte3)) / 10
elif analysis_method == 2:
return float((byte1 << 16) + (byte2 << 8) + byte3) / 10
else:
return float(byte1 * 10000 + byte2 * 100 + byte3) / 10
@staticmethod
def parse_consumption(analysis_method, byte1, byte2, byte3, byte4):
if analysis_method == 1:
return float(XC1MessageBody.parse_value(byte1) * 1000000 +
XC1MessageBody.parse_value(byte2) * 10000 +
XC1MessageBody.parse_value(byte3) * 100 +
XC1MessageBody.parse_value(byte4)) / 100
elif analysis_method == 2:
return float((byte1 << 32) + (byte2 << 16) + (byte3 << 8) + byte4) / 10
else:
return float(byte1 * 1000000 + byte2 * 10000 + byte3 * 100 + byte4) / 100
class XBBMessageBody(MessageBody):
def __init__(self, body):
super().__init__(body)
subprotocol_head = body[:6]
subprotocol_body = body[6:]
data_type = subprotocol_head[-1]
subprotocol_body_len = len(subprotocol_body)
if data_type == 0x20 or data_type == 0x11:
self.power = (subprotocol_body[0] & 0x1) > 0
self.dry = (subprotocol_body[0] & 0x10) > 0
self.boost_mode = (subprotocol_body[0] & 0x20) > 0
self.aux_heating = (subprotocol_body[1] & 0x40) > 0
self.sleep_mode = (subprotocol_body[2] & 0x80) > 0
try:
self.mode = BB_AC_MODES.index(subprotocol_body[5] + 1)
except ValueError:
self.mode = 0
self.target_temperature = (subprotocol_body[6] - 30) / 2
self.fan_speed = subprotocol_body[7]
self.timer = (subprotocol_body[25] & 0x04) > 0 if subprotocol_body_len > 27 else False
self.eco_mode = (subprotocol_body[25] & 0x40) > 0 if subprotocol_body_len > 27 else False
elif data_type == 0x10:
if subprotocol_body[8] & 0x80 == 0x80:
self.indoor_temperature = (0 - (~(subprotocol_body[7] + subprotocol_body[8] * 256) + 1) & 0xffff) / 100
else:
self.indoor_temperature = (subprotocol_body[7] + subprotocol_body[8] * 256) / 100
self.indoor_humidity = subprotocol_body[30]
self.sn8_flag = subprotocol_body[80] == 0x31
elif data_type == 0x12:
pass
elif data_type == 0x30:
if subprotocol_body[6] & 0x80 == 0x80:
self.outdoor_temperature = (0 - (~(subprotocol_body[5] + subprotocol_body[6] * 256) + 1) & 0xffff) / 100
else:
self.outdoor_temperature = (subprotocol_body[5] + subprotocol_body[6] * 256) / 100
elif data_type == 0x13 or data_type == 0x21:
pass
class MessageACResponse(MessageResponse):
def __init__(self, message, power_analysis_method=3):
super().__init__(message)
if self.message_type == MessageType.notify2 and self.body_type == 0xA0:
self.set_body(XA0MessageBody(super().body))
elif self.message_type == MessageType.notify1 and self.body_type == 0xA1:
self.set_body(XA1MessageBody(super().body))
elif self.message_type in [MessageType.query, MessageType.set, MessageType.notify2] and \
self.body_type in [0xB0, 0xB1, 0xB5]:
self.set_body(XBXMessageBody(super().body, self.body_type))
elif self.message_type in [MessageType.query, MessageType.set] and self.body_type == 0xC0:
self.set_body(XC0MessageBody(super().body))
elif self.message_type == MessageType.query and self.body_type == 0xC1:
self.set_body(XC1MessageBody(super().body, power_analysis_method))
elif self.message_type in [MessageType.set, MessageType.query, MessageType.notify2] and \
self.body_type == 0xBB and len(super().body) >= 21:
self.used_subprotocol = True
self.set_body(XBBMessageBody(super().body))
self.set_attr()