ML Edge for Crop Disease Detection: TensorFlow Lite na Raspberry Pi
Každoročně zničí choroby rostlin a škůdci plodin až 40 % produkce světové zemědělství, což způsobuje ekonomické ztráty větší než 220 miliard dolarů podle údajů FAO. V Itálii jsou vinařské země, sady a zeleninové zahrady zranitelné endemickými chorobami rostlin, jako je např padlí révy (Plasmopara viticola), padlí (Erysiphe necator) a strupovitost jablka (Venturia inaequalis), z nichž každý dokáže vymazat celý rok produktivní, pokud není včas zachycen.
Včasná diagnostika byla historicky svěřena vizuální zkušenosti farmáře nebo v lepším případě k pravidelným kontrolám agronomů. Oba přístupy trpí stejným omezením: latencí Mezi propuknutím onemocnění a fytosanitárním zásahem je interval, který je často dny nebo týdny během kterého se houba nebo bakterie šíří jako požár. Časové okno pro a účinné a minimální ošetření: například u padlí révy se redukuje na slunce 48-72 hodin od výskytu prvních lézí.
Umělá inteligence aplikovaná na počítačové vidění nabízí konkrétní odpověď: modely hluboké učení trénované na datovém souboru PlantVillage (54 309 obrázků, 38 tříd) dosahuje přesnosti vyšší než 97 % při rozpoznávání listových chorob. Skutečné úzké hrdlo už nejde o přesnost modelu, ale o nasazení v terénu. Na vinici na Apeninách Internetová konektivita často chybí nebo je nespolehlivá. Odesílání obrázků do cloudu pro odvození je nepraktické. Řešení aHrana ML: model běží přímo na zařízení, v terénu, bez konektivity, s latencí menší než za druhé.
Tento článek od základu buduje kompletní kanál detekce nemocí na Raspberry Pi 5 pomocí TensorFlow Lite: od základního modelového školení s přenosem učení na MobileNetV3 až po int8 kvantování, k nasazení s Python inferencí v reálném čase, k odesílání výsledků pryč MQTT na vzdálený řídicí panel. Každý krok je doprovázen pracovním kódem, čísly benchmarků ověřené a praktické úvahy pro italský zemědělský kontext.
Co se dozvíte v tomto článku
- proč edge ML překonává cloud pro zemědělské aplikace na otevřeném poli
- Jak připravit a rozšířit datovou sadu PlantVillage pro efektivní školení
- Přeneste učení s MobileNetV3Small na TensorFlow/Keras, krok za krokem
- Konverze modelu na TFLite s kvantizací po tréninku (float32 → int8)
- Nastavení hardwaru Raspberry Pi 5 s kamerovým modulem 3 a úplným odvozením Pythonu
- Integrovaný kanál: zachycení → předzpracování → odvození → MQTT → řídicí panel
- Srovnávací testy latence a spotřeby energie na Pi 4, Pi 5, Coral TPU, Jetson Nano
- Hlavní italské choroby rostlin a prioritní cílové plodiny
- Osvědčené postupy nasazení a běžné anti-vzorce
Pozice v řadě FoodTech
Toto je druhý článek ze série FoodTech, věnující se digitálním technologiím pro zemědělsko-potravinářský dodavatelský řetězec. Série pokrývá zemědělský IoT, počítačové vidění, blockchain pro sledovatelnost, prognózování poptávky a mnoho dalšího.
FoodTech Series – všechny články
| # | Titul | Úroveň | Stát |
|---|---|---|---|
| 1 | Zemědělský IoT Pipeline: Senzory, MQTT a časové řady na InfluxDB | Střední | K dispozici |
| 2 | ML Edge for Crop Disease Detection: TFLite na Raspberry Pi — Jste tady | Moderní | Proud |
| 3 | Satelitní API a NDVI: Sledování plodin pomocí Sentinel-2 | Střední | K dispozici |
| 4 | Počítačové vidění pro kvalitu potravin: vady, hmotnost a třídění | Moderní | K dispozici |
| 5 | Sledovatelnost blockchainových potravin: Hyperledger Fabric a GS1 | Moderní | K dispozici |
| 6 | Shoda s FSMA 204: Digitální sledovatelnost s Pythonem a API | Střední | K dispozici |
| 7 | Automatizace vertikálního zemědělství: Řízení klimatu pomocí PLC a ML | Moderní | K dispozici |
| 8 | Prognóza poptávky po jídle: Prophet, LSTM a Feature Engineering | Střední | K dispozici |
| 9 | Farm Dashboard v reálném čase: Angular, WebSocket a InfluxDB | Střední | K dispozici |
| 10 | Odolnost dodavatelského řetězce: Optimalizace pomocí OR-Tools a AI | Moderní | K dispozici |
Edge ML Fundamentals: Cloud vs Edge pro zemědělství
Před napsáním řádku kódu je nezbytné pochopit, proč je okrajový přístup jediný praktické pro detekci onemocnění v otevřeném poli. Toto není stylistická volba: a architektonická nutnost vynucená fyzickým kontextem zemědělství.
Hranice mraku v zemědělství
Aplikace pro detekci nemocí v cloudu by fungovala takto: Fotoaparát cvakne fotografii, odešle ji přes síť na cloudový server, model běží na serveru, výsledek se vrátí do zařízení. V kanceláři s optickými vlákny tento cyklus trvá méně než 200 ms. Na vinici v Toskánsku nebo v citrusovém háji v Kalábrii se proměnné radikálně mění:
Cloud vs Edge: Srovnání podle zemědělského kontextu
| Charakteristický | Cloudový přístup | Okrajový přístup |
|---|---|---|
| Je vyžadována konektivita | Nepřetržité, pásmo 1+ Mbps | Volitelné, pouze pro synchronizaci dat |
| Inferenční latence | 200 ms - 5 s (závisí na síti) | 50–500 ms (místní) |
| Provozní náklady | Vysoká (volání API, úložiště, šířka pásma) | Nízká (jednorázový hardware) |
| Soukromí poskytnuto | Firemní obrázky v cloudu | Data zůstávají na místě |
| Offline provoz | Nemožné | Kompletní |
| Škálovatelnost modelu | Neomezený | Omezeno RAM/CPU |
| Aktualizace modelu | Bezprostřední | Vyžaduje fyzické nasazení nebo nasazení OTA |
| Spotřeba energie | Nízká (zařízení) + vysoká (datové centrum) | Vše na zařízení (3-15W) |
TensorFlow vs TensorFlow Lite: Klíčové rozdíly
TensorFlow (TF) je kompletní tréninkový a inferenční rámec, optimalizovaný pro GPU a servery. TensorFlow Lite (TFLite) je komprimovaná a optimalizovaná verze pro vestavěná a mobilní zařízení. Praktické rozdíly jsou významné:
- Velikost runtime: Runtime TFLite a asi 1 MB v C++ (oproti 400 MB+ pro plné TF). Na Raspberry Pi se rozdíl promítá do obsazení paměti RAM 50–100 MB namísto 2–4 GB.
- Podporovaní operátoři: TFLite podporuje podmnožinu operací TF. Modelky s vlastními operacemi nebo nestandardními vrstvami vyžadují delegování nebo záložní řešení na CPU.
- Formát .tflite: Schéma FlatBuffers, optimalizované pro přímý přístup do paměti bez analýzy. Model je paměťově mapován bez deserializace.
- Hardwarová akcelerace: TFLite podporuje delegáty XNNPACK (SIMD na ARM), Delegát GPU, delegát Coral Edge TPU a NNAPI v systému Android.
Kvantování: float32 až int8
Nejefektivnější technika pro optimalizaci hran a kvantování: snížení numerické přesnosti vah z 32 bitů s pohyblivou řádovou čárkou na 8 bitová celá čísla. Efekt praktické a trojité:
Účinky kvantizace int8 na MobileNetV3Small
| Metrický | plovák32 | int8 (PTQ) | Delta |
|---|---|---|---|
| Velikost modelu | 9,8 MB | 2,6 MB | -73 % |
| Inference RAM (Pi 5) | ~180 MB | ~52 MB | -71 % |
| Inferenční latence (Pi 5) | ~180 ms | ~65 ms | -64 % |
| Přesnost PlantVillage | 97,2 % | 96,8 % | -0,4 % |
| Spotřeba energie | ~5,2W | ~3,8W | -27 % |
Benchmark na Raspberry Pi 5 (2,4 GHz, 8 GB RAM). Hodnoty naměřené s vyladěným modelem na 38 třídách PlantVillage.
Ztráta přesnosti 0,4 % je v oboru prakticky irelevantní: přirozená variabilita světelné podmínky, úhel a stadium onemocnění vnášejí nejistoty mnohem větší. Zisk v latenci a velikosti je však pro embedded nasazení rozhodující.
PlantVillage Dataset: Příprava a augmentace
Referenční datový soubor pro detekci listových chorob a PlantVillage, vytvořili Hughes a Salathé (Penn State University) a vydali v roce 2016. Obsahuje více 54 309 snímků fotografováno v kontrolovaných laboratorních podmínkách, organizované v 38 tříd pokrývající 14 druhů rostlin a příbuzných chorob.
Struktura datové sady
Třídy PlantVillage: Druhy a hlavní choroby
| Druh | Nemoc | Kauzální agent | Obrázky |
|---|---|---|---|
| Vinná réva (hroznová) | Černá hniloba | Guignardia bidwellii | 1,180 |
| Vinná réva (hroznová) | Návnada / Černé spalničky | Phaeoacremonium spp. | 1,383 |
| Vinná réva (hroznová) | Listová plíseň | Pseudocercospora vitis | 1,076 |
| Vinná réva (hroznová) | Zdravý | — | 423 |
| Rajče (rajče) | Early Blight | Alternaria solani | 1 000 |
| Rajče (rajče) | Late Blight | Phytophthora infestans | 1,909 |
| Rajče (rajče) | Listová plíseň | Tawny Passalora | 952 |
| Rajče (rajče) | Septoria Leaf Spot | Septoria lycopersici | 1,771 |
| Jabloň (Apple) | Strupovitost jablečná (strupovitost) | Venturia inaequalis | 630 |
| Jabloň (Apple) | Černá hniloba | Botryosphaeria obtusa | 621 |
| Jabloň (Apple) | Cedr Apple Rust | Gymnosporangium juniperi | 275 |
| Kukuřice (kukuřice) | Rez obecná | Puccinia sorghi | 1,192 |
| Brambory (brambory) | Early Blight | Alternaria solani | 1 000 |
| Brambory (brambory) | Late Blight | Phytophthora infestans | 1 000 |
| ... | + 24 dalších tříd | — | — |
Stažení a příprava datové sady
Dataset je k dispozici zdarma na Kaggle a Hugging Face. Níže je skript ke stažení, strukturování do složek a rozdělení vlaku/ověření/testu:
# 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}")Techniky augmentace dat pro obrázky listů
Obrázky PlantVillage se pořizují na jednotném pozadí v laboratoři. Abych zobecnil reálným podmínkám v terénu (kolísání světla, přirozené pozadí, různé úhly), augmentační techniky jsou základní. Kromě geometrických a fotometrických transformací standardy již zahrnuté ve výše uvedeném kódu, existují specifické techniky pro zemědělskou oblast:
- Náhodné mazání / výřez: Ztmaví náhodné obdélníky obrázku (simuluje okluze z překrývajících se listů, stínů, pracovních strojů). Snižuje nadměrné nošení lokální funkce a zlepšuje robustnost modelu o 2-3 % na skutečných obrázcích.
- MixUp a CutMix: Zkombinujte dva obrázky s náhodnými váhami (MixUp) nebo vyměňte oblast s patchem z jiného obrázku (CutMix). Zvláště efektivní pro třídy vizuálně podobné (např. raná plíseň vs. plíseň na rajčatech).
- Agresivní barevný jitter: Listové choroby vykazují barevné příznaky různé v závislosti na jevišti, slunečním záření a klimatických podmínkách. Zvýšení rozsahu variace odstínu/sytosti pomáhá zobecnění.
- Náhrada pozadí: Pokročilé techniky nahrazují bílé pozadí laboratoře se skutečnými snímky z terénu. Vyžaduje předběžnou sémantickou segmentaci ale výrazně zlepšuje výkon na snímcích v terénu.
Model Training: Transfer Learning with MobileNetV3
MobileNetV3Small je ideální architektura pro tento případ použití: navržená speciálně pro odvození na mobilních a vestavěných zařízeních nabízí vynikající rovnováhu mezi přesností a výpočetní náročnost. Ve srovnání s ResNet50 nebo EfficientNetB4 vyžaduje 100x méně operací odvozením při zachování konkurenční přesnosti.
Volby architektury a designu
Přeneste učení z předem vyškoleného ImageNetu a vítězné strategie: první vrstvy sítě se již naučily rozpoznávat textury, hrany a obecné barevné vzory. U listových chorob jsou tyto nízkoúrovňové prvky přímo znovu použitelné. Jemné doladění posledních vrstev (a volitelně posledních bloků) přizpůsobí model na konkrétní doménu.
# 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()Očekávané výsledky školení
Na kompletní datové sadě PlantVillage (38 tříd) byl model MobileNetV3Small doladěn obvykle dosahuje:
- Fáze 1 (15 epoch): Přesnost Val 91-93%, trénink ~45 min na T4 GPU
- Fáze 2 (20 epoch): Přesnost Val 96-97%, trénink ~2h na T4 GPU
- Testovací sady: Přesnost 96,5-97,5 %, průměrné F1-skóre 0,964
- Obtížné třídy: Raná plíseň vs pozdní plíseň na rajčatech (F1 ~0,91)
- Snadné lekce: Zdravé versus nemocné rostliny (F1 ~0,99)
Převod na TFLite s post-tréninkovou kvantizací
Natrénovaný model Keras je třeba pro provedení převést do formátu .tflite na Raspberry Pi. Proces převodu volitelně zahrnuje kvantování, což dramaticky zmenšuje velikost a zlepšuje rychlost vyvozování.
Tři režimy kvantizace
TFLite podporuje tři úrovně kvantizace s různými kompromisy mezi přesností a výkonem:
- Kvantifikace dynamického rozsahu: Kvantují se pouze váhy (float32 → int8). Aktivace zůstávají za běhu float32. Není vyžadována žádná kalibrační datová sada. Snížení velikost o 75 %, latence se zlepšila o 20–30 %. Doporučený výchozí bod.
- Plná celočíselná kvantizace (int8): Váhy AND aktivace kvantované v int8. Vyžaduje reprezentativní sadu dat kalibrace (100–500 snímků). Zmenšení velikosti 75 %, latence se zlepšila o 50-70 %. Vyžadováno pro hardwarově akcelerované (Coral TPU).
- Kvantizace Float16: Závaží v plováku 16. Užitečné hlavně pro GPU. Na ARM CPU nepřináší výrazné výhody oproti 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.")Nasazení na Raspberry Pi 5: Nastavení hardwaru a softwaru
Doporučený hardware
Raspberry Pi 5 (spuštěno v říjnu 2023, aktualizováno na 16GB verzi v roce 2024) a doporučená okrajová platforma pro tento případ použití. Oproti Pi 4 procesor Cortex-A76 na 2,4 GHz nabízí výkon téměř TFLite inference 5x vyšší, díky instrukcím ARM v8.2-A dot-product optimalizovaným pro operace int8.
Seznam hardwaru pro kompletní systém
| Komponent | Doporučený model | Orientační cena | Poznámky |
|---|---|---|---|
| Hlavní SBC | Raspberry Pi 5 (8 GB) | ~80 EUR | 8GB verze pro prostor RAM |
| Napájení | Oficiální RPi 5 PSU 27W USB-C | ~12 EUR | Nezbytné pro stabilní napájení |
| Skladování | MicroSD A2 64GB (Samsung Pro Endurance) | ~15 EUR | A2 pro lepší náhodné I/O |
| Fotoaparát | Raspberry Pi Camera Module 3 (12MP) | ~25 EUR | Autofokus, CSI-2 |
| Optika | 120° širokoúhlý objektiv | ~8 EUR | Pro pořízení z dronu nebo tyče |
| Domy | RPi 5 Active Cooler + kryt IP65 | ~20 EUR | Krytí IP65 pro venkovní použití |
| Síť | 4G LTE HAT (Waveshare SIM7600G) | ~45 EUR | Pro oblasti bez WiFi |
| Polní napájení | 20W solární panel + 10000mAh LiPo baterie | ~35 EUR | Autonomie ~ 3 dny bez slunce |
| Celkový systém | ~240 EUR | Pro samostatnou instalaci v terénu |
Instalační software na Raspberry Pi OS
# 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 elaborazioneKompletní Field Inference Pipeline
S převedeným modelem a nakonfigurovaným hardwarem je čas sestavit potrubí kompletní: pořízení snímku → předzpracování → TFLite inference → postprocessing → odeslání výsledku přes MQTT → dashboard. Toto je srdce produkční aplikace.
# 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: Porovnání platforem
Volba hardwarové platformy závisí na rozpočtu, požadovaném výkonu, spotřebě energetická a integrační složitost. Níže je aktualizované srovnání pro roky 2024–2025 na základě publikovaných benchmarků a vlastních měření na modelech klasifikace onemocnění.
Porovnání okrajových platforem pro zemědělské ML
| Platforma | CPU | BERAN | Latence MobileNetV3S int8 | Spotřeba | Náklady | Poznámky |
|---|---|---|---|---|---|---|
| Raspberry Pi 4B (4 GB) | Cortex-A72 1,8 GHz | 4GB LPDDR4 | ~320 ms | ~6W | ~55 EUR | Ekonomická základna |
| Raspberry Pi 5 (8 GB) | Cortex-A76 2,4 GHz | 8GB LPDDR4X | ~65 ms | ~8W | ~80 EUR | Doporučeno |
| RPi 4 + Coral USB TPU | Cortex-A72 + Edge TPU | 4 GB | ~15 ms | ~8W | ~95 EUR | Vyžaduje úplný model int8 |
| Google Coral Dev Board | Cortex-A53 + Edge TPU | 1 GB | ~12 ms | ~4W | ~120 EUR | Pouze kompilace TFLite + Coral |
| NVIDIA Jetson Nano (4 GB) | Cortex-A57 + 128 CUDA jader | 4GB LPDDR4 | ~8 ms | ~10W | ~149 USD | Overkill pro jednoduchou klasifikaci |
| Arduino Portenta H7 | Cortex-M7 480 MHz | 8 MB SDRAM | ~2000 ms | ~0,5W | ~100 EUR | Pouze malé modely (TFLite Micro) |
| Sipeed M1s Dock | BL808 RV64 480 MHz | 768 kB SRAM | ~800 ms (mikro) | ~0,3W | ~7 USD | Malé modely s ultranízkým výkonem |
Latence měřená na modelu MobileNetV3Small int8 (2,6 MB) pro jeden obrázek 224x224. Hodnoty Pi 5 z benchmarku Hackster.io 2024, Coral z benchmarku Georgia Southern University 2024.
Tepelné škrcení na Raspberry Pi 5
Raspberry Pi 5 bez aktivního chlazení zažije tepelné škrcení po 5-10 minutách kontinuálního vyvozování, se snížením výkonu 20–30 %. Pro polní nasazení (venkovní teploty do 40°C v létě), chlazení aktivní e povinný. Oficiální pouzdro s Active Cooler drží CPU pod 70°C i při trvalé zátěži.
Sledujte svou teplotu pomocí: vcgencmd measure_temp
Italské choroby rostlin: Prioritní cíle pro Edge ML
Italský zemědělský kontext má specifické vlastnosti, které vedou k výběru nasazení. Itálie je třetím největším producentem vína na světě (po Francii a Španělsku), prvním producentem vína a oliv z EU a má jednu z nejvyšších biologických rozmanitostí plodin v Evropě. Fytopatie ekonomicky nejpůsobivější pro italské zemědělství jsou:
Hlavní italské choroby rostlin a ekonomický dopad
| Nemoc | Činidlo | Postižené plodiny | Potenciální ztráta | Léčebné okénko |
|---|---|---|---|---|
| Plíseň révy vinná | Plasmopara viticola | Všechny hrozny DOC/DOCG | 20-100% produkce | 48-72h od vzhledu |
| Padlí réva | Erysiphe necator | Vinná réva, tykvovité | 15-60% produkce | 5-7 dní |
| Strupovitost jabloně | Venturia inaequalis | Ale Apple | 30-80% produkce | 3-5 dní |
| Plíseň rajčat | Phytophthora infestans | Rajče, brambory | 50-100% produkce | 24-48h |
| Botrytis (plíseň šedá) | Botrytis cinerea | Réva, jahoda, rajče | 10-40% produkce | 7-10 dní |
| Xylella fastidiosa | Xylella fastidiosa | Olivovník (Puglia) | Celá rostlina | Nenávratný |
V případě plísně hroznové je největší výzvou to, že PlantVillage nezahrnuje obrázky specifické pro italskou révu DOC/DOCG (Sangiovese, Nebbiolo, Primitivo, Nero d'Avola). Díky tomu je to nutné adaptace domény: shromáždit datovou sadu doplnění reálných obrázků z italských vinic a doladění modelu PlantVillage-předtrénovaný na místní datové sadě.
Fondy a pobídky pro italské digitální zemědělství
Il Strategický plán CAP 2023-2027 (Společná zemědělská politika) přiděluje zvláštní fondy na digitalizaci zemědělství prostřednictvím ekologických programů a intervencí sektorové. Intervence SRA22 („Řízení rizik“) a SRA29 („Přesné zemědělství“) poskytovat odměny společnostem, které přijmou digitální fytosanitární monitorovací systémy. Na národní úrovni má PNRR prostřednictvím opatření M2C4 alokovány finanční prostředky na inovace v zemědělství jako součást zelené a digitální transformace.
U zemědělských malých a středních podniků (společnosti s obratem < 2 mil. EUR) náklady na okrajový ML systém jako ten popsaný v tomto článku (~240 EUR hardware + vývoj) a odepisovatelné v jediné sezóně, pokud uvážíme, že včasné odhalení peronospory na vinice o rozloze 5 hektarů může ušetřit produkci v hodnotě 15 000–30 000 EUR.
Metriky výkonu a sledování
Systém ML ve výrobě bez monitorování je systém určený k degradaci tiše. Metriky, které mají být sledovány, jsou rozděleny do dvou kategorií: metriky model (přesnost) a systémové metriky (latence, uptime).
# 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
}Osvědčené postupy a anti-vzorce
Nejlepší postupy
8 zlatých pravidel pro Agricultural Edge ML
- Před nasazením vždy kvantujte: Rozdíl mezi float32 a int8 z hlediska latence na ARM a 50-70%, se ztrátou přesnosti pod 1%. Není důvod používat float32 ve výrobě na Pi 5.
- Kalibrace s reálnými daty z terénu: Soubor dat PlantVillage e získané v laboratoři. Nasbírejte alespoň 500 skutečných obrázků z cílového pole (stejná odrůda hroznů, stejné světlo, stejný úhel záběru) pro kalibraci int8 a pro jemné doladění. Rozdíl v přesnosti v terénu může být o 10-20% bez tohoto kroku.
- Zvládněte zpětnou vazbu: Implementujte mechanismus proč agronomové mohou potvrdit nebo opravit zjištění. Každá oprava je zlatá: se stanou tréninkovými daty pro další cyklus.
- Distribuce vstupu monitoru: Pokud model viděl pouze Sangiovese při výcviku, ale ve výrobě splňuje Primitivo, přesnost se zhroutí. Vykreslete rozložení vstupních vložení pro detekci posunu dat.
- Plán OTA aktualizace: Šablony musí být aktualizovatelné bez fyzického přístupu k zařízení. Implementujte mechanismus MQTT pro stahování a swap souboru .tflite s automatickým vrácením zpět v případě chyby.
- Redundance a režim offline: Systém musí správně fungovat i bez připojení. Pro detekce použijte místní vyrovnávací paměť (soubor SQLite nebo JSON). a synchronizovat, když se připojení obnoví.
- Kalibrujte práh spolehlivosti pro kulturu: Hranice 70 % pro plíseň (vysoce destruktivní, ekonomické ošetření) a odlišné od jednoho prahová hodnota pro padlí (méně naléhavá, nákladná léčba). Přizpůsobte práh k asymetrické náklady na falešně pozitivní vs.
- Zdokumentujte kontext akvizice: Každý obrázek musí být metadata s: denní dobou, povětrnostními podmínkami, fenologickým stádiem, GPS pozicí. Tato metadata jsou zásadní pro ladění a budoucí školení.
Anti-vzory, kterým je třeba se vyhnout
5 nejčastějších chyb v zemědělských Edge ML projektech
- Předpokládejme, že PlantVillage = výroba připravena: Soubor dat e získané za ideálních podmínek (rovnoměrné světlo, bílé pozadí, izolované listy). v terénu, obrázky mají stíny, zelené pozadí, kapky vody, překrývající se hmyz. Modelka trénovaný pouze na PlantVillage bez dolaďování v terénu má obvykle přesnost 60-70 % v reálném poli, oproti 97 % na laboratorní testovací sadě. Tato mezera a známý jako posun domény a to je hlavní praktický problém.
-
Ignorování vyvážení třídy: PlantVillage má mnoho tříd
nevyvážený (Tomato Late Blight: 1909 snímků; Cedr Apple Rust: 275 snímků).
Bez vážení tříd nebo převzorkování bude model orientován na početné třídy.
Použití
class_weightv Keras fit() neboWeightedRandomSampler. - Použití Pi jako webového serveru: Zpracovat synchronní požadavky HTTP pro závěr o Pi je špatná architektonická volba. Pi musí být producent Samostatný MQTT, nikoli server čekající na požadavky. Správa připojení a analýza HTTP zvyšuje režii a zbytečnou složitost.
- Nezvládejte extrémní světelné podmínky: srpna ve 12:00 Puglia, přímé světlo nasytí obrazy a listy vypadají zcela bílé. Za svítání s vysokou vlhkostí jsou listy mokré a odrážejí se jinak. Implementovat kontrola kvality obrazu (průměrný jas, sytost) předtím spusťte odvození a zahoďte/znovu zkuste snímky mimo rozsah.
- Kvantizovat bez ověření přesnosti: Plná kvantizace int8 vyžaduje reprezentativní soubor kalibračních dat. Použijte kalibrační datovou sadu příliš malé (méně než 100 obrázků) nebo nereprezentativní může vést ke ztrátě přesnost 3-5 % místo očekávaných 0,4 %. Vždy ověřte model kvantovány na testovací sadě před nasazením.
Aktualizace modelu Over-the-Air
S desítkami nebo stovkami zařízení nasazených v terénu ručně aktualizujte modelové a nepraktické. Distribuci umožňuje systém OTA (Over-the-Air) prostřednictvím MQTT nové modely bez fyzického přístupu k zařízením.
# 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()Závěry a další kroky
Vybudovali jsme kompletní systém detekce chorob plodin založený na Edge ML: od trénování modelu MobileNetV3Small na PlantVillage s jemným doladěním až po konverzi TFLite s kvantizací int8, nasazení na Raspberry Pi 5 s integrovaným potrubím Python pro zachycení, vyvození a odeslání MQTT. Systém dosahuje latence 65 ms za závěr, spotřebuje méně než 8W celkového výkonu a funguje zcela offline.
Čísla, která jsou pro zemědělského podnikatele skutečně důležitá, jsou různá: ztráty z nemocí plodiny jsou drahé 220 miliard dolarů celosvětově každý rok (FAO), a včasná detekce může snížit ztráty o 60–80 % při včasné léčbě. Systém jako ten popsaný stojí méně než 250 EUR v hardwaru a lze jej odepsat během jediné sezóny na jakékoli cenné plodině. Očekává se, že trh s umělou inteligencí v zemědělství poroste 5,9 miliardy USD v roce 2025 na 61,3 miliardy v roce 2035, s CAGR 26,3 %.
Největší mezera, kterou je třeba zaplnit, zůstává posun domény: PlantVillage pro nasazení v terénu to samo o sobě nestačí. Prioritou pro ty, kteří chtějí dát do výroby tohoto systému a sbírat skutečné obrázky z vašich plodin, označovat je s důvěryhodným agronomem a proveďte jemné doladění. Také 500 skutečných obrázků, správně použitých, dokáže odlišit systém s přesností 97 % v laboratoři a 65 % v terénu, vs. systém s 95% stabilitou v obou kontextech.
Shrnutí: Kompletní balíček technologií
| Vrstvy | Technologie | verze 2025 |
|---|---|---|
| Výcvik | TensorFlow / Keras | TF 2,17+ |
| Datové sady | PlantVillage (Kaggle/HuggingFace) | 54 309 snímků |
| Architektura | MobileNetV3Small | Přeneste učení ImageNet |
| Nasadit formát | TensorFlow Lite (.tflite) | int8 kvantováno |
| Železářské zboží | Raspberry Pi 5 (8 GB) | ARM Cortex-A76 |
| Pokoj | Modul fotoaparátu 3 (12MP) | Picamera2 + libcamera |
| Zasílání zpráv | MQTT (paho-mqtt) | Eclipse Mosquitto 2.x |
| Strava | Solární + LiPo | Autonomie 3+ dny |
| OTA | Ověření MQTT + SHA256 | Bezdotykové nasazení |
Připravované články v sérii FoodTech
- Článek 3 – Satelitní API a NDVI: Jak používat Sentinel-2 API (Copernicus) k monitorování zdraví plodin na poli pomocí dat bezplatný satelit, výpočet NDVI v Pythonu a integrace s IoT pipeline.
- Článek 4 – Počítačové vidění pro kvalitu potravin: Hluboké učení pro automatickou klasifikaci vad na výrobních linkách, třídění pro hmotnost/rozměr/barva a integrace s automatickými třídicími systémy.
- Článek 5 – Blockchain Food Sledovatelnost: Implementace systému sledovatelnosti z pole po vidlici s Hyperledger Fabric a standardy GS1 Digital Link pro soulad s legislativou EU o sledovatelnosti potravin.
Zdroje a reference
- PlantVillage Dataset: Hughes, D. P., & Salathé, M. (2016). Otevřené úložiště obrázků o zdraví rostlin, které umožňuje vývoj mobilních zařízení diagnostika onemocnění. ArXiv předtisk arXiv:1511.08060. K dispozici na Kaggle PlantVillage.
- Ztráty plodin FAO: FAO Plant Production and Protection – roční ztráty způsobené škůdci a chorobami
- TFLite na Raspberry Pi 5: Hackster.io — Srovnávání TFLite na Raspberry Pi 5
- Benchmarky Edge AI 2024: Georgia Southern University (2024). Srovnávání platforem Edge AI: Analýza výkonu NVIDIA Jetson a Raspberry Pi 5 s Coral TPU. Sborník konference IEEE. DOI: 10.1109/10971592.
- PlantVillage optimalizovaný pro MobileNetV3: Vědecké zprávy – Optimalizovaný MobileNet pro lehkou detekci chorob listů rostlin (2025)
- Itálie přesné zemědělství: Rural Hack — Zemědělství 4.0 Itálie 2025: Růst trhu a digitální vyspělost