ML Edge pentru detectarea bolilor culturilor: TensorFlow Lite pe Raspberry Pi
În fiecare an, bolile plantelor și dăunătorii culturilor distrug până la 40% din producție agricultura mondială, provocând pierderi economice mai mari decât 220 de miliarde de dolari conform datelor FAO. În Italia, țările viticole, livezile și grădinile de legume sunt vulnerabile la boli endemice ale plantelor, cum ar fi mucegaiul viței de vie (Plasmopara viticola), mucegaiul praf (Erysiphe necator) și crusta de mere (Venturia inaequalis), fiecare capabil să șteargă un an întreg productivă dacă nu este interceptată la timp.
Diagnosticul precoce a fost încredințat istoric experienței vizuale a fermierului sau, în cel mai bun caz, la controale periodice de către agronomi. Ambele abordări suferă de aceeași limitare: latența Există un interval între debutul bolii și intervenția fitosanitar și este adesea de zile sau săptămâni timp în care ciuperca sau bacteria se răspândește ca un incendiu. Fereastra de timp pentru a tratament eficient și minim: pentru mucegaiul pufos al viței de vie, de exemplu, este redus la soare 48-72 ore de la apariţia primelor leziuni.
Inteligența artificială aplicată vederii computerizate oferă un răspuns concret: modele de învățarea profundă instruită pe setul de date PlantVillage (54.309 imagini, 38 clase) obține acuratețe mai mare decât 97% în recunoaşterea bolilor foliare. Adevăratul blocaj nu mai este exactitatea modelului, ci desfășurare pe teren. Într-o vie de pe Apenini Toscan-Emilian sau pe o livadă de citrice din Sicilia, conexiunea la internet este adesea absentă sau nesigură. Trimiterea imaginilor către cloud pentru deducere nu este practic. Soluția șiEdge ML: modelul ruleaza direct pe dispozitiv, in teren, fara conectivitate, cu latenta mai mica de al doilea.
Acest articol construiește o conductă completă de detectare a bolilor pe Raspberry Pi 5 de la zero folosind TensorFlow Lite: de la antrenament de bază pe model cu transfer de învățare pe MobileNetV3, la cuantizarea int8, la implementarea cu inferență Python în timp real, la trimiterea rezultatelor MQTT la un tablou de bord la distanță. Fiecare pas este însoțit de cod de lucru, numere de referință considerații verificate și practice pentru contextul agricol italian.
Ce veți învăța în acest articol
- de ce edge ML depășește cloud-ul pentru aplicațiile agricole în câmp deschis
- Cum să pregătiți și să măriți setul de date PlantVillage pentru un antrenament eficient
- Transferați învățarea cu MobileNetV3Small pe TensorFlow/Keras, pas cu pas
- Conversie model în TFLite cu cuantificare post-antrenament (float32 → int8)
- Configurare hardware Raspberry Pi 5 cu Camera Module 3 și inferență Python completă
- Conductă integrată: captură → preprocesare → inferență → MQTT → tablou de bord
- Benchmark-uri de latență și consum de energie pe Pi 4, Pi 5, Coral TPU, Jetson Nano
- Principalele boli ale plantelor italiene și culturile vizate prioritare
- Cele mai bune practici de implementare și anti-modele comune
Poziție în seria FoodTech
Acesta este al doilea articol din serie FoodTech, dedicat tehnologiilor digitale pentru lanțul de aprovizionare agroalimentară. Seria acoperă IoT agricol, viziune computerizată, blockchain pentru trasabilitate, prognoza cererii și multe altele.
Seria FoodTech - Toate articolele
| # | Titlu | Nivel | Stat |
|---|---|---|---|
| 1 | Conducta agricolă IoT: senzori, MQTT și serii temporale pe InfluxDB | Intermediar | Disponibil |
| 2 | ML Edge pentru detectarea bolilor culturilor: TFLite pe Raspberry Pi — Eşti aici | Avansat | Actual |
| 3 | Satellite API și NDVI: monitorizarea culturilor cu Sentinel-2 | Intermediar | Disponibil |
| 4 | Viziunea computerizată pentru calitatea alimentelor: defecte, greutate și gradare | Avansat | Disponibil |
| 5 | Trasabilitatea alimentelor Blockchain: Hyperledger Fabric și GS1 | Avansat | Disponibil |
| 6 | Conformitate FSMA 204: Trasabilitate digitală cu Python și API | Intermediar | Disponibil |
| 7 | Automatizarea agriculturii pe verticală: Controlul climei cu PLC și ML | Avansat | Disponibil |
| 8 | Prognoza cererii de alimente: Prophet, LSTM și Feature Engineering | Intermediar | Disponibil |
| 9 | Tabloul de bord al fermei în timp real: Angular, WebSocket și InfluxDB | Intermediar | Disponibil |
| 10 | Reziliența lanțului de aprovizionare: optimizare cu instrumente OR-Tools și AI | Avansat | Disponibil |
Elemente fundamentale Edge ML: Cloud vs Edge pentru agricultură
Înainte de a scrie o linie de cod, este esențial să înțelegeți de ce abordarea edge este singura practice pentru detectarea bolilor în câmp deschis. Aceasta nu este o alegere stilistică: și o necesitate arhitecturală impusă de contextul fizic al agriculturii.
Limitele norului în agricultură
O aplicație de detectare a bolilor pe cloud ar funcționa astfel: camera face clic o fotografie, o trimite prin rețea la serverul cloud, modelul rulează pe server, rezultatul revine la dispozitiv. Într-un birou cu fibră optică, acest ciclu durează mai puțin de 200 ms. Într-o vie în Toscana sau într-o plantație de citrice din Calabria, variabilele se schimbă radical:
Cloud vs Edge: comparație în funcție de context agricol
| Caracteristică | Abordarea cloud | Abordarea marginii |
|---|---|---|
| Este necesară conectivitate | Continuă, bandă de 1+ Mbps | Opțional, numai pentru sincronizarea datelor |
| Latența de inferență | 200 ms - 5s (depinde de rețea) | 50-500 ms (local) |
| Cost de exploatare | Ridicat (apeluri API, stocare, lățime de bandă) | Scăzut (hardware unic) |
| Confidențialitate acordată | Imagini corporative în cloud | Datele rămân on-premise |
| Operare offline | Imposibil | Complet |
| Scalabilitatea modelului | Nelimitat | Limitat de RAM/CPU |
| Actualizare model | Imediat | Necesită implementare fizică sau OTA |
| Consumul de energie | Scăzut (dispozitiv) + ridicat (centru de date) | Totul de pe dispozitiv (3-15W) |
TensorFlow vs TensorFlow Lite: diferențe cheie
TensorFlow (TF) este cadrul complet de instruire și inferență, optimizat pentru GPU și servere. TensorFlow Lite (TFLite) este versiunea comprimată și optimizată pentru dispozitive încorporate și mobile. Diferențele practice sunt semnificative:
- Dimensiunea timpului de rulare: Timp de rulare TFLite și aproximativ 1MB în C++ (vs 400MB+ pentru TF complet). Pe Raspberry Pi, diferența se traduce într-o ocupare RAM de 50-100MB în loc de 2-4GB.
- Operatori acceptați: TFLite acceptă un subset de operațiuni TF. Modele cu operațiuni personalizate sau straturi non-standard necesită delegare sau alternativă la CPU.
- format .tflite: Schema FlatBuffers, optimizată pentru acces direct la memorie fără analizare. Modelul este mapat în memorie fără deserializare.
- Accelerare hardware: TFLite acceptă delegații XNNPACK (SIMD pe ARM), Delegat GPU, delegat Coral Edge TPU și NNAPI pe Android.
Cuantizare: float32 la int8
Cea mai de impact tehnică pentru optimizarea marginilor și cuantizarea: cel reducerea preciziei numerice a greutăților de la virgulă mobilă de 32 de biți la numere întregi de 8 biți. Efectul practic și triplu:
Efectele cuantizării int8 pe MobileNetV3Small
| Metric | plutitor32 | int8 (PTQ) | Delta |
|---|---|---|---|
| Dimensiunea modelului | 9,8 MB | 2,6 MB | -73% |
| Inferență RAM (Pi 5) | ~180 MB | ~52 MB | -71% |
| Latența de inferență (Pi 5) | ~180 ms | ~65 ms | -64% |
| Precizia PlantVillage | 97,2% | 96,8% | -0,4% |
| Consumul de energie | ~5,2 W | ~3,8W | -27% |
Benchmark pe Raspberry Pi 5 (2,4 GHz, 8 GB RAM). Valori măsurate cu modelul reglat fin pe 38 de clase PlantVillage.
Pierderea de 0,4% a preciziei este practic irelevantă în domeniu: variabilitatea naturală condițiile de iluminare, unghiul și stadiul bolii introduc incertitudini mult mai mare. Cu toate acestea, câștigul în latență și dimensiune este decisiv pentru implementarea încorporată.
Set de date PlantVillage: pregătire și creștere
Setul de date de referință pentru detectarea bolilor foliare și Plant Village, creat de Hughes și Salathé (Penn State University) și publicat în 2016. Conține mai multe 54.309 imagini fotografiat in conditii controlate de laborator, organizat in 38 de clase acoperind 14 specii de plante și boli aferente.
Structura setului de date
Clasele PlantVillage: Specii și principalele boli
| Specie | Boală | Agent cauzal | Imagini |
|---|---|---|---|
| Viță de vie (struguri) | Putregaiul negru | Guignardia bidwellii | 1.180 |
| Viță de vie (struguri) | Momeală / Rujeola Neagră | Phaeoacremonium spp. | 1.383 |
| Viță de vie (struguri) | Brâncirea frunzelor | Pseudocercospora vitis | 1.076 |
| Viță de vie (struguri) | Sănătos | — | 423 |
| roșie (roșie) | Blight timpuriu | Alternaria solani | 1.000 |
| roșie (roșie) | Târzie | Phytophthora infestans | 1.909 |
| roșie (roșie) | Mucegai pentru frunze | Tawny Passalora | 952 |
| roșie (roșie) | Septoria Leaf Spot | Septoria lycopersici | 1.771 |
| Măr (măr) | Crusta de măr (Scab) | Venturia inaequalis | 630 |
| Măr (măr) | Putregaiul negru | Botryosphaeria obtusa | 621 |
| Măr (măr) | Rugina de mere de cedru | Gymnosporangium juniperi | 275 |
| Porumb (Porumb) | Rugina comună | Puccinia sorghi | 1.192 |
| Cartofi (cartofi) | Blight timpuriu | Alternaria solani | 1.000 |
| Cartofi (cartofi) | Târzie | Phytophthora infestans | 1.000 |
| ... | + alte 24 de clase | — | — |
Descărcarea și pregătirea setului de date
Setul de date este disponibil gratuit pe Kaggle și Hugging Face. Mai jos este scriptul de descărcare, structurarea în foldere și trenul/validarea/testul împărțit:
# setup_dataset.py
# Download e preparazione dataset PlantVillage per training TFLite
import os
import shutil
import random
from pathlib import Path
import tensorflow as tf
import numpy as np
# ---- CONFIGURAZIONE ----
DATASET_ROOT = Path("./data/plantvillage")
SPLITS = {"train": 0.70, "val": 0.15, "test": 0.15}
IMG_SIZE = (224, 224)
SEED = 42
random.seed(SEED)
np.random.seed(SEED)
def create_splits(source_dir: Path, dest_dir: Path, splits: dict) -> dict:
"""
Crea split train/val/test da una cartella organizzata per classe.
Ritorna statistiche sullo split.
"""
stats = {"train": 0, "val": 0, "test": 0}
for class_dir in sorted(source_dir.iterdir()):
if not class_dir.is_dir():
continue
images = list(class_dir.glob("*.jpg")) + list(class_dir.glob("*.JPG"))
random.shuffle(images)
n_total = len(images)
n_train = int(n_total * splits["train"])
n_val = int(n_total * splits["val"])
split_images = {
"train": images[:n_train],
"val": images[n_train:n_train + n_val],
"test": images[n_train + n_val:]
}
for split_name, split_imgs in split_images.items():
target_dir = dest_dir / split_name / class_dir.name
target_dir.mkdir(parents=True, exist_ok=True)
for img_path in split_imgs:
shutil.copy2(img_path, target_dir / img_path.name)
stats[split_name] += 1
return stats
def build_dataset(data_dir: Path, subset: str, batch_size: int = 32) -> tf.data.Dataset:
"""
Costruisce un tf.data.Dataset con preprocessing e augmentation.
"""
is_training = (subset == "train")
dataset = tf.keras.utils.image_dataset_from_directory(
str(data_dir / subset),
image_size=IMG_SIZE,
batch_size=batch_size,
shuffle=is_training,
seed=SEED,
label_mode="categorical"
)
# Preprocessing: normalizzazione in [-1, 1] per MobileNetV3
normalization = tf.keras.layers.Rescaling(1.0 / 127.5, offset=-1)
if is_training:
# Data augmentation solo sul training set
augmentation = tf.keras.Sequential([
tf.keras.layers.RandomFlip("horizontal_and_vertical"),
tf.keras.layers.RandomRotation(0.2),
tf.keras.layers.RandomZoom(0.15),
tf.keras.layers.RandomBrightness(0.2),
tf.keras.layers.RandomContrast(0.2),
])
dataset = dataset.map(
lambda x, y: (augmentation(normalization(x), training=True), y),
num_parallel_calls=tf.data.AUTOTUNE
)
else:
dataset = dataset.map(
lambda x, y: (normalization(x), y),
num_parallel_calls=tf.data.AUTOTUNE
)
return dataset.cache().prefetch(tf.data.AUTOTUNE)
def get_class_names(data_dir: Path) -> list:
"""Restituisce la lista ordinata delle classi."""
train_dir = data_dir / "train"
return sorted([d.name for d in train_dir.iterdir() if d.is_dir()])
if __name__ == "__main__":
# Assumiamo che PlantVillage raw sia in ./data/plantvillage_raw/
raw_dir = Path("./data/plantvillage_raw")
print("Creazione split dataset...")
stats = create_splits(raw_dir, DATASET_ROOT, SPLITS)
print(f"Split completato:")
print(f" Train: {stats['train']} immagini")
print(f" Val: {stats['val']} immagini")
print(f" Test: {stats['test']} immagini")
print(f" Totale: {sum(stats.values())} immagini")
classes = get_class_names(DATASET_ROOT)
print(f"\nClassi trovate ({len(classes)}):")
for i, cls in enumerate(classes):
print(f" {i:2d}. {cls}")Tehnici de creștere a datelor pentru imaginile frunzelor
Imaginile PlantVillage sunt achiziționate pe un fundal uniform în laborator. A generaliza la condițiile reale de câmp (variații de lumină, fundal natural, unghiuri diferite), tehnicile de augmentare sunt fundamentale. Pe lângă transformările geometrice și fotometrice standardelor deja incluse în codul de mai sus, există tehnici specifice domeniului agricol:
- Ștergere aleatorie / Decupare: Întunecă dreptunghiuri aleatorii ale imaginii (simulează ocluzii de la suprapunerea frunzelor, umbrelor, mașinilor de lucru). Reduce supraadaptarea caracteristici locale și îmbunătățește robustețea modelului cu 2-3% pe imaginile reale.
- MixUp și CutMix: Combinați două imagini cu greutăți aleatorii (MixUp) sau înlocuiți o regiune cu un petic dintr-o altă imagine (CutMix). Deosebit de eficient pentru cursuri vizual asemănătoare (de exemplu, tutun timpuriu versus tutun târziu la tomate).
- Jitter agresiv de culoare: Bolile foliare prezintă simptome de culoare diferită în funcție de stadiu, expunerea la soare și condițiile climatice. Mărirea intervalului de variație a nuanței/saturației ajută la generalizare.
- Înlocuirea fundalului: Tehnici avansate înlocuiesc fundalul alb a laboratorului cu imagini reale de teren. Necesită segmentare semantică preliminară dar îmbunătățește drastic performanța imaginilor din câmp.
Model Training: Transfer Learning with MobileNetV3
MobileNetV3Small este arhitectura ideală pentru acest caz de utilizare: concepută special pentru inferența pe dispozitivele mobile și încorporate oferă un echilibru excelent între precizie și complexitatea de calcul. În comparație cu ResNet50 sau EfficientNetB4, necesită De 100 de ori mai puține operațiuni prin inferență, menținând în același timp acuratețe competitive.
Opțiuni de arhitectură și design
Transferați învățarea de la ImageNet pre-instruit și strategia câștigătoare: primele straturi ale rețelele au învățat deja să recunoască texturile, marginile și modelele generale de culoare. Pentru bolile foliare, aceste caracteristici de nivel scăzut sunt direct reutilizabile. Reglarea fină a ultimelor straturi (și opțional a ultimelor blocuri) adaptează modelul la domeniul specific.
# train_model.py
# Training MobileNetV3Small con transfer learning per classificazione malattie
import tensorflow as tf
import numpy as np
import json
from pathlib import Path
from datetime import datetime
from setup_dataset import build_dataset, get_class_names, DATASET_ROOT, IMG_SIZE
# ---- IPERPARAMETRI ----
NUM_CLASSES = 38 # Classi PlantVillage
BATCH_SIZE = 32
EPOCHS_FROZEN = 15 # Fase 1: solo classification head
EPOCHS_FINETUNE = 20 # Fase 2: ultimi blocchi sbloccati
BASE_LR = 1e-3
FINETUNE_LR = 1e-5
DROPOUT_RATE = 0.3
MODEL_DIR = Path("./models")
MODEL_DIR.mkdir(exist_ok=True)
def build_model(num_classes: int) -> tf.keras.Model:
"""
Costruisce il modello: MobileNetV3Small pretrained + classification head.
"""
# Base model senza top (include_top=False)
base_model = tf.keras.applications.MobileNetV3Small(
input_shape=(*IMG_SIZE, 3),
include_top=False,
weights="imagenet",
include_preprocessing=False # preprocessing già nel dataset pipeline
)
base_model.trainable = False # Freeze per fase 1
# Classification head
inputs = tf.keras.Input(shape=(*IMG_SIZE, 3))
x = base_model(inputs, training=False)
x = tf.keras.layers.GlobalAveragePooling2D()(x)
x = tf.keras.layers.BatchNormalization()(x)
x = tf.keras.layers.Dropout(DROPOUT_RATE)(x)
x = tf.keras.layers.Dense(256, activation="relu")(x)
x = tf.keras.layers.Dropout(DROPOUT_RATE / 2)(x)
outputs = tf.keras.layers.Dense(num_classes, activation="softmax")(x)
model = tf.keras.Model(inputs, outputs, name="mobilenetv3_plantdisease")
return model, base_model
def compile_model(model: tf.keras.Model, lr: float) -> tf.keras.Model:
"""Compila il modello con optimizer e metriche."""
model.compile(
optimizer=tf.keras.optimizers.Adam(learning_rate=lr),
loss="categorical_crossentropy",
metrics=[
"accuracy",
tf.keras.metrics.Precision(name="precision"),
tf.keras.metrics.Recall(name="recall"),
tf.keras.metrics.AUC(name="auc")
]
)
return model
def get_callbacks(phase: str) -> list:
"""Callbacks per entrambe le fasi di training."""
timestamp = datetime.now().strftime("%Y%m%d_%H%M")
return [
tf.keras.callbacks.ModelCheckpoint(
filepath=str(MODEL_DIR / f"best_{phase}.keras"),
monitor="val_accuracy",
save_best_only=True,
mode="max",
verbose=1
),
tf.keras.callbacks.EarlyStopping(
monitor="val_accuracy",
patience=5,
restore_best_weights=True,
verbose=1
),
tf.keras.callbacks.ReduceLROnPlateau(
monitor="val_loss",
factor=0.5,
patience=3,
min_lr=1e-7,
verbose=1
),
tf.keras.callbacks.TensorBoard(
log_dir=f"./logs/{phase}_{timestamp}",
histogram_freq=1
)
]
def train():
print("Caricamento dataset...")
train_ds = build_dataset(DATASET_ROOT, "train", BATCH_SIZE)
val_ds = build_dataset(DATASET_ROOT, "val", BATCH_SIZE)
test_ds = build_dataset(DATASET_ROOT, "test", BATCH_SIZE)
class_names = get_class_names(DATASET_ROOT)
print(f"Classi: {len(class_names)}")
# Salva class names per l'inferenza
with open(MODEL_DIR / "class_names.json", "w") as f:
json.dump(class_names, f, indent=2)
# ---- FASE 1: Training solo classification head ----
print("\n=== FASE 1: Training classification head (base model frozen) ===")
model, base_model = build_model(len(class_names))
model = compile_model(model, BASE_LR)
model.summary()
history_phase1 = model.fit(
train_ds,
validation_data=val_ds,
epochs=EPOCHS_FROZEN,
callbacks=get_callbacks("phase1"),
verbose=1
)
print(f"Fase 1 completata. Val accuracy: {max(history_phase1.history['val_accuracy']):.4f}")
# ---- FASE 2: Fine-tuning ultimi blocchi ----
print("\n=== FASE 2: Fine-tuning (ultimi 2 blocchi sbloccati) ===")
# Sblocca ultimi 30 layer del base model (gli ultimi 2 blocchi inv-res)
base_model.trainable = True
for layer in base_model.layers[:-30]:
layer.trainable = False
n_trainable = sum(1 for l in model.layers if l.trainable)
print(f"Layer addestrabili: {n_trainable}/{len(model.layers)}")
# LR molto basso per fine-tuning (evita catastrofic forgetting)
model = compile_model(model, FINETUNE_LR)
history_phase2 = model.fit(
train_ds,
validation_data=val_ds,
epochs=EPOCHS_FINETUNE,
callbacks=get_callbacks("phase2"),
verbose=1
)
best_val_acc = max(history_phase2.history["val_accuracy"])
print(f"Fase 2 completata. Val accuracy: {best_val_acc:.4f}")
# ---- VALUTAZIONE SU TEST SET ----
print("\n=== Valutazione finale su test set ===")
test_loss, test_acc, test_prec, test_rec, test_auc = model.evaluate(test_ds, verbose=1)
print(f"Test Accuracy: {test_acc:.4f} ({test_acc*100:.2f}%)")
print(f"Test Precision: {test_prec:.4f}")
print(f"Test Recall: {test_rec:.4f}")
print(f"Test AUC: {test_auc:.4f}")
# Salva modello finale in formato SavedModel
model.save(str(MODEL_DIR / "plantdisease_final.keras"))
print(f"\nModello salvato in {MODEL_DIR}/plantdisease_final.keras")
return model
if __name__ == "__main__":
# Configura GPU memory growth (evita OOM su GPU con poca VRAM)
gpus = tf.config.list_physical_devices("GPU")
for gpu in gpus:
tf.config.experimental.set_memory_growth(gpu, True)
trained_model = train()Rezultate așteptate de la antrenament
Pe un set complet de date PlantVillage (38 de clase), modelul MobileNetV3Small a fost reglat fin de obicei realizează:
- Faza 1 (15 epoci): Precizie Val 91-93%, antrenament ~45 min pe GPU T4
- Faza 2 (20 de epoci): Precizie Val 96-97%, antrenament ~2h pe GPU T4
- Seturi de testare: Precizie 96,5-97,5%, scor mediu F1 0,964
- Clase dificile: Tutun timpuriu versus tutun târziu la tomate (F1 ~0,91)
- Cursuri ușoare: Plante sănătoase vs plante bolnave (F1 ~0,99)
Conversie la TFLite cu cuantizare post-antrenament
Modelul Keras antrenat trebuie convertit în format .tflite pentru execuție pe Raspberry Pi. Procesul de conversie include opțional cuantificare, care reduce drastic dimensiunea și îmbunătățește viteza de inferență.
Trei moduri de cuantizare
TFLite acceptă trei niveluri de cuantizare, cu diferite compromisuri între precizie/performanță:
- Cuantificarea intervalului dinamic: Numai greutățile sunt cuantificate (float32 → int8). Activările rămân float32 în timpul execuției. Nu este necesar un set de date de calibrare. Reducere dimensiunea cu 75%, latența îmbunătățită cu 20-30%. Punctul de plecare recomandat.
- Cuantizarea întregului întreg (int8): Greutăți ȘI activări cuantificate în int8. Necesită un set de date de calibrare reprezentativ (100-500 de imagini). Reducerea dimensiunii 75%, latența îmbunătățită cu 50-70%. Necesar pentru accelerarea hardware (Coral TPU).
- Cuantificare Float16: Greutăți în float16. Util în principal pentru GPU-uri. Pe procesoarele ARM nu aduce avantaje semnificative față de float32.
# convert_to_tflite.py
# Conversione modello Keras a TFLite con quantizzazione int8
import tensorflow as tf
import numpy as np
import json
from pathlib import Path
from setup_dataset import build_dataset, DATASET_ROOT
MODEL_DIR = Path("./models")
TFLITE_DIR = Path("./tflite_models")
TFLITE_DIR.mkdir(exist_ok=True)
# Numero di batch per calibrazione (100-500 immagini raccomandate)
N_CALIBRATION_BATCHES = 10
BATCH_SIZE_CALIB = 16
def get_representative_dataset():
"""
Generatore per dataset di calibrazione int8.
Deve coprire la distribuzione reale dei dati di input.
"""
# Usa il validation set come dataset di calibrazione
val_ds = build_dataset(DATASET_ROOT, "val", batch_size=BATCH_SIZE_CALIB)
count = 0
for images, _ in val_ds.take(N_CALIBRATION_BATCHES):
for img in images:
# Input deve essere float32 nel range [-1, 1] (gia normalizzato dal dataset)
yield [tf.expand_dims(img, axis=0)]
count += 1
if count % 5 == 0:
print(f" Calibrazione batch {count}/{N_CALIBRATION_BATCHES}...")
def convert_float32(model_path: Path, output_path: Path) -> dict:
"""Conversione base senza quantizzazione (baseline)."""
converter = tf.lite.TFLiteConverter.from_keras_model(
tf.keras.models.load_model(str(model_path))
)
converter.optimizations = [] # Nessuna ottimizzazione
tflite_model = converter.convert()
output_path.write_bytes(tflite_model)
size_mb = len(tflite_model) / (1024 * 1024)
print(f"Float32 model: {size_mb:.2f} MB → {output_path}")
return {"path": str(output_path), "size_mb": size_mb, "type": "float32"}
def convert_dynamic_range(model_path: Path, output_path: Path) -> dict:
"""Dynamic range quantization: pesi int8, attivazioni float32."""
converter = tf.lite.TFLiteConverter.from_keras_model(
tf.keras.models.load_model(str(model_path))
)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
tflite_model = converter.convert()
output_path.write_bytes(tflite_model)
size_mb = len(tflite_model) / (1024 * 1024)
print(f"Dynamic range model: {size_mb:.2f} MB → {output_path}")
return {"path": str(output_path), "size_mb": size_mb, "type": "dynamic_range"}
def convert_int8_full(model_path: Path, output_path: Path) -> dict:
"""Full integer quantization: pesi E attivazioni in int8."""
converter = tf.lite.TFLiteConverter.from_keras_model(
tf.keras.models.load_model(str(model_path))
)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
converter.representative_dataset = get_representative_dataset
# Forza input/output in int8 per compatibilità con Coral TPU
converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]
converter.inference_input_type = tf.uint8
converter.inference_output_type = tf.uint8
print("Quantizzazione int8 con dataset di calibrazione...")
tflite_model = converter.convert()
output_path.write_bytes(tflite_model)
size_mb = len(tflite_model) / (1024 * 1024)
print(f"Int8 full model: {size_mb:.2f} MB → {output_path}")
return {"path": str(output_path), "size_mb": size_mb, "type": "int8_full"}
def benchmark_tflite_model(model_path: Path, test_images: np.ndarray, n_runs: int = 100) -> dict:
"""
Benchmarka un modello TFLite: latenza media, p95, p99.
"""
import time
interpreter = tf.lite.Interpreter(model_path=str(model_path))
interpreter.allocate_tensors()
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()
input_dtype = input_details[0]["dtype"]
input_scale, input_zero_point = input_details[0].get("quantization", (1.0, 0))
latencies = []
for i in range(n_runs):
img = test_images[i % len(test_images)]
# Prepara input nel formato corretto per il modello
if input_dtype == np.uint8:
# Modello int8: converti da float [-1,1] a uint8 [0,255]
img_input = ((img + 1.0) * 127.5).astype(np.uint8)
else:
img_input = img.astype(np.float32)
img_input = np.expand_dims(img_input, axis=0)
t_start = time.perf_counter()
interpreter.set_tensor(input_details[0]["index"], img_input)
interpreter.invoke()
output = interpreter.get_tensor(output_details[0]["index"])
t_end = time.perf_counter()
latencies.append((t_end - t_start) * 1000) # ms
latencies_arr = np.array(latencies)
return {
"n_runs": n_runs,
"mean_ms": float(np.mean(latencies_arr)),
"std_ms": float(np.std(latencies_arr)),
"p50_ms": float(np.percentile(latencies_arr, 50)),
"p95_ms": float(np.percentile(latencies_arr, 95)),
"p99_ms": float(np.percentile(latencies_arr, 99)),
"min_ms": float(np.min(latencies_arr)),
"max_ms": float(np.max(latencies_arr))
}
if __name__ == "__main__":
model_path = MODEL_DIR / "plantdisease_final.keras"
print("=== Conversione modelli TFLite ===\n")
# 1. Baseline float32
r1 = convert_float32(model_path, TFLITE_DIR / "plant_disease_f32.tflite")
# 2. Dynamic range
r2 = convert_dynamic_range(model_path, TFLITE_DIR / "plant_disease_dynrange.tflite")
# 3. Full int8 (per Coral TPU e massima performance su ARM)
r3 = convert_int8_full(model_path, TFLITE_DIR / "plant_disease_int8.tflite")
print("\n=== Riepilogo conversioni ===")
for r in [r1, r2, r3]:
print(f" {r['type']:20s} → {r['size_mb']:.2f} MB")
print("\nConversione completata. Modelli pronti per il deploy su Raspberry Pi.")Implementați pe Raspberry Pi 5: Configurare hardware și software
Hardware recomandat
Raspberry Pi 5 (lansat în octombrie 2023, actualizat cu versiunea de 16 GB în 2024) și platforma de margine recomandată pentru acest caz de utilizare. În comparație cu Pi 4, procesorul Cortex-A76 la 2,4 GHz oferă performanțe de inferență aproape de TFLite de 5 ori mai mare, datorită instrucțiunilor de produs punct ARM v8.2-A optimizate pentru operațiuni int8.
Lista de hardware pentru sistemul complet
| Componentă | Model recomandat | Cost orientativ | Note |
|---|---|---|---|
| SBC principal | Raspberry Pi 5 (8GB) | ~80 EUR | Versiune de 8 GB pentru spațiu de memorie RAM |
| Alimentare electrică | PSU oficial RPi 5 27W USB-C | ~12 EUR | Necesar pentru alimentare stabilă |
| Depozitare | MicroSD A2 64GB (Samsung Pro Endurance) | ~15 EUR | A2 pentru I/O aleatoare mai bune |
| Camera foto | Raspberry Pi Camera Modul 3 (12MP) | ~25 EUR | Focalizare automată, CSI-2 |
| Optica | Lentila cu unghi larg de 120° | ~8 EUR | Pentru achiziție de la dronă sau stâlp |
| Case | RPi 5 Active Cooler + carcasa IP65 | ~20 EUR | Protecție IP65 pentru utilizare în exterior |
| Net | 4G LTE HAT (Waveshare SIM7600G) | ~45 EUR | Pentru zonele fără WiFi |
| Alimentare de câmp | Panou solar 20W + baterie LiPo 10000mAh | ~35 EUR | Autonomie ~3 zile fara soare |
| Sistem total | ~240 EUR | Pentru instalare autonomă pe teren |
Configurați software-ul pe sistemul de operare Raspberry Pi
# Eseguire sul Raspberry Pi dopo aver installato Raspberry Pi OS 64-bit (Bookworm)
# 1. Aggiornamento sistema
sudo apt update && sudo apt full-upgrade -y
# 2. Dipendenze sistema
sudo apt install -y \
python3-pip python3-venv python3-dev \
libatlas-base-dev libjpeg-dev libopenjp2-7 \
libcamera-dev python3-picamera2 \
mosquitto mosquitto-clients \
git vim htop
# 3. Crea virtual environment
python3 -m venv /home/pi/cropai-env
source /home/pi/cropai-env/bin/activate
# 4. Installa TFLite Runtime (versione ottimizzata per ARM64)
# TFLite Runtime e molto più leggero del TF completo (~1MB vs ~400MB)
pip install tflite-runtime
# Alternativa: installa TensorFlow completo (necessario solo per retraining locale)
# pip install tensorflow # ~400MB, non consigliato per Pi se non necessario
# 5. Dipendenze progetto
pip install \
numpy pillow opencv-python-headless \
paho-mqtt \
requests \
RPi.GPIO # per LED status indicator
# 6. Verifica installazione TFLite
python3 -c "import tflite_runtime.interpreter as tflite; print('TFLite OK')"
# 7. Verifica camera
libcamera-hello --timeout 2000
# 8. Test fps camera
libcamera-vid -t 5000 --framerate 30 -o /dev/null --nopreview
# Output atteso: ~30 FPS senza elaborazioneCompletați pipeline de inferență de câmp
Cu modelul convertit și hardware-ul configurat, este timpul să asamblați conducta complet: achiziție de imagini → preprocesare → inferență TFLite → postprocesare → trimiterea rezultatului prin MQTT → tablou de bord. Aceasta este inima aplicației de producție.
# crop_disease_detector.py
# Pipeline completa: acquisizione → inferenza → MQTT su Raspberry Pi 5
import time
import json
import logging
import threading
from pathlib import Path
from datetime import datetime, timezone
from typing import Optional
import numpy as np
# TFLite Runtime (leggero, ottimizzato per ARM)
import tflite_runtime.interpreter as tflite
# Camera (Raspberry Pi Camera Module 3)
from picamera2 import Picamera2, Preview
from picamera2.encoders import JpegEncoder
from picamera2.outputs import FileOutput
# MQTT
import paho.mqtt.client as mqtt
# PIL per preprocessing
from PIL import Image
# ---- CONFIGURAZIONE ----
MODEL_PATH = Path("/home/pi/cropai/models/plant_disease_int8.tflite")
CLASS_NAMES_PATH = Path("/home/pi/cropai/models/class_names.json")
IMG_SIZE = (224, 224)
MQTT_BROKER = "192.168.1.100" # IP del broker locale o cloud
MQTT_PORT = 1883
MQTT_TOPIC_RESULTS = "cropai/field01/sensor01/detection"
MQTT_TOPIC_STATUS = "cropai/field01/sensor01/status"
MQTT_QOS = 1
CAPTURE_INTERVAL_SEC = 30.0 # Cattura ogni 30 secondi
CONFIDENCE_THRESHOLD = 0.70 # Soglia minima per alert
DEVICE_ID = "CROPAI-FIELD01-S01"
# Setup logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.FileHandler("/home/pi/cropai/logs/detector.log"),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class TFLiteInferenceEngine:
"""Wrapper per inferenza TFLite con gestione input uint8/float32."""
def __init__(self, model_path: Path, class_names: list):
self.class_names = class_names
# Inizializza interpreter con numero ottimale di thread
# Su RPi 5 (4 core), 4 thread migliora latenza del ~30%
self.interpreter = tflite.Interpreter(
model_path=str(model_path),
num_threads=4
)
self.interpreter.allocate_tensors()
self.input_details = self.interpreter.get_input_details()
self.output_details = self.interpreter.get_output_details()
self.input_dtype = self.input_details[0]["dtype"]
self.input_shape = self.input_details[0]["shape"] # [1, 224, 224, 3]
logger.info(f"Modello caricato: {model_path.name}")
logger.info(f"Input shape: {self.input_shape}, dtype: {self.input_dtype}")
logger.info(f"Classi: {len(self.class_names)}")
def preprocess(self, image: Image.Image) -> np.ndarray:
"""
Preprocessa l'immagine PIL per l'inferenza.
Gestisce sia modelli float32 che uint8 (int8 quantizzato).
"""
# Resize con alta qualità (LANCZOS per downscaling)
img_resized = image.resize(IMG_SIZE, Image.LANCZOS)
img_array = np.array(img_resized, dtype=np.float32)
if self.input_dtype == np.uint8:
# Modello int8: input uint8 in [0, 255]
return np.expand_dims(img_array.astype(np.uint8), axis=0)
else:
# Modello float32: normalizzazione in [-1, 1]
img_normalized = (img_array / 127.5) - 1.0
return np.expand_dims(img_normalized, axis=0)
def predict(self, image: Image.Image) -> dict:
"""
Esegue inferenza e restituisce top-3 predizioni con confidenze.
"""
t_start = time.perf_counter()
input_data = self.preprocess(image)
self.interpreter.set_tensor(self.input_details[0]["index"], input_data)
self.interpreter.invoke()
output_data = self.interpreter.get_tensor(self.output_details[0]["index"])
t_end = time.perf_counter()
latency_ms = (t_end - t_start) * 1000
# Decodifica output (gestione uint8 per modelli quantizzati)
if self.output_details[0]["dtype"] == np.uint8:
scale, zero_point = self.output_details[0]["quantization"]
probabilities = (output_data.flatten().astype(np.float32) - zero_point) * scale
else:
probabilities = output_data.flatten()
# Top-3 predizioni
top3_idx = np.argsort(probabilities)[::-1][:3]
predictions = [
{
"class": self.class_names[idx],
"confidence": float(probabilities[idx]),
"rank": rank + 1
}
for rank, idx in enumerate(top3_idx)
]
return {
"predictions": predictions,
"top1_class": predictions[0]["class"],
"top1_confidence": predictions[0]["confidence"],
"latency_ms": round(latency_ms, 2),
"is_healthy": "healthy" in predictions[0]["class"].lower(),
"alert": predictions[0]["confidence"] >= CONFIDENCE_THRESHOLD
and "healthy" not in predictions[0]["class"].lower()
}
class CropDiseaseDetector:
"""
Sistema completo: cattura immagini periodica, inferenza, invio MQTT.
"""
def __init__(self):
# Carica class names
with open(CLASS_NAMES_PATH) as f:
class_names = json.load(f)
# Inizializza inference engine
self.engine = TFLiteInferenceEngine(MODEL_PATH, class_names)
# Inizializza camera
self.camera = Picamera2()
camera_config = self.camera.create_still_configuration(
main={"size": (1920, 1080), "format": "RGB888"},
lores={"size": (640, 480)},
display="lores"
)
self.camera.configure(camera_config)
self.camera.start()
time.sleep(2) # Warmup autofocus
logger.info("Camera inizializzata")
# Inizializza MQTT
self.mqtt_client = mqtt.Client(
client_id=DEVICE_ID,
protocol=mqtt.MQTTv5
)
self.mqtt_client.on_connect = self._on_mqtt_connect
self.mqtt_client.on_disconnect = self._on_mqtt_disconnect
self.mqtt_connected = threading.Event()
self._connect_mqtt()
# Metriche sessione
self.session_stats = {
"total_captures": 0,
"total_alerts": 0,
"avg_latency_ms": 0.0,
"start_time": datetime.now(timezone.utc).isoformat()
}
def _on_mqtt_connect(self, client, userdata, flags, rc, properties=None):
if rc == 0:
logger.info(f"MQTT connesso al broker {MQTT_BROKER}")
self.mqtt_connected.set()
# Pubblica status online
self._publish_status("online")
else:
logger.error(f"MQTT connessione fallita: rc={rc}")
def _on_mqtt_disconnect(self, client, userdata, rc, properties=None):
logger.warning(f"MQTT disconnesso: rc={rc}. Reconnect automatico...")
self.mqtt_connected.clear()
def _connect_mqtt(self):
try:
self.mqtt_client.connect(MQTT_BROKER, MQTT_PORT, keepalive=60)
self.mqtt_client.loop_start()
# Attendi connessione (max 10s)
connected = self.mqtt_connected.wait(timeout=10)
if not connected:
logger.warning("Timeout MQTT. Continuo in modalità offline.")
except Exception as e:
logger.warning(f"MQTT non disponibile: {e}. Modalità offline attiva.")
def _publish_status(self, status: str):
payload = {
"device_id": DEVICE_ID,
"status": status,
"timestamp": datetime.now(timezone.utc).isoformat(),
"session_stats": self.session_stats
}
try:
self.mqtt_client.publish(
MQTT_TOPIC_STATUS,
json.dumps(payload),
qos=1,
retain=True # Retained: dashboard vede sempre ultimo stato
)
except Exception as e:
logger.debug(f"Publish status fallito: {e}")
def capture_and_infer(self) -> Optional[dict]:
"""Cattura immagine e lancia inferenza."""
try:
# Cattura frame dalla camera
frame = self.camera.capture_array("main")
# Converti numpy array in PIL Image
pil_image = Image.fromarray(frame)
# Salva immagine per debug/audit (opzionale)
timestamp_str = datetime.now().strftime("%Y%m%d_%H%M%S")
img_path = Path(f"/home/pi/cropai/captures/{timestamp_str}.jpg")
img_path.parent.mkdir(exist_ok=True)
pil_image.save(str(img_path), quality=85)
# Inferenza
result = self.engine.predict(pil_image)
result["device_id"] = DEVICE_ID
result["timestamp"] = datetime.now(timezone.utc).isoformat()
result["image_path"] = str(img_path)
# Aggiorna statistiche sessione
n = self.session_stats["total_captures"]
self.session_stats["total_captures"] += 1
self.session_stats["avg_latency_ms"] = (
(self.session_stats["avg_latency_ms"] * n + result["latency_ms"]) / (n + 1)
)
if result["alert"]:
self.session_stats["total_alerts"] += 1
return result
except Exception as e:
logger.error(f"Errore cattura/inferenza: {e}", exc_info=True)
return None
def publish_result(self, result: dict):
"""Pubblica risultato su MQTT."""
try:
payload = json.dumps(result, ensure_ascii=False)
qos = 2 if result.get("alert") else 1 # QoS 2 per alert critici
self.mqtt_client.publish(MQTT_TOPIC_RESULTS, payload, qos=qos)
if result["alert"]:
logger.warning(
f"ALERT: {result['top1_class']} "
f"(conf: {result['top1_confidence']:.2%}) "
f"- latenza: {result['latency_ms']:.0f}ms"
)
else:
logger.info(
f"OK: {result['top1_class']} "
f"(conf: {result['top1_confidence']:.2%}) "
f"- latenza: {result['latency_ms']:.0f}ms"
)
except Exception as e:
logger.error(f"Errore publish MQTT: {e}")
def run(self):
"""Loop principale di acquisizione e inferenza."""
logger.info(f"Sistema avviato. Intervallo cattura: {CAPTURE_INTERVAL_SEC}s")
logger.info(f"Soglia alert: {CONFIDENCE_THRESHOLD:.0%}")
try:
while True:
t_loop_start = time.time()
result = self.capture_and_infer()
if result:
self.publish_result(result)
# Pubblica stats ogni 10 catture
if self.session_stats["total_captures"] % 10 == 0:
self._publish_status("online")
# Rispetta intervallo configurato
elapsed = time.time() - t_loop_start
sleep_time = max(0, CAPTURE_INTERVAL_SEC - elapsed)
time.sleep(sleep_time)
except KeyboardInterrupt:
logger.info("Interruzione manuale. Shutdown graceful...")
finally:
self._publish_status("offline")
self.camera.stop()
self.mqtt_client.loop_stop()
self.mqtt_client.disconnect()
logger.info("Sistema arrestato correttamente")
if __name__ == "__main__":
detector = CropDiseaseDetector()
detector.run()Hardware Edge Benchmark: comparație platformă
Alegerea platformei hardware depinde de buget, performanța necesară, consum energie și complexitatea integrării. Mai jos este o comparație actualizată pentru 2024-2025 pe baza reperelor publicate și a măsurătorilor proprii asupra modelelor de clasificare a bolilor.
Comparația platformelor Edge pentru ML agricol
| Platformă | CPU | RAM | MobileNetV3S int8 latență | Consum | Cost | Note |
|---|---|---|---|---|---|---|
| Raspberry Pi 4B (4GB) | Cortex-A72 1.8GHz | 4 GB LPDDR4 | ~320 ms | ~6W | ~55 EUR | Linia de bază economică |
| Raspberry Pi 5 (8GB) | Cortex-A76 2.4GHz | 8 GB LPDDR4X | ~65 ms | ~8W | ~80 EUR | Recomandat |
| RPi 4 + Coral USB TPU | Cortex-A72 + Edge TPU | 4GB | ~15 ms | ~8W | ~95 EUR | Necesită modelul complet int8 |
| Google Coral Dev Board | Cortex-A53 + Edge TPU | 1 GB | ~12 ms | ~4W | ~120 EUR | Doar compilația TFLite + Coral |
| NVIDIA Jetson Nano (4 GB) | Cortex-A57 + 128 de nuclee CUDA | 4 GB LPDDR4 | ~8 ms | ~10W | ~149 USD | Exagerat pentru o clasificare simplă |
| Arduino Portenta H7 | Cortex-M7 480MHz | 8 MB SDRAM | ~2000 ms | ~0,5W | ~100 EUR | Numai modele mici (TFLite Micro) |
| Dock Sipeed M1s | BL808 RV64 480MHz | 768KB SRAM | ~800 ms (micro) | ~0,3W | ~7 USD | Putere ultra-scăzută, modele mici |
Latența măsurată pe modelul MobileNetV3Small int8 (2,6 MB) pentru o singură imagine de 224 x 224. Valorile Pi 5 din benchmark Hackster.io 2024, Coral de la Georgia Southern University 2024.
Reglare termică pe Raspberry Pi 5
Raspberry Pi 5 fără răcire activă experimentează o accelerare termică după 5-10 minute de inferență continuă, cu o reducere a performanței 20-30%. Pentru desfășurare pe teren (temperaturi exterioare de până la 40°C vara), răcire activ e obligatoriu. Carcasa oficială cu Active Cooler deține procesorul sub 70°C chiar și cu sarcină susținută.
Monitorizați-vă temperatura cu: vcgencmd measure_temp
Bolile plantelor italiene: Ținte prioritare pentru Edge ML
Contextul agricol italian are caracteristici specifice care ghidează alegerile de desfășurare. Italia este al treilea producator de vin din lume (dupa Franta si Spania), primul producator de vin și măsline din UE și are una dintre cele mai mari biodiversitate de culturi din Europa. Fitopatii cele mai de impact economic pentru agricultura italiană sunt:
Principalele boli ale plantelor italiene și impactul economic
| Boală | Agent | Culturile afectate | Pierdere potențială | Fereastra de tratament |
|---|---|---|---|---|
| Mucegaiul pufos al viței de vie | Plasmopara viticola | Toți strugurii DOC/DOCG | 20-100% producție | 48-72h de la apariție |
| Mucegaiul de viță de vie | Erysiphe necator | Viță de vie, cucurbitacee | 15-60% producție | 5-7 zile |
| Crusta de măr | Venturia inaequalis | Apple, totuși | 30-80% producție | 3-5 zile |
| Mucegaiul pufos al roșiilor | Phytophthora infestans | Roșii, cartofi | 50-100% producție | 24-48h |
| Botrytis (mucegai gri) | Botrytis cinerea | Viță de vie, căpșuni, roșii | 10-40% producție | 7-10 zile |
| Xylella fastidiosa | Xylella fastidiosa | Măslin (Puglia) | Planta intreaga | Irecuperabil |
Pentru mucegaiul strugurilor, cea mai mare provocare este că PlantVillage nu include imagini specific pentru viță-de-vie italiana DOC/DOCG (Sangiovese, Nebbiolo, Primitivo, Nero d'Avola). Acest lucru face necesar adaptarea domeniului: colectează un set de date suplimentarea imaginilor reale din podgoriile italiene și reglarea fină a modelului PlantVillage-preantrenat pe setul de date local.
Fonduri și stimulente pentru agricultura digitală italiană
Il Plan strategic PAC 2023-2027 (Politica agricolă comună) alocă fonduri specifice pentru digitalizarea agriculturii prin Ecoscheme și intervenții sectoriale. Intervenția SRA22 („Managementul riscurilor”) și SRA29 („Agricultura de precizie”) oferi recompense companiilor care adoptă sisteme digitale de monitorizare fitosanitar. La nivel național, PNRR prin măsura M2C4 a alocat fonduri pt inovarea în agricultură ca parte a tranziției ecologice și digitale.
Pentru IMM-urile agricole (companii cu cifra de afaceri < 2 milioane EUR), costul unui sistem edge ML ca cel descris in acest articol (~240 EUR hardware + dezvoltare) si amortizabil într-un singur sezon dacă avem în vedere că depistarea precoce a mucegaiului pufos pe o vie de 5 hectare poate economisi o producție în valoare de 15.000-30.000 EUR.
Măsuri de performanță și monitorizare
Un sistem ML în producție fără monitorizare este un sistem destinat să se degradeze în tăcere. Valorile care trebuie urmărite sunt împărțite în două categorii: valori model (acuratețe) și metrici de sistem (latență, timp de funcționare).
# metrics_collector.py
# Raccolta e invio metriche di sistema e modello via MQTT
import time
import json
import psutil
import subprocess
from datetime import datetime, timezone
from pathlib import Path
def get_cpu_temperature() -> float:
"""Legge temperatura CPU da vcgencmd (Raspberry Pi specifico)."""
try:
result = subprocess.run(
["vcgencmd", "measure_temp"],
capture_output=True, text=True, timeout=2
)
# Output: "temp=48.5'C"
temp_str = result.stdout.strip().replace("temp=", "").replace("'C", "")
return float(temp_str)
except Exception:
return -1.0
def get_system_metrics() -> dict:
"""Raccoglie metriche hardware del Raspberry Pi."""
cpu_freq = psutil.cpu_freq()
mem = psutil.virtual_memory()
disk = psutil.disk_usage("/")
return {
"timestamp": datetime.now(timezone.utc).isoformat(),
"cpu_percent": psutil.cpu_percent(interval=1),
"cpu_freq_mhz": cpu_freq.current if cpu_freq else 0,
"cpu_temp_c": get_cpu_temperature(),
"ram_used_mb": round(mem.used / 1024 / 1024, 1),
"ram_total_mb": round(mem.total / 1024 / 1024, 1),
"ram_percent": mem.percent,
"disk_used_gb": round(disk.used / 1024 / 1024 / 1024, 2),
"disk_free_gb": round(disk.free / 1024 / 1024 / 1024, 2),
"uptime_hours": round(time.time() / 3600, 1) # approssimato
}
class PerformanceTracker:
"""
Traccia accuracy in deployment tramite feedback umano.
Quando un agronomo conferma/smentisce una detection, il dato
viene usato per monitorare il data drift.
"""
def __init__(self, log_path: Path = Path("/home/pi/cropai/metrics/detections.jsonl")):
self.log_path = log_path
self.log_path.parent.mkdir(exist_ok=True)
self._confirmed = 0
self._rejected = 0
self._total = 0
def log_detection(self, result: dict, human_confirmed: bool = None):
"""Registra una detection con eventuale conferma umana."""
record = {
"timestamp": result.get("timestamp"),
"top1_class": result.get("top1_class"),
"top1_confidence": result.get("top1_confidence"),
"alert": result.get("alert"),
"latency_ms": result.get("latency_ms"),
"human_confirmed": human_confirmed
}
self._total += 1
if human_confirmed is True:
self._confirmed += 1
elif human_confirmed is False:
self._rejected += 1
with open(self.log_path, "a") as f:
f.write(json.dumps(record) + "\n")
def get_field_precision(self) -> float:
"""Precision in-field basata su feedback umano."""
labeled = self._confirmed + self._rejected
if labeled == 0:
return float("nan")
return self._confirmed / labeled
def get_summary(self) -> dict:
return {
"total_detections": self._total,
"human_confirmed": self._confirmed,
"human_rejected": self._rejected,
"field_precision": self.get_field_precision(),
"unlabeled": self._total - self._confirmed - self._rejected
}Cele mai bune practici și anti-modele
Cele mai bune practici
Cele 8 reguli de aur pentru Agricultural Edge ML
- Cuantificați întotdeauna înainte de implementare: Diferența dintre float32 și int8 în ceea ce privește latența pe ARM și 50-70%, cu pierderi de precizie sub 1%. Nu există niciun motiv pentru a utiliza float32 în producție pe Pi 5.
- Calibrați cu date reale de câmp: Setul de date PlantVillage e dobandite in laborator. Colectați cel puțin 500 de imagini reale din câmpul țintă (același soi de struguri, aceeași lumină, același unghi de tragere) pentru calibrare int8 și pentru reglaj fin. Diferența de precizie în câmp poate fi cu 10-20% fără acest pas.
- Gestionați bucla de feedback: Implementați un mecanism de ce agronomii pot confirma sau corecta depistarile. Fiecare fix este aur: devine date de antrenament pentru ciclul următor.
- Monitorizarea distribuției intrării: Dacă modelul a văzut numai Sangiovese în timpul antrenamentului, dar în producție îndeplinește Primitivo, precizia se prăbușește. Trasează distribuția înglobărilor de intrare pentru a detecta deriva de date.
- Programați actualizarea OTA: Șabloanele trebuie să fie actualizate fără acces fizic la dispozitiv. Implementați un mecanism MQTT pentru descărcări și schimbarea fișierului .tflite cu rollback automat în caz de eroare.
- Redundanță și modul offline: Sistemul trebuie să funcționeze corect chiar și fără conectivitate. Utilizați un buffer local (fișier SQLite sau JSON) pentru detectări și sincronizați când conexiunea revine.
- Calibrați pragul de încredere pentru cultură: un prag de 70% pentru mucegaiul pufos (tratament foarte distructiv, economic) și diferit de unul prag pentru făinare (tratament mai puțin urgent, costisitor). Adaptați pragul la costul asimetric al fals pozitive vs fals negative.
- Documentați contextul achiziției: Fiecare imagine trebuie să fie metadate cu: ora din zi, condițiile meteo, stadiul fenologic, poziția GPS. Aceste metadate sunt esențiale pentru depanare și formare viitoare.
Anti-modele de evitat
Cele mai frecvente 5 greșeli în proiectele Agricultural Edge ML
- Să presupunem că PlantVillage = gata de producție: Setul de date e dobândit în condiții ideale (lumină uniformă, fundal alb, frunze izolate). Pe teren, imaginile au umbre, fundal verde, picături de apă, insecte suprapuse. Un model antrenat numai pe PlantVillage fără reglaj fin în câmp are de obicei o precizie de 60-70% în câmp real, față de 97% pe setul de teste de laborator. Acest decalaj și cunoscut ca schimbare de domeniu și este principala problemă practică.
-
Ignorarea echilibrului clasei: PlantVillage are o mulțime de clase
dezechilibrat (Tomato Late Blight: 1909 imagini; Cedar Apple Rust: 275 imagini).
Fără ponderare a clasei sau supraeșantionare, modelul va fi orientat către clase abundente.
Utilizare
class_weightîn Keras fit() sauWeightedRandomSampler. - Folosind Pi ca server web: Gestionați solicitările HTTP sincrone pentru inferența despre Pi este o alegere arhitecturală proastă. Pi trebuie să fie un producător MQTT autonom, nu un server care așteaptă solicitări. Managementul conexiunii și analiza HTTP adaugă o suprasarcină și o complexitate inutilă.
- Nu manipulați condiții de lumină extremă: La ora 12:00 în august în Puglia, lumina directă saturează imaginile și frunzele par complet albe. În zorii cu umiditate ridicată, frunzele sunt umede și se reflectă diferit. Implementează o verificare a calității imaginii (luminozitate medie, saturație) înainte lansați inferența și eliminați/reîncercați imaginile în afara intervalului.
- Cuantificați fără a valida acuratețea: Cuantizarea completă int8 necesită un set de date de calibrare reprezentativ. Utilizați un set de date de calibrare prea mic (mai puțin de 100 de imagini) sau nereprezentativ poate duce la o pierdere precizie de 3-5% în loc de 0,4% așteptat. Validați întotdeauna modelul cuantificat pe setul de testare înainte de implementare.
Actualizare a modelului over-the-air
Cu zeci sau sute de dispozitive implementate pe teren, actualizați manual modele și impracticabile. Un sistem OTA (Over-the-Air) prin MQTT permite distribuția modele noi fără acces fizic la dispozitive.
# model_ota_updater.py
# Aggiornamento OTA del modello TFLite via MQTT
import json
import hashlib
import logging
import threading
import tempfile
import shutil
from pathlib import Path
import urllib.request
import paho.mqtt.client as mqtt
logger = logging.getLogger(__name__)
MODEL_PATH = Path("/home/pi/cropai/models/plant_disease_int8.tflite")
MODEL_BACKUP_PATH = Path("/home/pi/cropai/models/plant_disease_int8.tflite.bak")
MQTT_TOPIC_OTA = "cropai/ota/commands"
DEVICE_ID = "CROPAI-FIELD01-S01"
class OTAUpdater:
"""
Ascolta comandi OTA via MQTT e aggiorna il modello TFLite.
Formato comando MQTT:
{
"command": "update_model",
"model_url": "https://...",
"sha256": "abc123...",
"version": "2.1.0",
"target_devices": ["CROPAI-FIELD01-S01", "CROPAI-FIELD02-S01"]
}
"""
def __init__(self, mqtt_client: mqtt.Client, reload_callback):
self.mqtt_client = mqtt_client
self.reload_callback = reload_callback # Funzione chiamata dopo update
self._lock = threading.Lock()
# Sottoscrivi al topic OTA
mqtt_client.subscribe(MQTT_TOPIC_OTA, qos=2)
mqtt_client.message_callback_add(MQTT_TOPIC_OTA, self._on_ota_command)
logger.info(f"OTA updater attivo. Topic: {MQTT_TOPIC_OTA}")
def _on_ota_command(self, client, userdata, message):
"""Handler per comandi OTA."""
try:
command = json.loads(message.payload.decode())
# Verifica che il comando sia per questo device
target_devices = command.get("target_devices", [])
if target_devices and DEVICE_ID not in target_devices:
return
if command.get("command") == "update_model":
# Esegui update in thread separato (non bloccare il loop MQTT)
threading.Thread(
target=self._do_update,
args=(command,),
daemon=True
).start()
except Exception as e:
logger.error(f"Errore parsing comando OTA: {e}")
def _do_update(self, command: dict):
"""Esegue il download e swap del modello."""
with self._lock:
model_url = command["model_url"]
expected_sha256 = command["sha256"]
version = command.get("version", "unknown")
logger.info(f"OTA update avviato. Versione: {version}")
try:
# Download in file temporaneo
with tempfile.NamedTemporaryFile(suffix=".tflite", delete=False) as tmp:
tmp_path = Path(tmp.name)
logger.info(f"Download da {model_url}...")
urllib.request.urlretrieve(model_url, str(tmp_path))
# Verifica integrità SHA256
sha256 = hashlib.sha256(tmp_path.read_bytes()).hexdigest()
if sha256 != expected_sha256:
logger.error(f"SHA256 mismatch! Atteso: {expected_sha256}, Ottenuto: {sha256}")
tmp_path.unlink()
return
logger.info("SHA256 verificato. Backup modello corrente...")
# Backup modello corrente
if MODEL_PATH.exists():
shutil.copy2(MODEL_PATH, MODEL_BACKUP_PATH)
# Swap atomico
shutil.move(str(tmp_path), str(MODEL_PATH))
logger.info(f"Modello aggiornato a versione {version}")
# Ricarica modello nel detector
if self.reload_callback:
self.reload_callback(MODEL_PATH)
logger.info("Modello ricaricato con successo")
except Exception as e:
logger.error(f"OTA update fallito: {e}. Rollback...")
# Rollback al backup
if MODEL_BACKUP_PATH.exists():
shutil.copy2(MODEL_BACKUP_PATH, MODEL_PATH)
logger.info("Rollback completato")
if "tmp_path" in locals() and tmp_path.exists():
tmp_path.unlink()Concluzii și pașii următori
Am construit un sistem complet de detectare a bolilor culturilor bazat pe Edge ML: de la antrenarea modelului MobileNetV3Small pe PlantVillage cu reglaj fin, până la conversie TFLite cu cuantizare int8, implementat pe Raspberry Pi 5 cu pipeline Python integrat pentru captarea, inferența și trimiterea MQTT. Sistemul realizează latențe de 65 ms per deducere, consumă mai puțin de 8 W de putere totală și funcționează complet offline.
Cifrele care contează cu adevărat pentru antreprenorul agricol sunt diferite: pierderile din cauza bolilor a culturilor sunt scumpe 220 de miliarde de dolari la nivel global în fiecare an (FAO), iar depistarea precoce poate reduce pierderile cu 60-80% cu tratamente în timp util. Un sistem ca cel descris costa mai putin de 250 EUR in hardware, si poate fi amortizat intr-un singur sezon pe orice cultură valoroasă. Piața AI în agricultură este de așteptat să crească din 5,9 miliarde USD în 2025 până la 61,3 miliarde USD în 2035, cu un CAGR de 26,3%.
Cel mai mare gol de umplut rămâne cel al schimbare de domeniu: Plant Village nu este suficient în sine pentru desfășurare pe teren. Prioritatea pentru cei care vor să pună în producție acest sistem și colectați imagini reale din culturile dvs., etichetați-le cu un agronom de încredere și faceți reglajul fin. De asemenea, 500 de imagini reale, utilizate corect, poate face diferența între un sistem cu o precizie de 97% în laborator și 65% pe teren, vs un sistem cu 95% stabil în ambele contexte.
Rezumat: tehnologia completă
| Straturi | Tehnologie | Versiunea 2025 |
|---|---|---|
| Antrenamentul | TensorFlow / Keras | TF 2.17+ |
| Seturi de date | Plant Village (Kaggle/HuggingFace) | 54.309 imagini |
| Arhitectură | MobileNetV3Small | Transferați învățarea ImageNet |
| Format de implementare | TensorFlow Lite (.tflite) | int8 cuantizat |
| Hardware | Raspberry Pi 5 (8GB) | ARM Cortex-A76 |
| Cameră | Modulul camerei 3 (12MP) | Picamera2 + libcamera |
| Mesaje | MQTT (paho-mqtt) | Eclipse Mosquitto 2.x |
| Dieta | Solar + LiPo | Autonomie 3+ zile |
| OTA | Verificare MQTT + SHA256 | Implementare fără atingere |
Articole viitoare din seria FoodTech
- Articolul 3 – Satellite API și NDVI: Cum să utilizați API-ul Sentinel-2 (Copernicus) pentru a monitoriza sănătatea culturilor la scară de câmp cu date satelit gratuit, calcul NDVI în Python și integrare cu pipeline IoT.
- Articolul 4 — Viziunea computerizată pentru calitatea alimentelor: Învățare profundă pentru clasificarea automata a defectelor pe liniile de productie, gradare pt greutate/dimensiune/culoare și integrare cu sisteme automate de sortare.
- Articolul 5 – Trasabilitatea alimentelor în blockchain: Implementarea a unui sistem de trasabilitate field-to-furk cu Hyperledger Fabric și standardele GS1 Digital Link pentru conformitatea cu legislația UE privind trasabilitatea alimentelor.
Resurse și referințe
- Set de date PlantVillage: Hughes, D. P. și Salathé, M. (2016). Un depozit cu acces deschis de imagini despre sănătatea plantelor pentru a permite dezvoltarea dispozitivelor mobile diagnosticul bolii. ArXiv preprint arXiv:1511.08060. Disponibil pe Kaggle Plant Village.
- Pierderi de recolte FAO: Producția și protecția plantelor FAO — pierderi anuale cauzate de dăunători și boli
- TFLite pe Raspberry Pi 5: Hackster.io — Evaluarea comparativă a TFLite pe Raspberry Pi 5
- Valori de referință Edge AI 2024: Universitatea de Sud din Georgia (2024). Benchmarking Edge AI Platforms: Analiza performanței NVIDIA Jetson și Raspberry Pi 5 cu Coral TPU. Actele conferinței IEEE. DOI: 10.1109/10971592.
- PlantVillage optimizat MobileNetV3: Rapoarte științifice — MobileNet optimizat pentru detectarea bolilor ușoare ale frunzelor plantelor (2025)
- Agricultura de precizie Italia: Rural Hack — Agricultura 4.0 Italia 2025: Creșterea pieței și maturitatea digitală