I (basically ChatGPT) managed to create some kind of quirk to this based on generic Tuya Thermostat quirk and the work that was done in Z2MQTT with this thermostat (TS0601_TZE284_rlytpmij).
The model I have is this https://moeshouse.com/products/zigbee-smart-thermostat-programmable-temperature-controller-water-boiler-electric-heating?variant=50590497964347
I’m still testing this and appreciate corrections and suggestions. I don’t have any idea if the code is reasonable or pretty. The code has also some Finnish comments 
# Moes/Tuya TS0601 (_TZE284_rlytpmij) — ZHA quirk,
# - Builder-polku (quirks v2): tuya dp 1/111/117/47
# - Fallback (EF00): ep_attribute="tuya_ef00", time-callbackit varmistettu, DP-aliasit tuettu
# - Päivittää Thermostat.local_temperature ja TemperatureMeasurement.measured_value (0.01°C)
# - Replacement altistaa Thermostat + TemperatureMeasurement; EF00 klusteriluokkana
import logging
from zigpy.quirks import CustomDevice
from zigpy.profiles import zha
from zigpy.zcl.clusters.general import Basic, Identify, PowerConfiguration, Ota, Groups, Scenes
from zigpy.zcl.clusters.hvac import Thermostat, RunningState
from zigpy.zcl.clusters.measurement import TemperatureMeasurement
_LOGGER = logging.getLogger(__name__)
# ---------- Yritä builder-APIa (quirks v2) ----------
_builder_ok = False
try:
# v2 builder
from zigpy.quirks.v2 import BinarySensorDeviceClass, EntityType
from zigpy.quirks.v2.homeassistant.sensor import SensorDeviceClass, SensorStateClass
from zigpy.quirks.v2.homeassistant import UnitOfTemperature
from zigpy.types import t
from zigpy.zcl import foundation
from zhaquirks.tuya import TUYA_SET_TIME, TuyaTimePayload
from zhaquirks.tuya.builder import TuyaQuirkBuilder
from zhaquirks.tuya.mcu import TuyaAttributesCluster, TuyaMCUCluster
_builder_ok = True
except Exception:
_builder_ok = False
# ---------- Builder-pohjainen toteutus ----------
if _builder_ok:
class TuyaThermostatCluster(Thermostat, TuyaAttributesCluster):
"""Tuya-local Thermostat cluster; Heating only, poistetut attribuutit yhtenäisesti."""
_CONSTANT_ATTRIBUTES = {
Thermostat.AttributeDefs.ctrl_sequence_of_oper.id:
Thermostat.ControlSequenceOfOperation.Heating_Only
}
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Thermostat vakiot: merkkaa eksplisiittisesti unsupported, jos HA/ZHA yrittää lukea
self.add_unsupported_attribute(Thermostat.AttributeDefs.setpoint_change_source.id)
self.add_unsupported_attribute(Thermostat.AttributeDefs.setpoint_change_source_timestamp.id)
self.add_unsupported_attribute(Thermostat.AttributeDefs.pi_heating_demand.id)
# Kalibrointi uusissa toteutuksissa tuodaan usein erikseen
# (ZHA voi tarjota eri polkua; jätetään unsupported ellei DP 19 sidota builderilla)
# self.add_unsupported_attribute(Thermostat.AttributeDefs.local_temperature_calibration.id)
class NoManufTimeNoVersionRespTuyaMCUCluster(TuyaMCUCluster):
"""Tuya Manufacturer Cluster, jossa set_time on julkinen ja MCU-version vastaus ei kaada."""
class ServerCommandDefs(TuyaMCUCluster.ServerCommandDefs):
set_time = foundation.ZCLCommandDef(
id=TUYA_SET_TIME,
schema={"time": TuyaTimePayload},
is_manufacturer_specific=False,
)
def handle_mcu_version_response(self, payload: TuyaMCUCluster.MCUVersion) -> foundation.Status: # type: ignore
return foundation.Status.SUCCESS
# Rakennetaan quirk täsmälleen sinun valmistaja+mallille (DP:t Z2M-mallin mukaan)
(
TuyaQuirkBuilder("_TZE284_rlytpmij", "TS0601")
.tuya_dp( # SystemMode: 1 bool -> Heat/Off
dp_id=1,
ep_attribute=TuyaThermostatCluster.ep_attribute,
attribute_name=TuyaThermostatCluster.AttributeDefs.system_mode.name,
converter=lambda x: {True: Thermostat.SystemMode.Heat, False: Thermostat.SystemMode.Off}[x],
dp_converter=lambda x: {Thermostat.SystemMode.Heat: True, Thermostat.SystemMode.Off: False}[x],
)
.tuya_dp( # Setpoint: 111 -> OccupiedHeatingSetpoint (0.01°C)
dp_id=111,
ep_attribute=TuyaThermostatCluster.ep_attribute,
attribute_name=TuyaThermostatCluster.AttributeDefs.occupied_heating_setpoint.name,
converter=lambda x: int(x * 10), # DP 0.1°C -> ZCL 0.01°C
dp_converter=lambda x: int(x // 10), # ZCL 0.01°C -> DP 0.1°C
)
.tuya_dp( # Local temperature: 117 -> local_temperature (0.01°C)
dp_id=117,
ep_attribute=TuyaThermostatCluster.ep_attribute,
attribute_name=TuyaThermostatCluster.AttributeDefs.local_temperature.name,
converter=lambda x: int(x * 10), # 0.1°C -> 0.01°C
)
.tuya_dp( # Running state: 47 -> 0=heat,1=idle (Z2M mallissa)
dp_id=47,
ep_attribute=TuyaThermostatCluster.ep_attribute,
attribute_name=TuyaThermostatCluster.AttributeDefs.running_state.name,
converter=lambda x: RunningState.Heat_State_On if (x in (0, False)) else RunningState.Idle,
)
.adds(TuyaThermostatCluster) # lisää local Thermostat -klusteri
.skip_configuration() # ei lähetetä config-kyselyitä
.add_to_registry(replacement_cluster=NoManufTimeNoVersionRespTuyaMCUCluster) # set_time-fix
)
else:
# ---------- EF00-fallback (vanhempi ympäristö) ----------
try:
from zhaquirks.tuya import TuyaManufCluster
except Exception as e:
TuyaManufCluster = None
_LOGGER.warning("TuyaManufCluster ei saatavilla: %s", e)
DP_TEMP_IDS = (117, 1, 3)
DP_SETPOINT_IDS = (111, 2)
DP_HEATSTATE_IDS = (47, 12, 13)
def _tuya01_to_c(v) -> float:
if isinstance(v, (bytes, bytearray)):
try:
v = int.from_bytes(v, "big")
except Exception:
v = 0
return float(int(v)) / 10.0
class MoesTuyaCluster(TuyaManufCluster): # type: ignore
"""EF00-klusteri: parser + ZCL-mapping + time-callback-fixit"""
cluster_id = 0xEF00
ep_attribute = "tuya_ef00"
# Luokkataso no-op, varalle
@staticmethod
def set_time(*args, **kwargs): return None
@staticmethod
def set_time_local_offset(*args, **kwargs): return None
def handle_cluster_request(self, hdr, args, dst_addressing=None):
# Instanssitaso varmistus ennen basea
if not hasattr(self, "set_time") or self.set_time is None:
self.set_time = lambda *a, **kw: None
if not hasattr(self, "set_time_local_offset") or self.set_time_local_offset is None:
self.set_time_local_offset = lambda *a, **kw: None
return super().handle_cluster_request(hdr, args, dst_addressing=dst_addressing)
def _parse_dp_payload(self, data: bytes):
idx, ln, out = 0, len(data), []
while idx + 2 <= ln:
dp_id = data[idx]; dp_type = data[idx+1]; idx += 2
if idx >= ln: break
length = data[idx]
if (idx + 1 + length) > ln and (idx + 2) <= ln:
length = (length << 8) | data[idx + 1]; idx += 2
else:
idx += 1
if idx + length > ln: break
raw = data[idx: idx + length]; idx += length
if dp_type == 0x01: val = bool(raw[0]) if raw else False
elif dp_type == 0x02: val = int.from_bytes(raw, "big", signed=(len(raw) == 2))
elif dp_type == 0x04: val = raw[0] if raw else 0
else: val = raw
out.append((dp_id, dp_type, val, raw))
return out
def _apply(self, dp_id, dp_type, val, raw):
ep1 = getattr(self.endpoint.device, "endpoints", {}).get(1)
thermo: Thermostat = getattr(ep1, "thermostat", None) if ep1 else None
tmeas: TemperatureMeasurement = getattr(ep1, "temperature", None) if ep1 else None
def up(cluster, attr, value):
if cluster is None: return
try: cluster._update_attribute(attr, value)
except Exception as e: _LOGGER.debug("attr 0x%04X update failed: %s", attr, e)
# temp
if (dp_id in DP_TEMP_IDS):
c = _tuya01_to_c(val)
if thermo: up(thermo, 0x0000, int(round(val * 10)))
if tmeas: up(tmeas, 0x0000, int(round(val * 10)))
# setpoint
elif (dp_id in DP_SETPOINT_IDS) and thermo:
c = _tuya01_to_c(val)
up(thermo, 0x0012, int(round(val * 10)))
# mode
elif dp_id in (1, 2) and thermo:
sys = 0x04 if val in (True, 1, "heat", "manual") else 0x00
up(thermo, 0x001C, sys)
# running
elif (dp_id in DP_HEATSTATE_IDS) and thermo:
heat_on = val not in (1, "idle", "Idle", False)
up(thermo, 0x0029, 0x0001 if heat_on else 0x0000)
# calib
elif dp_id == 19 and thermo:
try: up(thermo, 0x0010, int(val))
except Exception: pass
def cluster_command(self, tsn, command_id, args):
try:
raw = bytes(args[0]) if args and isinstance(args[0], (bytes, bytearray)) else bytes(args or b"")
payload = raw
for cut in (0, 2, 4):
test = raw[cut:] if len(raw) > cut else raw
if self._parse_dp_payload(test): payload = test; break
for dp_id, dp_type, val, raw_dp in self._parse_dp_payload(payload):
_LOGGER.debug("Tuya DP recv: id=%s type=%s val=%s", dp_id, dp_type, val)
self._apply(dp_id, dp_type, val, raw_dp)
except Exception as e:
_LOGGER.exception("EF00 parse error: %s", e)
try:
return super().cluster_command(tsn, command_id, args)
except Exception:
return
class MoesThermostat(CustomDevice): # Fallback device class
signature = {
"models_info": [("_TZE284_rlytpmij", "TS0601")],
"endpoints": {
1: {
"profile_id": 0x0104,
"device_type": 0x0051,
"input_clusters": [0x0000, 0x0004, 0x0005, 0xED00, 0xEF00],
"output_clusters": [0x000A, 0x0019],
},
242: {
"profile_id": 0xA1E0,
"device_type": 0x0061,
"input_clusters": [],
"output_clusters": [0x0021],
},
},
}
replacement = {
"endpoints": {
1: {
"profile_id": 0x0104,
"device_type": 0x0301, # THERMOSTAT
"input_clusters": [
0x0000, 0x0003, 0x0001, 0x0004, 0x0005,
0x0201, 0x0402,
MoesTuyaCluster, # EF00 custom-luokka
],
"in_clusters": [
0x0000, 0x0003, 0x0001, 0x0004, 0x0005,
0x0201, 0x0402,
MoesTuyaCluster,
],
"output_clusters": [0x0019],
"out_clusters": [0x0019],
},
242: {
"profile_id": 0xA1E0,
"device_type": 0x0061,
"input_clusters": [],
"in_clusters": [],
"output_clusters": [0x0021],
"out_clusters": [0x0021],
},
}
}
def __init__(self, *a, **kw):
super().__init__(*a, **kw)
self._patch_ef00()
def setup(self):
self._patch_ef00()
def _patch_ef00(self):
# Instanssitasolle varmistetaan time-callbackit riippumatta siitä, kumpi EF00 on käytössä
try:
ep1 = self.endpoints.get(1)
ef00 = None
for attr in ("tuya_ef00", "tuya_cluster", "tuya_manufacturer", "manufacturer_specific"):
ef00 = getattr(ep1, attr, None)
if ef00 is not None:
break
if ef00 is None:
_LOGGER.debug("EF00 instance not found on ep1 during patch")
return
ef00.set_time = lambda *a, **kw: None
ef00.set_time_local_offset = lambda *a, **kw: None
_LOGGER.debug("EF00 time callbacks patched on instance")
except Exception as e:
_LOGGER.debug("EF00 patch failed: %s", e)
This took about 100 iterations/tries, really annoying.
I’m thinking of moving from ZHA to Z2M… A new Zigbee adapter should arrive today and I will build a parallel Z2M network for testing.
