Capstone: Production IoT Monitoring System

# ESP32 node: publish to its own topics, subscribe to commands
user esp32-node
topic write greenhouse/esp32-node/#
topic read greenhouse/cmd/esp32-node

# Pico W node: publish to its own topics, subscribe to commands
user pico-node
topic write greenhouse/pico-node/#
topic read greenhouse/cmd/pico-node

# STM32 node: publish to its own topics, subscribe to commands
user stm32-node
topic write greenhouse/stm32-node/#
topic read greenhouse/cmd/stm32-node

# Telegraf: read all greenhouse data
user telegraf
topic read greenhouse/#

# API server: read all data, write commands
user api-server
topic read greenhouse/#
topic write greenhouse/cmd/#

# Node-RED: read all data
user nodered
topic read greenhouse/#

#include <stdio.h>
#include <string.h>
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "esp_log.h"
#include "esp_wifi.h"
#include "esp_event.h"
#include "nvs_flash.h"
#include "mqtt_client.h"
#include "driver/i2c.h"
#include "driver/adc.h"

#define TAG "GREENHOUSE_ESP32"

/* Configuration */
#define PUBLISH_INTERVAL_MS   30000
#define BROKER_URI            "mqtts://192.168.1.100:8883"
#define MQTT_USERNAME         "esp32-node"
#define MQTT_PASSWORD         "your_password_here"
#define MAX_BACKOFF_MS        60000
#define OFFLINE_BUFFER_SIZE   60

/* BME280 I2C address */
#define BME280_ADDR           0x76
#define I2C_MASTER_NUM        I2C_NUM_0
#define I2C_SDA_PIN           21
#define I2C_SCL_PIN           22

/* Soil moisture ADC */
#define SOIL_ADC_CHANNEL      ADC1_CHANNEL_6  /* GPIO34 */
#define SOIL_DRY_VALUE        3200
#define SOIL_WET_VALUE        1400

/* Topic prefixes */
#define TOPIC_TEMP       "greenhouse/esp32-node/temperature"
#define TOPIC_HUMIDITY   "greenhouse/esp32-node/humidity"
#define TOPIC_PRESSURE   "greenhouse/esp32-node/pressure"
#define TOPIC_SOIL       "greenhouse/esp32-node/soil_moisture"
#define TOPIC_STATUS     "greenhouse/esp32-node/status"
#define TOPIC_CMD        "greenhouse/cmd/esp32-node"

/* State */
static bool mqtt_connected = false;
static esp_mqtt_client_handle_t mqtt_client = NULL;
static int backoff_ms = 1000;

/* Offline buffer */
typedef struct {
    float temperature;
    float humidity;
    float pressure;
    int   soil_pct;
} sensor_reading_t;

static sensor_reading_t offline_buffer[OFFLINE_BUFFER_SIZE];
static int buffer_head = 0;
static int buffer_count = 0;

/* ── BME280 driver (simplified, register-level) ── */

static esp_err_t bme280_init(void)
{
    i2c_config_t conf = {
        .mode = I2C_MODE_MASTER,
        .sda_io_num = I2C_SDA_PIN,
        .scl_io_num = I2C_SCL_PIN,
        .sda_pullup_en = GPIO_PULLUP_ENABLE,
        .scl_pullup_en = GPIO_PULLUP_ENABLE,
        .master.clk_speed = 100000,
    };
    i2c_param_config(I2C_MASTER_NUM, &conf);
    i2c_driver_install(I2C_MASTER_NUM, conf.mode, 0, 0, 0);

    /* Write config registers: normal mode, oversampling x1 */
    uint8_t ctrl_meas = 0x27;  /* osrs_t=1, osrs_p=1, mode=normal */
    uint8_t ctrl_hum  = 0x01;  /* osrs_h=1 */

    i2c_cmd_handle_t cmd = i2c_cmd_link_create();
    i2c_master_start(cmd);
    i2c_master_write_byte(cmd, (BME280_ADDR << 1) | I2C_MASTER_WRITE, true);
    i2c_master_write_byte(cmd, 0xF2, true);  /* ctrl_hum register */
    i2c_master_write_byte(cmd, ctrl_hum, true);
    i2c_master_stop(cmd);
    esp_err_t ret = i2c_master_cmd_begin(I2C_MASTER_NUM, cmd, pdMS_TO_TICKS(100));
    i2c_cmd_link_delete(cmd);
    if (ret != ESP_OK) return ret;

    cmd = i2c_cmd_link_create();
    i2c_master_start(cmd);
    i2c_master_write_byte(cmd, (BME280_ADDR << 1) | I2C_MASTER_WRITE, true);
    i2c_master_write_byte(cmd, 0xF4, true);  /* ctrl_meas register */
    i2c_master_write_byte(cmd, ctrl_meas, true);
    i2c_master_stop(cmd);
    ret = i2c_master_cmd_begin(I2C_MASTER_NUM, cmd, pdMS_TO_TICKS(100));
    i2c_cmd_link_delete(cmd);

    return ret;
}

