UE1.4 — Python pour l'IoT

Concept	Description
pyserial	Interface Python → port série USB/COM. Lit les données Arduino/ESP32
asyncio	Programmation asynchrone Python. Permet de lire plusieurs capteurs "simultanément"
Time-series	Données horodatées. Stockage optimal : InfluxDB, TimescaleDB ou CSV avec timestamp Unix

import serial
import csv
from datetime import datetime

ser = serial.Serial("/dev/ttyUSB0", 9600, timeout=1)

with open("donnees_iot.csv", "w", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(["timestamp", "temperature", "humidite"])

    for _ in range(100):
        ligne = ser.readline().decode("utf-8").strip()
        if "," in ligne:
            temp, hum = ligne.split(",")
            ts = datetime.now().isoformat()
            writer.writerow([ts, temp, hum])
            print(f"{ts} — {temp}°C, {hum}%")

ser.close()

Exercice 1 — Acquisition série

a) Écrivez un script Python qui lit les données de l'Arduino sur 60 secondes.

b) Sauvegardez les données dans un fichier CSV avec timestamp.

c) BONUS : Utilisez asyncio pour lire 2 ports série simultanément.

①

Charger un fichier CSV de données capteurs dans Pandas

Savoir expliquer

②

Effectuer un resample (secondes → minutes → heures)

Savoir faire

③

Détecter des anomalies avec z-score et IQR

Savoir analyser

Appliquer une moving average pour lisser les données

Savoir produire

1. Analyse de séries temporelles IoT

Les données IoT sont des séries temporelles. Pandas permet de les manipuler efficacement : chargement, nettoyage, agrégation et visualisation.

import pandas as pd
import matplotlib.pyplot as plt

df = pd.read_csv("donnees_iot.csv", parse_dates=["timestamp"])
df.set_index("timestamp", inplace=True)

# Resample : moyenne par minute
df_min = df.resample("1min").mean()

# Moving average sur 5 points
df["temp_lisse"] = df["temperature"].rolling(window=5).mean()

# Détection anomalies (z-score > 3)
from scipy import stats
z_scores = stats.zscore(df["temperature"].dropna())
anomalies = df[abs(z_scores) > 3]

print(f"Anomalies détectées : {len(anomalies)}")

plt.plot(df.index, df["temperature"], label="Brut")
plt.plot(df.index, df["temp_lisse"], label="Lissé")
plt.scatter(anomalies.index, anomalies["temperature"], color="red")
plt.legend()
plt.show()

Exercice 2 — Analyse Pandas

a) Chargez un fichier CSV de données capteurs dans Pandas.

b) Calculez la moyenne, min, max par heure.

c) Détectez les anomalies avec la méthode IQR.

d) BONUS : Créez un dashboard avec 3 graphiques.

①

Créer une API REST avec Flask

Savoir expliquer

②

Recevoir des données en POST au format JSON

Savoir faire

③

Stocker les données dans SQLite ou InfluxDB

Savoir analyser

④

Mettre en place un webhook pour ESP32

Savoir produire

from flask import Flask, request, jsonify
import sqlite3
from datetime import datetime

app = Flask(__name__)

def init_db():
    conn = sqlite3.connect("iot.db")
    conn.execute("""CREATE TABLE IF NOT EXISTS mesures (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        timestamp TEXT,
        device TEXT,
        temperature REAL,
        humidite REAL
    )""")
    conn.close()

@app.route("/api/mesures", methods=["POST"])
def recevoir_mesure():
    data = request.get_json()
    conn = sqlite3.connect("iot.db")
    conn.execute("INSERT INTO mesures (timestamp, device, temperature, humidite) VALUES (?, ?, ?, ?)",
                 (datetime.now().isoformat(), data.get("device"),
                  data.get("temp"), data.get("hum")))
    conn.commit()
    conn.close()
    return jsonify({"status": "ok"}), 201

@app.route("/api/mesures", methods=["GET"])
def lister_mesures():
    conn = sqlite3.connect("iot.db")
    rows = conn.execute("SELECT * FROM mesures ORDER BY id DESC LIMIT 100").fetchall()
    conn.close()
    return jsonify(rows)

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

