Local Metrics & Logic

[
  {
    "id": "tab_1",
    "type": "tab",
    "label": "Local Metrics & Logic",
    "disabled": false,
    "info": "",
    "env": []
  },
  {
    "id": "node_1",
    "type": "inject",
    "z": "tab_1",
    "name": "Poll Sensor API",
    "props": [
      {
        "p": "payload"
      },
      {
        "p": "topic",
        "vt": "str"
      }
    ],
    "repeat": "60",
    "crontab": "",
    "once": true,
    "onceDelay": 0.1,
    "topic": "",
    "payload": "",
    "payloadType": "date",
    "x": 120,
    "y": 580,
    "wires": [
      [
        "node_2"
      ]
    ]
  },
  {
    "id": "node_2",
    "type": "http request",
    "z": "tab_1",
    "name": "Get Channels from API",
    "method": "GET",
    "ret": "txt",
    "paytoqs": "ignore",
    "url": "https://api.example.com/channels?account_key=REDACTED_ACCOUNT_KEY",
    "tls": "",
    "persist": false,
    "proxy": "",
    "insecureHTTPParser": false,
    "authType": "",
    "senderr": false,
    "headers": [],
    "x": 320,
    "y": 580,
    "wires": [
      [
        "node_3"
      ]
    ]
  },
  {
    "id": "node_3",
    "type": "json",
    "z": "tab_1",
    "name": "Parse JSON",
    "property": "payload",
    "action": "",
    "pretty": false,
    "x": 510,
    "y": 580,
    "wires": [
      [
        "node_4"
      ]
    ]
  },
  {
    "id": "node_5",
    "type": "influxdb batch",
    "z": "tab_1",
    "influxdb": "influxdb_node",
    "precision": "ns",
    "retentionPolicy": "",
    "name": "Batch Push to InfluxDB",
    "database": "metrics_db",
    "precisionV18FluxV20": "ns",
    "retentionPolicyV18Flux": "",
    "x": 1630,
    "y": 560,
    "wires": []
  },
  {
    "id": "node_6",
    "type": "switch",
    "z": "tab_1",
    "name": "Zone Router",
    "property": "payload.tags.zone",
    "propertyType": "msg",
    "rules": [
      {
        "t": "eq",
        "v": "zone_weather",
        "vt": "str"
      },
      {
        "t": "eq",
        "v": "zone_prod1",
        "vt": "str"
      },
      {
        "t": "eq",
        "v": "zone_prod2",
        "vt": "str"
      },
      {
        "t": "eq",
        "v": "zone_prep",
        "vt": "str"
      },
      {
        "t": "eq",
        "v": "zone_server",
        "vt": "str"
      },
      {
        "t": "eq",
        "v": "zone_equipment",
        "vt": "str"
      },
      {
        "t": "else"
      }
    ],
    "checkall": "false",
    "repair": false,
    "outputs": 7,
    "x": 1130,
    "y": 560,
    "wires": [
      [
        "node_7"
      ],
      [
        "node_7"
      ],
      [
        "node_7"
      ],
      [
        "node_7"
      ],
      [
        "node_7"
      ],
      [
        "node_7"
      ],
      [
        "node_7"
      ]
    ]
  },
  {
    "id": "node_8",
    "type": "function",
    "z": "tab_1",
    "name": "Split Measurements",
    "func": "// Split the payload array into separate messages with parts for Join\nvar msgs = [];\nvar originalId = msg._msgid; // Keep the original msg ID\nvar count = msg.payload.length;\nfor (let i = 0; i < count; i++) {\n msgs.push({\n payload: msg.payload[i],\n parts: {\n id: originalId,\n index: i,\n count: count,\n type: 'array'\n }\n });\n}\nreturn [msgs];",
    "outputs": 1,
    "timeout": 0,
    "noerr": 0,
    "initialize": "",
    "finalize": "",
    "libs": [],
    "x": 940,
    "y": 560,
    "wires": [
      [
        "node_6"
      ]
    ]
  },
  {
    "id": "node_7",
    "type": "join",
    "z": "tab_1",
    "name": "Compile Batch Report",
    "mode": "auto",
    "build": "object",
    "property": "payload",
    "propertyType": "msg",
    "key": "topic",
    "joiner": "\\n",
    "joinerType": "str",
    "useparts": false,
    "accumulate": true,
    "timeout": "",
    "count": "",
    "reduceRight": false,
    "reduceExp": "",
    "reduceInit": "",
    "reduceInitType": "",
    "reduceFixup": "",
    "x": 1400,
    "y": 560,
    "wires": [
      [
        "node_5"
      ]
    ]
  },
  {
    "id": "node_9",
    "type": "mqtt out",
    "z": "tab_1",
    "name": "MQTT Push to HA",
    "topic": "",
    "qos": "0",
    "retain": "",
    "respTopic": "",
    "contentType": "",
    "userProps": "",
    "correl": "",
    "expiry": "",
    "broker": "mqtt_node",
    "x": 930,
    "y": 600,
    "wires": []
  },
  {
    "id": "node_10",
    "type": "inject",
    "z": "tab_1",
    "name": "Derive Wind&Moon - Trigger",
    "props": [
      {
        "p": "payload"
      }
    ],
    "repeat": "60",
    "crontab": "",
    "once": true,
    "onceDelay": 0.1,
    "topic": "",
    "payload": "",
    "payloadType": "date",
    "x": 170,
    "y": 640,
    "wires": [
      [
        "node_11"
      ]
    ]
  },
  {
    "id": "node_11",
    "type": "function",
    "z": "tab_1",
    "name": "Derive Wind & Moon Sensors",
    "func": "// Node-RED function: \"Derive Wind & Moon Sensors\"\n// This function expects global.sensorData to contain the latest weather fields.\n// Outputs:\n// 1 -> Discovery messages (array of discovery msg objects) [wired to MQTT Push to HA]\n// 2 -> State messages (array of state msg objects) [wired to MQTT Push to HA]\n\n// --- read sensor data stored by your flow ---\nconst fields = global.get('sensorData') || {};\n\n// helper: convert degrees to cardinal (16 points for detail)\nfunction degToCardinal(deg) {\n if (typeof deg !== 'number' || isNaN(deg)) return 'VAR';\n const dirs = ['N','NNE','NE','ENE','E','ESE','SE','SSE','S','SSW','SW','WSW','W','WNW','NW','NNW'];\n const idx = Math.round(((deg % 360) / 22.5) % 16);\n return dirs[idx];\n}\n\n// helper: moon phase (8-phase)\nfunction getMoonPhase() {\n const now = new Date();\n const baseDate = new Date(2001, 0, 1);\n const diff = now.getTime() - baseDate.getTime();\n const days = diff / (1000 * 60 * 60 * 24);\n const lunations = 0.20439731 + (days * 0.03386319269);\n const pos = lunations % 1;\n const index = Math.floor(pos * 8 + 0.5) % 8;\n const phases = [\"New Moon\",\"Waxing Crescent\",\"First Quarter\",\"Waxing Gibbous\",\"Full Moon\",\"Waning Gibbous\",\"Last Quarter\",\"Waning Crescent\"];\n return phases[index];\n}\n\nlet degHistory = flow.get('wind_deg_history') || [];\nconst wdCandidates = ['wind_direction','winddir','wind_dir','winddir_deg'];\nlet rawDeg = NaN;\nfor (let k of wdCandidates) {\n if (k in fields && fields[k] !== null && fields[k] !== undefined) {\n const v = parseFloat(fields[k]);\n if (!isNaN(v)) { rawDeg = v; break;\n }\n }\n}\nif (!isNaN(rawDeg)) {\n degHistory.push(rawDeg);\n if (degHistory.length > 5) degHistory.shift();\n flow.set('wind_deg_history', degHistory);\n}\nconst smoothedDeg = degHistory.length > 0 ? degHistory.reduce((a, b) => a + b, 0) / degHistory.length : rawDeg;\nconst cardinal = degToCardinal(smoothedDeg);\nconst moonPhase = getMoonPhase();\n\nlet speedHistory = flow.get('wind_speed_history') || [];\nconst currentSpeed = parseFloat(fields.wind_speed) || 0;\nconst currentTime = Date.now();\nif (!isNaN(currentSpeed)) {\n speedHistory.push({ time: currentTime, speed: currentSpeed });\n speedHistory = speedHistory.filter(entry => currentTime - entry.time <= 600000);\n flow.set('wind_speed_history', speedHistory);\n}\nconst derivedGust = speedHistory.length > 0 ? Math.max(...speedHistory.map(entry => entry.speed)) : currentSpeed;\n\nconst discoverySentWind = flow.get('discovery_sent_wind_cardinal') || false;\nconst discoverySentWindSmooth = flow.get('discovery_sent_wind_smooth') || false;\nconst discoverySentWindGust = flow.get('discovery_sent_wind_gust') || false;\nconst discoverySentMoon = flow.get('discovery_sent_moon_phase') || false;\n\nconst arrayForDiscovery = [];\nif (!discoverySentWind) {\n const payload = {\n name: \"Sensor Node Wind Direction (Cardinal)\",\n unique_id: \"sensor_node_wind_dir_letter\",\n object_id: \"sensor_node_wind_dir_letter\",\n state_topic: \"nodered/zone_weather/wind_dir_letter\",\n icon: \"mdi:compass\",\n device: {\n identifiers: [\"sensor_node\"],\n name: \"Sensor Node\",\n manufacturer: \"SensorVendor\"\n }\n };\n arrayForDiscovery.push({ topic: 'homeassistant/sensor/sensor_node_wind_dir_letter/config', payload: JSON.stringify(payload), retain: true });\n flow.set('discovery_sent_wind_cardinal', true);\n}\nif (!discoverySentWindSmooth) {\n const payload = {\n name: \"Sensor Node Wind Direction (Smoothed)\",\n unique_id: \"sensor_node_wind_direction_smoothed\",\n object_id: \"sensor_node_wind_direction_smoothed\",\n state_topic: \"nodered/zone_weather/wind_direction_smoothed\",\n unit_of_measurement: \"°\",\n icon: \"mdi:compass\",\n device: {\n identifiers: [\"sensor_node\"],\n name: \"Sensor Node\",\n manufacturer: \"SensorVendor\"\n }\n };\n arrayForDiscovery.push({ topic: 'homeassistant/sensor/sensor_node_wind_direction_smoothed/config', payload: JSON.stringify(payload), retain: true });\n flow.set('discovery_sent_wind_smooth', true);\n}\nif (!discoverySentWindGust) {\n const payload = {\n name: \"Sensor Node Wind Gust (Derived)\",\n unique_id: \"sensor_node_wind_gust_derived\",\n object_id: \"sensor_node_wind_gust_derived\",\n state_topic: \"nodered/zone_weather/wind_gust_derived\",\n unit_of_measurement: \"km/h\",\n device_class: \"wind_speed\",\n icon: \"mdi:weather-windy-variant\",\n device: {\n identifiers: [\"sensor_node\"],\n name: \"Sensor Node\",\n manufacturer: \"SensorVendor\"\n }\n };\n arrayForDiscovery.push({ topic: 'homeassistant/sensor/sensor_node_wind_gust_derived/config', payload: JSON.stringify(payload), retain: true });\n flow.set('discovery_sent_wind_gust', true);\n}\nif (!discoverySentMoon) {\n const payload = {\n name: \"Sensor Node Moon Phase\",\n unique_id: \"sensor_node_moon_phase\",\n object_id: \"sensor_node_moon_phase\",\n state_topic: \"nodered/zone_weather/moon_phase\",\n icon: \"mdi:weather-night\",\n device: {\n identifiers: [\"sensor_node\"],\n name: \"Sensor Node\",\n manufacturer: \"SensorVendor\"\n }\n };\n arrayForDiscovery.push({ topic: 'homeassistant/sensor/sensor_node_moon_phase/config', payload: JSON.stringify(payload), retain: true });\n flow.set('discovery_sent_moon_phase', true);\n}\n\nconst arrayForState = [];\narrayForState.push({ topic: 'nodered/zone_weather/wind_dir_letter', payload: String(cardinal) });\narrayForState.push({ topic: 'nodered/zone_weather/wind_direction_smoothed', payload: String(smoothedDeg) });\narrayForState.push({ topic: 'nodered/zone_weather/wind_gust_derived', payload: String(derivedGust) });\narrayForState.push({ topic: 'nodered/zone_weather/moon_phase', payload: String(moonPhase) });\n\nreturn [ (arrayForDiscovery.length ? arrayForDiscovery : null), arrayForState ];",
    "outputs": 2,
    "timeout": 0,
    "noerr": 0,
    "initialize": "",
    "finalize": "",
    "libs": [],
    "x": 450,
    "y": 640,
    "wires": [
      [
        "node_9"
      ],
      [
        "node_9"
      ]
    ]
  },
  {
    "id": "node_4",
    "type": "function",
    "z": "tab_1",
    "name": "Process Sensor Data",
    "func": "// Process Sensor Data (sanitized)\n// Output 1: Influx payload (only when new timestamp / new data for channel)\n// Output 2: MQTT discovery + state messages for Home Assistant (always sends updated values on change)\n\nfunction getSensorDetails(key) {\n  const details = {\n    'temperature': { name: 'Temperature', class: 'temperature', unit: '°C' },\n    'r_h': { name: 'Humidity', class: 'humidity', unit: '%' },\n    'air_pressure': { name: 'Air Pressure', class: 'pressure', unit: 'hPa' },\n    'wind_speed': { name: 'Wind Speed', class: 'wind_speed', unit: 'km/h' },\n    'wind_direction': { name: 'Wind Direction', class: 'None', unit: '°' },\n    'rainfall': { name: 'Rainfall Total', class: 'precipitation', unit: 'mm' },\n    'rain_delta': { name: 'Rainfall Delta', class: 'precipitation', unit: 'mm' },\n    'rain_rate': { name: 'Rainfall Rate', class: 'precipitation_intensity', unit: 'mm/h' },\n    'dew_point_temp': { name: 'Dew Point', class: 'temperature', unit: '°C' },\n    'external_light': { name: 'Light Level', class: 'illuminance', unit: 'lx' },\n    'socket_power': { name: 'Power', class: 'power', unit: 'W' },\n    'socket_voltage': { name: 'Voltage', class: 'voltage', unit: 'V' },\n    'socket_current': { name: 'Current', class: 'current', unit: 'A' },\n    'switch_status': { name: 'Switch', class: 'switch', unit: '' },\n    'in_use': { name: 'In Use', class: 'power' },\n    'optimal_state': { name: 'Optimal State', class: 'None' },\n    'storm_incoming': { name: 'Storm Incoming', class: 'problem' },\n    'fog_risk': { name: 'Fog Risk', class: 'problem' },\n    'frost_risk': { name: 'Frost Risk', class: 'problem' },\n    'wind_damage_risk': { name: 'Wind Damage Risk', class: 'problem' },\n    'check_cooker': { name: 'Check Cooker', class: 'problem' },\n    'check_humidifier': { name: 'Check Humidifier', class: 'problem' },\n    'temp_alert_level': { name: 'Temperature Alert Level', class: 'None' },\n    'rh_alert_level': { name: 'Humidity Alert Level', class: 'None' },\n    'co2_alert_level': { name: 'CO2 Alert Level', class: 'None' },\n    'toxic_hazard_level': { name: 'Toxic Hazard Level', class: 'None' }\n  };\n  return details[key] || { name: key.replace(/_/g, ' ').replace(/\\b\\w/g, l => l.toUpperCase()), class: 'None', unit: '' };\n}\n\nif (!msg.payload || msg.payload.result !== \"success\") {\n  return null;\n}\n\nvar lastTimestamps = flow.get('lastTimestamps') || {};\nvar discoverySent = flow.get('discoverySent') || {};\nvar prevStates = flow.get('lastStates') || {};\n\nvar channels = msg.payload.channels || [];\nvar measurementsForInflux = [];\nvar messagesForHA = [];\n\nfor (var c = 0; c < channels.length; c++) {\n  var channel = channels[c];\n  var channel_id = channel.channel_id;\n  var room = channel.name || 'unknown';\n\n  // timestamp (use channel.last_entry_date if present, otherwise now)\n  var timestamp = channel.last_entry_date ? new Date(channel.last_entry_date).getTime() * 1000000 : Date.now() * 1000000;\n  var isNewTimestamp = (lastTimestamps[channel_id] !== timestamp);\n  if (isNewTimestamp) {\n    lastTimestamps[channel_id] = timestamp;\n  }\n\n  // parse last_values robustly (coerce numeric strings to numbers)\n  var last_values = {};\n  try {\n    last_values = JSON.parse(channel.last_values || \"{}\");\n  } catch (e) {\n    last_values = {};\n  }\n\n  var zone = (room || 'unknown').toLowerCase().trim().replace(/ /g, '_');\n  var fields = {};\n  for (var i = 1; i <= 15; i++) {\n    var label = channel['field' + i];\n    if (!label) continue;\n    var value_obj = last_values['field' + i];\n    if (!value_obj || value_obj.value === undefined || value_obj.value === null) continue;\n\n    var normalized_label = label.toLowerCase().replace(/[/ ]/g, '_');\n\n    var coerced = Number(value_obj.value);\n    if (!isNaN(coerced)) {\n      fields[normalized_label] = coerced;\n    } else {\n      fields[normalized_label] = value_obj.value;\n    }\n  }\n\n  // Correct a common pressure scale issue\n  if ('air_pressure' in fields && fields.air_pressure > 50 && fields.air_pressure < 200) {\n    fields.air_pressure = fields.air_pressure * 10;\n  }\n  // Explicitly set wind_speed to 0 if the field is not present.\n  if (!('wind_speed' in fields)) {\n    fields.wind_speed = 0;\n  }\n\n  if (zone === 'zone_weather' && 'wind_speed' in fields) {\n    fields.wind_speed = fields.wind_speed * 3.6;\n  }\n\n  if (zone === 'zone_weather') {\n    var syncMinutes = 1;\n    var reported = 0;\n\n    if ('rainfall' in fields) {\n      reported = Number(fields.rainfall) || 0;\n    } else if ('rain' in fields) {\n      reported = Number(fields.rain) || 0;\n    } else if ('rain_mm' in fields) {\n      reported = Number(fields.rain_mm) || 0;\n    } else {\n      reported = 0;\n    }\n\n    fields.rain_delta = reported;\n    fields.rain_rate = (reported > 0) ? (reported * (60 / syncMinutes)) : 0;\n    global.set('sensorData', fields);\n  }\n\n  var derived = {};\n  if (zone === 'zone_weather') {\n    var dryTemp = (fields.dry_bulb_temp !== undefined) ? fields.dry_bulb_temp : (fields.temperature !== undefined ? fields.temperature : undefined);\n    derived.fog_risk = !!(fields.r_h > 95 && (fields.wind_speed || 0) < 5 && dryTemp !== undefined && Math.abs(dryTemp - (fields.dew_point_temp || 0)) < 2);\n    derived.frost_risk = !!(dryTemp !== undefined && dryTemp < 2);\n    derived.wind_damage_risk = !!((fields.wind_speed || 0) > 40);\n  }\n\n  var allMetrics = Object.assign({}, fields, derived);\n\n  for (const [key, value] of Object.entries(allMetrics)) {\n    const details = getSensorDetails(key);\n    const component = (typeof value === 'boolean') ? 'binary_sensor' : 'sensor';\n    const sensorId = `sensor_node_${zone}_${key}`;\n    const stateTopic = `nodered/${zone}/${key}`;\n\n    if (!discoverySent[sensorId]) {\n      const discoveryPayload = {\n        name: `${room} ${details.name}`,\n        unique_id: sensorId,\n        object_id: sensorId,\n        state_topic: stateTopic,\n        device_class: (details.class && details.class !== 'None') ? details.class : undefined,\n        unit_of_measurement: details.unit && details.unit !== '' ? details.unit : undefined,\n        state_class: (typeof value === 'number') ? 'measurement' : undefined,\n        device: {\n          identifiers: [`sensor_node_${channel_id}`],\n          name: room,\n          manufacturer: 'SensorVendor'\n        }\n      };\n      messagesForHA.push({ topic: `homeassistant/${component}/${sensorId}/config`, payload: JSON.stringify(discoveryPayload), retain: true });\n      discoverySent[sensorId] = true;\n    }\n\n    let stateValue = (typeof value === 'boolean') ? (value ? 'ON' : 'OFF') : value;\n    const prev = prevStates[sensorId];\n    const changed = String(prev) !== String(stateValue);\n    if (isNewTimestamp || changed) {\n      messagesForHA.push({ topic: stateTopic, payload: String(stateValue) });\n      prevStates[sensorId] = stateValue;\n    }\n  }\n\n  if (isNewTimestamp && Object.keys(fields).length > 0) {\n    let influxMetrics = Object.assign({}, allMetrics);\n    measurementsForInflux.push({\n      measurement: \"sensor_metrics\",\n      tags: { room: room || \"unknown\", zone: zone, channel_id: channel_id },\n      fields: influxMetrics,\n      timestamp: timestamp\n    });\n  }\n}\n\nflow.set('lastTimestamps', lastTimestamps);\nflow.set('discoverySent', discoverySent);\nflow.set('lastStates', prevStates);\n\nvar out1 = (measurementsForInflux.length > 0) ? { payload: measurementsForInflux } : null;\nvar out2 = (messagesForHA.length > 0) ? messagesForHA : null;\n\nreturn [out1, out2];",
    "outputs": 2,
    "timeout": 0,
    "noerr": 0,
    "initialize": "",
    "finalize": "",
    "libs": [],
    "x": 700,
    "y": 580,
    "wires": [
      [
        "node_8"
      ],
      [
        "node_9"
      ]
    ]
  },
  {
    "id": "influxdb_node",
    "type": "influxdb",
    "hostname": "influxdb-host",
    "port": "8086",
    "protocol": "http",
    "database": "metrics_db",
    "name": "InfluxDB",
    "usetls": false,
    "tls": "",
    "influxdbVersion": "1.x",
    "url": "http://influxdb-host:8086",
    "timeout": "10",
    "rejectUnauthorized": false
  },
  {
    "id": "mqtt_node",
    "type": "mqtt-broker",
    "name": "MQTT Broker",
    "broker": "mqtt-broker-host",
    "port": "1883",
    "clientid": "",
    "autoConnect": true,
    "usetls": false,
    "protocolVersion": "5",
    "keepalive": "60",
    "cleansession": true,
    "autoUnsubscribe": true,
    "birthTopic": "",
    "birthQos": "0",
    "birthRetain": "false",
    "birthPayload": "",
    "birthMsg": {},
    "closeTopic": "",
    "closeQos": "0",
    "closeRetain": "false",
    "closePayload": "",
    "closeMsg": {},
    "willTopic": "",
    "willQos": "0",
    "willRetain": "false",
    "willPayload": "",
    "willMsg": {},
    "userProps": "",
    "sessionExpiry": ""
  }
]