Humidity Control: The Ultimate Guide to Everything You Need

import machine
import time
import ujson
from bme280 import BME280
from dht import DHT22
from umqtt.simple import MQTTClient
import network
import gc

# Configuration constants - update these for your deployment
WIFI_SSID = "your-ssid"
WIFI_PASS = "your-pass"
MQTT_BROKER = "mqtt.your-domain.com"
MQTT_TOPIC = b"humidity/sensor/readings"
SENSOR_TYPE = "BME280"  # Options: BME280, DHT22
I2C_ADDR = 0x76  # BME280 default I2C address
DHT_PIN = 4  # GPIO4 for DHT22 data line
CALIBRATION_INTERVAL = 86400  # Calibrate every 24 hours (seconds)
REPORT_INTERVAL = 1  # Report readings every 1 second

class HumiditySensor:
    def __init__(self, sensor_type):
        self.sensor_type = sensor_type
        self.last_calibration = 0
        self.calibration_offset = 0.0
        self.init_sensor()

    def init_sensor(self):
        """Initialize sensor hardware based on configured type"""
        if self.sensor_type == "BME280":
            # Initialize I2C bus on ESP32-S3 default pins (SCL=GPIO8, SDA=GPIO9)
            self.i2c = machine.I2C(0, scl=machine.Pin(8), sda=machine.Pin(9), freq=400000)
            # Scan I2C bus to verify sensor is present
            devices = self.i2c.scan()
            if I2C_ADDR not in devices:
                raise RuntimeError(f"BME280 not found at I2C address {hex(I2C_ADDR)}")
            self.sensor = BME280(i2c=self.i2c, address=I2C_ADDR)
        elif self.sensor_type == "DHT22":
            self.pin = machine.Pin(DHT_PIN, machine.Pin.IN, machine.Pin.PULL_UP)
            self.sensor = DHT22(self.pin)
        else:
            raise ValueError(f"Unsupported sensor type: {self.sensor_type}")

    def read(self):
        """Read humidity from sensor with error handling and retries"""
        retries = 3
        for attempt in range(retries):
            try:
                if self.sensor_type == "BME280":
                    # BME280 returns (temperature, pressure, humidity)
                    _, _, humidity = self.sensor.read_compensated_data()
                    # Apply calibration offset from last calibration cycle
                    humidity += self.calibration_offset
                    return round(humidity, 2)
                elif self.sensor_type == "DHT22":
                    # DHT22 requires 2 second warmup after init, handled in init
                    self.sensor.measure()
                    humidity = self.sensor.humidity()
                    humidity += self.calibration_offset
                    return round(humidity, 2)
            except OSError as e:
                print(f"Sensor read error (attempt {attempt+1}/{retries}): {e}")
                time.sleep(0.5)
        # All retries failed, return last known good value or raise
        raise RuntimeError("Failed to read humidity sensor after 3 retries")

    def calibrate(self, reference_rh):
        """Calibrate sensor against reference hygrometer reading"""
        current_rh = self.read()
        self.calibration_offset = reference_rh - current_rh
        self.last_calibration = time.time()
        print(f"Calibrated: offset set to {self.calibration_offset:.2f}% RH")
        return self.calibration_offset

    def check_calibration(self):
        """Check if calibration is due and run if needed"""
        if time.time() - self.last_calibration > CALIBRATION_INTERVAL:
            print("Calibration interval reached. Please enter reference RH value:")
            # In production, this would read from a reference sensor or user input
            # For this example, we assume reference is entered via REPL
            reference_rh = float(input("Reference RH %: "))
            self.calibrate(reference_rh)

def connect_wifi():
    """Connect to WiFi with retry logic"""
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        print(f"Connecting to WiFi: {WIFI_SSID}")
        wlan.connect(WIFI_SSID, WIFI_PASS)
        for _ in range(20):
            if wlan.isconnected():
                break
            time.sleep(1)
        if not wlan.isconnected():
            raise RuntimeError("WiFi connection failed")
    print(f"WiFi connected: {wlan.ifconfig()[0]}")

def main():
    connect_wifi()
    sensor = HumiditySensor(SENSOR_TYPE)
    # Initialize MQTT client with unique ID based on MAC address
    mac = machine.unique_id()
    client_id = f"humidity-sensor-{ujson.dumps(mac)}"
    mqtt = MQTTClient(client_id, MQTT_BROKER, port=1883)
    mqtt.connect()
    print("MQTT connected to broker")

    last_report = 0
    while True:
        try:
            # Check if calibration is due
            sensor.check_calibration()
            # Read current humidity
            current_rh = sensor.read()
            # Prepare telemetry payload
            payload = ujson.dumps({
                "rh": current_rh,
                "sensor_type": SENSOR_TYPE,
                "calibration_offset": sensor.calibration_offset,
                "timestamp": time.time()
            })
            # Publish to MQTT topic
            mqtt.publish(MQTT_TOPIC, payload)
            print(f"Published: {payload}")
            # Sleep until next report interval
            time.sleep(REPORT_INTERVAL)
            gc.collect()  # Free memory on constrained devices
        except Exception as e:
            print(f"Main loop error: {e}")
            time.sleep(5)