Exercice 3 — Flask API

a) Créez une API Flask qui reçoit les données IoT en POST.

b) Configurez l'ESP32 pour envoyer les données à cette API.

c) BONUS : Ajoutez une route GET avec filtre par date.

①

Comprendre l'architecture Grafana + InfluxDB

Savoir expliquer

②

Configurer une source de données InfluxDB

Savoir faire

③

Créer des panels (graphique, jauge, stat)

Savoir analyser

④

Utiliser des variables Grafana pour filtrer

Savoir produire

1. Stack de monitoring IoT

Grafana est la plateforme de visualisation open-source la plus utilisée pour l'IoT. Elle se connecte à InfluxDB (base de données time-series) pour afficher des dashboards en temps réel.

version: "3"
services:
  influxdb:
    image: influxdb:2
    ports:
      - "8086:8086"

  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"
    depends_on:
      - influxdb

Exercice 4 — Dashboard Grafana

a) Installez Grafana et InfluxDB avec Docker.

b) Créez un dashboard avec température et humidité.

c) BONUS : Ajoutez une alerte par email si température > 40°C.

①

Comprendre le concept d'anomaly detection pour IoT

Savoir expliquer

②

Utiliser Isolation Forest pour détecter des anomalies

Savoir faire

③

Comprendre TinyML et TensorFlow Lite Micro

Savoir analyser

④

Déployer un modèle léger sur Arduino

Savoir produire

Concept	Description
Anomaly detection	Identifier des comportements anormaux dans une série de données capteur
Isolation Forest	Algorithme non supervisé qui isole les points aberrants par divisions aléatoires
TinyML	Machine Learning sur microcontrôleur. Modèles compressés < 100KB

import pandas as pd
from sklearn.ensemble import IsolationForest
import matplotlib.pyplot as plt

df = pd.read_csv("donnees_iot.csv")
X = df[["temperature", "humidite"]].values

model = IsolationForest(contamination=0.05, random_state=42)
df["anomalie"] = model.fit_predict(X)

# -1 = anomalie, 1 = normal
anomalies = df[df["anomalie"] == -1]
normaux   = df[df["anomalie"] == 1]

plt.scatter(normaux.index, normaux["temperature"], c="blue", label="Normal")
plt.scatter(anomalies.index, anomalies["temperature"], c="red", label="Anomalie")
plt.legend()
plt.title("Détection d'anomalies — Isolation Forest")
plt.show()

Exercice 5 — ML pour IoT

a) Appliquez Isolation Forest sur un jeu de données capteur.

b) Visualisez les anomalies détectées.

c) BONUS : Déployez un modèle TF Lite sur ESP32.

①

Comprendre l'architecture du Raspberry Pi

Savoir expliquer

②

Utiliser les GPIO du Raspberry Pi

Savoir faire

③

Créer un flux Node-RED pour IoT

Savoir analyser

④

Connecter capteurs et API avec Node-RED

Savoir produire

Concept	Description
Raspberry Pi	Mini-ordinateur Linux ARM. 4 cœurs, WiFi/BT, 40 pins GPIO. Idéal hub IoT
Node-RED	Outil de programmation par flux visuel. Connecte dispositifs IoT sans code
GPIO	Broches du RPi pour capteurs/actionneurs. Contrôle via RPi.GPIO ou pigpio

import RPi.GPIO as GPIO
import time

LED_PIN = 18
GPIO.setmode(GPIO.BCM)
GPIO.setup(LED_PIN, GPIO.OUT)

try:
    while True:
        GPIO.output(LED_PIN, GPIO.HIGH)
        time.sleep(1)
        GPIO.output(LED_PIN, GPIO.LOW)
        time.sleep(1)
except KeyboardInterrupt:
    GPIO.cleanup()

Exercice 6 — Raspberry Pi

a) Installez Node-RED sur Raspberry Pi.

b) Créez un flux qui lit un capteur MQTT et l'affiche.

c) BONUS : Connectez Node-RED à Grafana via HTTP.