static esp_err_t bme280_read(float *temp, float *hum, float *press)
{
    uint8_t data[8];
    /* Read registers 0xF7..0xFE (pressure, temperature, humidity) */
    i2c_cmd_handle_t cmd = i2c_cmd_link_create();
    i2c_master_start(cmd);
    i2c_master_write_byte(cmd, (BME280_ADDR << 1) | I2C_MASTER_WRITE, true);
    i2c_master_write_byte(cmd, 0xF7, true);
    i2c_master_start(cmd);
    i2c_master_write_byte(cmd, (BME280_ADDR << 1) | I2C_MASTER_READ, true);
    i2c_master_read(cmd, data, 8, I2C_MASTER_LAST_NACK);
    i2c_master_stop(cmd);
    esp_err_t ret = i2c_master_cmd_begin(I2C_MASTER_NUM, cmd, pdMS_TO_TICKS(100));
    i2c_cmd_link_delete(cmd);
    if (ret != ESP_OK) return ret;

    /* Simplified compensation (use full Bosch algorithm in production) */
    int32_t adc_t = ((int32_t)data[3] << 12) | ((int32_t)data[4] << 4) | (data[5] >> 4);
    int32_t adc_p = ((int32_t)data[0] << 12) | ((int32_t)data[1] << 4) | (data[2] >> 4);
    int32_t adc_h = ((int32_t)data[6] << 8) | data[7];

    /* Placeholder compensation (replace with calibration data in production) */
    *temp  = (float)adc_t / 16384.0f - 20.0f;
    *press = (float)adc_p / 256.0f;
    *hum   = (float)adc_h / 512.0f;

    return ESP_OK;
}

/* ── Soil moisture ── */

static int read_soil_moisture_pct(void)
{
    adc1_config_width(ADC_WIDTH_BIT_12);
    adc1_config_channel_atten(SOIL_ADC_CHANNEL, ADC_ATTEN_DB_11);
    int raw = adc1_get_raw(SOIL_ADC_CHANNEL);

    /* Map raw ADC to 0-100% (dry=0%, wet=100%) */
    if (raw >= SOIL_DRY_VALUE) return 0;
    if (raw <= SOIL_WET_VALUE) return 100;
    return (int)(100.0f * (SOIL_DRY_VALUE - raw) / (SOIL_DRY_VALUE - SOIL_WET_VALUE));
}

/* ── Offline buffer ── */

static void buffer_reading(float temp, float hum, float press, int soil)
{
    offline_buffer[buffer_head].temperature = temp;
    offline_buffer[buffer_head].humidity    = hum;
    offline_buffer[buffer_head].pressure    = press;
    offline_buffer[buffer_head].soil_pct    = soil;
    buffer_head = (buffer_head + 1) % OFFLINE_BUFFER_SIZE;
    if (buffer_count < OFFLINE_BUFFER_SIZE) buffer_count++;
    ESP_LOGW(TAG, "Buffered reading (%d in buffer)", buffer_count);
}

static void flush_buffer(esp_mqtt_client_handle_t client)
{
    if (buffer_count == 0) return;
    ESP_LOGI(TAG, "Flushing %d buffered readings", buffer_count);

    int start = (buffer_head - buffer_count + OFFLINE_BUFFER_SIZE) % OFFLINE_BUFFER_SIZE;
    for (int i = 0; i < buffer_count; i++) {
        int idx = (start + i) % OFFLINE_BUFFER_SIZE;
        char val[16];
        snprintf(val, sizeof(val), "%.1f", offline_buffer[idx].temperature);
        esp_mqtt_client_publish(client, TOPIC_TEMP, val, 0, 1, 1);
        snprintf(val, sizeof(val), "%.1f", offline_buffer[idx].humidity);
        esp_mqtt_client_publish(client, TOPIC_HUMIDITY, val, 0, 1, 1);
        snprintf(val, sizeof(val), "%.1f", offline_buffer[idx].pressure);
        esp_mqtt_client_publish(client, TOPIC_PRESSURE, val, 0, 1, 1);
        snprintf(val, sizeof(val), "%d", offline_buffer[idx].soil_pct);
        esp_mqtt_client_publish(client, TOPIC_SOIL, val, 0, 1, 1);
        vTaskDelay(pdMS_TO_TICKS(50));  /* Brief delay to avoid flooding */
    }
    buffer_count = 0;
    buffer_head = 0;
}

/* ── MQTT event handler ── */

