Mahsul Hastalıklarının Tespiti için ML Edge: Raspberry Pi'de TensorFlow Lite
Her yıl bitki hastalıkları ve ürün zararlıları Üretimin %40'ı dünya tarımıdaha büyük ekonomik kayıplara neden 220 milyar dolar FAO verilerine göre. İtalya'da bağcılık yapılan ülkeler, meyve bahçeleri ve sebze bahçeleri aşağıdakiler gibi endemik bitki hastalıklarına karşı hassastır: asma tüylü küfü (Plazmopara viticola), külleme (Erysiphe necator) ve elma kabuğu (Venturia inaequalis), her biri bir yılı tamamen silme kapasitesine sahip zamanında müdahale edilmezse üretkendir.
Erken tanı, tarihsel olarak çiftçinin görsel deneyimine veya en iyi durumda, çiftçinin görsel deneyimine emanet edilmiştir. Ziraat mühendisleri tarafından periyodik denetimler. Her iki yaklaşım da aynı sınırlamadan muzdariptir: gecikme Hastalığın başlangıcı ile bitki sağlığı müdahalesi arasında genellikle günler veya haftalar süren bir aralık vardır. Bu sırada mantar veya bakteri kontrol edilemeyen bir yangın gibi yayılır. Bir zaman penceresi etkili ve minimum düzeyde tedavi: örneğin asma küfü için güneşe indirgenir 48-72 saat ilk lezyonların ortaya çıkmasından itibaren.
Bilgisayar görüşüne uygulanan yapay zeka somut bir cevap sunuyor: PlantVillage veri kümesi (54.309 görüntü, 38 sınıf) üzerinde eğitilen derin öğrenme, doğruluk elde ediyor daha yüksek %97 Yaprak hastalıklarının tanınmasında. Gerçek darboğaz artık modelin doğruluğu değil, saha dağıtımı. Apeninler'deki bir üzüm bağında Toskana-Emilian'da veya Sicilya'daki bir narenciye bahçesinde internet bağlantısı genellikle yoktur veya güvenilmezdir. Çıkarım için görüntüleri buluta göndermek pratik değildir. Çözüm veKenar ML: model, sahada, bağlantı olmadan, gecikme süresinden daha az bir süre ile doğrudan cihaz üzerinde çalışır. ikinci.
Bu makale sıfırdan Raspberry Pi 5 üzerinde eksiksiz bir hastalık tespit hattı oluşturuyor TensorFlow Lite kullanımı: MobileNetV3'te transfer öğrenimi ile temel model eğitiminden, int8 niceleme, gerçek zamanlı Python çıkarımıyla konuşlandırma, sonuçları gönderme Uzak bir kontrol paneline MQTT. Her adıma çalışma kodu, kıyaslama numaraları eşlik eder İtalyan tarım bağlamı için doğrulanmış ve pratik hususlar.
Bu Makalede Neler Öğreneceksiniz?
- Edge ML neden açık alan tarım uygulamaları için buluttan daha iyi performans gösteriyor?
- Etkili eğitim için PlantVillage veri kümesi nasıl hazırlanır ve genişletilir?
- TensorFlow/Keras'ta MobileNetV3Small ile öğrenmeyi adım adım aktarın
- Eğitim sonrası niceleme ile modelin TFLite'a dönüştürülmesi (float32 → int8)
- Kamera Modülü 3 ve tam Python çıkarımıyla Raspberry Pi 5 donanım kurulumu
- Entegre işlem hattı: yakalama → ön işleme → çıkarım → MQTT → kontrol paneli
- Pi 4, Pi 5, Coral TPU ve Jetson Nano'da gecikme ve güç tüketimi karşılaştırmaları
- Başlıca İtalyan bitki hastalıkları ve öncelikli hedef ürünler
- En iyi dağıtım uygulamaları ve yaygın anti-kalıplar
FoodTech Serisindeki Konumu
Bu serinin ikinci makalesi Gıda Teknolojisi, dijital teknolojilere adanmış tarım-gıda tedarik zinciri için. Bu seri tarımsal Nesnelerin İnterneti, bilgisayar görüşü ve blockchain konularını kapsıyor. izlenebilirlik, talep tahmini ve çok daha fazlası.
FoodTech Serisi - Tüm Makaleler
| # | Başlık | Seviye | Durum |
|---|---|---|---|
| 1 | Tarımsal IoT Boru Hattı: InfluxDB'de Sensörler, MQTT ve Zaman Serisi | Orta seviye | Mevcut |
| 2 | Mahsul Hastalıklarının Tespiti için ML Edge: Raspberry Pi'de TFLite — Buradasınız | Gelişmiş | Akım |
| 3 | Uydu API ve NDVI: Sentinel-2 ile Mahsul İzleme | Orta seviye | Mevcut |
| 4 | Gıda Kalitesi için Bilgisayarlı Görme: Kusurlar, Ağırlık ve Sınıflandırma | Gelişmiş | Mevcut |
| 5 | Blockchain Gıda İzlenebilirliği: Hyperledger Fabric ve GS1 | Gelişmiş | Mevcut |
| 6 | FSMA 204 Uyumluluğu: Python ve API ile Dijital İzlenebilirlik | Orta seviye | Mevcut |
| 7 | Dikey Tarım Otomasyonu: PLC ve ML ile İklim Kontrolü | Gelişmiş | Mevcut |
| 8 | Gıda Talep Tahmini: Peygamber, LSTM ve Özellik Mühendisliği | Orta seviye | Mevcut |
| 9 | Çiftlik Kontrol Paneli Gerçek Zamanlı: Angular, WebSocket ve InfluxDB | Orta seviye | Mevcut |
| 10 | Tedarik Zinciri Esnekliği: OR-Tools ve Yapay Zeka ile Optimizasyon | Gelişmiş | Mevcut |
Edge ML Temelleri: Tarım için Bulut vs Edge
Bir kod satırı yazmadan önce, Edge yaklaşımının neden tek yaklaşım olduğunu anlamak önemlidir. Açık alanda hastalık tespiti için pratiktir. Bu stilistik bir seçim değil: ve tarımın fiziksel bağlamının dayattığı mimari bir gereklilik.
Tarımda Bulutun Sınırları
Bulut öncelikli bir hastalık tespit uygulaması şu şekilde çalışır: Kamera tıklar bir fotoğraf, onu ağ üzerinden bulut sunucusuna gönderir, model sunucuda çalışır, sonuç cihaz. Fiber optikli bir ofiste bu döngü 200 ms'den az sürer. Bir üzüm bağında Toskana'da veya Calabria'daki bir narenciye bahçesinde değişkenler kökten değişiyor:
Bulut ve Edge: Tarımsal Bağlama Göre Karşılaştırma
| karakteristik | Bulut yaklaşımı | Kenar yaklaşımı |
|---|---|---|
| Bağlantı gerekli | Sürekli, 1+ Mbps bant | İsteğe bağlı, yalnızca veri senkronizasyonu için |
| Çıkarım gecikmesi | 200ms - 5s (ağa bağlı) | 50-500 ms (yerel) |
| İşletme maliyeti | Yüksek (API çağrıları, depolama, bant genişliği) | Düşük (tek seferlik donanım) |
| Verilen gizlilik | Buluttaki kurumsal görseller | Veriler şirket içinde kalır |
| Çevrimdışı işlem | İmkansız | Tamamlamak |
| Model ölçeklenebilirliği | Sınırsız | RAM/CPU ile sınırlıdır |
| Model güncellemesi | hemen | Fiziksel veya OTA dağıtımı gerektirir |
| Enerji tüketimi | Düşük (cihaz) + yüksek (veri merkezi) | Cihazdaki her şey (3-15W) |
TensorFlow ve TensorFlow Lite: Temel Farklılıklar
TensorFlow (TF), GPU'lar ve sunucular için optimize edilmiş eksiksiz bir eğitim ve çıkarım çerçevesidir. TensorFlow Lite (TFLite), gömülü ve mobil cihazlar için sıkıştırılmış ve optimize edilmiş versiyondur. Pratik farklılıklar önemlidir:
- Çalışma Zamanı Boyutu: TFLite çalışma zamanı ve C++'da yaklaşık 1 MB (tam TF için 400 MB+'a karşılık). Raspberry Pi'de fark, 2-4GB yerine 50-100MB RAM doluluğuna dönüşüyor.
- Desteklenen operatörler: TFLite, TF işlemlerinin bir alt kümesini destekler. Modeller özel işlemler veya standart dışı katmanlar, CPU'ya yetki veya geri dönüş gerektirir.
- .tflite biçimi: Doğrudan bellek erişimi için optimize edilmiş FlatBuffers şeması ayrıştırmadan. Model, seri durumdan çıkarma olmadan bellek eşlemelidir.
- Donanım hızlandırma: TFLite, XNNPACK delegelerini destekler (ARM'de SIMD), GPU temsilcisi, Coral Edge TPU temsilcisi ve Android'de NNAPI.
Niceleme: float32'den int8'e
Kenar optimizasyonu için en etkili teknik ve nicemleme: ağırlıkların sayısal hassasiyetinin 32 bitlik kayan noktalardan 8 bitlik tam sayılara düşürülmesi. Etki pratik ve üç yönlü:
İnt8 Nicelemenin MobileNetV3Small Üzerindeki Etkileri
| Metrik | kayan nokta32 | int8 (PTQ) | Delta |
|---|---|---|---|
| Modeli boyutu | 9,8 MB | 2,6 MB | -73% |
| RAM çıkarımı (Pi 5) | ~180MB | ~52MB | -71% |
| Çıkarım gecikmesi (Pi 5) | ~180 ms | ~65 ms | -64% |
| PlantVillage doğruluğu | %97,2 | %96,8 | -%0,4 |
| Enerji tüketimi | ~5,2W | ~3,8W | -27% |
Raspberry Pi 5 (2,4 GHz, 8 GB RAM) ile karşılaştırma. Değerler 38 PlantVillage sınıfında ince ayarlı modelle ölçülmüştür.
%0,4'lük doğruluk kaybının sahada pratikte hiçbir önemi yoktur: doğal değişkenlik Aydınlatma koşulları, hastalığın açısı ve evresi belirsizliklere neden oluyor çok daha büyük. Ancak gecikme ve boyuttaki kazanç, yerleşik dağıtım için belirleyicidir.
PlantVillage Veri Kümesi: Hazırlama ve Büyütme
Yaprak hastalıklarını tespit etmek için referans veri seti ve Bitki Köyü, Hughes ve Salathé (Penn State Üniversitesi) tarafından oluşturulmuş ve 2016'da yayınlanmıştır. Daha fazlasını içerir 54.309 görsel kontrollü laboratuvar koşullarında fotoğraflandı, organize edildi içinde 38 sınıf 14 bitki türünü ve ilgili hastalıkları kapsamaktadır.
Veri kümesi yapısı
PlantVillage Sınıfları: Türler ve Ana Hastalıklar
| Türler | Hastalık | Nedensel Ajan | Görseller |
|---|---|---|---|
| Asma (Üzüm) | Siyah Çürük | Guignardia bidwellii | 1.180 |
| Asma (Üzüm) | Yem / Kara Kızamık | Phaeoacremonium spp. | 1.383 |
| Asma (Üzüm) | Yaprak Yanıklığı | Pseudocercospora vitis | 1.076 |
| Asma (Üzüm) | Sağlıklı | — | 423 |
| Domates (Domates) | Erken Yanıklık | Alternaria solani | 1.000 |
| Domates (Domates) | Geç Yanıklık | Fitofthora infestans | 1.909 |
| Domates (Domates) | Yaprak Kalıbı | Sarımsı Passalora | 952 |
| Domates (Domates) | Septoria Yaprak Lekesi | Septoria lycopersici | 1.771 |
| Elma ağacı (Elma) | Elma Kabuğu (Kabuk) | Venturia inaequalis | 630 |
| Elma ağacı (Elma) | Siyah Çürük | Botryosphaeria obtusa | 621 |
| Elma ağacı (Elma) | Sedir Elması Pası | Gymnosporangium ardıç | 275 |
| Mısır (Mısır) | Ortak Pas | Puccinia sorghi | 1.192 |
| Patates (Patates) | Erken Yanıklık | Alternaria solani | 1.000 |
| Patates (Patates) | Geç Yanıklık | Fitofthora infestans | 1.000 |
| ... | + 24 diğer sınıf | — | — |
Veri Kümesi İndirme ve Hazırlama
Veri kümesi Kaggle ve Hugging Face'te ücretsiz olarak mevcuttur. İndirme betiği aşağıdadır, klasörler halinde yapılanma ve bölünmüş eğitim/doğrulama/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}")Yaprak Görüntüleri için Veri Arttırma Teknikleri
PlantVillage görüntüleri laboratuvarda tek tip bir arka planda elde edilir. genelleştirmek gerçek saha koşullarına (ışık değişimleri, doğal arka plan, farklı açılar) göre büyütme teknikleri esastır. Geometrik ve fotometrik dönüşümlerin yanı sıra Yukarıdaki kodda zaten yer alan standartların yanı sıra, tarım alanına yönelik özel teknikler de bulunmaktadır:
- Rastgele Silme / Kesme: Görüntünün rastgele dikdörtgenlerini koyulaştırır (simüle eder) üst üste gelen yapraklardan, gölgelerden, iş makinelerinden kaynaklanan tıkanmalar). Aşırı uyumu azaltır yerel özellikler sunar ve model sağlamlığını gerçek görüntülere göre %2-3 artırır.
- MixUp ve CutMix: İki görüntüyü rastgele ağırlıklarla birleştirin (MixUp) veya değiştirin başka bir görüntüden (CutMix) yamanın bulunduğu bir bölge. Özellikle dersler için etkili görsel olarak benzer (örneğin domateste erken yanıklık ve geç yanıklık).
- Agresif Renk Değişimi: Yaprak hastalıkları renk belirtileri gösterir sahneye, güneşe maruz kalma durumuna ve iklim koşullarına bağlı olarak farklılık gösterir. Ton/doygunluk değişim aralığını artırmak genellemeye yardımcı olur.
- Arka plan değişikliği: Beyaz arka planın yerini gelişmiş teknikler alıyor Laboratuvarın gerçek saha görüntüleri ile. Ön anlamsal bölümleme gerektirir ancak saha içi görüntülerde performansı büyük ölçüde artırır.
Model Eğitimi: MobileNetV3 ile Öğrenmeyi Aktarın
MobileNetV3Small bu kullanım durumu için ideal mimaridir: özellikle mobil ve gömülü cihazlardaki çıkarımlar doğruluk arasında mükemmel bir denge sunar ve hesaplama karmaşıklığı. ResNet50 veya EfficientNetB4 ile karşılaştırıldığında, 100 kat daha az işlem Rekabetçi doğrulukları korurken çıkarım yoluyla.
Mimari ve Tasarım Seçimleri
Önceden eğitilmiş ImageNet'ten öğrenimi ve kazanma stratejisini aktarın: İlk katmanlar ağlar dokuları, kenarları ve genel renk desenlerini tanımayı zaten öğrenmiştir. Yaprak hastalıkları için bu düşük seviyeli özellikler doğrudan yeniden kullanılabilir. Son katmanlara (ve isteğe bağlı olarak son bloklara) ince ayar yapılması modeli uyarlar belirli bir etki alanına.
# 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()Eğitimden Beklenen Sonuçlar
Eksiksiz bir PlantVillage veri kümesinde (38 sınıf), MobileNetV3Small modelinde ince ayarlar yapılmıştır genellikle şunları başarır:
- Aşama 1 (15 dönem): Değer doğruluğu %91-93, T4 GPU'da eğitim ~45 dakika
- Aşama 2 (20 dönem): Değer doğruluğu %96-97, T4 GPU'da eğitim ~2 saat
- Test setleri: Doğruluk %96,5-97,5, ortalama F1 puanı 0,964
- Zor sınıflar: Domateste erken yanıklık ve geç yanıklık (F1 ~0.91)
- Kolay Sınıflar: Sağlıklı ve hastalıklı bitkiler (F1 ~0,99)
Eğitim Sonrası Niceleme ile TFLite'a Dönüştürme
Eğitilen Keras modelinin yürütülebilmesi için .tflite biçimine dönüştürülmesi gerekir Raspberry Pi'de. Dönüştürme işlemi isteğe bağlı olarak nicelemeyi içerir, bu da boyutu önemli ölçüde azaltır ve çıkarım hızını artırır.
Üç Niceleme Modu
TFLite, farklı doğruluk/performans değişimleriyle üç düzeyli nicelemeyi destekler:
- Dinamik Aralık Kuantizasyonu: Yalnızca ağırlıklar nicelenir (float32 → int8). Etkinleştirmeler çalışma zamanında float32 olarak kalır. Kalibrasyon veri seti gerekmez. Azaltma boyut %75 arttı, gecikme %20-30 arttı. Önerilen başlangıç noktası.
- Tam Tam Sayı Niceleme (int8): Ağırlıklar VE aktivasyonlar int8'de nicelendirilmiştir. Temsili bir kalibrasyon veri seti gerektirir (100-500 görüntü). Boyut küçültme %75, gecikme %50-70 oranında iyileştirildi. Donanım hızlandırmalı (Coral TPU) için gereklidir.
- Float16 Niceleme: Float16'daki ağırlıklar. Esas olarak GPU'lar için kullanışlıdır. ARM CPU'larda float32'ye göre önemli avantajlar getirmez.
# 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.")Raspberry Pi 5'te Dağıtım: Donanım ve Yazılım Kurulumu
Önerilen Donanım
Raspberry Pi 5 (Ekim 2023'te piyasaya sürüldü, 2024'te 16GB sürümle güncellendi) ve bu kullanım durumu için önerilen uç platform. Pi 4 ile karşılaştırıldığında işlemci 2,4 GHz'de Cortex-A76, TFLite'a yakın çıkarım performansı sunuyor 5 kat daha yüksek, int8 işlemleri için optimize edilmiş ARM v8.2-A nokta ürün talimatları sayesinde.
Komple Sistem için Donanım Listesi
| Bileşen | Önerilen model | Gösterge maliyeti | Notlar |
|---|---|---|---|
| Ana KKG | Ahududu Pi 5 (8GB) | ~80 Avro | RAM boşluğu için 8GB versiyonu |
| Güç kaynağı | Resmi RPi 5 PSU 27W USB-C | ~12 Avro | Kararlı güç kaynağı için gerekli |
| Depolamak | MicroSD A2 64GB (Samsung Pro Dayanıklılık) | ~15 Avro | Daha iyi rastgele G/Ç için A2 |
| Kamera | Raspberry Pi Kamera Modülü 3 (12MP) | ~25 Avro | Otomatik odaklama, CSI-2 |
| Optik | 120° geniş açılı mercek | ~8 Avro | Drone veya direkte satın almak için |
| Evler | RPi 5 Aktif Soğutucu + IP65 kasa | ~20 Avro | Dış mekan kullanımı için IP65 koruması |
| Açık | 4G LTE ŞAPKA (Waveshare SIM7600G) | ~45 Avro | WiFi olmayan alanlar için |
| Saha güç kaynağı | 20W güneş paneli + 10000mAh LiPo pil | ~35 Avro | Özerklik ~3 gün güneşsiz |
| Toplam sistem | ~240 Avro | Sahada bağımsız kurulum için |
Raspberry Pi OS'de Kurulum Yazılımı
# 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 elaborazioneTam Alan Çıkarım Hattı
Model dönüştürüldükten ve donanım yapılandırıldıktan sonra boru hattını birleştirmenin zamanı geldi tamamlama: görüntü edinme → ön işleme → TFLite çıkarımı → son işleme → MQTT → kontrol paneli aracılığıyla sonuç gönderiliyor. Bu, üretim uygulamasının kalbidir.
# 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()Donanım Kenarı Karşılaştırması: Platform Karşılaştırması
Donanım platformunun seçimi bütçeye, gerekli performansa ve tüketime bağlıdır Enerji ve entegrasyon karmaşıklığı. Aşağıda 2024-2025 için güncellenmiş bir karşılaştırma bulunmaktadır hastalık sınıflandırma modellerine ilişkin yayınlanmış kıyaslamalara ve kendi ölçümlerine dayanmaktadır.
Tarımsal ML için Uç Platformların Karşılaştırılması
| platformu | İşlemci | Veri deposu | MobileNetV3S int8 gecikmesi | Tüketim | Maliyet | Notlar |
|---|---|---|---|---|---|---|
| Ahududu Pi 4B (4GB) | Cortex-A72 1.8GHz | 4 GB LPDDR4 | ~320 ms | ~6W | ~55 Avro | Ekonomik temel |
| Ahududu Pi 5 (8GB) | Cortex-A76 2,4 GHz | 8GB LPDDR4X | ~65 ms | ~8W | ~80 Avro | Tavsiye edilen |
| RPi 4 + Mercan USB TPU | Cortex-A72 + Kenar TPU | 4 cigabayt | ~15 ms | ~8W | ~95 Avro | İnt8 tam modelini gerektirir |
| Google Coral Geliştirme Kurulu | Cortex-A53 + Kenar TPU | 1 GB | ~12 ms | ~4W | ~120 Avro | Yalnızca TFLite + Coral derlemesi |
| NVIDIA Jetson Nano (4 GB) | Cortex-A57 + 128 CUDA çekirdeği | 4 GB LPDDR4 | ~8 ms | ~10W | ~149 ABD Doları | Basit sınıflandırma için aşırıya kaçma |
| Arduino Portenta H7 | Cortex-M7 480MHz | 8 MB SDRAM | ~2000 ms | ~0,5W | ~100 Avro | Yalnızca küçük modeller (TFLite Micro) |
| Sipeed M1s Bağlantı İstasyonu | BL808 RV64 480MHz | 768KB SRAM | ~800 ms (mikro) | ~0,3W | ~7 ABD Doları | Ultra düşük güçlü, küçük modeller |
Gecikme, tek 224x224 görüntü için MobileNetV3Small int8 modelinde (2,6 MB) ölçülmüştür. Hackster.io 2024 kıyaslamasından Pi 5 değerleri, Georgia Southern University 2024 kıyaslamasından Coral.
Raspberry Pi 5'te Termal Azaltma
Aktif soğutması olmayan Raspberry Pi 5, 5-10 dakika sonra termal daralma yaşıyor performansında bir azalma ile sürekli çıkarım %20-30. Sahada kullanım için (yaz aylarında dış ortam sıcaklıkları 40°C'ye kadar), soğutma aktif e zorunlu. Active Cooler'ın bulunduğu resmi kasa CPU'yu barındırıyor Sürekli yükte bile 70°C'nin altında.
Sıcaklığınızı şununla izleyin: vcgencmd measure_temp
İtalyan bitki hastalıkları: Edge ML için Öncelikli Hedefler
İtalyan tarım bağlamı, dağıtım seçimlerini yönlendiren belirli özelliklere sahiptir. İtalya, dünyanın en büyük üçüncü şarap üreticisi (Fransa ve İspanya'dan sonra), ilk şarap üreticisidir. AB'den gelen şarap ve zeytin, Avrupa'daki en yüksek mahsul biyoçeşitliliğinden birine sahiptir. Fitopatiler İtalyan tarımına ekonomik açıdan en etkili olanlar şunlardır:
Başlıca İtalyan bitki hastalıkları ve ekonomik etkisi
| Hastalık | Ajan | Etkilenen mahsuller | Potansiyel Kayıp | Tedavi penceresi |
|---|---|---|---|---|
| Asma tüylü küf | Plazmopara viticola | Tüm DOC/DOCG üzümleri | %20-100 üretim | görünümden itibaren 48-72 saat |
| Asma küllemesi | Erysiphe necator | Asma, kabakgiller | %15-60 üretim | 5-7 gün |
| Elma ağacı kabuğu | Venturia inaequalis | Elma yine de | %30-80 üretim | 3-5 gün |
| Domates tüylü küf | Fitofthora infestans | Domates, patates | %50-100 üretim | 24-48 saat |
| Botrytis (Gri Küf) | Botrytis cinerea | Asma, çilek, domates | %10-40 üretim | 7-10 gün |
| Xylella fastidiosa | Xylella fastidiosa | Zeytin ağacı (Puglia) | Bütün bitki | Kurtarılamaz |
Üzüm küfü için en büyük zorluk PlantVillage'ın görsel içermemesidir İtalyan DOC/DOCG asmalarına (Sangiovese, Nebbiolo, Primitivo, Nero d'Avola) özeldir. Bu gerekli kılıyor etki alanı uyarlaması: bir veri kümesi topla İtalyan üzüm bağlarından gerçek görüntülerin eklenmesi ve modelde ince ayar yapılması PlantVillage yerel veri kümesi üzerinde önceden eğitilmiştir.
İtalyan Dijital Tarımına Yönelik Fonlar ve Teşvikler
Il CAP Stratejik Planı 2023-2027 (Ortak Tarım Politikası) tahsisleri Eko-programlar ve müdahaleler yoluyla tarımsal dijitalleşmeye yönelik özel fonlar sektörel. SRA22 ("Risk Yönetimi") ve SRA29 ("Hassas Tarım") müdahalesi Dijital bitki sağlığı izleme sistemlerini benimseyen şirketlere ödüller sağlayın. Ulusal düzeyde, PNRR M2C4 tedbiri aracılığıyla aşağıdakiler için fon tahsis etmiştir: Yeşil ve dijital geçişin bir parçası olarak tarımda inovasyon.
Tarımsal KOBİ'ler için (cirosu < 2 milyon Avro olan şirketler), uç makine öğrenimi sisteminin maliyeti bu makalede açıklanana benzer (~240 EUR donanım + geliştirme) ve amortismana tabi Tüylü küf hastalığının erken teşhisinin tek bir sezonda gerçekleştiğini düşünürsek 5 hektarlık bir bağ, 15.000-30.000 Euro tutarında üretim tasarrufu sağlayabilir.
Performans ve İzleme Metrikleri
Üretimde izleme olmadan çalışan bir makine öğrenimi sistemi, bozulmaya mahkum bir sistemdir sessizce. İzlenecek metrikler iki kategoriye ayrılır: metrikler model (doğruluk) ve sistem ölçümleri (gecikme, çalışma süresi).
# 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
}En İyi Uygulamalar ve Anti-Kalıplar
En İyi Uygulamalar
Tarımsal Edge ML için 8 Altın Kural
- Dağıtımdan önce daima niceleme yapın: Float32 ve int8 arasındaki fark ARM'deki gecikme açısından %50-70, doğruluk kaybı ise %1'in altında. Pi 5'te üretimde float32 kullanmanın hiçbir nedeni yok.
- Gerçek saha verileriyle kalibre edin: PlantVillage veri kümesi e laboratuvarda elde edildi. Hedef alandan en az 500 gerçek görüntü toplayın Kalibrasyon için (aynı üzüm çeşidi, aynı ışık, aynı çekim açısı) int8 ve ince ayar için. Alan içi doğruluktaki fark şunlar olabilir: bu adım olmadan %10-20 oranında.
- Geri bildirim döngüsünü yönetin: Nedenini açıklayan bir mekanizma uygulayın Ziraat uzmanları tespitleri onaylayabilir veya düzeltebilir. Her düzeltme altındır: bir sonraki döngü için eğitim verisi haline gelir.
- Giriş dağıtımını izleyin: Model gördüyse Eğitim sırasında sadece Sangiovese, üretimde ise Primitivo ile buluşuyor, doğruluk çöker. Veri kaymasını tespit etmek için giriş yerleştirmelerinin dağılımını çizin.
- OTA güncellemesini planlayın: Şablonlar güncellenebilir olmalıdır cihaza fiziksel erişim olmadan. İndirmeler için bir MQTT mekanizması uygulayın ve hata durumunda .tflite dosyasının otomatik geri alma ile değiştirilmesi.
- Yedeklilik ve çevrimdışı mod: Sistem düzgün çalışmalı bağlantı olmadan bile. Algılamalar için yerel bir arabellek (SQLite veya JSON dosyası) kullanın ve bağlantı geri geldiğinde senkronize edin.
- Kültür için güven eşiğini kalibre edin: %70 eşik tüylü küf için (oldukça yıkıcı, ekonomik tedavi) ve birinden farklı külleme eşiği (daha az acil, pahalı tedavi). Eşiği şuna uyarlayın: Yanlış pozitiflerin ve yanlış negatiflerin asimetrik maliyeti.
- Edinme bağlamını belgeleyin: Her resim olmalı meta veriler: günün saati, hava koşulları, fenolojik aşama, GPS konumu. Bu meta veriler hata ayıklama ve gelecekteki eğitim için kritik öneme sahiptir.
Kaçınılması Gereken Anti-Desenler
Tarımsal Edge ML Projelerinde En Yaygın 5 Hata
- PlantVillage'ın = üretime hazır olduğunu varsayalım: Veri kümesi e ideal koşullar altında elde edilmiştir (tekdüze ışık, beyaz arka plan, izole edilmiş yapraklar). Sahada, görüntülerde gölgeler, yeşil arka plan, su damlaları ve üst üste bindirilmiş böcekler var. Bir model sahada ince ayar yapılmadan yalnızca PlantVillage üzerinde eğitilmiş olması genellikle doğruluğa sahiptir gerçek alanda %60-70, laboratuvar test setinde ise %97. Bu boşluk ve olarak bilinir etki alanı kayması ve bu, temel pratik sorundur.
-
Sınıf dengesinin göz ardı edilmesi: PlantVillage'da çok sayıda sınıf var
dengesiz (Domates Geç Yanıklığı: 1909 görüntü; Sedir Elma Pas: 275 görüntü).
Sınıf ağırlıklandırması veya aşırı örnekleme olmadığında model, bol sınıflara karşı önyargılı olacaktır.
Kullanmak
class_weightKeras fit() veyaWeightedRandomSampler. - Pi'yi web sunucusu olarak kullanma: Senkron HTTP isteklerini işleme Pi ile ilgili çıkarım kötü bir mimari tercihtir. Pi yapımcı olmalı Bağımsız MQTT, istekleri bekleyen bir sunucu değil. Bağlantı yönetimi ve HTTP ayrıştırma ek yükü ve gereksiz karmaşıklığı artırır.
- Aşırı ışık koşullarına maruz kalmayın: Ağustos ayında saat 12:00'de Puglia, doğrudan ışık görüntüleri doyurur ve yapraklar tamamen beyaz görünür. Şafakta yüksek nem nedeniyle yapraklar ıslaktır ve farklı şekilde yansır. Uygula önce görüntü kalitesinin (ortalama parlaklık, doygunluk) kontrolü çıkarımı başlatın ve aralık dışı görüntüleri atın/yeniden deneyin.
- Doğruluğu doğrulamadan niceleme: int8 tam niceleme temsili bir kalibrasyon veri seti gerektirir. Kalibrasyon veri kümesi kullanın çok küçük (100 görüntüden az) veya temsili olmayan, kayba neden olabilir beklenen %0,4 yerine %3-5 doğruluk. Modeli her zaman doğrula Dağıtımdan önce test setinde nicemlenmiştir.
Havadan Model Güncellemesi
Sahada konuşlandırılmış düzinelerce veya yüzlerce cihaz varken manuel olarak güncelleyin modeller ve pratik değildir. MQTT aracılığıyla bir OTA (Over-the-Air) sistemi dağıtıma izin verir cihazlara fiziksel erişimi olmayan yeni modeller.
# 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()Sonuçlar ve Sonraki Adımlar
Edge ML'yi temel alan eksiksiz bir mahsul hastalığı tespit sistemi oluşturduk: MobileNetV3Small modelinin PlantVillage'de ince ayar ile eğitilmesinden dönüşüme kadar İnt8 nicemlemeli TFLite, entegre Python hattıyla Raspberry Pi 5'te konuşlandırılıyor MQTT yakalama, çıkarım ve gönderme için. Sistem şu gecikme sürelerine ulaşır: 65ms başına çıkarım, toplam güçten 8W'tan az tüketir ve tamamen çevrimdışı çalışır.
Tarım girişimcisi için asıl önemli olan rakamlar farklıdır: Hastalıklardan kaynaklanan kayıplar mahsullerin pahalı olması Dünya çapında 220 milyar dolar her yıl (FAO), Erken teşhis, zamanında yapılan tedavilerle kayıpları %60-80 oranında azaltabilir. Bir sistem anlatılan gibi donanım maliyeti 250 Euro'dan azdır ve tek bir sezonda amortismana tabi tutulabilir herhangi bir değerli mahsul üzerinde. Tarımdaki yapay zeka pazarının büyümesi bekleniyor 2025'te 5,9 milyar ABD dolarından 2035'te 61,3 milyara, yıllık bileşik büyüme oranı ise %26,3'tür.
Doldurulması gereken en büyük boşluk hâlâ etki alanı kayması: Bitki Köyü saha içi dağıtım için tek başına yeterli değildir. Koymak isteyenler için öncelik Bu sistemi üretime alın ve ürünlerinizden gerçek görüntüler toplayın, bunları etiketleyin güvenilir bir tarım uzmanıyla görüşün ve ince ayar yapın. Ayrıca doğru kullanılan 500 gerçek görsel, Laboratuvarda %97 doğruluk oranı ile sahada %65 doğruluk oranına sahip bir sistem arasındaki farkı yaratabilen, her iki bağlamda da %95 kararlı bir sisteme karşı.
Özet: Tam Teknoloji Yığını
| Katmanlar | Teknoloji | 2025 versiyonu |
|---|---|---|
| Eğitim | TensorFlow / Keras | TF 2.17+ |
| Veri kümeleri | PlantVillage (Kaggle/HuggingFace) | 54.309 görsel |
| Mimarlık | MobileNetV3Küçük | Öğrenimi aktarın ImageNet |
| Formatı dağıt | TensorFlow Lite (.tflite) | int8 nicemlenmiş |
| Donanım | Ahududu Pi 5 (8GB) | ARM Cortex-A76 |
| Oda | Kamera Modülü 3 (12MP) | Picamera2 + libcamera |
| Mesajlaşma | MQTT (paho-mqtt) | Eclipse Sivrisinek 2.x |
| Diyet | Güneş + Lipo | Özerklik 3+ gün |
| OTA | MQTT + SHA256 doğrulaması | Sıfır dokunuşlu dağıtım |
FoodTech Serisinde Gelecek Makaleler
- Madde 3 — Uydu API'si ve NDVI: Sentinel-2 API'si nasıl kullanılır? (Copernicus) mahsul sağlığını tarla ölçeğinde verilerle izleyecek ücretsiz uydu, Python'da NDVI hesaplaması ve IoT boru hattıyla entegrasyon.
- Madde 4 — Gıda Kalitesine Yönelik Bilgisayarlı Görme: Derin öğrenme Üretim hatlarındaki kusurların otomatik olarak sınıflandırılması, ağırlık/boyut/renk ve otomatik sıralama sistemleriyle entegrasyon.
- Madde 5 - Blockchain Gıda İzlenebilirliği: Uygulama Hyperledger Fabric ve GS1 standartlarıyla sahadan çatala izlenebilirlik sisteminin geliştirilmesi Gıda izlenebilirliğine ilişkin AB mevzuatına uyum için Dijital Bağlantı.
Kaynaklar ve Referanslar
- PlantVillage Veri Kümesi: Hughes, D.P. ve Salathé, M. (2016). Mobil teknolojilerin geliştirilmesini sağlamak için bitki sağlığına ilişkin açık erişimli görsel deposu hastalık tanısı. ArXiv ön baskısı arXiv:1511.08060. Şu tarihte mevcut: Kaggle PlantKöy.
- FAO Mahsul Kayıpları: FAO Bitki Üretimi ve Koruması — zararlılardan ve hastalıklardan kaynaklanan yıllık kayıplar
- Raspberry Pi 5'te TFLite: Hackster.io — Raspberry Pi 5'te TFLite'ın Kıyaslanması
- Edge AI 2024 Karşılaştırmaları: Georgia Güney Üniversitesi (2024). Uç Yapay Zeka Platformlarının Kıyaslanması: NVIDIA Jetson ve Raspberry Pi 5'in Coral TPU ile Performans Analizi. IEEE Konferans Bildirileri. DOI: 10.1109/10971592.
- MobileNetV3 için optimize edilmiş PlantVillage: Bilimsel Raporlar — Hafif bitki yaprağı hastalığının tespiti için optimize edilmiş MobileNet (2025)
- İtalya Hassas Tarım: Kırsal Hack — Tarım 4.0 İtalya 2025: Pazar Büyümesi ve Dijital Olgunluk