node-red-contrib-myio-data-fetcher

A Node-RED node that listens to MQTT messages for energy and water timestamp requests, queries PostgreSQL/TimescaleDB for aggregated data, and publishes results back to MQTT topics.

Features

• MQTT-to-MQTT Processing: Standalone node that processes requests and publishes responses via MQTT
• PostgreSQL/TimescaleDB Integration: Executes complex time-series queries with time bucketing
• Multiplier Support: Automatically extracts and applies multipliers from channel names (e.g., "Water Meter x100")
• Health Monitoring: Visual status indicators and periodic MQTT status heartbeats
• Gap Detection: Handles missing data with future timestamp validation
• Auto-Reconnection: Automatic reconnection for both MQTT and PostgreSQL connections
• No Outputs: Pure MQTT communication, no Node-RED wiring needed

Installation

npm install node-red-contrib-myio-data-fetcher

Configuration

Environment Variables Required

• CENTRAL_UUID - Gateway UUID for MQTT topics
• DB_HOST - PostgreSQL host
• DB_PORT - PostgreSQL port
• DB_DATABASE - PostgreSQL database name
• DB_USERNAME - PostgreSQL username
• DB_PASSWORD - PostgreSQL password

MQTT Configuration

Configure the MQTT broker settings in the node properties:

• Broker: MQTT broker hostname/IP
• Port: MQTT broker port (default: 1883)
• Username/Password: Optional authentication

Usage

1. Add the node to your Node-RED flow
2. Configure MQTT broker settings
3. Set environment variables for PostgreSQL connection
4. Deploy - The node runs automatically as a background service

MQTT Topics

Subscribes to:

• gateway/<UUID>/fetch - Listens for timestamp requests

Publishes to:

• gateway/<UUID>/data - Energy query results
• gateway/<UUID>/water - Water query results
• gateway/<UUID>/status - Status heartbeats (every 30 seconds)

Message Formats

Energy Request:

{
  "last_energy_ts": "2025-08-30T12:00:00.000Z"
}

Water Request:

{
  "last_water_ts": "2025-08-30T12:00:00.000Z",
  "lookahead_hours": 24
}

Response Format:

[
  {
    "timestamp": "2025-08-30T12:00:00.000Z",
    "slave_id": 1,
    "value": 1.25,
    "channel_id": 1
  }
]

Status Heartbeat:

{
  "status": "ok",
  "timestamp": "2025-08-30T12:00:00.000Z", 
  "metadata": {
    "slaves": [],
    "waterChannels": [1],
    "source": "node-red"
  }
}

Database Schema Requirements

Required Tables

slaves - Device information

CREATE TABLE slaves (
  id INTEGER PRIMARY KEY,
  type VARCHAR(255),
  name VARCHAR(255),
  clamp_type INTEGER
);

channels - Channel configuration with multipliers

CREATE TABLE channels (
  id INTEGER PRIMARY KEY,
  slave_id INTEGER REFERENCES slaves(id),
  channel INTEGER,
  type VARCHAR(255),
  name VARCHAR(255)  -- Contains multipliers like "Water Meter x100"
);

consumption_realtime - Energy consumption data

CREATE TABLE consumption_realtime (
  timestamp TIMESTAMPTZ,
  slave_id INTEGER,
  value NUMERIC,
  value_reactive NUMERIC
);

channel_pulse_log - Water flow pulse data

CREATE TABLE channel_pulse_log (
  timestamp TIMESTAMPTZ,
  slave_id INTEGER,
  channel INTEGER,
  value INTEGER
);

Multiplier Extraction

The node automatically extracts multipliers from channel names using the pattern:

• x([0-9]+\.?[0-9]*)
• Examples: "Water Meter x100" → 100, "Flow Sensor x2.5" → 2.5

Status Indicators

• 🟢 Green dot: Both MQTT and PostgreSQL connected
• 🔴 Red ring: Connection issues (hover for details)
• MQTT disconnected: MQTT broker connection lost
• PostgreSQL disconnected: Database connection lost
• PG & MQTT disconnected: Both connections lost

Testing

Test the node by publishing to the fetch topic:

# Energy request
mosquitto_pub -t "gateway/test-uuid/fetch" \
  -m '{"last_energy_ts":"2025-08-30T00:00:00.000Z"}'

# Water request  
mosquitto_pub -t "gateway/test-uuid/fetch" \
  -m '{"last_water_ts":"2025-08-30T00:00:00.000Z"}'

# Monitor responses
mosquitto_sub -t "gateway/test-uuid/#" -v

License

MIT

Support

For issues and feature requests, please use the GitHub Issues page.