static void mqtt_event_handler(void *args, esp_event_base_t base,
                                int32_t event_id, void *event_data)
{
    esp_mqtt_event_handle_t event = event_data;

    switch (event_id) {
    case MQTT_EVENT_CONNECTED:
        ESP_LOGI(TAG, "MQTT connected");
        mqtt_connected = true;
        backoff_ms = 1000;  /* Reset backoff */
        /* Publish online status */
        esp_mqtt_client_publish(event->client, TOPIC_STATUS, "online", 0, 1, 1);
        /* Subscribe to command topic */
        esp_mqtt_client_subscribe(event->client, TOPIC_CMD, 1);
        /* Flush any buffered readings */
        flush_buffer(event->client);
        break;

    case MQTT_EVENT_DISCONNECTED:
        ESP_LOGW(TAG, "MQTT disconnected, backoff %d ms", backoff_ms);
        mqtt_connected = false;
        vTaskDelay(pdMS_TO_TICKS(backoff_ms));
        if (backoff_ms < MAX_BACKOFF_MS) {
            backoff_ms *= 2;
        }
        break;

    case MQTT_EVENT_DATA:
        ESP_LOGI(TAG, "Received command on %.*s: %.*s",
                 event->topic_len, event->topic,
                 event->data_len, event->data);
        /* Parse command JSON here (e.g., change publish interval) */
        break;

    default:
        break;
    }
}

/* ── Main task ── */

static void sensor_publish_task(void *pvParameters)
{
    bme280_init();

    while (1) {
        float temp, hum, press;
        int soil_pct;

        if (bme280_read(&temp, &hum, &press) == ESP_OK) {
            soil_pct = read_soil_moisture_pct();

            if (mqtt_connected) {
                char val[16];
                snprintf(val, sizeof(val), "%.1f", temp);
                esp_mqtt_client_publish(mqtt_client, TOPIC_TEMP, val, 0, 1, 1);
                snprintf(val, sizeof(val), "%.1f", hum);
                esp_mqtt_client_publish(mqtt_client, TOPIC_HUMIDITY, val, 0, 1, 1);
                snprintf(val, sizeof(val), "%.1f", press);
                esp_mqtt_client_publish(mqtt_client, TOPIC_PRESSURE, val, 0, 1, 1);
                snprintf(val, sizeof(val), "%d", soil_pct);
                esp_mqtt_client_publish(mqtt_client, TOPIC_SOIL, val, 0, 1, 1);
                ESP_LOGI(TAG, "Published: T=%.1f H=%.1f P=%.1f Soil=%d%%",
                         temp, hum, press, soil_pct);
            } else {
                buffer_reading(temp, hum, press, soil_pct);
            }
        } else {
            ESP_LOGE(TAG, "BME280 read failed");
        }

        vTaskDelay(pdMS_TO_TICKS(PUBLISH_INTERVAL_MS));
    }
}

void app_main(void)
{
    nvs_flash_init();
    esp_netif_init();
    esp_event_loop_create_default();

    /* Wi-Fi connect (use your wifi_connect() from Lesson 3) */
    wifi_connect();

    /* MQTT client setup */
    extern const uint8_t server_cert_pem_start[] asm("_binary_ca_crt_start");

    esp_mqtt_client_config_t mqtt_cfg = {
        .broker = {
            .address.uri = BROKER_URI,
            .verification.certificate = (const char *)server_cert_pem_start,
        },
        .credentials = {
            .username = MQTT_USERNAME,
            .authentication.password = MQTT_PASSWORD,
            .client_id = "esp32-greenhouse",
        },
        .session = {
            .last_will = {
                .topic   = TOPIC_STATUS,
                .msg     = "offline",
                .msg_len = 7,
                .qos     = 1,
                .retain  = 1,
            },
            .keepalive = 30,
        },
    };

    mqtt_client = esp_mqtt_client_init(&mqtt_cfg);
    esp_mqtt_client_register_event(mqtt_client, ESP_EVENT_ANY_ID,
                                    mqtt_event_handler, NULL);
    esp_mqtt_client_start(mqtt_client);

    xTaskCreate(sensor_publish_task, "sensor_pub", 4096, NULL, 5, NULL);
}

import network
import time
import json
from machine import Pin, ADC
from umqtt.simple import MQTTClient

# Configuration
WIFI_SSID     = "your_ssid"
WIFI_PASS     = "your_password"
BROKER_IP     = "192.168.1.100"
BROKER_PORT   = 8883
MQTT_USER     = "pico-node"
MQTT_PASS     = "your_password_here"
CLIENT_ID     = "pico-greenhouse"
PUBLISH_INTERVAL = 30  # seconds

# Topics
TOPIC_LIGHT   = b"greenhouse/pico-node/light"
TOPIC_MOTION  = b"greenhouse/pico-node/motion"
TOPIC_STATUS  = b"greenhouse/pico-node/status"
TOPIC_CMD     = b"greenhouse/cmd/pico-node"

# Hardware
ldr = ADC(Pin(26))           # ADC0, GP26
pir = Pin(15, Pin.IN)        # PIR sensor on GP15
led = Pin("LED", Pin.OUT)    # Onboard LED for status

# State
last_motion = 0
backoff_s = 1
MAX_BACKOFF = 60

def connect_wifi():
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        print("Connecting to Wi-Fi...")
        wlan.connect(WIFI_SSID, WIFI_PASS)
        timeout = 20
        while not wlan.isconnected() and timeout > 0:
            time.sleep(1)
            timeout -= 1
    if wlan.isconnected():
        print("Wi-Fi connected:", wlan.ifconfig())
        led.on()
        return True
    else:
        print("Wi-Fi connection failed")
        led.off()
        return False

def read_light_lux():
    """Read LDR and convert ADC value to approximate lux."""
    raw = ldr.read_u16()  # 0-65535
    # Simple linear mapping (calibrate for your LDR + resistor divider)
    lux = int(raw * 1000 / 65535)
    return lux

def on_message(topic, msg):
    """Handle incoming commands."""
    print("Command received:", topic, msg)
    try:
        cmd = json.loads(msg)
        if "interval" in cmd:
            global PUBLISH_INTERVAL
            PUBLISH_INTERVAL = int(cmd["interval"])
            print("Publish interval changed to", PUBLISH_INTERVAL)
    except Exception as e:
        print("Command parse error:", e)

def mqtt_connect():
    """Connect to MQTT broker with TLS."""
    global backoff_s
    try:
        client = MQTTClient(
            CLIENT_ID,
            BROKER_IP,
            port=BROKER_PORT,
            user=MQTT_USER,
            password=MQTT_PASS,
            keepalive=30,
            ssl=True,
        )
        client.set_last_will(TOPIC_STATUS, b"offline", retain=True, qos=1)
        client.set_callback(on_message)
        client.connect()
        client.subscribe(TOPIC_CMD, 1)
        client.publish(TOPIC_STATUS, b"online", retain=True, qos=1)
        print("MQTT connected")
        backoff_s = 1  # Reset backoff
        return client
    except Exception as e:
        print("MQTT connect failed:", e, "backoff:", backoff_s)
        time.sleep(backoff_s)
        if backoff_s < MAX_BACKOFF:
            backoff_s *= 2
        return None

def main():
    global last_motion

    if not connect_wifi():
        return

    client = None
    last_publish = 0

    while True:
        # Reconnect if needed
        if client is None:
            client = mqtt_connect()
            if client is None:
                continue

        try:
            # Check for incoming messages (non-blocking)
            client.check_msg()

            # Publish light level periodically
            now = time.time()
            if now - last_publish >= PUBLISH_INTERVAL:
                lux = read_light_lux()
                client.publish(TOPIC_LIGHT, str(lux).encode(), retain=True, qos=1)
                print("Published light:", lux, "lux")
                last_publish = now

            # Publish motion on change (edge detection)
            current_motion = pir.value()
            if current_motion != last_motion:
                client.publish(TOPIC_MOTION, str(current_motion).encode(),
                               retain=False, qos=1)
                print("Motion event:", "detected" if current_motion else "clear")
                last_motion = current_motion

            time.sleep(0.1)  # Small poll interval for responsive motion detection

        except Exception as e:
            print("MQTT error:", e)
            client = None  # Force reconnect on next loop

main()

/* ── Includes and defines ── */
#include "main.h"
#include <stdio.h>
#include <string.h>

#define MQ135_ADC_CHANNEL   ADC_CHANNEL_0  /* PA0 */
#define PUBLISH_INTERVAL_MS 30000
#define UART_TIMEOUT        5000

/* MQTT topics */
static const char *TOPIC_AQ     = "greenhouse/stm32-node/air_quality";
static const char *TOPIC_STATUS = "greenhouse/stm32-node/status";

extern UART_HandleTypeDef huart2;  /* ESP-01 on USART2 */
extern ADC_HandleTypeDef  hadc1;

/* ── ESP-01 AT command helpers ── */

static int esp_send_cmd(const char *cmd, const char *expected,
                         uint32_t timeout_ms)
{
    char rx_buf[256] = {0};
    HAL_UART_Transmit(&huart2, (uint8_t *)cmd, strlen(cmd), 1000);
    HAL_UART_Transmit(&huart2, (uint8_t *)"\r\n", 2, 100);

    HAL_UART_Receive(&huart2, (uint8_t *)rx_buf, sizeof(rx_buf) - 1,
                      timeout_ms);

    if (strstr(rx_buf, expected) != NULL) {
        return 1;  /* Success */
    }
    return 0;  /* Failed */
}

