← FTP Data Upload from Node-RED to a Monitoring Platform

FTP Data Upload from Node-RED to a Monitoring Platform — full example

Generate timestamped CSV files with Python and upload them by FTP to an official monitoring platform. Complete ftplib script used on a real slope site.

Complete, runnable program for the Server (Node-RED) (ftp-data-upload-platform.py): wiring header, requirements and integration notes included.

Read-only preview.

#!/usr/bin/env python3
"""
COMPLETE EXAMPLE — Data delivery to the official monitoring platform via FTP

Hardware:  Server (Node-RED + Python) on a Raspberry Pi
Based on:  geotechnical slope monitoring project (4 stations,
           ESP32 PLC 14 Ethernet)

Architecture:
  - Node-RED aggregates the readings of the 4 stations into a JSON file
    (latest_readings.json) on every measurement interval.
  - This script (launched by a Node-RED exec node or by cron) generates
    an .est file — a CSV with record id, date, time and the N measurement
    channels — and uploads it via FTP to the official monitoring
    platform.
  - In the real project the file carries 169 channels; here they are
    generated from the JSON, leaving empty the channels without data.

Usage:
  python3 ftp-data-upload-platform.py
"""

import json
import os
import sys
from datetime import datetime
from ftplib import FTP

# --- Configuration (fill in with real credentials for production) -------
FTP_HOST = "ftp.remote-platform.example"
FTP_USER = "FTP_USER"            # placeholder: user assigned by the platform
FTP_PASS = "FTP_PASS"            # placeholder: assigned password
FTP_DIR = "/incoming"            # remote delivery folder

REG_ID = "SLOPE_STATION_01"      # agreed record identifier
NUM_CHANNELS = 169               # channels expected by the platform
READINGS_FILE = "/home/pi/slope/latest_readings.json"
OUTPUT_DIR = "/home/pi/slope/sent"


def read_readings(path):
    """Loads the JSON with the latest readings aggregated by Node-RED.

    Expected format: {"channel_001": 12.34, "channel_002": -0.07, ...}
    """
    try:
        with open(path) as f:
            return json.load(f)
    except (OSError, ValueError) as e:
        print(f"ERROR reading readings: {e}")
        return {}


def generate_est_file(readings):
    """Generates the .est file: CSV with timestamp and NUM_CHANNELS columns."""
    now = datetime.now()
    date_str = now.strftime("%d/%m/%Y")
    time_str = now.strftime("%H:%M:%S")

    # Ordered channels: channel_001 ... channel_169; empty when no data
    channels = []
    for i in range(1, NUM_CHANNELS + 1):
        value = readings.get(f"channel_{i:03d}", "")
        channels.append(f"{value:.4f}" if isinstance(value, (int, float)) else "")

    os.makedirs(OUTPUT_DIR, exist_ok=True)
    fname = os.path.join(
        OUTPUT_DIR, f"{REG_ID}_{now.strftime('%Y%m%d_%H%M%S')}.est"
    )

    # Same structure as the format agreed with the platform:
    # record_id,date,time, ,channel1,channel2,...,channelN
    with open(fname, "w") as f:
        f.write(REG_ID + "," + date_str + "," + time_str + ", ," + ",".join(channels) + "\n")

    print(f"Generated {fname} ({NUM_CHANNELS} channels)")
    return fname


def upload_via_ftp(fname):
    """Uploads the .est file to the official platform via FTP."""
    ftp = FTP(FTP_HOST, timeout=30)
    try:
        ftp.login(FTP_USER, FTP_PASS)
        ftp.cwd(FTP_DIR)
        with open(fname, "rb") as f:
            ftp.storbinary("STOR " + os.path.basename(fname), f)
        print(f"Uploaded {os.path.basename(fname)} to {FTP_HOST}{FTP_DIR}")
    finally:
        ftp.quit()


def main():
    readings = read_readings(READINGS_FILE)
    if not readings:
        # With no readings nothing is sent: the platform detects the gap
        # and the SMTP alert system reports the failure in parallel.
        sys.exit(1)

    fname = generate_est_file(readings)
    try:
        upload_via_ftp(fname)
    except Exception as e:
        # The .est file is not deleted: it stays in OUTPUT_DIR so it can
        # be retried manually.
        print(f"FTP ERROR: {e} — file kept at {fname}")
        sys.exit(2)


if __name__ == "__main__":
    main()