if __name__ == "__main__":
    main()

import time
import ujson
from umqtt.simple import MQTTClient
import machine

# PID configuration constants
SETPOINT_RH = 65.0  # Target humidity % RH
KP = 2.5  # Proportional gain
KI = 0.1  # Integral gain
KD = 1.2  # Derivative gain
PID_INTERVAL = 1  # PID loop runs every 1 second
ACTUATOR_PIN = 5  # GPIO5 for humidifier relay
ACTUATOR_INVERT = False  # Set to True if relay is active-low
MAX_INTEGRAL = 100.0  # Anti-windup limit for integral term
MQTT_TELEMETRY_TOPIC = b"humidity/pid/telemetry"
MQTT_SETPOINT_TOPIC = b"humidity/pid/setpoint"

class PIDController:
    def __init__(self, kp, ki, kd, setpoint):
        self.kp = kp
        self.ki = ki
        self.kd = kd
        self.setpoint = setpoint
        self.last_error = 0.0
        self.integral = 0.0
        self.last_time = time.time()

    def update(self, current_rh):
        """Calculate PID output based on current humidity reading"""
        now = time.time()
        dt = now - self.last_time
        if dt < 0.1:  # Avoid division by zero or tiny dt
            dt = 0.1

        error = self.setpoint - current_rh
        # Proportional term
        p_term = self.kp * error
        # Integral term with anti-windup
        self.integral += error * dt
        if self.integral > MAX_INTEGRAL:
            self.integral = MAX_INTEGRAL
        elif self.integral < -MAX_INTEGRAL:
            self.integral = -MAX_INTEGRAL
        i_term = self.ki * self.integral
        # Derivative term (derivative of error, not measurement to avoid spikes)
        d_term = self.kd * (error - self.last_error) / dt
        # Calculate total output (0-100% duty cycle for PWM or on/off threshold)
        output = p_term + i_term + d_term
        # Clamp output to 0-100 range
        output = max(0.0, min(100.0, output))

        self.last_error = error
        self.last_time = now
        return output

    def update_setpoint(self, new_setpoint):
        """Update PID setpoint with bounds checking"""
        if 0 <= new_setpoint <= 100:
            self.setpoint = new_setpoint
            print(f"Updated setpoint to {new_setpoint}% RH")
        else:
            print(f"Invalid setpoint {new_setpoint}, must be 0-100% RH")

class HumidifierActuator:
    def __init__(self, pin, invert=False):
        self.pin = machine.Pin(pin, machine.Pin.OUT)
        self.invert = invert
        self.state = False
        self.set_state(False)  # Start with actuator off

    def set_state(self, on):
        """Set actuator state, accounting for active-high/low relays"""
        self.state = on
        if self.invert:
            self.pin.value(not on)
        else:
            self.pin.value(on)

    def set_pwm(self, duty_cycle):
        """For PWM-controlled humidifiers, set duty cycle (0-100%)"""
        # ESP32 has 16 PWM channels, use channel 0 for actuator
        self.pwm = machine.PWM(machine.Pin(ACTUATOR_PIN), freq=1000)
        # Convert duty cycle % to 0-1023 range for 10-bit PWM
        duty = int((duty_cycle / 100) * 1023)
        self.pwm.duty(duty)
        print(f"Set humidifier duty cycle to {duty_cycle}%")

def mqtt_callback(topic, msg):
    """Handle incoming MQTT messages for setpoint updates"""
    try:
        if topic == MQTT_SETPOINT_TOPIC:
            new_setpoint = ujson.loads(msg).get("setpoint")
            if new_setpoint is not None:
                pid.update_setpoint(float(new_setpoint))
    except Exception as e:
        print(f"MQTT callback error: {e}")

