ML Edge do wykrywania chorób upraw: TensorFlow Lite na Raspberry Pi
Co roku choroby roślin i szkodniki upraw niszczą do 40% produkcji światowe rolnictwo, powodując straty gospodarcze większe niż 220 miliardów dolarów według danych FAO. We Włoszech kraje winiarskie, sady i ogrody warzywne są podatne na endemiczne choroby roślin, takie jak mączniak rzekomy winorośli (Plasmopara viticola), mączniak prawdziwy (Necator Erysiphe) i parch jabłoni (Venturia inaequalis), z których każdy może wymazać cały rok produktywny, jeśli nie zostanie przechwycony w czasie.
Historycznie rzecz biorąc, wczesną diagnozę powierzano doświadczeniu wizualnemu rolnika lub, w najlepszym przypadku, okresowym inspekcjom agronomów. Obydwa podejścia mają to samo ograniczenie: opóźnienie Pomiędzy wystąpieniem choroby a interwencją fitosanitarną występuje przerwa, która często wynosi dni lub tygodnie podczas którego grzyb lub bakteria rozprzestrzenia się błyskawicznie. Okno czasowe dla skuteczne i minimalne leczenie: na przykład mączniak rzekomy winorośli jest redukowany do światła słonecznego 48-72 godziny od pojawienia się pierwszych zmian.
Sztuczna inteligencja zastosowana do widzenia komputerowego oferuje konkretną odpowiedź: modele głębokie uczenie się przeszkolone w oparciu o zbiór danych PlantVillage (54 309 obrazów, 38 klas) osiąga dokładność wyższy niż 97% w rozpoznawaniu chorób liści. Prawdziwe wąskie gardło nie chodzi już o dokładność modelu, ale o rozmieszczenie w terenie. W winnicy w Apeninach Toskanii-Emilian lub w gaju cytrusowym na Sycylii łączność z Internetem jest często nieobecna lub zawodna. Wysyłanie obrazów do chmury w celu wyciągnięcia wniosków jest niepraktyczne. Rozwiązanie iKrawędź ML: model działa bezpośrednio na urządzeniu, w terenie, bez łączności, z opóźnieniami mniejszymi niż drugi.
W tym artykule omówiono od podstaw kompletny proces wykrywania chorób na Raspberry Pi 5 przy użyciu TensorFlow Lite: od podstawowego szkolenia modelowego z nauką transferową na MobileNetV3, do int8, po wdrażanie z wnioskowaniem w języku Python w czasie rzeczywistym i wysyłanie wyników MQTT do zdalnego pulpitu nawigacyjnego. Każdemu krokowi towarzyszy działający kod i numery testów porównawczych zweryfikowane i praktyczne rozważania dotyczące włoskiego kontekstu rolniczego.
Czego dowiesz się w tym artykule
- dlaczego Edge ML przewyższa chmurę w zastosowaniach rolniczych na otwartym terenie
- Jak przygotować i rozszerzyć zbiór danych PlantVillage w celu skutecznego szkolenia
- Przenieś naukę za pomocą MobileNetV3Small na TensorFlow/Keras, krok po kroku
- Konwersja modelu do TFLite z kwantyzacją po treningu (float32 → int8)
- Konfiguracja sprzętowa Raspberry Pi 5 z modułem kamery 3 i pełnym wnioskowaniem w języku Python
- Zintegrowany potok: przechwytywanie → przetwarzanie wstępne → wnioskowanie → MQTT → dashboard
- Testy porównawcze opóźnień i zużycia energii na Pi 4, Pi 5, Coral TPU, Jetson Nano
- Główne włoskie choroby roślin i priorytetowe uprawy docelowe
- Najlepsze praktyki w zakresie wdrażania i typowe anty-wzorce
Pozycja w serii FoodTech
To drugi artykuł z tej serii Technologia żywności, poświęcony technologiom cyfrowym dla łańcucha dostaw rolno-spożywczych. Seria obejmuje rolniczy IoT, wizję komputerową, blockchain dla identyfikowalność, prognozowanie popytu i wiele więcej.
Seria FoodTech - Wszystkie artykuły
| # | Tytuł | Poziom | Państwo |
|---|---|---|---|
| 1 | Rurociąg IoT w rolnictwie: czujniki, MQTT i szeregi czasowe w InfluxDB | Mediator | Dostępny |
| 2 | ML Edge do wykrywania chorób upraw: TFLite na Raspberry Pi — Jesteś tutaj | Zaawansowany | Aktualny |
| 3 | Satelitarne API i NDVI: Monitorowanie upraw za pomocą Sentinel-2 | Mediator | Dostępny |
| 4 | Wizja komputerowa dla jakości żywności: wady, waga i klasyfikacja | Zaawansowany | Dostępny |
| 5 | Identyfikowalność żywności w łańcuchu bloków: tkanina Hyperledger i GS1 | Zaawansowany | Dostępny |
| 6 | Zgodność z FSMA 204: Cyfrowa identyfikowalność za pomocą Pythona i API | Mediator | Dostępny |
| 7 | Pionowa automatyzacja rolnictwa: kontrola klimatu za pomocą PLC i ML | Zaawansowany | Dostępny |
| 8 | Prognozowanie popytu na żywność: Prorok, LSTM i inżynieria cech | Mediator | Dostępny |
| 9 | Panel farmy w czasie rzeczywistym: Angular, WebSocket i InfluxDB | Mediator | Dostępny |
| 10 | Odporność łańcucha dostaw: optymalizacja za pomocą narzędzi OR i sztucznej inteligencji | Zaawansowany | Dostępny |
Podstawy Edge ML: Chmura kontra Edge dla rolnictwa
Przed napisaniem linii kodu konieczne jest zrozumienie, dlaczego podejście brzegowe jest jedynym rozwiązaniem praktyczne do wykrywania chorób na otwartym terenie. To nie jest wybór stylistyczny: oraz konieczność architektoniczna narzucona przez fizyczny kontekst rolnictwa.
Granice chmury w rolnictwie
Aplikacja do wykrywania chorób działająca w chmurze działałaby w następujący sposób: Kamera klika zdjęcie, wysyła je przez sieć na serwer w chmurze, model uruchamia się na serwerze, wynik wraca do urządzenie. W biurze wyposażonym w światłowód cykl ten trwa mniej niż 200 ms. W winnicy w Toskanii czy w gaju cytrusowym w Kalabrii zmienne zmieniają się radykalnie:
Chmura kontra Edge: porównanie w kontekście rolniczym
| Charakterystyczny | Podejście chmurowe | Podejście krawędziowe |
|---|---|---|
| Wymagana łączność | Ciągłe, pasmo 1+ Mbps | Opcjonalnie, tylko do synchronizacji danych |
| Opóźnienie wnioskowania | 200 ms - 5 s (w zależności od sieci) | 50-500 ms (lokalnie) |
| Koszt operacyjny | Wysoki (wywołania API, pamięć, przepustowość) | Niski (sprzęt jednorazowy) |
| Zapewniona prywatność | Obrazy korporacyjne w chmurze | Dane pozostają lokalne |
| Działanie offline | Niemożliwe | Kompletny |
| Skalowalność modelu | Nieograniczony | Ograniczone przez pamięć RAM/procesor |
| Aktualizacja modelu | Natychmiastowy | Wymaga wdrożenia fizycznego lub OTA |
| Zużycie energii | Niski (urządzenie) + wysoki (centrum danych) | Wszystko na urządzeniu (3-15W) |
TensorFlow vs TensorFlow Lite: kluczowe różnice
TensorFlow (TF) to kompletna platforma szkoleniowa i wnioskowania zoptymalizowana pod kątem procesorów graficznych i serwerów. TensorFlow Lite (TFLite) to skompresowana i zoptymalizowana wersja dla urządzeń wbudowanych i mobilnych. Praktyczne różnice są znaczące:
- Rozmiar środowiska wykonawczego: Środowisko wykonawcze TFLite i około 1 MB w C++ (w porównaniu z 400 MB + w przypadku pełnego TF). W przypadku Raspberry Pi różnica przekłada się na zajętość pamięci RAM na poziomie 50-100MB zamiast 2-4GB.
- Obsługiwani operatorzy: TFLite obsługuje podzbiór operacji TF. Modele z niestandardowymi operacjami lub niestandardowymi warstwami wymagają delegowania lub powrotu do procesora.
- Format .tflite: Schemat FlatBuffers, zoptymalizowany pod kątem bezpośredniego dostępu do pamięci bez analizowania. Model jest mapowany w pamięci bez deserializacji.
- Przyspieszenie sprzętowe: TFLite obsługuje delegatów XNNPACK (SIMD na ARM), Delegat GPU, delegat TPU Coral Edge i NNAPI na Androidzie.
Kwantyzacja: float32 do int8
Najbardziej wpływowa technika optymalizacji krawędzi i kwantyzacja: zmniejszenie precyzji numerycznej wag z 32-bitowych liczb zmiennoprzecinkowych do 8-bitowych liczb całkowitych. Efekt praktyczny i potrójny:
Wpływ kwantyzacji int8 na MobileNetV3Small
| Metryczny | pływak32 | int8 (PTQ) | Delta |
|---|---|---|---|
| Rozmiar modelu | 9,8 MB | 2,6 MB | -73% |
| Wnioskowanie z pamięci RAM (Pi 5) | ~180MB | ~52MB | -71% |
| Opóźnienie wnioskowania (Pi 5) | ~180 ms | ~65 ms | -64% |
| Dokładność PlantVillage | 97,2% | 96,8% | -0,4% |
| Zużycie energii | ~5,2 W | ~3,8 W | -27% |
Test porównawczy na Raspberry Pi 5 (2,4 GHz, 8 GB RAM). Wartości zmierzone za pomocą precyzyjnie dostrojonego modelu w 38 klasach PlantVillage.
Strata dokładności wynosząca 0,4% jest praktycznie nieistotna w terenie: naturalna zmienność warunki oświetleniowe, kąt i stadium choroby wprowadzają niepewność znacznie większy. Jednak w przypadku wdrożenia osadzonego decydujące znaczenie ma wzrost opóźnienia i rozmiaru.
Zbiór danych PlantVillage: przygotowanie i powiększanie
Referencyjny zbiór danych do wykrywania chorób liści i RoślinnaWieś, stworzony przez Hughesa i Salathé (Penn State University) i opublikowany w 2016 r. Zawiera więcej 54 309 obrazów sfotografowane w kontrolowanych warunkach laboratoryjnych, zorganizowane w 38 klas obejmujące 14 gatunków roślin i chorób pokrewnych.
Struktura zbioru danych
Klasy PlantVillage: gatunki i główne choroby
| Gatunek | Choroba | Agent przyczynowy | Obrazy |
|---|---|---|---|
| Winorośl (winogrono) | Czarna zgnilizna | Guignardia bidwellii | 1180 |
| Winorośl (winogrono) | Przynęta / Czarna odra | Phaeoacremonium spp. | 1383 |
| Winorośl (winogrono) | Zaraza liści | Pseudocercospora vitis | 1076 |
| Winorośl (winogrono) | Zdrowy | — | 423 |
| Pomidor (Pomidor) | Wczesna zaraza | Alternaria solani | 1000 |
| Pomidor (Pomidor) | Późna zaraza | Phytophthora infestans | 1909 |
| Pomidor (Pomidor) | Pleśń Liść | Tawny Passalora | 952 |
| Pomidor (Pomidor) | Plamistość liści Septorii | Septoria lycopersici | 1771 |
| Jabłoń (Jabłko) | Parch jabłoni (parch) | Venturia inaequalis | 630 |
| Jabłoń (Jabłko) | Czarna zgnilizna | Botryosphaeria obtusa | 621 |
| Jabłoń (Jabłko) | Rdza cedrowo-jabłkowa | Gymnosporangium jałowiec | 275 |
| Kukurydza (kukurydza) | Rdza zwyczajna | Puccinia Sorghi | 1192 |
| Ziemniak (Ziemniak) | Wczesna zaraza | Alternaria solani | 1000 |
| Ziemniak (Ziemniak) | Późna zaraza | Phytophthora infestans | 1000 |
| ... | + 24 inne klasy | — | — |
Pobieranie i przygotowanie zbioru danych
Zbiór danych jest dostępny bezpłatnie w Kaggle i Hugging Face. Poniżej znajduje się skrypt do pobrania, uporządkowanie w foldery i podzielony pociąg/walidacja/test:
# 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}")Techniki powiększania danych w przypadku obrazów liści
Obrazy PlantVillage są uzyskiwane w laboratorium na jednolitym tle. Uogólniać do rzeczywistych warunków terenowych (zmiany oświetlenia, naturalne tło, różne kąty), techniki augmentacji są podstawą. Oprócz przekształceń geometrycznych i fotometrycznych standardy już zawarte w powyższym kodzie, istnieją specyficzne techniki dla dziedziny rolnictwa:
- Losowe usuwanie/wycinanie: Przyciemnia losowe prostokąty obrazu (symuluje przesłonięcia spowodowane nakładającymi się liśćmi, cienie, maszyny robocze). Redukuje nadmierne dopasowanie lokalnych cech i poprawia niezawodność modelu o 2-3% w stosunku do rzeczywistych obrazów.
- MixUp i CutMix: Połącz dwa obrazy z losowymi wagami (MixUp) lub zamień region z łatką z innego obrazu (CutMix). Szczególnie skuteczny na zajęciach wizualnie podobne (np. zaraza wczesna vs zaraza późna pomidora).
- Agresywne drgania kolorów: Choroby liści wykazują objawy barwne różnią się w zależności od sceny, nasłonecznienia i warunków klimatycznych. Zwiększenie zakresu zmian odcienia/nasycenia pomaga w uogólnianiu.
- Podstawienie tła: Zaawansowane techniki zastępują białe tło laboratorium z rzeczywistymi obrazami terenowymi. Wymaga wstępnej segmentacji semantycznej ale drastycznie poprawia wydajność obrazów w terenie.
Szkolenie modelowe: transfer nauki za pomocą MobileNetV3
MobileNetV3Small to idealna architektura dla tego przypadku użycia: zaprojektowana specjalnie dla wnioskowanie na urządzeniach mobilnych i wbudowanych zapewnia doskonałą równowagę pomiędzy dokładnością i złożoność obliczeniowa. W porównaniu do ResNet50 lub EfficientNetB4 wymaga 100 razy mniej operacji poprzez wnioskowanie przy zachowaniu konkurencyjnej dokładności.
Wybór architektury i projektu
Przenieś naukę z wstępnie przeszkolonego ImageNet i zwycięską strategię: pierwsze warstwy sieci nauczyły się już rozpoznawać tekstury, krawędzie i ogólne wzorce kolorów. W przypadku chorób liści te funkcje niskiego poziomu można bezpośrednio ponownie wykorzystać. Dopracowanie ostatnich warstw (i opcjonalnie ostatnich bloków) dostosowuje model do określonej domeny.
# 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()Oczekiwane rezultaty szkolenia
Na kompletnym zestawie danych PlantVillage (38 klas) dopracowano model MobileNetV3Small zazwyczaj osiąga:
- Faza 1 (15 epok): Dokładność Val 91-93%, trening ~45 min na GPU T4
- Faza 2 (20 epok): Dokładność Val 96-97%, trening ~2h na GPU T4
- Zestawy testowe: Dokładność 96,5-97,5%, średni wynik F1 0,964
- Trudne zajęcia: Zaraza wczesna vs zaraza późna pomidorów (F1 ~0,91)
- Łatwe zajęcia: Rośliny zdrowe vs chore (F1 ~0,99)
Konwersja do TFLite z kwantyzacją po treningu
Aby można było wykonać wyszkolony model Keras, należy go przekonwertować na format .tflite na Raspberry Pi. Proces konwersji opcjonalnie obejmuje kwantyzację, co radykalnie zmniejsza rozmiar i poprawia szybkość wnioskowania.
Trzy tryby kwantyzacji
TFLite obsługuje trzy poziomy kwantyzacji, z różnymi kompromisami w zakresie dokładności i wydajności:
- Kwantyzacja zakresu dynamiki: Kwantyzowane są tylko wagi (float32 → int8). Aktywacje pozostają typu float32 w czasie wykonywania. Nie jest wymagany zestaw danych kalibracyjnych. Redukcja rozmiar o 75%, opóźnienie zmniejszone o 20-30%. Zalecany punkt wyjścia.
- Pełna kwantyzacja liczb całkowitych (int8): Wagi ORAZ aktywacje skwantowane w int8. Wymaga reprezentatywnego zestawu danych kalibracyjnych (100–500 obrazów). Zmniejszenie rozmiaru 75%, opóźnienie zmniejszone o 50-70%. Wymagane w przypadku akceleracji sprzętowej (Coral TPU).
- Kwantyzacja Float16: Ciężary w pływaku16. Przydatne głównie dla procesorów graficznych. Na procesorach ARM nie zapewnia znaczących korzyści w porównaniu z 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.")Wdróż na Raspberry Pi 5: konfiguracja sprzętu i oprogramowania
Zalecany sprzęt
Raspberry Pi 5 (wprowadzony na rynek w październiku 2023 r., zaktualizowany o wersję 16 GB w 2024 r.) i zalecana platforma brzegowa dla tego przypadku użycia. W porównaniu do Pi 4, procesor Cortex-A76 przy 2,4 GHz oferuje wydajność wnioskowania bliską TFLite 5x wyższy, dzięki instrukcjom produktu kropkowego ARM v8.2-A zoptymalizowanym pod kątem operacji int8.
Lista sprzętu dla kompletnego systemu
| Część | Polecany model | Orientacyjny koszt | Notatki |
|---|---|---|---|
| Główny SBC | Raspberry Pi 5 (8 GB) | ~80 EUR | Wersja 8 GB zapewniająca zapas pamięci RAM |
| Zasilanie | Oficjalny zasilacz RPi 5 27 W USB-C | ~12 EUR | Niezbędne do stabilnego zasilania |
| Składowanie | MicroSD A2 64 GB (Samsung Pro Endurance) | ~15 EUR | A2 dla lepszego losowego wejścia/wyjścia |
| Kamera | Moduł kamery Raspberry Pi 3 (12 MP) | ~25 EUR | Autofokus, CSI-2 |
| Optyka | Obiektyw szerokokątny 120° | ~8 EUR | Do nabycia z drona lub słupa |
| Domy | Aktywny Cooler RPi 5 + obudowa IP65 | ~20 EUR | Stopień ochrony IP65 do użytku na zewnątrz |
| Internet | CZAPKA 4G LTE (Waveshare SIM7600G) | ~45 EUR | Dla obszarów bez Wi-Fi |
| Zasilanie terenowe | Panel słoneczny 20W + akumulator LiPo 10000mAh | ~35 EUR | Autonomia ~3 dni bez słońca |
| Totalny system | ~240 EUR | Do samodzielnego montażu w terenie |
Skonfiguruj oprogramowanie w systemie operacyjnym 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 elaborazioneKompletny potok wnioskowania o terenie
Po przekonwertowaniu modelu i skonfigurowaniu sprzętu nadszedł czas na złożenie rurociągu ukończone: akwizycja obrazu → przetwarzanie wstępne → wnioskowanie TFLite → przetwarzanie końcowe → wysłanie wyniku poprzez MQTT → dashboard. To jest serce aplikacji produkcyjnej.
# 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()Test porównawczy sprzętu Edge: porównanie platform
Wybór platformy sprzętowej zależy od budżetu, wymaganej wydajności, zużycia złożoność energetyczna i integracyjna. Poniżej znajduje się zaktualizowane porównanie dla lat 2024–2025 w oparciu o opublikowane benchmarki i własne pomiary modeli klasyfikacji chorób.
Porównanie platform brzegowych dla rolnictwa ML
| Platforma | Procesor | BARAN | Opóźnienie MobileNetV3S int8 | Konsumpcja | Koszt | Notatki |
|---|---|---|---|---|---|---|
| Raspberry Pi 4B (4 GB) | Cortex-A72 1,8 GHz | 4GB LPDDR4 | ~320 ms | ~6W | ~55 EUR | Podstawa ekonomiczna |
| Raspberry Pi 5 (8 GB) | Cortex-A76 2,4 GHz | 8GB LPDDR4X | ~65 ms | ~8W | ~80 EUR | Zalecony |
| RPi 4 + koralowy TPU USB | Cortex-A72 + Edge TPU | 4 GB | ~15 ms | ~8W | ~95 EUR | Wymaga pełnego modelu int8 |
| Rada deweloperów Google Coral | Cortex-A53 + Edge TPU | 1 GB | ~12 ms | ~4W | ~120 EUR | Tylko kompilacja TFLite + Coral |
| NVIDIA Jetson Nano (4 GB) | Cortex-A57 + 128 rdzeni CUDA | 4GB LPDDR4 | ~8 ms | ~10W | ~149 USD | Przesada w przypadku prostej klasyfikacji |
| Arduino Portenta H7 | Cortex-M7 480 MHz | 8MB SDRAMU | ~2000 ms | ~0,5 W | ~100 EUR | Tylko małe modele (TFLite Micro) |
| Stacja dokująca Sipeed M1s | BL808 RV64 480 MHz | 768 KB pamięci SRAM | ~800 ms (mikro) | ~0,3 W | ~7 USD | Bardzo małe modele o bardzo niskim poborze mocy |
Opóźnienie mierzone w modelu MobileNetV3Small int8 (2,6MB) dla pojedynczego obrazu 224x224. Wartości Pi 5 z benchmarku Hackster.io 2024, Coral z benchmarku Georgia Southern University 2024.
Throttling termiczny w Raspberry Pi 5
Raspberry Pi 5 bez aktywnego chłodzenia doświadcza dławienia termicznego po 5-10 minutach ciągłego wnioskowania, ze zmniejszeniem wydajności 20-30%. Do stosowania w terenie (temperatura zewnętrzna do 40°C w lecie), chłodzenie aktywny tj obowiązkowy. Oficjalna obudowa z Active Coolerem mieści procesor poniżej 70°C, nawet przy ciągłym obciążeniu.
Monitoruj swoją temperaturę za pomocą: vcgencmd measure_temp
Włoskie choroby roślin: cele priorytetowe dla Edge ML
Kontekst włoskiego rolnictwa ma specyficzne cechy, które wpływają na decyzje dotyczące rozmieszczenia. Włochy są trzecim co do wielkości producentem wina na świecie (po Francji i Hiszpanii), pierwszym producentem wina i oliwek z UE i charakteryzuje się jedną z najwyższych różnorodności biologicznej upraw w Europie. Fitopatie najbardziej wpływowe ekonomicznie dla włoskiego rolnictwa to:
Główne włoskie choroby roślin i skutki gospodarcze
| Choroba | Agent | Dotknięte uprawy | Potencjalna strata | Okno leczenia |
|---|---|---|---|---|
| Mączniak rzekomy winorośli | Plasmopara viticola | Wszystkie winogrona DOC/DOCG | Produkcja 20-100%. | 48-72h od pojawienia się |
| Mączniak prawdziwy winorośli | Necator Erysiphe | Winorośl, dyniowate | 15-60% produkcji | 5-7 dni |
| Parch jabłoni | Venturia inaequalis | Jednak jabłko | 30-80% produkcji | 3-5 dni |
| Mączniak rzekomy pomidora | Phytophthora infestans | Pomidor, ziemniak | Produkcja 50-100%. | 24-48h |
| Botrytis (szara pleśń) | Botrytis cinerea | Winorośl, truskawka, pomidor | 10-40% produkcji | 7-10 dni |
| Xylella fastidiosa | Xylella fastidiosa | Drzewo oliwne (Apulia) | Cała roślina | Nie do odzyskania |
W przypadku mączniaka rzekomego winogron największym wyzwaniem jest to, że PlantVillage nie zawiera obrazów specyficzne dla włoskich winorośli DOC/DOCG (Sangiovese, Nebbiolo, Primitivo, Nero d'Avola). To sprawia, że jest to konieczne adaptacja domeny: zbierz zbiór danych uzupełnienie rzeczywistych zdjęć z włoskich winnic i dopracowanie modelu PlantVillage — wstępnie przeszkolony na lokalnym zestawie danych.
Fundusze i zachęty dla włoskiego rolnictwa cyfrowego
Il Plan strategiczny WPR na lata 2023-2027 (Wspólna Polityka Rolna) przydziela specjalne fundusze na cyfryzację rolnictwa poprzez ekoprogramy i interwencje sektorowy. Interwencja SRA22 („Zarządzanie ryzykiem”) i SRA29 („Rolnictwo precyzyjne”) zapewnia nagrody dla firm, które wdrażają cyfrowe systemy monitorowania fitosanitarnego. Na poziomie krajowym PNRR poprzez działanie M2C4 przeznaczyło środki na innowacje w rolnictwie w ramach transformacji ekologicznej i cyfrowej.
Dla małych i średnich przedsiębiorstw rolniczych (firm o obrotach < 2 mln EUR) koszt systemu Edge ML jak ten opisany w tym artykule (~240 EUR sprzęt + rozwój) i podlega amortyzacji w ciągu jednego sezonu, jeśli weźmiemy pod uwagę, że wczesne wykrycie mączniaka rzekomego winnica o powierzchni 5 hektarów może zaoszczędzić na produkcji o wartości 15 000–30 000 euro.
Wskaźniki wydajności i monitorowania
System ML w produkcji bez monitorowania jest systemem skazanym na degradację cicho. Metryki, które należy śledzić, są podzielone na dwie kategorie: metryki model (dokładność) i metryki systemu (opóźnienie, czas pracy).
# 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
}Najlepsze praktyki i anty-wzorce
Najlepsze praktyki
8 złotych zasad rolnictwa Edge ML
- Zawsze kwantyzuj przed wdrożeniem: Różnica między float32 i int8 pod względem opóźnienia na ARM i 50-70%, przy utracie dokładności poniżej 1%. Nie ma powodu używać float32 w produkcji na Pi 5.
- Kalibracja przy użyciu rzeczywistych danych terenowych: Zbiór danych PlantVillage, np nabyte w laboratorium. Zbierz co najmniej 500 prawdziwych obrazów z pola docelowego (ta sama odmiana winogron, to samo światło, ten sam kąt fotografowania) do kalibracji int8 i do dostrajania. Różnica w dokładności w terenie może być o 10-20% bez tego etapu.
- Obsługuj pętlę sprzężenia zwrotnego: Zaimplementuj mechanizm dlaczego agronomowie mogą potwierdzić lub skorygować ustalenia. Każde rozwiązanie jest złotem: staje się danymi treningowymi na następny cykl.
- Monitoruj dystrybucję sygnału wejściowego: Jeśli model widział tylko Sangiovese podczas treningu, ale w produkcji spełnia Primitivo, dokładność upada. Narysuj rozkład osadzania danych wejściowych, aby wykryć dryf danych.
- Zaplanuj aktualizację OTA: Szablony muszą być aktualizowalne bez fizycznego dostępu do urządzenia. Zaimplementuj mechanizm MQTT do pobierania i zamiana pliku .tflite z automatycznym przywracaniem w przypadku błędu.
- Redundancja i tryb offline: System musi działać poprawnie nawet bez łączności. Do wykrywania użyj lokalnego bufora (pliku SQLite lub JSON). i zsynchronizuj, gdy połączenie zostanie przywrócone.
- Skalibruj próg ufności dla kultury: Próg 70% na mączniaka rzekomego (wysoce niszczycielskie, ekonomiczne leczenie) i różni się od jednego próg dla mączniaka prawdziwego (mniej pilne, droższe leczenie). Dostosuj próg do asymetryczny koszt wyników fałszywie pozytywnych i fałszywie negatywnych.
- Udokumentuj kontekst przejęcia: Każdy obraz musi taki być metadane zawierające: porę dnia, warunki pogodowe, stopień fenologiczny, pozycję GPS. Te metadane mają kluczowe znaczenie dla debugowania i przyszłego szkolenia.
Anty-wzorce, których należy unikać
5 najczęstszych błędów w projektach Agricultural Edge ML
- Załóżmy, że PlantVillage = produkcja jest gotowa: Zbiór danych np uzyskane w idealnych warunkach (jednolite światło, białe tło, pojedyncze liście). Na polu, obrazy mają nałożone cienie, zielone tło, krople wody i owady. Modelka wyszkolony tylko w PlantVillage bez dostrajania w terenie, zazwyczaj ma dokładność 60-70% w rzeczywistym polu, w porównaniu do 97% na zestawie laboratoryjnym. Ta luka i znany jako zmiana domeny i to jest główny problem praktyczny.
-
Ignorowanie równowagi klas: PlantVillage ma wiele klas
niezrównoważony (Późna zaraza pomidora: 1909 obrazów; Rdza cedrowo-jabłkowa: 275 obrazów).
Bez ważenia klas i nadmiernego próbkowania model będzie stronniczy w kierunku klas obfitujących.
Używać
class_weightw Keras fit() lubWeightedRandomSampler. - Używanie Pi jako serwera WWW: Obsługuj synchroniczne żądania HTTP dla wnioskowanie o Pi jest złym wyborem architektonicznym. Pi musi być producentem Samodzielny MQTT, a nie serwer czekający na żądania. Zarządzanie połączeniami i analizowanie HTTP zwiększają obciążenie i niepotrzebną złożoność.
- Nie narażaj się na ekstremalne warunki oświetleniowe: O godzinie 12:00 w sierpniu w Apulia, bezpośrednie światło nasyca obrazy, a liście wydają się całkowicie białe. O świcie przy dużej wilgotności liście są mokre i inaczej odbijają światło. Wdrożyć wcześniejsze sprawdzenie jakości obrazu (średnia jasność, nasycenie). uruchom wnioskowanie i odrzuć/spróbuj ponownie obrazy spoza zakresu.
- Kwantyzacja bez sprawdzania dokładności: Pełna kwantyzacja int8 wymaga reprezentatywnego zestawu danych kalibracyjnych. Użyj zestawu danych kalibracyjnych zbyt mały (mniej niż 100 zdjęć) lub niereprezentatywny może prowadzić do strat dokładność 3-5% zamiast oczekiwanych 0,4%. Zawsze sprawdzaj model skwantowane na zestawie testowym przed wdrożeniem.
Aktualizacja modelu drogą bezprzewodową
W przypadku dziesiątek lub setek urządzeń wdrożonych w terenie należy ręcznie zaktualizować modele i niepraktyczne. System OTA (Over-the-Air) poprzez MQTT umożliwia dystrybucję nowe modele bez fizycznego dostępu do urządzeń.
# 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()Wnioski i dalsze kroki
Zbudowaliśmy kompletny system wykrywania chorób upraw w oparciu o Edge ML: od szkolenia modelu MobileNetV3Small w PlantVillage z dostrajaniem, po konwersję TFLite z kwantyzacją int8, wdrażany na Raspberry Pi 5 ze zintegrowanym potokiem Pythona do przechwytywania, wnioskowania i wysyłania protokołu MQTT. System osiąga opóźnienia rzędu 65 ms na wnioskowanie, zużywa niecałe 8W całkowitej mocy i działa całkowicie offline.
Liczby, które naprawdę liczą się dla przedsiębiorcy rolnego, są różne: straty spowodowane chorobami upraw są drogie 220 miliardów dolarów na całym świecie co roku (FAO), a wczesne wykrycie może zmniejszyć straty o 60–80%, jeśli leczenie zostanie podjęte w odpowiednim czasie. System taki jak ten opisany kosztuje mniej niż 250 EUR w sprzęcie i można go amortyzować w ciągu jednego sezonu na każdej wartościowej uprawie. Oczekuje się, że rynek sztucznej inteligencji w rolnictwie będzie się rozwijał 5,9 mld USD w 2025 r. do 61,3 mld USD w 2035 r., przy CAGR na poziomie 26,3%.
Największą luką do wypełnienia pozostaje tzw zmiana domeny: PlantVillage sam w sobie nie wystarczy do wdrożenia w terenie. Priorytet dla tych, którzy chcą postawić do produkcji tego systemu i zbieraj prawdziwe obrazy swoich upraw, oznacz je z zaufanym agronomem i dokonaj dostrojenia. Również 500 prawdziwych obrazów, używanych prawidłowo, może sprawić różnicę pomiędzy systemem z dokładnością 97% w laboratorium i 65% w terenie, w porównaniu z systemem stabilnym w 95% w obu kontekstach.
Podsumowanie: Kompletny stos technologii
| Warstwy | Technologia | Wersja 2025 |
|---|---|---|
| Szkolenie | TensorFlow/Keras | TF 2.17+ |
| Zbiory danych | PlantVillage (Kaggle/HuggingFace) | 54 309 obrazów |
| Architektura | MobileNetV3Small | Transfer nauki ImageNet |
| Wdróż format | TensorFlow Lite (.tflite) | int8 skwantowany |
| Sprzęt komputerowy | Raspberry Pi 5 (8 GB) | Kora ARM-A76 |
| Pokój | Moduł kamery 3 (12 MP) | Picamera2 + libcamera |
| Wiadomości | MQTT (paho-mqtt) | Eclipse Mosquitto 2.x |
| Dieta | Energia słoneczna + LiPo | Autonomia 3+ dni |
| OTA | Sprawdź MQTT + SHA256 | Wdrożenie bezdotykowe |
Nadchodzące artykuły z serii FoodTech
- Artykuł 3 — Satelitarne API i NDVI: Jak korzystać z API Sentinel-2 (Copernicus) do monitorowania stanu upraw w skali polowej za pomocą danych darmowy satelita, obliczenia NDVI w Pythonie i integracja z rurociągiem IoT.
- Artykuł 4 – Wizja komputerowa dla jakości żywności: Głębokie uczenie się do automatycznej klasyfikacji wad na liniach produkcyjnych, klasyfikacja dla waga/wymiar/kolor oraz integracja z automatycznymi systemami sortowania.
- Artykuł 5 – Możliwość śledzenia żywności w łańcuchu bloków: Wdrożenie systemu identyfikowalności od pola do stołu, zgodnego ze standardami Hyperledger Fabric i GS1 Łącze cyfrowe zapewniające zgodność z przepisami UE dotyczącymi identyfikowalności żywności.
Zasoby i referencje
- Zbiór danych PlantVillage: Hughes, DP i Salathé, M. (2016). Repozytorium obrazów na temat zdrowia roślin o otwartym dostępie, umożliwiające rozwój urządzeń mobilnych diagnoza choroby. Wydruk wstępny ArXiv arXiv:1511.08060. Dostępne na Wioska Kaggle Plant.
- Straty w uprawach FAO: Produkcja i ochrona roślin FAO — roczne straty spowodowane szkodnikami i chorobami
- TFLite na Raspberry Pi 5: Hackster.io — Test porównawczy TFLite na Raspberry Pi 5
- Testy porównawcze Edge AI 2024: Uniwersytet Południowy Georgia (2024). Benchmarking platform Edge AI: analiza wydajności NVIDIA Jetson i Raspberry Pi 5 z Coral TPU. Materiały konferencyjne IEEE. DOI: 10.1109/10971592.
- PlantVillage zoptymalizowane pod kątem MobileNetV3: Raporty naukowe — Zoptymalizowana sieć mobilna do wykrywania lekkich chorób liści roślin (2025)
- Włochy Rolnictwo precyzyjne: Rural Hack – Rolnictwo 4.0 Włochy 2025: rozwój rynku i dojrzałość cyfrowa