static int esp_connect_wifi(const char *ssid, const char *pass)
{
    char cmd[128];
    esp_send_cmd("AT+CWMODE=1", "OK", 2000);
    snprintf(cmd, sizeof(cmd), "AT+CWJAP=\"%s\",\"%s\"", ssid, pass);
    return esp_send_cmd(cmd, "CONNECTED", 15000);
}

static int esp_mqtt_connect(const char *broker, uint16_t port,
                             const char *client_id,
                             const char *user, const char *pass)
{
    char cmd[256];
    snprintf(cmd, sizeof(cmd),
             "AT+MQTTUSERCFG=0,1,\"%s\",\"%s\",\"%s\",0,0,\"\"",
             client_id, user, pass);
    esp_send_cmd(cmd, "OK", 2000);

    /* Set last will */
    snprintf(cmd, sizeof(cmd),
             "AT+MQTTCONNCFG=0,30,0,\"%s\",\"offline\",1,1",
             TOPIC_STATUS);
    esp_send_cmd(cmd, "OK", 2000);

    snprintf(cmd, sizeof(cmd), "AT+MQTTCONN=0,\"%s\",%d,1",
             broker, port);
    return esp_send_cmd(cmd, "OK", 10000);
}

static int esp_mqtt_publish(const char *topic, const char *payload,
                             int qos, int retain)
{
    char cmd[256];
    snprintf(cmd, sizeof(cmd), "AT+MQTTPUB=0,\"%s\",\"%s\",%d,%d",
             topic, payload, qos, retain);
    return esp_send_cmd(cmd, "OK", 3000);
}

/* ── Sensor reading ── */

static int read_air_quality(void)
{
    HAL_ADC_Start(&hadc1);
    HAL_ADC_PollForConversion(&hadc1, 100);
    uint32_t raw = HAL_ADC_GetValue(&hadc1);
    HAL_ADC_Stop(&hadc1);

    /* Map 12-bit ADC (0-4095) to AQI (0-500) */
    int aqi = (int)((float)raw * 500.0f / 4095.0f);
    return aqi;
}

/* ── Main loop ── */

int main(void)
{
    HAL_Init();
    SystemClock_Config();
    MX_GPIO_Init();
    MX_USART2_UART_Init();
    MX_ADC1_Init();

    /* Wait for ESP-01 to boot */
    HAL_Delay(3000);

    /* Connect Wi-Fi */
    while (!esp_connect_wifi("your_ssid", "your_password")) {
        HAL_Delay(5000);
    }

    /* Connect MQTT */
    int connected = 0;
    int backoff = 1000;
    while (!connected) {
        connected = esp_mqtt_connect("192.168.1.100", 8883,
                                      "stm32-greenhouse",
                                      "stm32-node", "your_password");
        if (!connected) {
            HAL_Delay(backoff);
            if (backoff < 60000) backoff *= 2;
        }
    }

    /* Publish online status */
    esp_mqtt_publish(TOPIC_STATUS, "online", 1, 1);

    /* Main sensor loop */
    while (1) {
        int aqi = read_air_quality();
        char payload[8];
        snprintf(payload, sizeof(payload), "%d", aqi);
        esp_mqtt_publish(TOPIC_AQ, payload, 1, 1);

        HAL_Delay(PUBLISH_INTERVAL_MS);
    }
}

# Global agent configuration
[agent]
  interval = "10s"
  round_interval = true
  flush_interval = "10s"
  flush_jitter = "2s"

# MQTT consumer input
[[inputs.mqtt_consumer]]
  servers = ["ssl://mosquitto:8883"]
  topics = ["greenhouse/#"]
  username = "telegraf"
  password = "your_telegraf_password"
  client_id = "telegraf-greenhouse"
  qos = 1
  persistent_session = true

  # TLS configuration
  tls_ca = "/etc/telegraf/certs/ca.crt"
  insecure_skip_verify = false

  # Parse topic to extract node and metric
  # Topic format: greenhouse/<node>/<metric>
  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "greenhouse/+/+"
    measurement = "measurement/_/_"
    tags = "_/node/metric"

  # Data format: plain string values
  data_format = "value"
  data_type = "float"

# InfluxDB v2 output
[[outputs.influxdb_v2]]
  urls = ["http://influxdb:8086"]
  token = "your_influxdb_token"
  organization = "greenhouse"
  bucket = "sensor_data"

  # Tag filtering
  [outputs.influxdb_v2.tagpass]
    node = ["esp32-node", "pico-node", "stm32-node"]

option task = {name: "downsample_hourly", every: 1h}

from(bucket: "sensor_data")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "mqtt_consumer")
  |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
  |> to(bucket: "sensor_data_longterm", org: "greenhouse")

from(bucket: "sensor_data")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r.node == "esp32-node" and r.metric == "temperature")
  |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
  |> yield(name: "temperature")

from(bucket: "sensor_data")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r.node == "esp32-node" and r.metric == "humidity")
  |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
  |> yield(name: "humidity")

from(bucket: "sensor_data")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r.node == "esp32-node" and r.metric == "soil_moisture")
  |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
  |> yield(name: "soil_moisture")

from(bucket: "sensor_data")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r.node == "pico-node" and r.metric == "light")
  |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
  |> yield(name: "light")

from(bucket: "sensor_data")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r.node == "stm32-node" and r.metric == "air_quality")
  |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
  |> yield(name: "air_quality")

from(bucket: "sensor_data")
  |> range(start: -5m)
  |> filter(fn: (r) => r.metric == "status")
  |> last()
  |> group(columns: ["node"])
  |> yield(name: "status")

import os
import json
from datetime import datetime, timedelta
from flask import Flask, request, jsonify, abort
from influxdb_client import InfluxDBClient
import paho.mqtt.client as mqtt

app = Flask(__name__)

# Configuration
INFLUX_URL    = os.getenv("INFLUX_URL", "http://influxdb:8086")
INFLUX_TOKEN  = os.getenv("INFLUX_TOKEN", "your_influxdb_token")
INFLUX_ORG    = os.getenv("INFLUX_ORG", "greenhouse")
INFLUX_BUCKET = os.getenv("INFLUX_BUCKET", "sensor_data")
API_KEY       = os.getenv("API_KEY", "your_api_key_here")

MQTT_BROKER   = os.getenv("MQTT_BROKER", "mosquitto")
MQTT_PORT     = int(os.getenv("MQTT_PORT", "8883"))
MQTT_USER     = os.getenv("MQTT_USER", "api-server")
MQTT_PASS     = os.getenv("MQTT_PASS", "your_password")

# InfluxDB client
influx_client = InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG)
query_api = influx_client.query_api()

# MQTT client for publishing commands
mqtt_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="flask-api")
mqtt_client.username_pw_set(MQTT_USER, MQTT_PASS)
mqtt_client.tls_set(ca_certs="/app/certs/ca.crt")
mqtt_client.connect(MQTT_BROKER, MQTT_PORT, keepalive=60)
mqtt_client.loop_start()

# Known devices
DEVICES = {
    "esp32-node":  {"name": "ESP32 Greenhouse", "sensors": ["temperature", "humidity", "pressure", "soil_moisture"]},
    "pico-node":   {"name": "Pico W Greenhouse", "sensors": ["light", "motion"]},
    "stm32-node":  {"name": "STM32 Greenhouse", "sensors": ["air_quality"]},
}

def require_api_key(f):
    """Simple API key authentication decorator."""
    from functools import wraps
    @wraps(f)
    def decorated(*args, **kwargs):
        key = request.headers.get("X-API-Key")
        if key != API_KEY:
            abort(401, description="Invalid or missing API key")
        return f(*args, **kwargs)
    return decorated

@app.route("/api/devices", methods=["GET"])
@require_api_key
def list_devices():
    """List all registered devices and their sensors."""
    return jsonify({"devices": DEVICES})

@app.route("/api/devices/<node_id>/data", methods=["GET"])
@require_api_key
def get_device_data(node_id):
    """Query historical data for a device.

    Query params:
      - metric: sensor metric name (required)
      - range: time range, e.g., "1h", "24h", "7d" (default: "1h")
      - aggregate: window period, e.g., "1m", "5m" (default: "1m")
    """
    if node_id not in DEVICES:
        abort(404, description=f"Device {node_id} not found")

    metric = request.args.get("metric")
    if not metric or metric not in DEVICES[node_id]["sensors"]:
        abort(400, description=f"Invalid metric. Available: {DEVICES[node_id]['sensors']}")

    time_range = request.args.get("range", "1h")
    aggregate = request.args.get("aggregate", "1m")

    query = f'''
    from(bucket: "{INFLUX_BUCKET}")
      |> range(start: -{time_range})
      |> filter(fn: (r) => r.node == "{node_id}" and r.metric == "{metric}")
      |> aggregateWindow(every: {aggregate}, fn: mean, createEmpty: false)
    '''

    tables = query_api.query(query)
    records = []
    for table in tables:
        for record in table.records:
            records.append({
                "time": record.get_time().isoformat(),
                "value": record.get_value(),
            })

    return jsonify({
        "device": node_id,
        "metric": metric,
        "range": time_range,
        "aggregate": aggregate,
        "count": len(records),
        "data": records,
    })

