esp32micropythonarduino-esp32

How to Listen to Firebase Realtime Database as a Stream (Subscription) on ESP32 Micro Controller using MicroPython

How to Listen to Firebase Realtime Database as a Stream (Subscription) on ESP32 Micro Controller (mini computers) for IOT Projects using MicroPython

Solution

  • After spending three days searching and troubleshooting, I finally found a way to subscribe to Firebase Realtime Database updates as a persistent stream on an ESP32 using MicroPython. Here’s how I did it:
    Listen to Firebase Realtime Database as a Stream (Subscription) on ESP32 using MicroPython

    WiFi Connection (wifi.py)
    This script connects the ESP32 to WiFi.

    import network

def connect_wifi():
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        print(f"Connecting to {"network"}...")
        wlan.connect("network", "Pass1!@#")
        while not wlan.isconnected():
            pass
    print("Connected!", wlan.ifconfig())

    Firebase Stream Listener (main.py)

    This script sets up a persistent connection to Firebase and listens for real-time updates.

    import network
import urequests
import ujson
import wifi
import time

# 🔧 Configuration (Keep sensitive details private)
FIREBASE_URL = "<YOUR_FIREBASE_URL>.json"  # Replace with your actual Firebase URL
WATCH_KEYS = ["<YOUR_KEY>"]  # Replace with specific keys you want to track (or leave empty to track all)

def process_firebase_update(path, data):
    """Handles Firebase updates based on the received path and data."""
    if isinstance(data, dict):
        for key, value in data.items():
            print(f"🔄 Update: {key} → {value}")
    elif isinstance(data, str):
        print(f"🔄 Direct Update: {path} → {data}")

def listen_to_firebase():
    """Subscribe to Firebase stream (persistent connection)."""
    while True:
        try:
            print("🔥 Listening to Firebase Stream (Persistent)...")

            headers = {
                "Accept": "text/event-stream",
                "Connection": "keep-alive"
            }

            response = urequests.get(FIREBASE_URL, headers=headers, stream=True)
            print("✅ [Connected to Firebase Stream]")

            while True:
                line = response.raw.readline()
                if not line:
                    raise Exception("⚠️ Connection lost. Reconnecting...")

                line = line.strip()
                if line.startswith(b"data: "):  
                    try:
                        json_data = line[6:]  
                        parsed_data = ujson.loads(json_data)

                        if parsed_data is None:
                            continue

                        print("🔥 Firebase Update:", parsed_data)

                        # Extracting data
                        if isinstance(parsed_data, dict) and "data" in parsed_data:
                            data = parsed_data["data"]
                            path = parsed_data.get("path", "/")

                            # If watching specific keys, filter updates
                            if not WATCH_KEYS or any(key in path for key in WATCH_KEYS):
                                process_firebase_update(path, data)

                    except Exception as e:
                        print("❌ JSON Parse Error:", str(e))

        except Exception as e:
            print("❌ Firebase Stream Error:", str(e))
            print("🔄 Reconnecting in 5 seconds...")
            time.sleep(5)

# Run
wifi.connect_wifi()
listen_to_firebase()

    This solution: