IoT Data Analytics Pipeline

Project Overview:

This project shows a real-time Internet of Things(IoT) data pipeline, combining hardware, software, and visualization tools to capture, process, and analyze environmental data as efficiently and safely as possible by separating functions.

Why This Project?

This project helped me explore key tools and techniques in data engineering and embedded systems while also creating something modular and practical to monitor my house's air conditioning.

Skills learned in this project can be practically applied to monitor and maintain smart homes, gardens, and security applications

More broadly, these skills are useful to embedded systems, data engineering, data analysis, and data science roles

Key Features:

System Design

This system was built to be flexible, scalable, and secure. Each part of the setup has a clear job, working together to capture data, process it efficiently, and show it in a way that's easy to understand. The focus was on making something that works well in real-time and can grow when new sensors or features are added later.

One potential application of this skills from this project is in a smart gardening system. Sensors can monitor soil moisture and sunlight levels, while weather data is integrated to predict rainfall. Based on these inputs, the system can decide when to activate sprinklers or deploy a tarp to protect crops from harsh conditions.

Data Flow

1. Sensors: DHT11 and BME280 sensors measure environmental data
2. Microcontroller: ESP32s sends JSON payloads via MQTT over WiFi to the API on my laptop
3. Backend: FastAPI processes the data and stores it in InfluxDB bucket on laptop
4. Visualization: Grafana queries the database and visualizes the metrics

Implementation

Approaching the project was an iterative process. I'd saved comma-separated values to an SD card and used Bluetooth in past projects and wanted to improve on data transmission protocols, data engineering principles, and security

I tackled this as an end-to-end project to explore new software tools, focusing on affordability, scalability and security. To this end I chose the technical stack and measures that I have.

Hardware

• DHT11 Sensor: Measures temperature and humidity. Chosen because it's familiar and cheap
• BME280 Sensor: Measures temperature and humidity more accurately than the DHT11, as well as barometric pressure. Chosen because it's different and relevant
• Laptop: a good testing ground before moving to cloud services. Chosen because it's available and I wanted to work with open-source, local options
• ESP32 Microcontrollers: Reads sensor data, connects to WiFi and MQTT broker, sends data. Chosen because it's a relatively cheap and powerful edge device
1. Device A: Captures data from a DHT11 sensor
2. Device B: Captures data from a BME280 sensor, extendible to other sensors

Software

I wanted more experience with C++ and Python so I worked with those. I got a chance to work with JSON data, learning how to build and parse them in both languages

void loop() {
  // Check if the ESP32 is connected to the MQTT broker
  if (!client.connected()) {
    reconnect();
  }
  client.loop();

  unsigned long currentMillis = millis();

  // Check if it's time to publish data
  if (currentMillis - lastPublishTime >= publishInterval) {
    lastPublishTime = currentMillis;

    // Read temperature and humidity from DHT11 sensor
    float humidity = dht.readHumidity();
    float temperature = dht.readTemperature();

    // Check if the readings are valid
    if (isnan(humidity) || isnan(temperature)) {
      Serial.println("Failed to read from DHT sensor!");
      return;  // Skip the rest if the sensor readings are invalid
    }

    // Create JSON object with sensor data and metadata
    StaticJsonDocument<200> doc;
    doc["deviceID"] = device;
    doc["sensorName"] = sensorName;
    doc["temperature"] = temperature;
    doc["humidity"] = humidity;
    doc["timestamp"] = currentMillis;  // Timestamp for when the data was read, additional logic optional like deep sleep

    // Serialize the JSON object to a buffer
    char jsonBuffer[512];
    serializeJson(doc, jsonBuffer);

    // Publish the data to the MQTT broker
    Serial.print("Publishing message: ");
    Serial.println(jsonBuffer);
    if (client.publish(mqtt_topic, jsonBuffer)) {
      Serial.println("Message published successfully!");
    } else {
      Serial.println("Failed to publish message.");
    }
  }
}
    

# MQTT message handler
def on_message(client, userdata, msg):
    global latest_sensor_data
    try:
        # Parse the JSON payload
        sensor_data = json.loads(msg.payload.decode())
        
        # Store the latest sensor data
        latest_sensor_data = sensor_data

        # Insert the parsed data into InfluxDB
        asyncio.run(insert_data(sensor_data))
    except Exception as e:
        print(f"Error parsing message: {e}")

# API endpoint to latest reading
@app.get("/latest-data")
async def get_latest_data():
    if latest_sensor_data:
        return {"sensor_data": latest_sensor_data}
    return {"message": "No data received yet."}

Protocols

FULL CODE AND WALKTHROUGH ON GITHUB

Results

Live Monitoring Using Insightful Dashboards: Sensor data from multiple ESP32 devices was transmitted via MQTT and visualized in real-time using Grafana, with a 10-second refresh rate. Grafana dashboards featured time-series graphs, gauge panels, and threshold markers, making trends like temperature fluctuations and light exposure easy to track at a glance.

System Efficiency & Scalability: Lightweight MQTT and asynchronous FastAPI processing ensured low latency and high responsiveness, even with multiple devices sending data. The modular design allowed seamless addition of sensors and integration with external APIs, such as weather services for smarter irrigation decisions. Cloud services can enhance that scalability

Tons of Practice: From tasks relevant to project planning, system design, data engineering, and data analysis, I had the pleasure to do it all at once and learn along the way. I feel so much more confident in setting up data pipelines, secure connections, and using unfamiliar technologies. I can't wait to expand on what I learned here.

Possible Future Improvements

1. API Improvements: Enhance security and explore Node-RED for simplified data ingestion
2. Threshold Alerts: Configure automated notifications for abnormal temperature or humidity levels
3. Network Configuration: Enable remote access to Grafana dashboards for better accessibility
4. Cloud Integration: Use cloud services like AWS Lamda, S3, and Glue
5. Expanded Sensors: Add air quality, motion, or other environmental sensors for broader monitoring capabilities

• Send an E-Mail
• (727) 542-5243
• Wimauma, FL 33598