def main():
    # Initialize PID controller
    global pid
    pid = PIDController(KP, KI, KD, SETPOINT_RH)
    # Initialize actuator
    actuator = HumidifierActuator(ACTUATOR_PIN, ACTUATOR_INVERT)
    # Initialize MQTT client
    mac = machine.unique_id()
    client_id = f"humidity-pid-{ujson.dumps(mac)}"
    mqtt = MQTTClient(client_id, "mqtt.your-domain.com", port=1883)
    mqtt.set_callback(mqtt_callback)
    mqtt.connect()
    mqtt.subscribe(MQTT_SETPOINT_TOPIC)
    print("PID controller started, subscribed to setpoint topic")

    last_pid_run = 0
    while True:
        try:
            # Check for incoming MQTT messages
            mqtt.check_msg()
            # Run PID loop at specified interval
            if time.time() - last_pid_run >= PID_INTERVAL:
                # Simulate reading current RH (in production, read from sensor)
                current_rh = 60.0  # Replace with actual sensor read
                pid_output = pid.update(current_rh)
                # Control actuator: on/off if output > 50%, else off (or PWM)
                if pid_output > 50:
                    actuator.set_state(True)
                else:
                    actuator.set_state(False)
                # Publish telemetry
                payload = ujson.dumps({
                    "setpoint": pid.setpoint,
                    "current_rh": current_rh,
                    "pid_output": pid_output,
                    "actuator_state": actuator.state,
                    "timestamp": time.time()
                })
                mqtt.publish(MQTT_TELEMETRY_TOPIC, payload)
                last_pid_run = time.time()
            time.sleep(0.1)
        except Exception as e:
            print(f"PID main loop error: {e}")
            time.sleep(5)

if __name__ == "__main__":
    main()

import paho.mqtt.client as mqtt
import json
import time
import sqlite3
import os
from datetime import datetime
import requests

# Configuration
MQTT_BROKER = "mqtt.your-domain.com"
MQTT_PORT = 1883
MQTT_TOPIC = "humidity/+/readings"  # Wildcard to subscribe to all sensor nodes
DB_PATH = "/var/lib/humidity/telemetry.db"
GRAFANA_URL = "http://grafana.your-domain.com/api/datasources/proxy/1/write"
API_KEY = "your-grafana-api-key"
AGGREGATION_INTERVAL = 60  # Aggregate readings every 60 seconds

class TelemetryStore:
    def __init__(self, db_path):
        self.db_path = db_path
        os.makedirs(os.path.dirname(db_path), exist_ok=True)
        self.init_db()

    def init_db(self):
        """Initialize SQLite database for local telemetry storage"""
        conn = sqlite3.connect(self.db_path)
        c = conn.cursor()
        c.execute('''CREATE TABLE IF NOT EXISTS readings
                     (id INTEGER PRIMARY KEY AUTOINCREMENT,
                      sensor_id TEXT,
                      rh REAL,
                      calibration_offset REAL,
                      timestamp INTEGER,
                      created_at TEXT)''')
        c.execute('''CREATE INDEX IF NOT EXISTS idx_sensor_timestamp 
                     ON readings(sensor_id, timestamp)''')
        conn.commit()
        conn.close()

    def store_reading(self, sensor_id, rh, calibration_offset, timestamp):
        """Store a single sensor reading in the database"""
        conn = sqlite3.connect(self.db_path)
        c = conn.cursor()
        created_at = datetime.utcnow().isoformat()
        c.execute('''INSERT INTO readings (sensor_id, rh, calibration_offset, timestamp, created_at)
                     VALUES (?, ?, ?, ?, ?)''',
                  (sensor_id, rh, calibration_offset, timestamp, created_at))
        conn.commit()
        conn.close()

    def get_aggregated(self, start_time, end_time):
        """Get aggregated RH readings per sensor for a time range"""
        conn = sqlite3.connect(self.db_path)
        c = conn.cursor()
        c.execute('''SELECT sensor_id, AVG(rh), MIN(rh), MAX(rh), COUNT(*)
                     FROM readings
                     WHERE timestamp BETWEEN ? AND ?
                     GROUP BY sensor_id''',
                  (start_time, end_time))
        results = c.fetchall()
        conn.close()
        return results

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print(f"Connected to MQTT broker at {MQTT_BROKER}:{MQTT_PORT}")
        client.subscribe(MQTT_TOPIC)
    else:
        print(f"MQTT connection failed with code {rc}")

def on_message(client, userdata, msg):
    """Handle incoming sensor telemetry messages"""
    try:
        payload = json.loads(msg.payload.decode())
        # Extract sensor ID from topic (format: humidity/sensor//readings)
        topic_parts = msg.topic.split("/")
        if len(topic_parts) != 3:
            print(f"Invalid topic format: {msg.topic}")
            return
        sensor_id = topic_parts[1]
        # Validate payload fields
        required_fields = ["rh", "sensor_type", "calibration_offset", "timestamp"]
        for field in required_fields:
            if field not in payload:
                print(f"Missing field {field} in payload from {sensor_id}")
                return
        # Store reading in local DB
        store = userdata["store"]
        store.store_reading(
            sensor_id=sensor_id,
            rh=payload["rh"],
            calibration_offset=payload["calibration_offset"],
            timestamp=payload["timestamp"]
        )
        print(f"Stored reading from {sensor_id}: {payload['rh']}% RH")
    except json.JSONDecodeError:
        print(f"Invalid JSON payload from topic {msg.topic}")
    except Exception as e:
        print(f"Error processing message: {e}")

