Practical case: MQTT rain gauge with Raspberry Pi 5

#!/usr/bin/env python3
"""
Objective: rain-gauge-mqtt-logger
Device: Raspberry Pi 5 + tipping bucket rain gauge + DS3231 RTC + local SQLite log
"""

import os
import time
import json
import sqlite3
import queue
import threading
from datetime import datetime, timezone
import paho.mqtt.client as mqtt

DB_FILE = "rain_log.db"
MQTT_BROKER = "localhost"  # Change to your broker IP if external
MQTT_PORT = 1883
MQTT_TOPIC = "weather/rain_gauge"
MM_PER_TIP = 0.2794  # Calibration metric: mm of rain per bucket tip
RAIN_PIN = 17

# Hardware Abstraction / Mocking
MOCK_MODE = os.environ.get("MOCK_HARDWARE", "0") == "1"

if MOCK_MODE:
    print("[INFO] Running in MOCK mode. Hardware GPIO bypassed.")
    class MockButton:
        def __init__(self, pin, pull_up, bounce_time):
            self.pin = pin
            self.when_pressed = None
            self._running = True
            self._thread = threading.Thread(target=self._simulate_rain, daemon=True)
            self._thread.start()

        def _simulate_rain(self):
            # Simulate a rain tip every 5 seconds
            while self._running:
                time.sleep(5)
                if self.when_pressed:
                    self.when_pressed()

    Button = MockButton
else:
    from gpiozero import Button

# Global Queue for thread-safe interrupt handling
tip_queue = queue.Queue()

def setup_database():
    """Initializes the SQLite database and creates the schema if it doesn't exist."""
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS rain_events (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT NOT NULL,
            daily_tips INTEGER NOT NULL,
            daily_mm REAL NOT NULL
        )
    ''')
    conn.commit()
    return conn

def bucket_tipped():
    """Hardware interrupt callback. Kept extremely short."""
    # Put a timestamp into the queue. The OS time is synced via the DS3231 RTC.
    tip_time = datetime.now(timezone.utc).isoformat()
    tip_queue.put(tip_time)

def publish_mqtt(client, payload):
    """Publishes the JSON payload to the MQTT broker."""
    try:
        client.publish(MQTT_TOPIC, json.dumps(payload), qos=1)
        print(f"[MQTT] Published: {payload}")
    except Exception as e:
        print(f"[MQTT ERROR] Failed to publish: {e}")

def main():
    print("[INFO] Starting Rain Gauge MQTT Logger...")

    # Initialize Database
    db_conn = setup_database()
    db_cursor = db_conn.cursor()

    # Initialize MQTT Client
    mqtt_client = mqtt.Client()
    try:
        mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)
        mqtt_client.loop_start()
        print(f"[INFO] Connected to MQTT Broker at {MQTT_BROKER}:{MQTT_PORT}")
    except Exception as e:
        print(f"[WARN] Could not connect to MQTT broker: {e}. Running in local-only mode.")

    # Initialize Hardware Interrupt
    # bounce_time handles the mechanical bouncing of the reed switch
    rain_gauge = Button(RAIN_PIN, pull_up=True, bounce_time=0.1)
    rain_gauge.when_pressed = bucket_tipped

    daily_tips = 0
    current_date = datetime.now(timezone.utc).date()

    print("[INFO] System ready. Waiting for rain events...")

    try:
        while True:
            try:
                # Block until an event occurs (timeout allows clean exit on KeyboardInterrupt)
                tip_time_str = tip_queue.get(timeout=1.0)

                # Check for day rollover to reset daily counters
                event_date = datetime.fromisoformat(tip_time_str).date()
                if event_date > current_date:
                    daily_tips = 0
                    current_date = event_date

                daily_tips += 1
                daily_mm = round(daily_tips * MM_PER_TIP, 4)

                # 1. Save to local SQLite database
                db_cursor.execute(
                    "INSERT INTO rain_events (timestamp, daily_tips, daily_mm) VALUES (?, ?, ?)",
                    (tip_time_str, daily_tips, daily_mm)
                )
                db_conn.commit()
                print(f"[DB] Logged tip {daily_tips} at {tip_time_str} ({daily_mm} mm total)")

                # 2. Publish to MQTT
                payload = {
                    "timestamp": tip_time_str,
                    "event": "bucket_tip",
                    "daily_tips": daily_tips,
                    "daily_mm": daily_mm
                }
                publish_mqtt(mqtt_client, payload)

            except queue.Empty:
                continue

    except KeyboardInterrupt:
        print("\n[INFO] Shutting down logger...")
    finally:
        db_conn.close()
        mqtt_client.loop_stop()
        mqtt_client.disconnect()

if __name__ == "__main__":
    main()