[RELEASE] Generic Zigbee Contact/Leak Sensor Driver for Tuya Leak Sensors

I got tired of my many Tuya Zigbee leak sensors I got from Ali Express reporting closed for dry and open for wet, using the generic Zigbee contact sensor driver. There could be a driver out there in the community I don't know about that reports correctly for these, and I just don't know where it is...

So I just had AI quickly write me a Generic Zigbee contact sensor driver, and then I modified it to also be a water sensor, that reports wet and dry correctly for these Tuya sensors.

In case anyone else is also using these type of sensors with the Generic contact sensor driver, but wants it to report wet and dry as an actual water sensor
image

Generic Zigbee Contact/Leak Sensor Driver
/**
 *  Generic Zigbee Contact/Leak Sensor Driver for Hubitat Elevation
 *
 *  Supports standard Zigbee contact sensors using the IAS Zone cluster (0x0500)
 *  and optionally reports battery status from the Power Configuration cluster.
 *  Modified to also be a WaterSensor, to report wet and dry correctly for some Tuya  Zigbee sensors (where closed = dry and open = wet)
 */

import hubitat.zigbee.zcl.DataType
import hubitat.device.HubMultiAction
import hubitat.device.Protocol

metadata {
    definition (name: "Generic Zigbee Contact/Leak Sensor", namespace: "Hubitat", author: "Deepseek AI", mnmn: "SmartThingsCommunity", vid: "generic-contact-v1") {
        capability "ContactSensor"
        capability "WaterSensor"
        capability "Battery"
        capability "Health Check"
        capability "Configuration"

        command "refresh"
        fingerprint profileId: "0104", inClusters: "0000,0001,0500", outClusters: "0019"
    }

    preferences {
        input name: "debugLogging", type: "bool", title: "Enable debug logging", defaultValue: true
        input name: "batteryReportingInterval", type: "number", title: "Battery Reporting Interval (minutes)", defaultValue: 1440
        input name: "zoneStatusReportingInterval", type: "number", title: "Zone Status Reporting Interval (seconds)", defaultValue: 60
    }
}

def installed() { initialize() }
def updated() { unschedule(); initialize() }

def initialize() {
    if (debugLogging) runIn(1800, logsOff)
    configure()
}

def logsOff() {
    log.warn "debug logging disabled"
    device.updateSetting("debugLogging", [value: "false", type: "bool"])
}

def configure() {
    log.debug "Configuring device..."
    if (!device.currentValue("checkInterval")) {
        sendEvent(name: "checkInterval", value: 12 * 60 * 60, displayed: false,
                  data: [protocol: "zigbee", hubHardwareId: device.hub.hardwareID])
    }
    def cmds = []
    def zoneMin = (zoneStatusReportingInterval ?: 60).toInteger()
    def battMin = 300
    def battMax = ((batteryReportingInterval ?: 1440).toInteger()) * 60

    cmds += zigbee.configureReporting(zigbee.IAS_ZONE_CLUSTER, 0x0000, DataType.UINT16, zoneMin, 3600, 0x01)
    cmds += zigbee.configureReporting(zigbee.POWER_CONFIGURATION_CLUSTER, 0x0020, DataType.UINT8, battMin, battMax, 0x01)
    cmds += zigbee.configureReporting(zigbee.POWER_CONFIGURATION_CLUSTER, 0x0021, DataType.UINT8, battMin, battMax, 0x01)

    sendHubCommand(new HubMultiAction(cmds, Protocol.ZIGBEE))
    refresh()
}

def refresh() {
    def cmds = []
    cmds += zigbee.readAttribute(zigbee.IAS_ZONE_CLUSTER, 0x0000)
    cmds += zigbee.readAttribute(zigbee.POWER_CONFIGURATION_CLUSTER, 0x0020)
    cmds += zigbee.readAttribute(zigbee.POWER_CONFIGURATION_CLUSTER, 0x0021)
    sendHubCommand(new HubMultiAction(cmds, Protocol.ZIGBEE))
}

def parse(String description) {
    if (debugLogging) log.debug "parse: $description"

    // Handle human-readable zone status messages (e.g., "zone status 0x0001 ...")
    if (description.startsWith("zone status")) {
        def parts = description.split()
        if (parts.size() >= 3) {
            def zoneStatus = Integer.decode(parts[2])
            def contact = (zoneStatus & 0x01) ? "open" : "closed"
            def water = (zoneStatus & 0x01) == 0x01 ? "wet" : "dry"
            sendEvent(name: "contact", value: contact, isStateChange: true)
            sendEvent(name: "water", value: water, isStateChange: true)
        }
        return
    }

    // Parse the Zigbee message
    def msg = zigbee.parse(description)
    if (!msg) {
        if (debugLogging) log.debug "Failed to parse message"
        return
    }

    switch (msg.clusterId) {
        case zigbee.IAS_ZONE_CLUSTER:
            handleIasZone(msg)
            break
        case zigbee.POWER_CONFIGURATION_CLUSTER:
            handlePowerConfig(msg)
            break
        default:
            if (debugLogging) log.debug "Unhandled cluster: ${msg.clusterId}"
    }
}

private void handleIasZone(msg) {
    if (debugLogging) log.debug "IAS Zone: cmd=${msg.command}, attrId=${msg.attrId}, val=${msg.value}, data=${msg.data}"

    // Zone status change notification (command 0x00)
    if (msg.command == 0x00 && msg.data?.size() >= 2) {
        def zoneStatus = (msg.data[0] & 0xFF) + ((msg.data[1] & 0xFF) << 8)
        def contact = (zoneStatus & 0x01) ? "open" : "closed"
        def water = (zoneStatus & 0x01) == 0x01 ? "wet" : "dry"
        sendEvent(name: "contact", value: contact, isStateChange: true)
        sendEvent(name: "water", value: water, isStateChange: true)
        if (debugLogging) log.debug "Zone status change: ${zoneStatus} -> ${contact}"
    }
    // Attribute read response (command 0x01) or report (command 0x07) for zone status (attrId 0x0000)
    else if ((msg.command == 0x01 || msg.command == 0x07) && msg.attrId == 0x0000) {
        def zoneStatus = msg.value as Integer
        def contact = (zoneStatus & 0x01) ? "open" : "closed"
        def water = (zoneStatus & 0x01) == 0x01 ? "wet" : "dry"
        sendEvent(name: "contact", value: contact, isStateChange: true)
        sendEvent(name: "water", value: water, isStateChange: true)
        if (debugLogging) log.debug "Zone status attribute: ${zoneStatus} -> ${contact}"
    }
}

private void handlePowerConfig(msg) {
    if (debugLogging) log.debug "Power Config: cmd=${msg.command}, attrId=${msg.attrId}, val=${msg.value}"

    if (msg.attrId == 0x0020 && msg.value != null) {
        def voltage = msg.value as Integer
        def batteryVolts = voltage / 10.0
        def percent = Math.min(100, Math.max(0, ((batteryVolts - 2.4) / 0.6 * 100).round()))
        sendEvent(name: "battery", value: percent, unit: "%", isStateChange: true)
        if (debugLogging) log.debug "Battery voltage: ${batteryVolts}V -> ${percent}%"
    }
    else if (msg.attrId == 0x0021 && msg.value != null) {
        def raw = msg.value as Integer
        def percent = Math.min(100, (raw / 2).round())
        sendEvent(name: "battery", value: percent, unit: "%", isStateChange: true)
        if (debugLogging) log.debug "Battery raw: ${raw} -> ${percent}%"
    }
}

The "Tuya NEO Coolcam Zigbee Water Leak Sensor" driver has worked fine for my Tuya leak sensors, but good to have choices!

1 Like

Yup, I figured there was a driver that worked with them out there already, good to know now what that is. I didn't spend much time on the AI driver, it only took three iterations with AI to get it working.