def push_to_grafana(aggregated_data):
    """Push aggregated data to Grafana Cloud via InfluxDB line protocol"""
    lines = []
    for sensor_id, avg_rh, min_rh, max_rh, count in aggregated_data:
        # InfluxDB line protocol format: measurement,tag_set field_set timestamp
        line = f"humidity_agg,sensor_id={sensor_id} avg_rh={avg_rh},min_rh={min_rh},max_rh={max_rh},count={count} {int(time.time())}"
        lines.append(line)
    if not lines:
        return
    try:
        response = requests.post(
            GRAFANA_URL,
            headers={"Authorization": f"Bearer {API_KEY}"},
            data="\n".join(lines),
            timeout=10
        )
        if response.status_code == 204:
            print(f"Pushed {len(lines)} aggregated readings to Grafana")
        else:
            print(f"Grafana push failed: {response.status_code} {response.text}")
    except Exception as e:
        print(f"Error pushing to Grafana: {e}")

def main():
    # Initialize telemetry store
    store = TelemetryStore(DB_PATH)
    # Initialize MQTT client
    client = mqtt.Client(userdata={"store": store})
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(MQTT_BROKER, MQTT_PORT, 60)
    # Start MQTT loop in background thread
    client.loop_start()

    last_aggregation = time.time()
    while True:
        try:
            # Run aggregation at specified interval
            if time.time() - last_aggregation >= AGGREGATION_INTERVAL:
                start_time = int(last_aggregation)
                end_time = int(time.time())
                aggregated = store.get_aggregated(start_time, end_time)
                if aggregated:
                    push_to_grafana(aggregated)
                last_aggregation = time.time()
            time.sleep(1)
        except KeyboardInterrupt:
            print("Shutting down...")
            client.loop_stop()
            break
        except Exception as e:
            print(f"Main loop error: {e}")
            time.sleep(5)

if __name__ == "__main__":
    main()

def median_filter(readings):
    """Return median of 3+ sensor readings, reject invalid values first"""
    # Reject readings outside valid 0-100% RH range
    valid = [r for r in readings if 0 <= r <= 100]
    if len(valid) < 2:
        raise ValueError("Not enough valid sensor readings")
    sorted_readings = sorted(valid)
    n = len(sorted_readings)
    if n % 2 == 1:
        return sorted_readings[n//2]
    else:
        return (sorted_readings[n//2 -1] + sorted_readings[n//2]) / 2

def pid_update(self, current_rh, dt):
    error = self.setpoint - current_rh
    # Integral with anti-windup
    self.integral += error * dt
    self.integral = max(-MAX_INTEGRAL, min(MAX_INTEGRAL, self.integral))
    # Derivative of error (not measurement)
    d_error = (error - self.last_error) / dt
    output = (self.kp * error) + (self.ki * self.integral) + (self.kd * d_error)
    # Clamp output to valid range
    return max(0.0, min(100.0, output))

def store_reading(self, sensor_id, rh, offset, timestamp):
    conn = sqlite3.connect(self.db_path)
    c = conn.cursor()
    c.execute('''INSERT INTO readings (sensor_id, rh, calibration_offset, timestamp)
                 VALUES (?, ?, ?, ?)''', (sensor_id, rh, offset, timestamp))
    conn.commit()
    conn.close()

ultimate-humidity-control/
├── firmware/
│   ├── esp32-sensor/          # MicroPython code for ESP32-S3 sensor nodes
│   │   ├── sensor_reading.py  # Code example 1: sensor interfacing
│   │   ├── pid_controller.py  # Code example 2: PID control loop
│   │   └── boot.py            # WiFi/MQTT boot configuration
├── gateway/
│   ├── telemetry_aggregator.py # Code example 3: edge telemetry aggregation
│   └── requirements.txt       # Python dependencies for gateway
├── grafana/
│   ├── dashboard.json         # Pre-built Grafana dashboard
│   └── datasource.json        # InfluxDB datasource config
├── mosquitto/
│   ├── mosquitto.conf         # Secured Mosquitto broker config
│   └── acl.conf               # MQTT access control lists
└── README.md                  # Deployment and calibration instructions