@app.route("/api/devices/<node_id>/command", methods=["POST"])
@require_api_key
def send_command(node_id):
    """Send a command to a device via MQTT.

    Body (JSON):
      - Any key-value pairs to send as command payload
      - Example: {"interval": 60}
    """
    if node_id not in DEVICES:
        abort(404, description=f"Device {node_id} not found")

    payload = request.get_json()
    if not payload:
        abort(400, description="Request body must be JSON")

    topic = f"greenhouse/cmd/{node_id}"
    result = mqtt_client.publish(topic, json.dumps(payload), qos=1)

    if result.rc == 0:
        return jsonify({"status": "sent", "topic": topic, "payload": payload})
    else:
        abort(500, description="Failed to publish command")

@app.route("/api/health", methods=["GET"])
def health_check():
    """Health check endpoint (no auth required)."""
    return jsonify({"status": "ok", "timestamp": datetime.utcnow().isoformat()})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

[
    {
        "id": "mqtt_in_greenhouse",
        "type": "mqtt in",
        "topic": "greenhouse/#",
        "broker": "mosquitto_broker",
        "qos": "1",
        "name": "Greenhouse MQTT"
    },
    {
        "id": "parse_topic",
        "type": "function",
        "name": "Parse Topic and Value",
        "func": "const parts = msg.topic.split('/');\nmsg.node = parts[1];\nmsg.metric = parts[2];\nmsg.value = parseFloat(msg.payload);\nmsg.timestamp = Date.now();\n\n// Store last-seen timestamp per node\nconst ctx = global.get('lastSeen') || {};\nctx[msg.node] = msg.timestamp;\nglobal.set('lastSeen', ctx);\n\nreturn msg;"
    },
    {
        "id": "threshold_check",
        "type": "switch",
        "name": "Threshold Check",
        "property": "metric",
        "rules": [
            {"t": "eq", "v": "temperature"},
            {"t": "eq", "v": "humidity"},
            {"t": "eq", "v": "soil_moisture"},
            {"t": "eq", "v": "air_quality"}
        ],
        "outputs": 4
    },
    {
        "id": "temp_alert",
        "type": "function",
        "name": "Temperature Alert",
        "func": "if (msg.value > 35) {\n    msg.payload = {\n        text: `CRITICAL: Temperature is ${msg.value}C on ${msg.node} (threshold: 35C). Check greenhouse ventilation.`\n    };\n    return msg;\n}\nif (msg.value < 5) {\n    msg.payload = {\n        text: `CRITICAL: Temperature is ${msg.value}C on ${msg.node} (threshold: 5C). Check greenhouse heating.`\n    };\n    return msg;\n}\nreturn null;"
    },
    {
        "id": "humidity_alert",
        "type": "function",
        "name": "Humidity Alert",
        "func": "if (msg.value < 20) {\n    msg.payload = {\n        text: `WARNING: Humidity is ${msg.value}% on ${msg.node} (threshold: <20%). Consider misting.`\n    };\n    return msg;\n}\nif (msg.value > 90) {\n    msg.payload = {\n        text: `WARNING: Humidity is ${msg.value}% on ${msg.node} (threshold: >90%). Check for condensation.`\n    };\n    return msg;\n}\nreturn null;"
    },
    {
        "id": "soil_alert",
        "type": "function",
        "name": "Soil Moisture Alert",
        "func": "if (msg.value < 15) {\n    msg.payload = {\n        text: `WARNING: Soil moisture is ${msg.value}% on ${msg.node}. Water the plants.`\n    };\n    return msg;\n}\nreturn null;"
    },
    {
        "id": "aqi_alert",
        "type": "function",
        "name": "Air Quality Alert",
        "func": "if (msg.value > 150) {\n    msg.payload = {\n        text: `CRITICAL: Air quality index is ${msg.value} on ${msg.node} (threshold: 150). Open ventilation.`\n    };\n    return msg;\n}\nreturn null;"
    },
    {
        "id": "slack_webhook",
        "type": "http request",
        "method": "POST",
        "url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK",
        "name": "Send to Slack"
    },
    {
        "id": "inactivity_check",
        "type": "inject",
        "repeat": "60",
        "name": "Check Every 60s"
    },
    {
        "id": "inactivity_eval",
        "type": "function",
        "name": "Inactivity Detection",
        "func": "const ctx = global.get('lastSeen') || {};\nconst now = Date.now();\nconst threshold = 5 * 60 * 1000; // 5 minutes\nconst nodes = ['esp32-node', 'pico-node', 'stm32-node'];\nconst alerts = [];\n\nfor (const node of nodes) {\n    const lastSeen = ctx[node];\n    if (!lastSeen || (now - lastSeen) > threshold) {\n        alerts.push(node);\n    }\n}\n\nif (alerts.length > 0) {\n    msg.payload = {\n        text: `INACTIVITY ALERT: No data received from ${alerts.join(', ')} for over 5 minutes. Check device connectivity.`\n    };\n    return msg;\n}\nreturn null;"
    }
]

