Hi all, I have this device installed in my home QIACHIP Upgrade Universal WIFI Ceiling Fan and I wanted to get local control over it using MQTT. Of course its a Tuya device. So, I have a NAS from QNAP which I run Container Station and created a MQTT server and used MQTT Client for local Tuya devices by Volker76 on github here. Follow the instructions there on setup. Here's the YML configuration file I used.
YML Code:
services:
tuya-mqtt:
image: volkerhaensel/tuya_mqtt.net:latest
container_name: tuya-mqtt
networks:
macvlan_net:
ipv4_address: YOUR.IP.ADDRESS.HERE # Assign a unique LAN IP for Tuya-MQTT (e.g., 192.168.1.230)
volumes:
- tuya.net_config:/app/DataDir # DataDir is case-sensitive
restart: unless-stopped
expose:
- "80/tcp" # Web UI → http://YOUR.IP.ADDRESS.HERE
- "6666/udp" # Tuya discovery (unencrypted)
- "6667/udp" # Tuya discovery (encrypted)
healthcheck:
test: ["CMD-SHELL", "wget -q -O /dev/null http://localhost:80 || exit 1"]
interval: 60s
timeout: 5s
retries: 5
start_period: 20s
mosquitto:
image: eclipse-mosquitto:2
container_name: tuya-mosquitto
networks:
macvlan_net:
ipv4_address: YOUR.IP.ADDRESS.HERE # Assign a DIFFERENT unique LAN IP for Mosquitto (e.g., 192.168.1.232)
expose:
- "1883/tcp" # Clients connect → tcp://YOUR.IP.ADDRESS.HERE:1883
volumes:
- ./mosquitto:/mosquitto # Place mosquitto.conf in ./mosquitto/config/
restart: unless-stopped
networks:
macvlan_net:
driver: macvlan
driver_opts:
parent: br0 # Change if your LAN interface is different
ipam:
config:
- subnet: 192.168.1.0/24 # Match your LAN subnet (change if needed)
gateway: 192.168.1.1 # Change to your router’s gateway IP
volumes:
tuya.net_config:
I entered into the Tuya-MQTT, set that up as per Volker76's github page. I was able to identified the device in my network and connected to the device. Of note, you may need a Tuya IOT login for this, again, see Volker76's github page for details.
After the fact I created this Driver using our AI overlord's help. Installed the driver into Hubitat, added a Virtual Device using the driver and coifgured the device under the device's preferences. It will ask for your mqtt Mosquitto server IP address (not the MQTT-Tuya IP address) and the Device ID, IP or Name.
You initialize and then you will then be able to control your Tuya device.
Driver code:
/**
* Tuya MQTT.NET Fan+Light (Local) + DP Learn
* Platform: Hubitat
* Target bridge: volkerhaensel/tuya_mqtt.net
* License: MIT
*/
import groovy.transform.Field
import groovy.json.JsonOutput
@Field static final Integer PRESENCE_TIMEOUT_SEC = 3600
@Field static final Integer RESEND_DELAY_SEC = 2
@Field static final Integer WATCHDOG_PERIOD_SEC = 30
@Field static final Integer BACKOFF_MIN_SEC = 5
@Field static final Integer BACKOFF_MAX_SEC = 300
@Field static final Integer CONNECT_STALL_SEC = 20
@Field static final Integer RESUBSCRIBE_AFTER_SILENCE = 300
@Field static final Integer OBSERVED_MAX_DPS = 64
@Field static final List<String> SPEED_ORDER = ["off", "low", "medium", "high"]
@Field static final List<Integer> DPs_PRIMARY = [1, 3, 9]
@Field static final String RECONNECT_JOB = "reconnectStep"
@Field static final String BOOTSTRAP_TOPIC_KEY = "__bootstrapWildcard__"
@Field static final String CMD_SEGMENT = "command"
metadata {
definition(name: "Tuya MQTT.NET Fan+Light (DP Learn)", namespace: "public", author: "you") {
capability "Initialize"
capability "Switch"
capability "FanControl"
capability "Refresh"
capability "PresenceSensor"
command "cycleSpeed"
command "reconnectNow"
command "clearObservedDPs"
command "clearLearnedRoot"
attribute "supportedFanSpeeds", "JSON_OBJECT"
attribute "speed", "STRING"
attribute "lastSeen", "STRING"
attribute "mqttStatus", "STRING"
attribute "activeRoot", "STRING"
attribute "mqttConnected", "STRING"
attribute "observedDPs", "JSON_OBJECT"
attribute "dpLast", "JSON_OBJECT"
}
preferences {
input name: "brokerHost", type: "text", title: "MQTT Broker Host", defaultValue: "192.168.4.232", required: true
input name: "brokerPort", type: "text", title: "MQTT Broker Port", defaultValue: "1883", required: true
input name: "username", type: "text", title: "MQTT Username (optional)"
input name: "password", type: "password", title: "MQTT Password (optional)"
input name: "baseTopic", type: "text", title: "tuya_mqtt.net MQTT Topic / Base topic", defaultValue: ""
input name: "identifierMode", type: "enum", title: "Identifier Mode",
options: ["Auto", "ID/Name", "IP"], defaultValue: "ID/Name"
input name: "deviceId", type: "text", title: "tuya_mqtt.net Device Name or ID", defaultValue: "Fan", required: false
input name: "deviceIp", type: "text", title: "Device IP (optional)", required: false
input name: "enableLearning", type: "bool", title: "Learn active MQTT root from wildcard subscription", defaultValue: false
input name: "legacyCommandTopic", type: "bool", title: "Also publish legacy command/n topics", defaultValue: false
input name: "createLightChild", type: "bool", title: "Expose DP9 as child light", defaultValue: true
input name: "logEnable", type: "bool", title: "Enable debug logging", defaultValue: true
}
}
def installed() { initialize() }
def updated() { initialize() }
def uninstalled() {
try { interfaces.mqtt.disconnect() } catch (ignored) {}
unschedule()
}
def initialize() {
unschedule()
try { interfaces.mqtt.disconnect() } catch (ignored) {}
validatePrefs()
def prefRoots = candidateIdentifiers().collect { rootFor(it) }.findAll { it }
if (state.activeRoot && prefRoots && !prefRoots.contains(state.activeRoot as String)) {
state.remove("activeRoot")
}
state.lastSeenEpoch = 0L
state.lastMqttOkEpoch = now()
state.connected = false
state.remove("connectingAt")
state.backoff = BACKOFF_MIN_SEC
state.reconnectAtEpoch = 0L
state.subscribeRetryAt = 0L
state.lastResubscribeEpoch = now()
state.remove("pendingSpeedRaw")
state.remove("stabilizeUntil")
if (state.observed == null) state.observed = []
if (state.dpLast == null) state.dpLast = [:]
sendEvent(name: "mqttConnected", value: "false")
sendEvent(name: "activeRoot", value: state.activeRoot ?: "")
connectMqtt()
runIn(3, "publishSupportedSpeeds")
runIn(4, "ensureLightChild")
runIn(5, "startWatchdogLoop")
runEvery1Minute("presenceTick")
runIn(10, "connectionGuard")
if (logEnable) log.debug "Initialized Tuya MQTT.NET Fan+Light driver"
}
def refresh() {
if (logEnable) log.debug "Refresh requested; waiting for next MQTT DP/TS update"
subscribeTopics()
}
def clearObservedDPs() {
state.observed = []
state.dpLast = [:]
sendEvent(name: "observedDPs", value: "[]", isStateChange: true)
sendEvent(name: "dpLast", value: "{}", isStateChange: true)
}
def clearLearnedRoot() {
state.remove("activeRoot")
sendEvent(name: "activeRoot", value: "", isStateChange: true)
subscribeTopics()
}
private boolean mqttIsConnected() {
return (state.connected == true) ||
(state.connected?.toString() == "true") ||
(device.currentValue("mqttConnected")?.toString() == "true")
}
private void connectMqtt() {
state.connected = false
final Integer portInt = safePort()
final String uri = "tcp://${brokerHost}:${portInt}"
final String clientId = "hubitat-tuya-fan-${device.id}"
if (logEnable) log.debug "Connecting MQTT ${uri} as ${clientId}"
try {
sendEvent(name: "mqttStatus", value: "connecting")
sendEvent(name: "mqttConnected", value: "false")
state.connectingAt = now()
interfaces.mqtt.connect(uri, clientId, username ?: null, password ?: null)
} catch (e) {
if (logEnable) log.warn "MQTT connect threw: ${e?.message ?: e}"
state.remove("connectingAt")
scheduleReconnect("connect threw")
}
}
private void scheduleReconnect(String reason = null) {
if (mqttIsConnected()) return
long nowMs = now()
long nextAt = (state.reconnectAtEpoch ?: 0L) as long
if (nextAt > nowMs) {
if (logEnable && reason) {
long remaining = Math.max(1L, (nextAt - nowMs) / 1000L)
log.debug "Reconnect already scheduled in ${remaining}s; not rescheduling (${reason})"
}
return
}
Integer delay = (state.backoff ?: BACKOFF_MIN_SEC) as Integer
delay = Math.max(BACKOFF_MIN_SEC, Math.min(delay, BACKOFF_MAX_SEC))
Integer jitter = Math.max(1, (int)(delay * 0.2))
Integer withJitter = Math.max(1, delay + (new Random().nextInt(jitter * 2 + 1) - jitter))
try { unschedule(RECONNECT_JOB) } catch (ignored) {}
if (logEnable) {
log.warn "Scheduling reconnect in ${withJitter}s (backoff=${delay}s)" + (reason ? " reason=${reason}" : "")
}
runIn(withJitter, RECONNECT_JOB)
state.reconnectAtEpoch = nowMs + (withJitter * 1000L)
state.backoff = Math.min(delay * 2, BACKOFF_MAX_SEC)
}
private void immediateReconnect(String reason = null) {
if (logEnable) log.warn "Immediate reconnect" + (reason ? " (${reason})" : "")
try { unschedule(RECONNECT_JOB) } catch (ignored) {}
state.reconnectAtEpoch = 0L
state.backoff = BACKOFF_MIN_SEC
state.connected = false
state.remove("connectingAt")
try { interfaces.mqtt.disconnect() } catch (ignored) {}
pauseExecution(250)
connectMqtt()
}
def reconnectStep() {
if (mqttIsConnected()) {
if (logEnable) log.debug "Reconnect skipped: already connected"
return
}
try { interfaces.mqtt.disconnect() } catch (ignored) {}
pauseExecution(250)
connectMqtt()
}
def reconnectNow() {
if (logEnable) log.warn "Manual reconnect requested"
immediateReconnect("manual")
}
private boolean publishMqtt(String topic, String payload, Boolean retain = false, Integer qos = 1) {
if (!mqttIsConnected()) {
if (logEnable) log.warn "Publish blocked while disconnected: ${topic}"
scheduleReconnect("publish while disconnected")
return false
}
try {
interfaces.mqtt.publish(topic, payload, qos, retain)
return true
} catch (e) {
String msg = (e?.message ?: e?.toString() ?: "").toLowerCase()
state.connected = false
sendEvent(name: "mqttConnected", value: "false")
if (msg.contains("client is not connected")) {
immediateReconnect("publish: client not connected")
} else {
if (logEnable) log.warn "Publish failed ${topic}: ${e?.message ?: e}"
scheduleReconnect("publish exception")
}
return false
}
}
private String rootFor(String ident) {
def b = cleanTopicPart(baseTopic)
def i = cleanTopicPart(ident)
if (!i) return null
return b ? "${b}/${i}" : i
}
private List<String> candidateIdentifiers() {
def ids = []
def mode = (identifierMode ?: "ID/Name").toString()
String did = deviceId?.trim()
String dip = deviceIp?.trim()
switch (mode) {
case "ID/Name":
if (did) ids << did
break
case "IP":
if (dip) ids << dip
break
default:
if (did) ids << did
if (!did && dip) ids << dip
break
}
return ids.findAll { it }.unique()
}
private List<String> configuredRoots() {
def roots = []
if (state.activeRoot) roots << (state.activeRoot as String)
candidateIdentifiers().each { ident ->
def r = rootFor(ident)
if (r) roots << r
}
return roots.findAll { it }.unique()
}
private String activeRootOrNull() {
if (state.activeRoot) return state.activeRoot as String
def roots = configuredRoots()
return roots ? roots[0] : null
}
private String tStateDP(String root, int dp) { "${root}/DP${dp}" }
private String tStateDps(String root, int dp) { "${root}/dps/${dp}" }
private String tCmdDP(String root, int dp) { "${root}/DP${dp}/${CMD_SEGMENT}" }
private String tCmdLegacy(String root, int dp) { "${root}/${CMD_SEGMENT}/${dp}" }
private String tDeviceTs(String root) { "${root}/TS" }
private boolean topicMatchesDp(String topic, String root, int dp) {
return topic == tStateDP(root, dp) || topic == tStateDps(root, dp)
}
def subscribeTopics() {
if (!mqttIsConnected()) {
long nowMs = now()
long retryAt = (state.subscribeRetryAt ?: 0L) as long
if (retryAt <= nowMs) {
state.subscribeRetryAt = nowMs + 3000L
runIn(3, "subscribeTopics")
}
return
}
state.subscribeRetryAt = 0L
def roots = configuredRoots()
if (roots) {
roots.each { r -> subscribeRoot(r) }
if (logEnable) log.debug "Subscribed to roots: ${roots}"
}
if (learningEnabled()) {
def wild = bootstrapWildcard()
try {
interfaces.mqtt.subscribe(wild)
state[BOOTSTRAP_TOPIC_KEY] = wild
if (logEnable) log.debug "Root learning subscribed to ${wild}"
} catch (e) {
handleSubscribeFailure(e, "bootstrap subscribe")
}
}
}
private void subscribeRoot(String root) {
try {
interfaces.mqtt.subscribe("${root}/#")
DPs_PRIMARY.each { dp ->
interfaces.mqtt.subscribe(tStateDP(root, dp))
interfaces.mqtt.subscribe(tStateDps(root, dp))
}
interfaces.mqtt.subscribe(tDeviceTs(root))
} catch (e) {
handleSubscribeFailure(e, "subscribe ${root}")
}
}
private void handleSubscribeFailure(def e, String context) {
String msg = (e?.message ?: e?.toString() ?: "").toLowerCase()
state.connected = false
sendEvent(name: "mqttConnected", value: "false")
if (msg.contains("client is not connected")) {
immediateReconnect("${context}: client not connected")
} else {
if (logEnable) log.warn "${context} failed: ${e?.message ?: e}"
scheduleReconnect("${context} failed")
}
}
private String bootstrapWildcard() {
def b = cleanTopicPart(baseTopic)
return b ? "${b}/#" : "#"
}
private void learnActiveRootFromTopic(String topic) {
if (state.activeRoot) return
if (isCommandTopic(topic)) return
String root = null
def m = (topic =~ /^(.*)\/DP(\d+)$/)
if (m.matches()) root = m[0][1]
if (!root) {
m = (topic =~ /^(.*)\/dps\/(\d+)$/)
if (m.matches()) root = m[0][1]
}
if (!root) return
state.activeRoot = root
sendEvent(name: "activeRoot", value: root, isStateChange: true)
if (logEnable) log.warn "Active MQTT root learned: ${root}"
def wild = (state[BOOTSTRAP_TOPIC_KEY] ?: null)
if (wild && wild != "${root}/#") {
try { interfaces.mqtt.unsubscribe(wild) } catch (ignored) {}
state.remove(BOOTSTRAP_TOPIC_KEY)
}
if (mqttIsConnected()) subscribeRoot(root)
}
def mqttClientStatus(String status) {
sendEvent(name: "mqttStatus", value: status)
if (logEnable) log.debug "MQTT status: ${status}"
String s = (status ?: "").toString().toLowerCase()
boolean ok =
s.contains("connection succeeded") ||
s.contains("connection succeed") ||
s.contains("succeeded") ||
s.contains("succeed")
boolean bad =
s.startsWith("error") ||
s.contains("lost") ||
s.contains("disconnected") ||
s.contains("connection failed") ||
s.contains("fail") ||
s.contains("client is not connected")
if (ok) {
state.connected = true
state.backoff = BACKOFF_MIN_SEC
state.lastMqttOkEpoch = now()
state.lastResubscribeEpoch = now()
state.remove("connectingAt")
state.reconnectAtEpoch = 0L
try { unschedule(RECONNECT_JOB) } catch (ignored) {}
sendEvent(name: "mqttConnected", value: "true")
presenceMark(true)
try { subscribeTopics() } catch (ignored) {}
runIn(2, "subscribeTopics")
runIn(30, "connectionGuard")
return
}
if (bad) {
state.connected = false
state.remove("connectingAt")
sendEvent(name: "mqttConnected", value: "false")
if (s.contains("client is not connected")) {
immediateReconnect("status: client not connected")
} else {
scheduleReconnect("mqttClientStatus bad: ${status}")
}
runIn(10, "connectionGuard")
return
}
runIn(10, "connectionGuard")
}
def parse(String description) {
def msg = interfaces.mqtt.parseMessage(description)
if (!msg?.topic) return
final String topic = msg.topic.toString()
final String payload = (msg.payload == null) ? "" : msg.payload.toString().trim()
state.lastMqttOkEpoch = now()
if (!state.activeRoot) learnActiveRootFromTopic(topic)
final String root = activeRootOrNull()
if (!root) return
if (!topic.startsWith(root + "/")) return
if (isCommandTopic(topic)) return
final String rel = topic.substring(root.length() + 1)
if (rel == "TS") {
markSeen()
return
}
Integer observedDp = dpFromRelative(rel)
if (observedDp != null) {
markSeen()
rememberObservedDp(observedDp, payload)
}
if (topicMatchesDp(topic, root, 1)) {
final boolean onNow = asBool(payload)
sendEvent(name: "switch", value: onNow ? "on" : "off")
if (!onNow) sendEvent(name: "speed", value: "off")
return
}
if (topicMatchesDp(topic, root, 3)) {
final String raw = payload.replaceAll("\"", "")
final String name = (raw == "1") ? "low" : (raw == "2") ? "medium" : (raw == "3") ? "high" : "off"
final long nowMs = now()
if (state.pendingSpeedRaw && (state.pendingSpeedRaw in ["2", "3"]) && raw == "1") {
final long until = (state.stabilizeUntil ?: 0L) as long
if (nowMs <= until) {
publishDP3(state.pendingSpeedRaw as String)
return
}
}
sendEvent(name: "speed", value: name)
if (state.pendingSpeedRaw && raw == state.pendingSpeedRaw) {
state.remove("pendingSpeedRaw")
state.remove("stabilizeUntil")
}
return
}
if (topicMatchesDp(topic, root, 9)) {
final boolean onNow = asBool(payload)
childLight()?.sendEvent(name: "switch", value: onNow ? "on" : "off")
return
}
}
def on() {
sendEvent(name: "switch", value: "on")
publishDP1(true)
}
def off() {
publishDP1(false)
sendEvent(name: "switch", value: "off")
sendEvent(name: "speed", value: "off")
state.remove("pendingSpeedRaw")
state.remove("stabilizeUntil")
}
def setSpeed(String speed) {
speed = (speed ?: "off").toLowerCase()
switch (speed) {
case "off":
off()
return
case "low":
case "medium":
case "high":
final String v = (speed == "low") ? "1" : (speed == "medium") ? "2" : "3"
publishDP1(true)
state.pendingSpeedRaw = v
state.stabilizeUntil = now() + 5000
runInMillisSafe(500, "publishInitialSpeed")
runIn(RESEND_DELAY_SEC, "reassertPendingSpeed")
sendEvent(name: "switch", value: "on")
sendEvent(name: "speed", value: speed)
return
default:
setSpeed("medium")
return
}
}
def cycleSpeed() {
def cur = (device.currentValue("speed") ?: "off").toLowerCase()
def idx = SPEED_ORDER.indexOf(cur)
if (idx < 0) idx = 0
setSpeed(SPEED_ORDER[(idx + 1) % SPEED_ORDER.size()])
}
def publishInitialSpeed() {
def v = state.pendingSpeedRaw
if (v) publishDP3(v as String)
}
def reassertPendingSpeed() {
def v = state.pendingSpeedRaw
if (v) publishDP3(v as String)
}
private void publishDP1(boolean val) {
final String root = activeRootOrNull()
if (!root) {
log.warn "No active MQTT root for DP1 command"
return
}
final String p = val ? "true" : "false"
publishMqtt(tCmdDP(root, 1), p, false, 1)
if (legacyCommandTopic == true) {
publishMqtt(tCmdLegacy(root, 1), p, false, 1)
}
}
private void publishDP3(String v) {
final String root = activeRootOrNull()
if (!root) {
log.warn "No active MQTT root for DP3 command"
return
}
final String payload = JsonOutput.toJson((v ?: "").replaceAll("\"", ""))
publishMqtt(tCmdDP(root, 3), payload, false, 1)
if (legacyCommandTopic == true) {
publishMqtt(tCmdLegacy(root, 3), payload, false, 1)
}
}
private void publishDP9(boolean val) {
final String root = activeRootOrNull()
if (!root) {
log.warn "No active MQTT root for DP9 command"
return
}
final String p = val ? "true" : "false"
publishMqtt(tCmdDP(root, 9), p, false, 1)
if (legacyCommandTopic == true) {
publishMqtt(tCmdLegacy(root, 9), p, false, 1)
}
}
private void publishSupportedSpeeds() {
sendEvent(name: "supportedFanSpeeds", value: '["off","low","medium","high"]', isStateChange: true)
}
private void ensureLightChild() {
if (createLightChild != true) return
if (!childLight()) {
final String dni = "${device.deviceNetworkId}-light"
try {
addChildDevice("hubitat", "Generic Component Switch", dni,
[name: "${device.displayName} Light", label: "${device.displayName} Light", isComponent: true])
} catch (e) {
log.warn "Cannot create child light: ${e}"
}
}
}
private childLight() {
getChildDevice("${device.deviceNetworkId}-light")
}
def componentOn(cd) {
publishDP9(true)
childLight()?.sendEvent(name: "switch", value: "on")
}
def componentOff(cd) {
publishDP9(false)
childLight()?.sendEvent(name: "switch", value: "off")
}
def componentRefresh(cd) {}
private void markSeen() {
state.lastSeenEpoch = now()
sendEvent(name: "lastSeen", value: formatEpoch(state.lastSeenEpoch))
presenceMark(true)
}
private void presenceTick() {
final long last = (state.lastSeenEpoch ?: 0L) as long
if (last <= 0L) {
presenceMark(mqttIsConnected())
return
}
final long ageS = (now() - last) / 1000L
presenceMark(ageS <= PRESENCE_TIMEOUT_SEC)
}
private void presenceMark(boolean present) {
final String want = present ? "present" : "not present"
if (device.currentValue("presence") != want) {
sendEvent(name: "presence", value: want)
}
}
def startWatchdogLoop() {
runIn(Math.max(5, WATCHDOG_PERIOD_SEC), "watchdogTick")
}
def watchdogTick() {
if (!mqttIsConnected()) {
scheduleReconnect("watchdog not connected")
} else {
final long lastOk = (state.lastMqttOkEpoch ?: now()) as long
final long idleSec = (now() - lastOk) / 1000L
final long lastSub = (state.lastResubscribeEpoch ?: now()) as long
final long sinceSub = (now() - lastSub) / 1000L
if (idleSec >= RESUBSCRIBE_AFTER_SILENCE && sinceSub >= RESUBSCRIBE_AFTER_SILENCE) {
if (logEnable) log.warn "MQTT quiet for ${idleSec}s; resubscribing without recycling socket"
state.lastResubscribeEpoch = now()
subscribeTopics()
}
}
runIn(Math.max(5, WATCHDOG_PERIOD_SEC), "watchdogTick")
}
def connectionGuard() {
if (mqttIsConnected()) {
state.remove("connectingAt")
runIn(30, "connectionGuard")
return
}
long ca = (state.connectingAt ?: 0L) as long
if (ca > 0L) {
long age = (now() - ca) / 1000L
if (age >= CONNECT_STALL_SEC) {
if (logEnable) log.warn "Connection stalled for ${age}s; forcing reconnect"
immediateReconnect("connection stalled")
return
}
} else {
scheduleReconnect("guard not connected")
}
runIn(10, "connectionGuard")
}
private void rememberObservedDp(Integer dp, String payload) {
if (dp == null) return
def obs = (state.observed ?: []) as List
def dpLast = (state.dpLast ?: [:]) as Map
if (!obs.contains(dp)) obs << dp
obs = obs.collect { it as Integer }.unique().sort()
if (obs.size() > OBSERVED_MAX_DPS) {
obs = obs.take(OBSERVED_MAX_DPS)
}
dpLast[dp.toString()] = payload
if (dpLast.size() > OBSERVED_MAX_DPS) {
def keys = dpLast.keySet().toList().sort().take(OBSERVED_MAX_DPS)
dpLast = keys.collectEntries { [(it): dpLast[it]] }
}
state.observed = obs
state.dpLast = dpLast
sendEvent(name: "observedDPs", value: JsonOutput.toJson(obs), isStateChange: true)
sendEvent(name: "dpLast", value: JsonOutput.toJson(dpLast), isStateChange: true)
}
private Integer dpFromRelative(String rel) {
if (!rel) return null
def m = (rel =~ /^DP(\d+)$/)
if (m.matches()) return (m[0][1] as Integer)
m = (rel =~ /^dps\/(\d+)$/)
if (m.matches()) return (m[0][1] as Integer)
return null
}
private boolean isCommandTopic(String topic) {
if (!topic) return false
return topic.endsWith("/${CMD_SEGMENT}") || topic.contains("/${CMD_SEGMENT}/")
}
private boolean learningEnabled() {
return enableLearning == null ? false : (enableLearning as Boolean)
}
private String cleanTopicPart(def v) {
return (v == null) ? "" : v.toString().trim().replaceAll(/^\/+|\/+$/, "")
}
private void validatePrefs() {
if (!brokerHost?.trim()) {
throw new IllegalArgumentException("MQTT Broker Host is required")
}
if (baseTopic != null) {
def norm = cleanTopicPart(baseTopic)
if (norm != baseTopic) device.updateSetting("baseTopic", [value: norm, type: "text"])
}
if (deviceId != null) {
def v = deviceId.trim()
if (v != deviceId) device.updateSetting("deviceId", [value: v, type: "text"])
}
if (deviceIp != null) {
def v = deviceIp.trim()
if (v != deviceIp) device.updateSetting("deviceIp", [value: v, type: "text"])
}
}
private Integer safePort() {
try {
def cleaned = (brokerPort?.toString()?.replaceAll("[^0-9]", "") ?: "1883")
Integer p = cleaned as Integer
if (p <= 0 || p > 65535) return 1883
return p
} catch (ignored) {
return 1883
}
}
private void runInMillisSafe(Integer ms, String handler) {
try {
runInMillis(ms, handler)
} catch (ignored) {
Integer sec = Math.max(1, (int)Math.ceil((ms ?: 0) / 1000.0))
runIn(sec, handler)
}
}
private String formatEpoch(long epochMs) {
try {
return new Date(epochMs).format("yyyy-MM-dd HH:mm:ss", location.timeZone)
} catch (ignored) {
return new Date(epochMs).format("yyyy-MM-dd HH:mm:ss")
}
}
private static boolean asBool(def v) {
def s = (v == null ? "" : v.toString().trim().toLowerCase())
return (s in ["true", "on", "1", "\"true\"", "\"on\"", "\"1\""])
}
Hope this helps someone out there. Please modify as needed. Don't ask me how this works, it was generated using ChatGPT Codex.