zha-device-handlers
zha-device-handlers copied to clipboard
Published
20 hours ago •
zigpy
zigpy
Error with custom quirk (4-pipe thermostat heat/cool Tuya _TZE204_mpbki2zm)
Bug description
I created a custom quirk for the BAC-006 series Zigbee thermostat "TZE204_mpbki2zm", which is a 4-pipe thermostat designed for both cooling and heating. Additionally, it features fan coil speed control with options for auto, low, medium, and high speeds.
While the device with the quirk works to some extent, it has a few bugs that render it ineffective.
1.Cooling Mode Error: When the thermostat is set to cool mode and the set point is lower than the current temperature, the state incorrectly shows "heating" instead of "cooling".
2.Heat Mode Configuration Issue: If you configure the device to heat mode via the touch screen, then turn it off and back on, the state in Home Assistant incorrectly switches to cool mode, while the physical device remains in heat mode.
The quirk:
"""Map from manufacturer to standard clusters for electric heating thermostats."""
"""Tuya MCU based thermostat."""
import logging
import asyncio
from typing import Optional, Union
from zigpy.profiles import zha
import zigpy.types as t
from zigpy.zcl.clusters.general import Basic, Groups, Ota, Scenes, Time, GreenPowerProxy
from zigpy.zcl.clusters.hvac import Thermostat, Fan
from zigpy.quirks import CustomDevice
from typing import Dict, Optional, Union
from zigpy.zcl import foundation
from zigpy.zcl.clusters.hvac import Thermostat
from zhaquirks import LocalDataCluster, Bus
from zhaquirks.const import (
DEVICE_TYPE,
ENDPOINTS,
INPUT_CLUSTERS,
MODELS_INFO,
OUTPUT_CLUSTERS,
PROFILE_ID,
)
from zhaquirks.tuya import (
TuyaManufClusterAttributes,
TuyaThermostat,
TuyaThermostatCluster,
TuyaUserInterfaceCluster,
NoManufacturerCluster,
TUYA_MCU_COMMAND,
TuyaLocalCluster,
)
from zhaquirks.tuya.mcu import (
DPToAttributeMapping,
TuyaClusterData,
TuyaMCUCluster,
)
class TuyaTC(t.enum8):
"""Tuya thermostat commands."""
OFF = 0x00
ON = 0x01
class ZclTC(t.enum8):
"""ZCL thermostat commands."""
OFF = 0x00
ON = 0x01
TUYA2ZB_COMMANDS = {
ZclTC.OFF: TuyaTC.OFF,
ZclTC.ON: TuyaTC.ON,
}
# MOESBHT6_TARGET_TEMP_ATTR = 0x0210 # [0,0,0,21] target room temp (degree)
# MOESBHT6_TEMPERATURE_ATTR = 0x0218 # [0,0,0,200] current room temp (decidegree)
# MOESBHT6_ENABLED_ATTR = 0x0101 # [0] off [1] on
# MOESBHT6_MODE_ATTR = 0x0402 # [0] manual [1] scheduled
# MOESBHT6_CHILD_LOCK_ATTR = 0x0128 # [0] unlocked [1] child-locked
# MOESBHT6_RUNNING_MODE_ATTR = 0x0424 # 1[] idle [0] heating
# MOESBHT6_RUNNING_STATE_ATTR = 0x0424 # [1] idle [0] heating
TUYA_FANCOIL_TARGET_TEMP_ATTR = 0x0210 # [0,0,0,21] target room temp (degree)
TUYA_FANCOIL_TEMPERATURE_ATTR = 0x0218 # [0,0,0,200] current room temp (decidegree)
TUYA_FANCOIL_ENABLED_ATTR = 0x0101 # [0] off [1] on
TUYA_FANCOIL_RUNNING_MODE_ATTR = 0x0402 # [0] cooling [1] heating [2] off
TUYA_FANCOIL_RUNNING_STATE_ATTR = 0x0424 # [0] cooling [1] heating [2] off
TUYA_FANCOIL_CHILD_LOCK_ATTR = 0x0128 # [0] unlocked [1] child-locked
TUYA_FANCOIL_FAN_MODE_ATTR = 0x041c# [0] manual [1] scheduled
_LOGGER = logging.getLogger(__name__)
class MoesBHT6ManufCluster(TuyaManufClusterAttributes, NoManufacturerCluster, TuyaLocalCluster):
"""Manufacturer Specific Cluster of some electric heating thermostats."""
attributes = {
TUYA_FANCOIL_TARGET_TEMP_ATTR: ("target_temperature", t.uint32_t, True),
TUYA_FANCOIL_TEMPERATURE_ATTR: ("temperature", t.uint32_t, True),
TUYA_FANCOIL_ENABLED_ATTR: ("enabled", t.uint8_t, True),
TUYA_FANCOIL_RUNNING_MODE_ATTR: ("running_mode", t.uint8_t, True),
TUYA_FANCOIL_RUNNING_STATE_ATTR: ("running_state", t.uint8_t, True),
TUYA_FANCOIL_CHILD_LOCK_ATTR: ("child_lock", t.uint8_t, True),
TUYA_FANCOIL_FAN_MODE_ATTR: ("fan_mode", t.uint8_t, True),
}
async def on_enabled_change(self, value):
_LOGGER.debug("Reading running_mode from device")
running_mode_id = self.attributes_by_name["running_state"].id
success, _ = await self.read_attributes((running_mode_id,), manufacturer=None)
current_running_mode = success[running_mode_id]
_LOGGER.debug("Current running_mode from device: " + str(current_running_mode))
self.endpoint.device.thermostat_bus.listener_event("enabled_change", value, current_running_mode)
async def on_running_change(self, value):
_LOGGER.debug("Reading running_mode from device")
running_mode_id = self.attributes_by_name["running_state"].id
success, _ = await self.read_attributes((running_mode_id,), manufacturer=None)
current_running_mode = success[running_mode_id]
_LOGGER.debug("Current running_mode from device: " + str(current_running_mode))
self.endpoint.device.thermostat_bus.listener_event("running_change", value, current_running_mode)
def _update_attribute(self, attrid, value):
super()._update_attribute(attrid, value)
if attrid == TUYA_FANCOIL_TARGET_TEMP_ATTR:
self.endpoint.device.thermostat_bus.listener_event(
"temperature_change",
"occupied_heating_setpoint",
value * 10, # degree to centidegree
)
self.endpoint.device.thermostat_bus.listener_event(
"temperature_change",
"occupied_cooling_setpoint",
value * 10, # degree to centidegree
)
elif attrid == TUYA_FANCOIL_TEMPERATURE_ATTR:
self.endpoint.device.thermostat_bus.listener_event(
"temperature_change",
"local_temperature",
value * 10, # decidegree to centidegree
)
elif attrid == TUYA_FANCOIL_FAN_MODE_ATTR:
if value == 0: # manual
self.endpoint.device.thermostat_bus.listener_event(
"program_change", "manual"
)
elif value == 1: # scheduled
self.endpoint.device.thermostat_bus.listener_event(
"program_change", "scheduled"
)
elif attrid == TUYA_FANCOIL_ENABLED_ATTR:
"""Needs to get running_mode async in order to set the correct state when device is enabled"""
asyncio.create_task(self.on_enabled_change(value))
self.endpoint.device.thermostat_bus.listener_event("enabled_change", value)
elif attrid == TUYA_FANCOIL_RUNNING_MODE_ATTR:
self.endpoint.device.thermostat_bus.listener_event("state_change", value)
elif attrid == TUYA_FANCOIL_RUNNING_STATE_ATTR:
asyncio.create_task(self.on_running_change(value))
self.endpoint.device.thermostat_bus.listener_event("running_change", value)
elif attrid == TUYA_FANCOIL_CHILD_LOCK_ATTR:
self.endpoint.device.ui_bus.listener_event("child_lock_change", value)
elif attrid == TUYA_FANCOIL_FAN_MODE_ATTR:
self.endpoint.device.fan_bus.listener_event("fan_mode_change", value)
class TuyaFancoilFanCluster(LocalDataCluster, Fan):
_CONSTANT_ATTRIBUTES = {0x0001: Fan.FanModeSequence.Low_Med_High_Auto}
def __init__(self, *args, **kwargs):
"""Init."""
super().__init__(*args, **kwargs)
self.endpoint.device.fan_bus.add_listener(self)
def fan_mode_change(self, value):
if value == 0:
self._update_attribute(self.attributes_by_name["fan_mode"].id, self.FanMode.Low)
elif value == 1:
self._update_attribute(self.attributes_by_name["fan_mode"].id, self.FanMode.Medium)
elif value == 2:
self._update_attribute(self.attributes_by_name["fan_mode"].id, self.FanMode.High)
elif value == 3:
self._update_attribute(self.attributes_by_name["fan_mode"].id, self.FanMode.Auto)
class TuyaFancoilThermostatCluster(TuyaThermostatCluster):
"""Thermostat cluster for Tuya fancoil thermostats."""
_CONSTANT_ATTRIBUTES = {
0x001B: Thermostat.ControlSequenceOfOperation.Cooling_and_Heating,
}
def state_change(self, value):
"""State update from device."""
if value == 0:
mode = self.RunningMode.Cool
state = self.RunningState.Cool_State_On
system_mode = self.SystemMode.Cool
elif value == 1:
mode = self.RunningMode.Heat
state = self.RunningState.Heat_State_On
system_mode = self.SystemMode.Heat
elif value == 2:
mode = self.RunningMode.Off
state = self.RunningState.Idle
system_mode = self.SystemMode.Fan_only
if mode is not None:
self._update_attribute(self.attributes_by_name["running_mode"].id, mode)
if state is not None:
self._update_attribute(self.attributes_by_name["running_state"].id, state)
if system_mode is not None:
self._update_attribute(self.attributes_by_name["system_mode"].id, system_mode)
# pylint: disable=W0236
async def command(
self,
command_id: Union[foundation.GeneralCommand, int, t.uint8_t],
*args,
manufacturer: Optional[Union[int, t.uint16_t]] = None,
expect_reply: bool = True,
tsn: Optional[Union[int, t.uint8_t]] = None,
):
"""Implement thermostat commands."""
if command_id != 0x0000:
return foundation.GENERAL_COMMANDS[
foundation.GeneralCommand.Default_Response
].schema(
command_id=command_id, status=foundation.Status.UNSUP_CLUSTER_COMMAND
)
mode, offset = args
if mode not in (self.SetpointMode.Heat, self.SetpointMode.Cool, self.SetpointMode.Both):
return foundation.GENERAL_COMMANDS[
foundation.GeneralCommand.Default_Response
].schema(command_id=command_id, status=foundation.Status.INVALID_VALUE)
heating_attrid = self.attributes_by_name["occupied_heating_setpoint"].id
cooling_attrid = self.attributes_by_name["occupied_cooling_setpoint"].id
success, _ = await self.read_attributes((heating_attrid, cooling_attrid), manufacturer=manufacturer)
try:
current_heat = success[heating_attrid]
current_cool = success[cooling_attrid]
except KeyError:
return foundation.Status.FAILURE
# offset is given in decidegrees, see Zigbee cluster specification
(res,) = await self.write_attributes(
{"occupied_heating_setpoint": current_heat + offset * 10,
"occupied_cooling_setpoint": current_cool + offset * 10},
manufacturer=manufacturer,
)
return foundation.GENERAL_COMMANDS[
foundation.GeneralCommand.Default_Response
].schema(command_id=command_id, status=res[0].status)
class TuyaFancoilThermostat(TuyaFancoilThermostatCluster):
"""Thermostat cluster for Tuya fancoil."""
def map_attribute(self, attribute, value):
_LOGGER.info("Metodo: map_attribute "+str(attribute))
"""Map standardized attribute value to dict of manufacturer values."""
if attribute == "occupied_heating_setpoint" or attribute == "occupied_cooling_setpoint":
# centidegree to degree
return {TUYA_FANCOIL_TARGET_TEMP_ATTR: round(value / 10)}
if attribute == "system_mode":
if value == self.SystemMode.Off:
return {TUYA_FANCOIL_ENABLED_ATTR: 0}
if value == self.SystemMode.Cool:
return {TUYA_FANCOIL_ENABLED_ATTR: 1, TUYA_FANCOIL_RUNNING_MODE_ATTR: 0}
if value == self.SystemMode.Heat:
return {TUYA_FANCOIL_ENABLED_ATTR: 1, TUYA_FANCOIL_RUNNING_MODE_ATTR: 1}
self.error("Unsupported value for SystemMode " + str(value))
elif attribute == "programing_oper_mode":
if value == self.ProgrammingOperationMode.Simple:
return {TUYA_FANCOIL_FAN_MODE_ATTR: 0}
if value == self.ProgrammingOperationMode.Schedule_programming_mode:
return {TUYA_FANCOIL_FAN_MODE_ATTR: 1}
self.error("Unsupported value for ProgrammingOperationMode")
elif attribute == "running_state":
if value == self.RunningState.Idle:
return {TUYA_FANCOIL_RUNNING_STATE_ATTR: 1}
if value == self.RunningState.Heat_State_On:
return {TUYA_FANCOIL_RUNNING_STATE_ATTR: 1}
if value == self.RunningState.Cool_State_On:
return {TUYA_FANCOIL_RUNNING_STATE_ATTR: 1}
self.error("Unsupported value for RunningState")
elif attribute == "running_mode":
if value == self.RunningMode.Off:
return {TUYA_FANCOIL_ENABLED_ATTR: 0, TUYA_FANCOIL_RUNNING_MODE_ATTR: 1}
if value == self.SystemMode.Cool:
return {TUYA_FANCOIL_ENABLED_ATTR: 1, TUYA_FANCOIL_RUNNING_MODE_ATTR: 0}
if value == self.RunningMode.Heat:
return {TUYA_FANCOIL_ENABLED_ATTR: 1, TUYA_FANCOIL_RUNNING_MODE_ATTR: 1}
self.error("Unsupported value for RunningMode")
return super().map_attribute(attribute, value)
def program_change(self, mode):
_LOGGER.info("Metodo: program_change "+str(mode))
"""Programming mode change."""
if mode == "manual":
value = self.ProgrammingOperationMode.Simple
else:
value = self.ProgrammingOperationMode.Schedule_programming_mode
self._update_attribute(self.attributes_by_name["programing_oper_mode"].id, value)
def enabled_change(self, value, running_state):
"""System mode change."""
if value == 0:
self._update_attribute(self.attributes_by_name["system_mode"].id, self.SystemMode.Off)
self._update_attribute(self.attributes_by_name["running_mode"].id, self.RunningMode.Off)
self._update_attribute(self.attributes_by_name["running_state"].id, self.RunningState.Idle)
else:
if running_state == 0:
self._update_attribute(self.attributes_by_name["system_mode"].id, self.SystemMode.Cool)
self._update_attribute(self.attributes_by_name["running_mode"].id, self.RunningMode.Cool)
self._update_attribute(self.attributes_by_name["running_state"].id, self.RunningState.Cool_State_On)
elif running_state == 1:
self._update_attribute(self.attributes_by_name["system_mode"].id, self.SystemMode.Heat)
self._update_attribute(self.attributes_by_name["running_mode"].id, self.RunningMode.Heat)
self._update_attribute(self.attributes_by_name["running_state"].id, self.RunningState.Heat_State_On)
elif running_state == 2:
self._update_attribute(self.attributes_by_name["system_mode"].id, self.SystemMode.Fan_only)
self._update_attribute(self.attributes_by_name["running_mode"].id, self.RunningMode.Off)
self._update_attribute(self.attributes_by_name["running_state"].id, self.RunningState.Fan_State_On)
def running_change(self, value, running_state):
"""Running state change."""
if value == 0:
_LOGGER.info("Modo Activo")
if running_state == 0:
mode = self.RunningMode.Heat
state = self.RunningState.Heat_State_On
elif running_state == 1:
mode = self.RunningMode.Cool
state = self.RunningState.Cool_State_On
if value == 1:
_LOGGER.info("Modo inactivo/idle")
mode = self.RunningMode.Off
state = self.RunningState.Idle
_LOGGER.info("Valor "+str(value))
_LOGGER.info("running estate "+str(running_state))
self._update_attribute(self.attributes_by_name["running_mode"].id, mode)
self._update_attribute(self.attributes_by_name["running_state"].id, state)
class TuyaFancoilUserInterface(TuyaUserInterfaceCluster):
"""HVAC User interface cluster for tuya electric heating thermostats."""
_CHILD_LOCK_ATTR = TUYA_FANCOIL_CHILD_LOCK_ATTR
class MoesBHT6(CustomDevice):
"""Tuya thermostat for devices like the Moes BHT-006GBZB Electric floor heating."""
"""Generic Tuya thermostat device."""
def __init__(self, *args, **kwargs):
"""Init device."""
self.thermostat_bus = Bus()
self.ui_bus = Bus()
self.battery_bus = Bus()
self.fan_bus = Bus()
super().__init__(*args, **kwargs)
signature = {
MODELS_INFO: [
("_TZE204_mpbki2zm", "TS0601"),
],
ENDPOINTS: {
# endpoint=1 profile=260 device_type=81 device_version=1 input_clusters=[0, 4, 5, 61184],
# output_clusters=[10, 25]
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.SMART_PLUG,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
TuyaManufClusterAttributes.cluster_id,
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
242:{
PROFILE_ID: 41440,
DEVICE_TYPE: 97,
INPUT_CLUSTERS: [],
OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
},
},
}
replacement = {
ENDPOINTS: {
1: {
PROFILE_ID: zha.PROFILE_ID,
DEVICE_TYPE: zha.DeviceType.THERMOSTAT,
INPUT_CLUSTERS: [
Basic.cluster_id,
Groups.cluster_id,
Scenes.cluster_id,
MoesBHT6ManufCluster,
TuyaFancoilThermostat,
TuyaFancoilUserInterface,
TuyaFancoilFanCluster
],
OUTPUT_CLUSTERS: [Time.cluster_id, Ota.cluster_id],
},
242: {
PROFILE_ID: 41440,
DEVICE_TYPE: 97,
INPUT_CLUSTERS: [],
OUTPUT_CLUSTERS: [GreenPowerProxy.cluster_id],
}
}
}
Steps to reproduce
1.Create a folder quirks in .config 2. Add the quirk I posted above inside the folder "quirks" 3. add this code to configuration.yaml
zha:
enable_quirks: true
custom_quirks_path: /config/quirks
Expected behavior
I hope the device will work properly without so many inconsistencies. I believe the quirk code is a bit messy and contains redundancies.
I developed this code based on other quirks and I am not very familiar with development in Zigpy.
Any contributions would be greatly appreciated.
This is the old post when I requested support for the device, but at the end I developed a partial solution https://github.com/zigpy/zha-device-handlers/issues/2529
Screenshots/Video
Screenshots/Video
[Paste/upload your media here]
Device signature
Device signature
[Paste the device signature here]
Diagnostic information
Diagnostic information
[Paste the diagnostic information here]
Logs
Logs
[Paste the logs here]
Additional information
No response