// At the top of each alert function, add:
const cooldown = 5 * 60 * 1000;  // 5 minutes between repeated alerts
const key = `alert_${msg.node}_${msg.metric}`;
const lastAlert = flow.get(key) || 0;
const now = Date.now();

if (now - lastAlert < cooldown) {
    return null;  // Suppress duplicate
}
flow.set(key, now);
// ... rest of alert logic

version: "3.8"

services:
  # MQTT broker with TLS
  mosquitto:
    image: eclipse-mosquitto:2
    container_name: greenhouse-mosquitto
    ports:
      - "8883:8883"
    volumes:
      - ./mosquitto/mosquitto.conf:/mosquitto/config/mosquitto.conf:ro
      - ./mosquitto/passwd:/mosquitto/config/passwd:ro
      - ./mosquitto/acl:/mosquitto/config/acl:ro
      - ./certs:/mosquitto/certs:ro
      - mosquitto_data:/mosquitto/data
      - mosquitto_log:/mosquitto/log
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "mosquitto_sub", "-t", "$$SYS/#", "-C", "1", "-u", "healthcheck", "-P", "healthcheck", "--cafile", "/mosquitto/certs/ca.crt", "-p", "8883"]
      interval: 30s
      timeout: 10s
      retries: 3

  # Telegraf: MQTT to InfluxDB
  telegraf:
    image: telegraf:1.30
    container_name: greenhouse-telegraf
    volumes:
      - ./telegraf/telegraf.conf:/etc/telegraf/telegraf.conf:ro
      - ./certs:/etc/telegraf/certs:ro
    depends_on:
      mosquitto:
        condition: service_healthy
      influxdb:
        condition: service_healthy
    restart: unless-stopped

  # InfluxDB: time-series database
  influxdb:
    image: influxdb:2.7
    container_name: greenhouse-influxdb
    ports:
      - "8086:8086"
    environment:
      DOCKER_INFLUXDB_INIT_MODE: setup
      DOCKER_INFLUXDB_INIT_USERNAME: admin
      DOCKER_INFLUXDB_INIT_PASSWORD: your_influxdb_password
      DOCKER_INFLUXDB_INIT_ORG: greenhouse
      DOCKER_INFLUXDB_INIT_BUCKET: sensor_data
      DOCKER_INFLUXDB_INIT_RETENTION: 30d
      DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: your_influxdb_token
    volumes:
      - influxdb_data:/var/lib/influxdb2
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "influx", "ping"]
      interval: 30s
      timeout: 10s
      retries: 3

  # Grafana: dashboards
  grafana:
    image: grafana/grafana:10.4.0
    container_name: greenhouse-grafana
    ports:
      - "3000:3000"
    environment:
      GF_SECURITY_ADMIN_USER: admin
      GF_SECURITY_ADMIN_PASSWORD: your_grafana_password
      GF_INSTALL_PLUGINS: grafana-clock-panel
    volumes:
      - grafana_data:/var/lib/grafana
      - ./grafana/provisioning:/etc/grafana/provisioning:ro
    depends_on:
      influxdb:
        condition: service_healthy
    restart: unless-stopped

  # Flask REST API
  flask-api:
    build: ./api
    container_name: greenhouse-api
    ports:
      - "5000:5000"
    environment:
      INFLUX_URL: http://influxdb:8086
      INFLUX_TOKEN: your_influxdb_token
      INFLUX_ORG: greenhouse
      INFLUX_BUCKET: sensor_data
      MQTT_BROKER: mosquitto
      MQTT_PORT: "8883"
      MQTT_USER: api-server
      MQTT_PASS: your_password
      API_KEY: your_api_key_here
    volumes:
      - ./certs:/app/certs:ro
    depends_on:
      - mosquitto
      - influxdb
    restart: unless-stopped

  # Node-RED: alerts and automation
  nodered:
    image: nodered/node-red:3.1
    container_name: greenhouse-nodered
    ports:
      - "1880:1880"
    volumes:
      - nodered_data:/data
      - ./nodered/flows.json:/data/flows.json:ro
      - ./certs:/data/certs:ro
    depends_on:
      - mosquitto
    restart: unless-stopped

volumes:
  mosquitto_data:
  mosquitto_log:
  influxdb_data:
  grafana_data:
  nodered_data: