bogdan Initial commit

7bb6a82 2 weeks ago History

module.exports = function(RED) {
  function Z2MSNZB05PHomeBusNode(config) {
    RED.nodes.createNode(this, config);
    var node = this;

    node.adapterId = "z2m-snzb-05p";
    node.sourceTopic = "zigbee2mqtt/SNZB-05P/#";
    node.subscriptionStarted = false;
    node.startTimer = null;
    node.site = normalizeToken(config.site || config.mqttSite);
    node.legacyRoom = normalizeToken(config.mqttRoom);
    node.legacySensor = normalizeToken(config.mqttSensor);
    node.publishCache = Object.create(null);
    node.retainedCache = Object.create(null);
    node.seenOptionalCapabilities = Object.create(null);
    node.detectedDevices = Object.create(null);
    node.stats = {
      processed_inputs: 0,
      devices_detected: 0,
      home_messages: 0,
      last_messages: 0,
      meta_messages: 0,
      home_availability_messages: 0,
      operational_messages: 0,
      invalid_messages: 0,
      invalid_topics: 0,
      invalid_payloads: 0,
      unmapped_messages: 0,
      adapter_exceptions: 0,
      errors: 0,
      dlq: 0
    };
    node.statsPublishEvery = 25;
    node.batteryLowThreshold = Number(config.batteryLowThreshold);
    if (!Number.isFinite(node.batteryLowThreshold)) node.batteryLowThreshold = 20;

    var CAPABILITY_MAPPINGS = [
      {
        sourceSystem: "zigbee2mqtt",
        sourceTopicMatch: "zigbee2mqtt/SNZB-05P/<site>/<location>/<device_id>",
        sourceFields: ["water_leak", "waterLeak", "leak"],
        targetBus: "home",
        targetCapability: "water_leak",
        core: true,
        stream: "value",
        payloadProfile: "scalar",
        dataType: "boolean",
        historianMode: "state",
        historianEnabled: true,
        read: function(payload) {
          return readBool(payload, this.sourceFields);
        }
      },
      {
        sourceSystem: "zigbee2mqtt",
        sourceTopicMatch: "zigbee2mqtt/SNZB-05P/<site>/<location>/<device_id>",
        sourceFields: ["battery"],
        targetBus: "home",
        targetCapability: "battery",
        core: true,
        stream: "value",
        payloadProfile: "scalar",
        dataType: "number",
        unit: "%",
        precision: 1,
        historianMode: "sample",
        historianEnabled: true,
        read: function(payload) {
          var value = readNumber(payload, this.sourceFields[0]);
          if (value === undefined) return undefined;
          return clamp(Math.round(value), 0, 100);
        }
      },
      {
        sourceSystem: "zigbee2mqtt",
        sourceTopicMatch: "zigbee2mqtt/SNZB-05P/<site>/<location>/<device_id>",
        sourceFields: ["battery_low", "batteryLow", "battery"],
        targetBus: "home",
        targetCapability: "battery_low",
        core: true,
        stream: "value",
        payloadProfile: "scalar",
        dataType: "boolean",
        historianMode: "state",
        historianEnabled: true,
        read: function(payload) {
          var raw = readBool(payload, this.sourceFields.slice(0, 2));
          if (raw !== undefined) return raw;
          var battery = readNumber(payload, this.sourceFields[2]);
          if (battery === undefined) return undefined;
          return battery <= node.batteryLowThreshold;
        }
      },
      {
        sourceSystem: "zigbee2mqtt",
        sourceTopicMatch: "zigbee2mqtt/SNZB-05P/<site>/<location>/<device_id>",
        sourceFields: ["tamper", "tampered", "tamper_alarm", "alarm_tamper"],
        targetBus: "home",
        targetCapability: "tamper",
        core: false,
        stream: "value",
        payloadProfile: "scalar",
        dataType: "boolean",
        historianMode: "state",
        historianEnabled: true,
        read: function(payload) {
          return readBool(payload, this.sourceFields);
        }
      }
    ];

    function normalizeToken(value) {
      if (value === undefined || value === null) return "";
      return String(value).trim();
    }

    function transliterate(value) {
      var s = normalizeToken(value);
      if (!s) return "";
      if (typeof s.normalize === "function") {
        s = s.normalize("NFKD").replace(/[\u0300-\u036f]/g, "");
      }
      return s;
    }

    function toKebabCase(value, fallback) {
      var s = transliterate(value).toLowerCase().replace(/[^a-z0-9]+/g, "-");
      s = s.replace(/^-+|-+$/g, "").replace(/-{2,}/g, "-");
      return s || fallback || "";
    }

    function clamp(n, min, max) {
      return Math.max(min, Math.min(max, n));
    }

    function topicTokens(topic) {
      if (typeof topic !== "string") return [];
      return topic.split("/").map(function(token) { return token.trim(); }).filter(function(token) { return !!token; });
    }

    function asBool(value) {
      if (typeof value === "boolean") return value;
      if (typeof value === "number") return value !== 0;
      if (typeof value === "string") {
        var v = value.trim().toLowerCase();
        if (v === "true" || v === "1" || v === "on" || v === "yes" || v === "online") return true;
        if (v === "false" || v === "0" || v === "off" || v === "no" || v === "offline") return false;
      }
      return null;
    }

    function pickFirst(obj, keys) {
      if (!obj || typeof obj !== "object") return undefined;
      for (var i = 0; i < keys.length; i++) {
        if (Object.prototype.hasOwnProperty.call(obj, keys[i])) return obj[keys[i]];
      }
      return undefined;
    }

    function pickPath(obj, path) {
      if (!obj || typeof obj !== "object") return undefined;
      var parts = path.split(".");
      var current = obj;
      for (var i = 0; i < parts.length; i++) {
        if (!current || typeof current !== "object" || !Object.prototype.hasOwnProperty.call(current, parts[i])) return undefined;
        current = current[parts[i]];
      }
      return current;
    }

    function pickFirstPath(obj, paths) {
      for (var i = 0; i < paths.length; i++) {
        var value = pickPath(obj, paths[i]);
        if (value !== undefined && value !== null && normalizeToken(value) !== "") return value;
      }
      return undefined;
    }

    function readBool(payload, fields) {
      var raw = asBool(pickFirst(payload, fields));
      return raw === null ? undefined : raw;
    }

    function readNumber(payload, field) {
      if (!payload || typeof payload !== "object") return undefined;
      var value = payload[field];
      if (typeof value !== "number" || !isFinite(value)) return undefined;
      return Number(value);
    }

    function toIsoTimestamp(value) {
      if (value === undefined || value === null) return "";
      if (value instanceof Date && !isNaN(value.getTime())) return value.toISOString();
      if (typeof value === "number" && isFinite(value)) {
        var ms = value < 100000000000 ? value * 1000 : value;
        var dateFromNumber = new Date(ms);
        return isNaN(dateFromNumber.getTime()) ? "" : dateFromNumber.toISOString();
      }
      if (typeof value === "string") {
        var trimmed = value.trim();
        if (!trimmed) return "";
        if (/^\d+$/.test(trimmed)) return toIsoTimestamp(Number(trimmed));
        var dateFromString = new Date(trimmed);
        return isNaN(dateFromString.getTime()) ? "" : dateFromString.toISOString();
      }
      return "";
    }

    function canonicalSourceTopic(topic) {
      var t = normalizeToken(topic);
      if (!t) return "";
      if (/\/availability$/i.test(t)) return t.replace(/\/availability$/i, "");
      return t;
    }

    function inferFromTopic(topic) {
      var tokens = topicTokens(topic);
      var result = {
        deviceType: "",
        site: "",
        location: "",
        deviceId: "",
        friendlyName: "",
        isAvailabilityTopic: false
      };

      if (tokens.length >= 2 && tokens[0].toLowerCase() === "zigbee2mqtt") {
        result.deviceType = tokens[1];

        if (tokens.length >= 6 && tokens[5].toLowerCase() === "availability") {
          result.site = tokens[2];
          result.location = tokens[3];
          result.deviceId = tokens[4];
          result.friendlyName = tokens.slice(1, 5).join("/");
          result.isAvailabilityTopic = true;
          return result;
        }

        if (tokens.length >= 5 && tokens[4].toLowerCase() !== "availability") {
          result.site = tokens[2];
          result.location = tokens[3];
          result.deviceId = tokens[4];
          result.friendlyName = tokens.slice(1, 5).join("/");
          return result;
        }

        result.friendlyName = tokens.slice(1).join("/");
        result.deviceId = tokens[1];
        result.isAvailabilityTopic = tokens.length >= 3 && tokens[tokens.length - 1].toLowerCase() === "availability";
        return result;
      }

      if (tokens.length >= 5 && tokens[1].toLowerCase() === "home") {
        result.site = tokens[0];
        result.location = tokens[2];
        result.deviceId = tokens[4];
        result.isAvailabilityTopic = tokens.length >= 6 && tokens[5].toLowerCase() === "availability";
        return result;
      }

      if (tokens.length >= 4) {
        result.site = tokens[0];
        result.location = tokens[2];
        result.deviceId = tokens[3];
      } else if (tokens.length >= 2) {
        result.location = tokens[tokens.length - 2];
        result.deviceId = tokens[tokens.length - 1];
      } else if (tokens.length === 1) {
        result.deviceId = tokens[0];
      }

      return result;
    }

    function resolveIdentity(msg, payload) {
      var safeMsg = (msg && typeof msg === "object") ? msg : {};
      var inferred = inferFromTopic(safeMsg.topic);
      var siteRaw = inferred.site
        || normalizeToken(pickFirst(payload, ["site", "homeSite"]))
        || normalizeToken(pickFirst(safeMsg, ["site", "homeSite"]))
        || node.site
        || "";

      var locationRaw = inferred.location
        || normalizeToken(pickFirst(payload, ["location", "room", "homeLocation", "mqttRoom"]))
        || normalizeToken(pickFirst(safeMsg, ["location", "room", "homeLocation", "mqttRoom"]))
        || node.legacyRoom
        || "";

      var deviceRaw = inferred.deviceId
        || normalizeToken(pickFirst(payload, ["deviceId", "device_id", "sensor", "mqttSensor", "friendly_name", "friendlyName", "name", "device"]))
        || normalizeToken(pickFirst(safeMsg, ["deviceId", "device_id", "sensor", "mqttSensor"]))
        || node.legacySensor
        || inferred.friendlyName
        || inferred.deviceType;

      var displayName = normalizeToken(pickFirst(payload, ["display_name", "displayName", "friendly_name", "friendlyName", "name"]))
        || normalizeToken(pickFirst(safeMsg, ["display_name", "displayName", "friendly_name", "friendlyName", "name"]))
        || inferred.friendlyName
        || deviceRaw
        || "SNZB-05P";

      var sourceRef = normalizeToken(pickFirstPath(payload, ["source_ref", "sourceRef", "ieee_address", "ieeeAddr", "device.ieee_address", "device.ieeeAddr"]))
        || normalizeToken(pickFirstPath(safeMsg, ["source_ref", "sourceRef", "ieee_address", "ieeeAddr", "device.ieee_address", "device.ieeeAddr"]))
        || canonicalSourceTopic(safeMsg.topic)
        || inferred.friendlyName
        || inferred.deviceType
        || deviceRaw
        || "z2m-snzb-05p";

      return {
        site: toKebabCase(siteRaw, "unknown"),
        location: toKebabCase(locationRaw, "unknown"),
        deviceId: toKebabCase(deviceRaw, toKebabCase(inferred.deviceType, "snzb-05p")),
        displayName: displayName,
        sourceRef: sourceRef,
        sourceTopic: canonicalSourceTopic(safeMsg.topic),
        isAvailabilityTopic: inferred.isAvailabilityTopic
      };
    }

    function noteDevice(identity) {
      if (!identity) return;
      var key = identity.site + "/" + identity.location + "/" + identity.deviceId;
      if (node.detectedDevices[key]) return;
      node.detectedDevices[key] = true;
      node.stats.devices_detected += 1;
    }

    function validateInboundTopic(topic) {
      var rawTopic = normalizeToken(topic);
      if (!rawTopic) {
        return {
          valid: false,
          reason: "Topic must be a non-empty string"
        };
      }

      var tokens = rawTopic.split("/").map(function(token) {
        return token.trim();
      });
      if (tokens.length < 5) {
        return {
          valid: false,
          reason: "Topic must match zigbee2mqtt/SNZB-05P/<site>/<location>/<device_id>[/availability]"
        };
      }
      if (tokens[0].toLowerCase() !== "zigbee2mqtt" || tokens[1].toLowerCase() !== "snzb-05p") {
        return {
          valid: false,
          reason: "Topic must start with zigbee2mqtt/SNZB-05P"
        };
      }
      if (!tokens[2] || !tokens[3] || !tokens[4]) {
        return {
          valid: false,
          reason: "Topic must contain non-empty site, location and device_id segments"
        };
      }
      if (tokens.length === 5) {
        return {
          valid: true,
          isAvailabilityTopic: false
        };
      }
      if (tokens.length === 6 && tokens[5].toLowerCase() === "availability") {
        return {
          valid: true,
          isAvailabilityTopic: true
        };
      }
      return {
        valid: false,
        reason: "Topic must not contain extra segments beyond an optional /availability suffix"
      };
    }

    function translatedMessageCount() {
      return node.stats.home_messages + node.stats.last_messages + node.stats.meta_messages + node.stats.home_availability_messages;
    }

    function updateNodeStatus(fill, shape, suffix) {
      var parts = [
        "dev " + node.stats.devices_detected,
        "in " + node.stats.processed_inputs,
        "tr " + translatedMessageCount()
      ];
      if (node.stats.operational_messages > 0) parts.push("op " + node.stats.operational_messages);
      if (node.stats.errors > 0) parts.push("err " + node.stats.errors);
      if (node.stats.invalid_topics > 0) parts.push("topic " + node.stats.invalid_topics);
      if (node.stats.invalid_messages > 0) parts.push("msg " + node.stats.invalid_messages);
      if (node.stats.invalid_payloads > 0) parts.push("payload " + node.stats.invalid_payloads);
      if (node.stats.unmapped_messages > 0) parts.push("unmapped " + node.stats.unmapped_messages);
      if (node.stats.adapter_exceptions > 0) parts.push("exc " + node.stats.adapter_exceptions);
      if (node.stats.dlq > 0) parts.push("dlq " + node.stats.dlq);
      if (suffix) parts.push(suffix);
      node.status({
        fill: fill,
        shape: shape,
        text: parts.join(" | ")
      });
    }

    function buildHomeTopic(identity, mapping, stream) {
      return identity.site + "/" + mapping.targetBus + "/" + identity.location + "/" + mapping.targetCapability + "/" + identity.deviceId + "/" + stream;
    }

    function buildSysTopic(site, stream) {
      return site + "/sys/adapter/" + node.adapterId + "/" + stream;
    }

    function makePublishMsg(topic, payload, retain) {
      return {
        topic: topic,
        payload: payload,
        qos: 1,
        retain: !!retain
      };
    }

    function makeSubscribeMsg(topic) {
      return {
        action: "subscribe",
        topic: [{
          topic: topic,
          qos: 2,
          rh: 0,
          rap: true
        }]
      };
    }

    function signature(value) {
      return JSON.stringify(value);
    }

    function shouldPublishLiveValue(cacheKey, payload) {
      var sig = signature(payload);
      if (node.publishCache[cacheKey] === sig) return false;
      node.publishCache[cacheKey] = sig;
      return true;
    }

    function shouldPublishRetained(cacheKey, payload) {
      var sig = signature(payload);
      if (node.retainedCache[cacheKey] === sig) return false;
      node.retainedCache[cacheKey] = sig;
      return true;
    }

    function humanizeCapability(capability) {
      return capability.replace(/_/g, " ").replace(/\b[a-z]/g, function(ch) { return ch.toUpperCase(); });
    }

    function resolveObservation(msg, payloadObject) {
      var safeMsg = (msg && typeof msg === "object") ? msg : {};
      var observedAtRaw = pickFirstPath(payloadObject, [
        "observed_at",
        "observedAt",
        "timestamp",
        "time",
        "ts",
        "last_seen",
        "lastSeen",
        "device.last_seen",
        "device.lastSeen"
      ]) || pickFirstPath(safeMsg, [
        "observed_at",
        "observedAt",
        "timestamp",
        "time",
        "ts"
      ]);
      var observedAt = toIsoTimestamp(observedAtRaw);
      if (observedAt) {
        return {
          observedAt: observedAt,
          quality: "good"
        };
      }
      return {
        observedAt: new Date().toISOString(),
        quality: "estimated"
      };
    }

    function buildMetaPayload(identity, mapping) {
      var payload = {
        schema_ref: "mqbus.home.v1",
        payload_profile: mapping.payloadProfile,
        stream_payload_profiles: {
          value: "scalar",
          last: "envelope"
        },
        data_type: mapping.dataType,
        adapter_id: node.adapterId,
        source: mapping.sourceSystem,
        source_ref: identity.sourceRef,
        source_topic: identity.sourceTopic,
        display_name: identity.displayName + " " + humanizeCapability(mapping.targetCapability),
        tags: [mapping.sourceSystem, "snzb-05p", mapping.targetBus],
        historian: {
          enabled: !!mapping.historianEnabled,
          mode: mapping.historianMode
        }
      };

      if (mapping.unit) payload.unit = mapping.unit;
      if (mapping.precision !== undefined) payload.precision = mapping.precision;

      return payload;
    }

    function buildLastPayload(value, observation) {
      var payload = {
        value: value,
        observed_at: observation.observedAt
      };
      if (observation.quality && observation.quality !== "good") payload.quality = observation.quality;
      return payload;
    }

    function noteMessage(kind) {
      if (!Object.prototype.hasOwnProperty.call(node.stats, kind)) return;
      node.stats[kind] += 1;
    }

    function noteErrorKind(code) {
      if (code === "invalid_message") {
        noteMessage("invalid_messages");
      } else if (code === "invalid_topic") {
        noteMessage("invalid_topics");
      } else if (code === "payload_not_object" || code === "invalid_availability_payload") {
        noteMessage("invalid_payloads");
      } else if (code === "no_mapped_fields") {
        noteMessage("unmapped_messages");
      } else if (code === "adapter_exception") {
        noteMessage("adapter_exceptions");
      }
    }

    function summarizeForLog(value) {
      if (value === undefined) return "undefined";
      if (value === null) return "null";
      if (typeof value === "string") {
        return value.length > 180 ? value.slice(0, 177) + "..." : value;
      }
      try {
        var serialized = JSON.stringify(value);
        if (serialized.length > 180) return serialized.slice(0, 177) + "...";
        return serialized;
      } catch (err) {
        return String(value);
      }
    }

    function logIssue(level, code, reason, msg, rawPayload) {
      var parts = [
        "[" + node.adapterId + "]",
        code + ":",
        reason
      ];
      var sourceTopic = msg && typeof msg === "object" ? normalizeToken(msg.topic) : "";
      if (sourceTopic) parts.push("topic=" + sourceTopic);
      if (rawPayload !== undefined) parts.push("payload=" + summarizeForLog(rawPayload));

      var text = parts.join(" ");
      if (level === "error") {
        node.error(text, msg);
        return;
      }
      node.warn(text);
    }

    function enqueueHomeMeta(messages, identity, mapping) {
      var topic = buildHomeTopic(identity, mapping, "meta");
      var payload = buildMetaPayload(identity, mapping);
      if (!shouldPublishRetained("meta:" + topic, payload)) return;
      messages.push(makePublishMsg(topic, payload, true));
      noteMessage("meta_messages");
    }

    function enqueueHomeAvailability(messages, identity, mapping, online) {
      var topic = buildHomeTopic(identity, mapping, "availability");
      var payload = online ? "online" : "offline";
      if (!shouldPublishRetained("availability:" + topic, payload)) return;
      messages.push(makePublishMsg(topic, payload, true));
      noteMessage("home_availability_messages");
    }

    function enqueueHomeLast(messages, identity, mapping, value, observation) {
      var topic = buildHomeTopic(identity, mapping, "last");
      var payload = buildLastPayload(value, observation);
      if (!shouldPublishRetained("last:" + topic, payload)) return;
      messages.push(makePublishMsg(topic, payload, true));
      noteMessage("last_messages");
    }

    function enqueueHomeValue(messages, identity, mapping, value) {
      var topic = buildHomeTopic(identity, mapping, "value");
      if (!shouldPublishLiveValue("value:" + topic, value)) return;
      messages.push(makePublishMsg(topic, value, false));
      noteMessage("home_messages");
    }

    function enqueueAdapterAvailability(messages, site, online) {
      var topic = buildSysTopic(site, "availability");
      var payload = online ? "online" : "offline";
      if (!shouldPublishRetained("sys:availability:" + topic, payload)) return;
      messages.push(makePublishMsg(topic, payload, true));
      noteMessage("operational_messages");
    }

    function enqueueAdapterStats(messages, site, force) {
      if (!force && node.stats.processed_inputs !== 1 && (node.stats.processed_inputs % node.statsPublishEvery) !== 0) return;
      var topic = buildSysTopic(site, "stats");
      var payload = {
        processed_inputs: node.stats.processed_inputs,
        devices_detected: node.stats.devices_detected,
        translated_messages: translatedMessageCount(),
        home_messages: node.stats.home_messages,
        last_messages: node.stats.last_messages,
        meta_messages: node.stats.meta_messages,
        home_availability_messages: node.stats.home_availability_messages,
        operational_messages: node.stats.operational_messages,
        invalid_messages: node.stats.invalid_messages,
        invalid_topics: node.stats.invalid_topics,
        invalid_payloads: node.stats.invalid_payloads,
        unmapped_messages: node.stats.unmapped_messages,
        adapter_exceptions: node.stats.adapter_exceptions,
        errors: node.stats.errors,
        dlq: node.stats.dlq
      };
      messages.push(makePublishMsg(topic, payload, true));
      noteMessage("operational_messages");
    }

    function enqueueError(messages, site, code, reason, sourceTopic) {
      var payload = {
        code: code,
        reason: reason,
        source_topic: normalizeToken(sourceTopic),
        adapter_id: node.adapterId
      };
      messages.push(makePublishMsg(buildSysTopic(site, "error"), payload, false));
      noteErrorKind(code);
      noteMessage("errors");
      noteMessage("operational_messages");
    }

    function enqueueDlq(messages, site, code, sourceTopic, rawPayload) {
      var payload = {
        code: code,
        source_topic: normalizeToken(sourceTopic),
        payload: rawPayload
      };
      messages.push(makePublishMsg(buildSysTopic(site, "dlq"), payload, false));
      noteMessage("dlq");
      noteMessage("operational_messages");
    }

    function activeMappings(payloadObject) {
      var result = [];
      for (var i = 0; i < CAPABILITY_MAPPINGS.length; i++) {
        var mapping = CAPABILITY_MAPPINGS[i];
        if (typeof mapping.applies === "function" && !mapping.applies(payloadObject)) continue;
        var value = payloadObject ? mapping.read(payloadObject) : undefined;
        var seen = mapping.core || node.seenOptionalCapabilities[mapping.targetCapability];
        if (value !== undefined) node.seenOptionalCapabilities[mapping.targetCapability] = true;
        if (mapping.core || seen || value !== undefined) {
          result.push({
            mapping: mapping,
            hasValue: value !== undefined,
            value: value
          });
        }
      }
      return result;
    }

    function resolveOnlineState(payloadObject, availabilityValue) {
      if (availabilityValue !== null && availabilityValue !== undefined) return !!availabilityValue;
      if (!payloadObject) return true;
      var active = true;
      if (typeof payloadObject.availability === "string") active = payloadObject.availability.trim().toLowerCase() !== "offline";
      if (typeof payloadObject.online === "boolean") active = payloadObject.online;
      return active;
    }

    function flush(send, done, controlMessages, publishMessages, error) {
      send([
        publishMessages.length ? publishMessages : null,
        controlMessages.length ? controlMessages : null
      ]);
      if (done) done(error);
    }

    function startSubscriptions() {
      if (node.subscriptionStarted) return;
      node.subscriptionStarted = true;
      node.send([null, makeSubscribeMsg(node.sourceTopic)]);
      updateNodeStatus("grey", "dot", "subscribed");
    }

    node.on("input", function(msg, send, done) {
      send = send || function() { node.send.apply(node, arguments); };
      node.stats.processed_inputs += 1;

      try {
        var controlMessages = [];
        if (!msg || typeof msg !== "object") {
          var invalidMessages = [];
          var invalidSite = resolveIdentity({}, null).site;
          enqueueAdapterAvailability(invalidMessages, invalidSite, true);
          enqueueError(invalidMessages, invalidSite, "invalid_message", "Input must be an object", "");
          enqueueDlq(invalidMessages, invalidSite, "invalid_message", "", null);
          enqueueAdapterStats(invalidMessages, invalidSite, true);
          logIssue("warn", "invalid_message", "Input must be an object", null, msg);
          updateNodeStatus("yellow", "ring", "bad msg");
          flush(send, done, controlMessages, invalidMessages);
          return;
        }

        var payload = msg.payload;
        var payloadObject = payload && typeof payload === "object" && !Array.isArray(payload) ? payload : null;
        var identity = resolveIdentity(msg, payloadObject);
        var observation = resolveObservation(msg, payloadObject);
        var messages = [];
        var topicValidation = validateInboundTopic(msg.topic);
        var availabilityValue = null;

        enqueueAdapterAvailability(messages, identity.site, true);

        if (!topicValidation.valid) {
          enqueueError(messages, identity.site, "invalid_topic", topicValidation.reason, msg.topic);
          enqueueDlq(messages, identity.site, "invalid_topic", msg.topic, payload);
          enqueueAdapterStats(messages, identity.site, true);
          logIssue("warn", "invalid_topic", topicValidation.reason, msg, payload);
          updateNodeStatus("yellow", "ring", "bad topic");
          flush(send, done, controlMessages, messages);
          return;
        }

        identity.isAvailabilityTopic = topicValidation.isAvailabilityTopic;
        noteDevice(identity);

        if (topicValidation.isAvailabilityTopic) {
          availabilityValue = asBool(payload);
          if (availabilityValue === null) {
            enqueueError(messages, identity.site, "invalid_availability_payload", "Availability payload must be online/offline or boolean", msg.topic);
            enqueueDlq(messages, identity.site, "invalid_availability_payload", msg.topic, payload);
            enqueueAdapterStats(messages, identity.site, true);
            logIssue("warn", "invalid_availability_payload", "Availability payload must be online/offline or boolean", msg, payload);
            updateNodeStatus("yellow", "ring", "invalid availability");
            flush(send, done, controlMessages, messages);
            return;
          }
        } else if (!payloadObject) {
          enqueueError(messages, identity.site, "payload_not_object", "Telemetry payload must be an object", msg.topic);
          enqueueDlq(messages, identity.site, "payload_not_object", msg.topic, payload);
          enqueueAdapterStats(messages, identity.site, true);
          logIssue("warn", "payload_not_object", "Telemetry payload must be an object", msg, payload);
          updateNodeStatus("yellow", "ring", "payload not object");
          flush(send, done, controlMessages, messages);
          return;
        }

        var online = resolveOnlineState(payloadObject, availabilityValue);
        var mappings = activeMappings(payloadObject);
        var hasMappedValue = false;
        for (var i = 0; i < mappings.length; i++) {
          if (mappings[i].hasValue) {
            hasMappedValue = true;
            break;
          }
        }
        var hasAvailabilityField = topicValidation.isAvailabilityTopic || (payloadObject && (typeof payloadObject.availability === "string" || typeof payloadObject.online === "boolean"));

        if (!hasMappedValue && !hasAvailabilityField) {
          enqueueError(messages, identity.site, "no_mapped_fields", "Payload did not contain any supported SNZB-05P fields", msg.topic);
          enqueueDlq(messages, identity.site, "no_mapped_fields", msg.topic, payloadObject);
          logIssue("warn", "no_mapped_fields", "Payload did not contain any supported SNZB-05P fields", msg, payloadObject);
        }

        for (var j = 0; j < mappings.length; j++) {
          enqueueHomeMeta(messages, identity, mappings[j].mapping);
          enqueueHomeAvailability(messages, identity, mappings[j].mapping, online);
          if (mappings[j].hasValue) {
            enqueueHomeLast(messages, identity, mappings[j].mapping, mappings[j].value, observation);
            enqueueHomeValue(messages, identity, mappings[j].mapping, mappings[j].value);
          }
        }

        enqueueAdapterStats(messages, identity.site, false);

        if (!hasMappedValue && !hasAvailabilityField) {
          updateNodeStatus("yellow", "ring", "unmapped");
        } else {
          updateNodeStatus(online ? "green" : "yellow", online ? "dot" : "ring", online ? null : "offline");
        }

        flush(send, done, controlMessages, messages);
      } catch (err) {
        var errorPayload = msg && msg.payload && typeof msg.payload === "object" && !Array.isArray(msg.payload) ? msg.payload : null;
        var errorIdentity = resolveIdentity(msg, errorPayload);
        noteDevice(errorIdentity);
        var errorMessages = [];
        enqueueAdapterAvailability(errorMessages, errorIdentity.site, true);
        enqueueError(errorMessages, errorIdentity.site, "adapter_exception", err.message, msg && msg.topic);
        enqueueAdapterStats(errorMessages, errorIdentity.site, true);
        logIssue("error", "adapter_exception", err.message, msg, msg && msg.payload);
        updateNodeStatus("red", "ring", "error");
        flush(send, done, [], errorMessages, err);
      }
    });

    node.on("close", function() {
      if (node.startTimer) {
        clearTimeout(node.startTimer);
        node.startTimer = null;
      }
    });

    node.startTimer = setTimeout(startSubscriptions, 250);
    updateNodeStatus("grey", "ring", "waiting");
  }

  RED.nodes.registerType("z2m-snzb-05p-homebus", Z2MSNZB05PHomeBusNode);
};