Try this quirk, its the one I used. Just copy and paste the code to a file.
I’ve had to split it over 2 posts to get within the character limit.
"""Beca TRV devices support."""
import logging
from typing import Optional, Union
import zigpy.types as t
from zhaquirks import Bus, LocalDataCluster
from zhaquirks.const import (
DEVICE_TYPE,
ENDPOINTS,
INPUT_CLUSTERS,
MODELS_INFO,
OUTPUT_CLUSTERS,
PROFILE_ID,
)
from zhaquirks.tuya import (
TuyaManufClusterAttributes,
TuyaPowerConfigurationCluster,
TuyaThermostat,
TuyaThermostatCluster,
TuyaUserInterfaceCluster,
)
from zigpy.profiles import zha
from zigpy.zcl import foundation
from zigpy.zcl.clusters.general import (
AnalogOutput,
Basic,
BinaryInput,
Groups,
OnOff,
Ota,
Scenes,
Time,
)
from zigpy.zcl.clusters.hvac import Thermostat
_LOGGER = logging.getLogger(__name__)
BECA_TARGET_TEMP_ATTR = 0x0202 # target room temp (degree)
BECA_TEMPERATURE_ATTR = 0x0203 # current room temp (decidegree)
BECA_MODE_ATTR = 0x0401 # [0] schedule [1] manual [2] temporary manual [3] away
BECA_CHILD_LOCK_ATTR = 0x010D # [0] unlocked [1] locked
BECA_TEMP_CALIBRATION_ATTR = 0x0269 # temperature calibration (degree)
BECA_MIN_TEMPERATURE_ATTR = 0x026D # minimum limit of temperature setting (degree)
BECA_MAX_TEMPERATURE_ATTR = 0x026C # maximum limit of temperature setting (degree)
BECA_WINDOW_DETECT_ATTR = 0x0409 # [1] alarm not active [0] alarm active
BECA_WINDOW_DETECT_A2_ATTR = 0x0108 # [0] function disactive [1] function active
BECA_BOOST_TIME_ATTR = 0x0267 # BOOST mode operating time in (sec)
BECA_BOOST_ATTR = 0x0104 # [0] off [1] on
BECA_BOOST_COUNTDOWN_ATTR = 0x0205 # (seconds)
BECA_ECO_TEMP_ATTR = 0x026B # eco mode temperature (degree)
BECA_ECO_MODE_ATTR = 0x016A # [0] off [1] on
BECA_VALVE_STATE_ATTR = 0x0268 # opening percentage
BECA_VALVE_STATE_ONOFF_ATTR = 0x0407 # [0] closed [1] opened
BECA_BATTERY_ATTR = 0x020E # battery percentage remaining 0-100%
BECA_SCHEDULE = 0x0065 # schedule
# [6, 0, 40, 11, 30, 42, 13, 30, 44, 17, 30, 46, 6, 0, 48, 12, 0, 46, 14, 30, 44, 17, 30, 42, 6, 0, 38, 12, 30, 40, 14, 30, 42, 18, 30, 40] for :
# Monday to friday : 06:00 20°C / 11:30 21°C / 13:30 22°C / 17:30 23°C
# Saturday : 06:00 24°C / 12:00 23°C / 14:30 22°C / 17:30 21°C
# Sunday : 06:00 19°C / 12:30 20°C / 14:30 21°C / 18:30 20°C )
BecaManufClusterSelf = {}
class data288(t.FixedList, item_type=t.uint8_t, length=36):
"""General data, Discrete, 288 bit."""
pass
class BecaManufCluster(TuyaManufClusterAttributes):
"""Manufacturer Specific Cluster of thermostatic valves."""
def __init__(self, *args, **kwargs):
"""Init."""
super().__init__(*args, **kwargs)
global BecaManufClusterSelf
BecaManufClusterSelf[self.endpoint.device.ieee] = self
set_time_offset = 1970
attributes = TuyaManufClusterAttributes.attributes.copy()
attributes.update(
{
BECA_TEMPERATURE_ATTR: ("temperature", t.uint32_t),
BECA_TARGET_TEMP_ATTR: ("target_temperature", t.uint32_t),
BECA_MODE_ATTR: ("mode", t.uint8_t),
BECA_CHILD_LOCK_ATTR: ("child_lock", t.uint8_t),
BECA_TEMP_CALIBRATION_ATTR: ("temperature_calibration", t.int32s),
BECA_MIN_TEMPERATURE_ATTR: ("min_temperature", t.uint32_t),
BECA_MAX_TEMPERATURE_ATTR: ("max_temperature", t.uint32_t),
BECA_WINDOW_DETECT_ATTR: ("window_detection", t.uint8_t),
BECA_WINDOW_DETECT_A2_ATTR: ("window_detection_A2_function", t.uint8_t),
BECA_BOOST_TIME_ATTR: ("boost_duration_seconds", t.uint32_t),
BECA_BOOST_ATTR: ("boost_enabled", t.uint8_t),
BECA_BOOST_COUNTDOWN_ATTR: ("boost_duration_seconds", t.uint32_t),
BECA_ECO_TEMP_ATTR: ("eco_mode_temperature", t.uint32_t),
BECA_ECO_MODE_ATTR: ("eco_mode_enabled", t.uint8_t),
BECA_VALVE_STATE_ATTR: ("valve_state", t.uint32_t),
BECA_VALVE_STATE_ONOFF_ATTR: ("valve_state_onoff", t.uint8_t),
BECA_BATTERY_ATTR: ("battery", t.uint32_t),
BECA_SCHEDULE: ("schedule", data288),
}
)
DIRECT_MAPPED_ATTRS = {
BECA_TEMPERATURE_ATTR: ("local_temperature", lambda value: value * 10),
BECA_TARGET_TEMP_ATTR: ("occupied_heating_setpoint", lambda value: value * 100),
BECA_TEMP_CALIBRATION_ATTR: ("local_temperature_calibration", None),
BECA_MIN_TEMPERATURE_ATTR: (
"min_heat_setpoint_limit",
lambda value: value * 100,
),
BECA_MAX_TEMPERATURE_ATTR: (
"max_heat_setpoint_limit",
lambda value: value * 100,
),
}
def _update_attribute(self, attrid, value):
"""Override default _update_attribute."""
super()._update_attribute(attrid, value)
if attrid in self.DIRECT_MAPPED_ATTRS:
self.endpoint.device.thermostat_bus.listener_event(
"temperature_change",
self.DIRECT_MAPPED_ATTRS[attrid][0],
value
if self.DIRECT_MAPPED_ATTRS[attrid][1] is None
else self.DIRECT_MAPPED_ATTRS[attrid][1](value),
)
if attrid == BECA_WINDOW_DETECT_ATTR:
self.endpoint.device.BecaWindowDetection_bus.listener_event(
"set_value", value
)
if attrid == BECA_WINDOW_DETECT_A2_ATTR:
self.endpoint.device.BecaWindowDetection_A2_bus.listener_event(
"switch_change", value
)
elif attrid == BECA_CHILD_LOCK_ATTR:
mode = 1 if value else 0
self.endpoint.device.ui_bus.listener_event("child_lock_change", mode)
self.endpoint.device.BecaChildLock_bus.listener_event("switch_change", mode)
elif attrid in (BECA_MODE_ATTR, BECA_BOOST_ATTR, BECA_ECO_MODE_ATTR):
if attrid == BECA_BOOST_ATTR and value == 1:
self.endpoint.device.thermostat_bus.listener_event("mode_change", 5)
elif attrid == BECA_ECO_MODE_ATTR and value == 1:
self.endpoint.device.thermostat_bus.listener_event("mode_change", 4)
elif attrid == BECA_MODE_ATTR:
self.endpoint.device.thermostat_bus.listener_event("mode_change", value)
elif attrid == BECA_VALVE_STATE_ATTR:
self.endpoint.device.BecaValveState_bus.listener_event("set_value", value)
elif attrid == BECA_TEMP_CALIBRATION_ATTR:
self.endpoint.device.BecaTempCalibration_bus.listener_event(
"set_value", value
)
elif attrid == BECA_BOOST_TIME_ATTR:
self.endpoint.device.BecaBoostTime_bus.listener_event("set_value", value)
elif attrid == BECA_BOOST_COUNTDOWN_ATTR:
self.endpoint.device.BecaBoostCountdown_bus.listener_event(
"set_value", value
)
elif attrid == BECA_ECO_TEMP_ATTR:
self.endpoint.device.BecaEcoTemp_bus.listener_event("set_value", value)
elif attrid == BECA_BATTERY_ATTR:
self.endpoint.device.battery_bus.listener_event("battery_change", value)
elif attrid == BECA_MIN_TEMPERATURE_ATTR:
self.endpoint.device.BecaMinTemp_bus.listener_event("set_value", value)
elif attrid == BECA_MAX_TEMPERATURE_ATTR:
self.endpoint.device.BecaMaxTemp_bus.listener_event("set_value", value)
elif attrid == BECA_SCHEDULE:
self.endpoint.device.thermostat_bus.listener_event("schedule_change", value)
# elif attrid in (BECA_TEMPERATURE_ATTR, BECA_TARGET_TEMP_ATTR):
# self.endpoint.device.thermostat_bus.listener_event(
# "hass_climate_state_change", attrid, value
# )
elif attrid == BECA_VALVE_STATE_ONOFF_ATTR:
self.endpoint.device.thermostat_bus.listener_event(
"state_change", not value
)
class BecaThermostat(TuyaThermostatCluster):
"""Thermostat cluster for thermostatic valves."""
class Preset(t.enum8):
"""Working modes of the thermostat."""
Away = 0x00
Schedule = 0x01
Manual = 0x02
Comfort = 0x03
Eco = 0x04
Boost = 0x05
Complex = 0x06
TempManual = 0x07
class WorkDays(t.enum8):
"""Workday configuration for scheduler operation mode."""
MonToFri = 0x00
MonToSat = 0x01
MonToSun = 0x02
class ForceValveState(t.enum8):
"""Force valve state option."""
Normal = 0x00
Open = 0x01
Close = 0x02
_CONSTANT_ATTRIBUTES = {
0x001B: Thermostat.ControlSequenceOfOperation.Heating_Only,
0x001C: Thermostat.SystemMode.Heat,
}
attributes = TuyaThermostatCluster.attributes.copy()
attributes.update(
{
0x4002: ("operation_preset", Preset),
0x4110: ("schedule_workday_1_hour", t.uint8_t),
0x4111: ("schedule_workday_1_minute", t.uint8_t),
0x4112: ("schedule_workday_1_temperature", t.uint8_t),
0x4113: ("schedule_workday_2_hour", t.uint8_t),
0x4114: ("schedule_workday_2_minute", t.uint8_t),
0x4115: ("schedule_workday_2_temperature", t.uint8_t),
0x4116: ("schedule_workday_3_hour", t.uint8_t),
0x4117: ("schedule_workday_3_minute", t.uint8_t),
0x4118: ("schedule_workday_3_temperature", t.uint8_t),
0x4119: ("schedule_workday_4_hour", t.uint8_t),
0x4120: ("schedule_workday_4_minute", t.uint8_t),
0x4121: ("schedule_workday_4_temperature", t.uint8_t),
0x4122: ("schedule_saturday_1_hour", t.uint8_t),
0x4123: ("schedule_saturday_1_minute", t.uint8_t),
0x4124: ("schedule_saturday_1_temperature", t.uint8_t),
0x4125: ("schedule_saturday_2_hour", t.uint8_t),
0x4126: ("schedule_saturday_2_minute", t.uint8_t),
0x4127: ("schedule_saturday_2_temperature", t.uint8_t),
0x4128: ("schedule_saturday_3_hour", t.uint8_t),
0x4129: ("schedule_saturday_3_minute", t.uint8_t),
0x4130: ("schedule_saturday_3_temperature", t.uint8_t),
0x4131: ("schedule_saturday_4_hour", t.uint8_t),
0x4132: ("schedule_saturday_4_minute", t.uint8_t),
0x4133: ("schedule_saturday_4_temperature", t.uint8_t),
0x4134: ("schedule_sunday_1_hour", t.uint8_t),
0x4135: ("schedule_sunday_1_minute", t.uint8_t),
0x4136: ("schedule_sunday_1_temperature", t.uint8_t),
0x4137: ("schedule_sunday_2_hour", t.uint8_t),
0x4138: ("schedule_sunday_2_minute", t.uint8_t),
0x4139: ("schedule_sunday_2_temperature", t.uint8_t),
0x4140: ("schedule_sunday_3_hour", t.uint8_t),
0x4141: ("schedule_sunday_3_minute", t.uint8_t),
0x4142: ("schedule_sunday_3_temperature", t.uint8_t),
0x4143: ("schedule_sunday_4_hour", t.uint8_t),
0x4144: ("schedule_sunday_4_minute", t.uint8_t),
0x4145: ("schedule_sunday_4_temperature", t.uint8_t),
}
)
DIRECT_MAPPING_ATTRS = {
"min_heat_setpoint_limit": (
BECA_MIN_TEMPERATURE_ATTR,
lambda value: round(value / 100),
),
"max_heat_setpoint_limit": (
BECA_MAX_TEMPERATURE_ATTR,
lambda value: round(value / 100),
),
"local_temperature_calibration": (
BECA_TEMP_CALIBRATION_ATTR,
lambda value: value,
),
}
SCHEDULE_ATTRS = {
"schedule_sunday_4_temperature": 20,
"schedule_sunday_4_minute": 30,
"schedule_sunday_4_hour": 18,
"schedule_sunday_3_temperature": 21,
"schedule_sunday_3_minute": 30,
"schedule_sunday_3_hour": 14,
"schedule_sunday_2_temperature": 20,
"schedule_sunday_2_minute": 30,
"schedule_sunday_2_hour": 12,
"schedule_sunday_1_temperature": 19,
"schedule_sunday_1_minute": 0,
"schedule_sunday_1_hour": 6,
"schedule_saturday_4_temperature": 21,
"schedule_saturday_4_minute": 30,
"schedule_saturday_4_hour": 17,
"schedule_saturday_3_temperature": 22,
"schedule_saturday_3_minute": 30,
"schedule_saturday_3_hour": 14,
"schedule_saturday_2_temperature": 23,
"schedule_saturday_2_minute": 00,
"schedule_saturday_2_hour": 12,
"schedule_saturday_1_temperature": 24,
"schedule_saturday_1_minute": 0,
"schedule_saturday_1_hour": 6,
"schedule_workday_4_temperature": 23,
"schedule_workday_4_minute": 30,
"schedule_workday_4_hour": 17,
"schedule_workday_3_temperature": 22,
"schedule_workday_3_minute": 30,
"schedule_workday_3_hour": 13,
"schedule_workday_2_temperature": 21,
"schedule_workday_2_minute": 30,
"schedule_workday_2_hour": 11,
"schedule_workday_1_temperature": 20,
"schedule_workday_1_minute": 0,
"schedule_workday_1_hour": 6,
}
def map_attribute(self, attribute, value):
"""Map standardized attribute value to dict of manufacturer values."""
if attribute in self.DIRECT_MAPPING_ATTRS:
return {
self.DIRECT_MAPPING_ATTRS[attribute][0]: value
if self.DIRECT_MAPPING_ATTRS[attribute][1] is None
else self.DIRECT_MAPPING_ATTRS[attribute][1](value)
}
if attribute == "occupied_heating_setpoint":
mode = self._attr_cache.get(self.attributes_by_name["operation_preset"].id)
if mode == self.Preset.Schedule:
return {BECA_MODE_ATTR: 2, BECA_TARGET_TEMP_ATTR: value / 100}
else:
return {BECA_TARGET_TEMP_ATTR: value / 100}
if attribute == "operation_preset":
if value == 0:
return {BECA_MODE_ATTR: 3, BECA_BOOST_ATTR: 0, BECA_ECO_MODE_ATTR: 0}
if value == 1:
return {BECA_MODE_ATTR: 0, BECA_BOOST_ATTR: 0, BECA_ECO_MODE_ATTR: 0}
if value == 2:
return {BECA_MODE_ATTR: 1, BECA_BOOST_ATTR: 0, BECA_ECO_MODE_ATTR: 0}
if value == 4:
return {BECA_BOOST_ATTR: 0, BECA_ECO_MODE_ATTR: 1}
if value == 5:
return {BECA_BOOST_ATTR: 1, BECA_ECO_MODE_ATTR: 0}
if value == 7:
return {BECA_MODE_ATTR: 2, BECA_BOOST_ATTR: 0, BECA_ECO_MODE_ATTR: 0}
if attribute in ("programing_oper_mode", "occupancy"):
if attribute == "occupancy":
occupancy = value
oper_mode = self._attr_cache.get(
self.attributes_by_name["programing_oper_mode"].id,
self.ProgrammingOperationMode.Simple,
)
else:
occupancy = self._attr_cache.get(
self.attributes_by_name["occupancy"].id, self.Occupancy.Occupied
)
oper_mode = value
if occupancy == self.Occupancy.Unoccupied:
return {BECA_MODE_ATTR: 3}
if occupancy == self.Occupancy.Occupied:
if oper_mode == self.ProgrammingOperationMode.Schedule_programming_mode:
return {BECA_MODE_ATTR: 0}
if oper_mode == self.ProgrammingOperationMode.Simple:
return {BECA_MODE_ATTR: 1}
self.error("Unsupported value for ProgrammingOperationMode")
else:
self.error("Unsupported value for Occupancy")
if attribute in self.SCHEDULE_ATTRS:
data = data288()
for num, (attr, default) in enumerate(self.SCHEDULE_ATTRS.items()):
if num % 3 == 0:
if attr == attribute:
val = round(value * 2)
else:
val = round(
self._attr_cache.get(
self.attributes_by_name[attr].id, default
)
* 2
)
else:
if attr == attribute:
val = value
else:
val = self._attr_cache.get(
self.attributes_by_name[attr].id, default
)
data.append(val)
return {BECA_SCHEDULE: data}
def hass_climate_state_change(self, attrid, value):
"""Update of the HASS Climate gui state according to temp difference."""
if attrid == BECA_TEMPERATURE_ATTR:
temp_current = value * 10
temp_set = self._attr_cache.get(
self.attributes_by_name["occupied_heating_setpoint"].id
)
else:
temp_set = value * 100
temp_current = self._attr_cache.get(
self.attributes_by_name["local_temperature"].id
)
state = 0 if (int(temp_current) >= int(temp_set)) else 1
self.endpoint.device.thermostat_bus.listener_event("state_change", state)
def mode_change(self, value):
"""System Mode change."""
if value == 1:
operation_preset = self.Preset.Manual
prog_mode = self.ProgrammingOperationMode.Simple
occupancy = self.Occupancy.Occupied
elif value == 2:
operation_preset = self.Preset.TempManual
prog_mode = self.ProgrammingOperationMode.Simple
occupancy = self.Occupancy.Occupied
elif value == 3:
operation_preset = self.Preset.Away
prog_mode = self.ProgrammingOperationMode.Simple
occupancy = self.Occupancy.Unoccupied
elif value == 4:
operation_preset = self.Preset.Eco
prog_mode = self.ProgrammingOperationMode.Economy_mode
occupancy = self.Occupancy.Occupied
elif value == 5:
operation_preset = self.Preset.Boost
prog_mode = self.ProgrammingOperationMode.Simple
occupancy = self.Occupancy.Occupied
else:
operation_preset = self.Preset.Schedule
prog_mode = self.ProgrammingOperationMode.Schedule_programming_mode
occupancy = self.Occupancy.Occupied
self._update_attribute(
self.attributes_by_name["programing_oper_mode"].id, prog_mode
)
self._update_attribute(self.attributes_by_name["occupancy"].id, occupancy)
self._update_attribute(
self.attributes_by_name["operation_preset"].id, operation_preset
)
def schedule_change(self, value):
"""Scheduler attribute change."""
self._update_attribute(
self.attributes_by_name["schedule_workday_1_hour"].id, value[35]
)
self._update_attribute(
self.attributes_by_name["schedule_workday_1_minute"].id, value[34]
)
self._update_attribute(
self.attributes_by_name["schedule_workday_1_temperature"].id, value[33] / 2
)
self._update_attribute(
self.attributes_by_name["schedule_workday_2_hour"].id, value[32]
)
self._update_attribute(
self.attributes_by_name["schedule_workday_2_minute"].id, value[31]
)
self._update_attribute(
self.attributes_by_name["schedule_workday_2_temperature"].id, value[30] / 2
)
self._update_attribute(
self.attributes_by_name["schedule_workday_3_hour"].id, value[29]
)
self._update_attribute(
self.attributes_by_name["schedule_workday_3_minute"].id, value[28]
)
self._update_attribute(
self.attributes_by_name["schedule_workday_3_temperature"].id, value[27] / 2
)
self._update_attribute(
self.attributes_by_name["schedule_workday_4_hour"].id, value[26]
)
self._update_attribute(
self.attributes_by_name["schedule_workday_4_minute"].id, value[25]
)
self._update_attribute(
self.attributes_by_name["schedule_workday_4_temperature"].id, value[24] / 2
)
self._update_attribute(
self.attributes_by_name["schedule_saturday_1_hour"].id, value[23]
)
self._update_attribute(
self.attributes_by_name["schedule_saturday_1_minute"].id, value[22]
)
self._update_attribute(
self.attributes_by_name["schedule_saturday_1_temperature"].id, value[21] / 2
)
self._update_attribute(
self.attributes_by_name["schedule_saturday_2_hour"].id, value[20]
)
self._update_attribute(
self.attributes_by_name["schedule_saturday_2_minute"].id, value[19]
)
self._update_attribute(
self.attributes_by_name["schedule_saturday_2_temperature"].id, value[18] / 2
)
self._update_attribute(
self.attributes_by_name["schedule_saturday_3_hour"].id, value[17]
)
self._update_attribute(
self.attributes_by_name["schedule_saturday_3_minute"].id, value[16]
)
self._update_attribute(
self.attributes_by_name["schedule_saturday_3_temperature"].id, value[15] / 2
)
self._update_attribute(
self.attributes_by_name["schedule_saturday_4_hour"].id, value[14]
)
self._update_attribute(
self.attributes_by_name["schedule_saturday_4_minute"].id, value[13]
)
self._update_attribute(
self.attributes_by_name["schedule_saturday_4_temperature"].id, value[12] / 2
)
self._update_attribute(
self.attributes_by_name["schedule_sunday_1_hour"].id, value[11]
)
self._update_attribute(
self.attributes_by_name["schedule_sunday_1_minute"].id, value[10]
)
self._update_attribute(
self.attributes_by_name["schedule_sunday_1_temperature"].id, value[9] / 2
)
self._update_attribute(
self.attributes_by_name["schedule_sunday_2_hour"].id, value[8]
)
self._update_attribute(
self.attributes_by_name["schedule_sunday_2_minute"].id, value[7]
)
self._update_attribute(
self.attributes_by_name["schedule_sunday_2_temperature"].id, value[6] / 2
)
self._update_attribute(
self.attributes_by_name["schedule_sunday_3_hour"].id, value[5]
)
self._update_attribute(
self.attributes_by_name["schedule_sunday_3_minute"].id, value[4]
)
self._update_attribute(
self.attributes_by_name["schedule_sunday_3_temperature"].id, value[3] / 2
)
self._update_attribute(
self.attributes_by_name["schedule_sunday_4_hour"].id, value[2]
)
self._update_attribute(
self.attributes_by_name["schedule_sunday_4_minute"].id, value[1]
)
self._update_attribute(
self.attributes_by_name["schedule_sunday_4_temperature"].id, value[0] / 2
)
class BecaUserInterface(TuyaUserInterfaceCluster):
"""HVAC User interface cluster for tuya electric heating thermostats."""
_CHILD_LOCK_ATTR = BECA_CHILD_LOCK_ATTR
class BecaWindowDetection(LocalDataCluster, BinaryInput):
"""On/Off cluster for the window detect function of the electric heating thermostats."""
def __init__(self, *args, **kwargs):
"""Init."""
super().__init__(*args, **kwargs)
self.endpoint.device.BecaWindowDetection_bus.add_listener(self)
def set_value(self, value):
"""Set value."""
self._update_attribute(self.attributes_by_name["present_value"].id, value)
