作物の病気検出のための ML Edge: Raspberry Pi 上の TensorFlow Lite
毎年、植物の病気や作物の害虫が被害をもたらします。 生産量の40% 世界の農業、より大きな経済的損失を引き起こす 2,200億ドル FAOのデータによると。イタリアでは、ワイン生産国、果樹園、菜園が次のような風土病の植物病に対して脆弱です。 つるべと病 (マラリア原虫）、うどんこ病（エリシフェ・ネカトル）と リンゴのかさぶた（ベンチュリア・イナエクアリス）、それぞれ 1 年全体を消去できます 時間内に阻止されなければ生産的です。
早期診断は歴史的に農家の視覚的経験に委ねられてきましたが、最良の場合には、 農学者による定期検査まで。どちらのアプローチも同じ制限、つまり遅延に悩まされます。 病気の発症と植物検疫の介入の間には間隔があり、多くの場合、数日から数週間かかります。 その間、真菌や細菌は山火事のように広がります。の時間枠 効果的かつ最小限の治療法：ブドウのべと病の場合、例えば日光に当てれば軽減されます。 48～72時間 最初の病変の出現から.
コンピューター ビジョンに適用された人工知能は、具体的な答えを提供します。 PlantVillage データセット (54,309 枚の画像、38 クラス) でトレーニングされたディープラーニングにより高い精度が達成されます。 より高い 97% 葉の病気の認識において。本当のボトルネック それはもはやモデルの精度ではなく、 フィールド展開。アペニン山脈のブドウ畑で トスカーナ州とエミリア州、またはシチリア島の柑橘類の果樹園では、インターネット接続が存在しないか、信頼性が低いことがよくあります。 推論のために画像をクラウドに送信することは現実的ではありません。解決策とエッジML: モデルは、接続なしでフィールドでデバイス上で直接実行され、レイテンシーは 2番目。
この記事では、Raspberry Pi 5 上に完全な病気検出パイプラインをゼロから構築します。 TensorFlow Lite の使用: MobileNetV3 での転移学習による基本的なモデルのトレーニングから、 int8 量子化、リアルタイム Python 推論によるデプロイ、結果の送信まで MQTT からリモート ダッシュボードへ。各ステップには、動作するコード、ベンチマークの数値が付いています。 イタリアの農業事情に関する検証済みかつ実践的な考慮事項。
この記事で学べること
- オープンフィールド農業アプリケーションにおいてエッジ ML がクラウドより優れたパフォーマンスを発揮する理由
- 効果的なトレーニングのために PlantVillage データセットを準備および拡張する方法
- TensorFlow/Keras での MobileNetV3Small による転移学習、ステップバイステップ
- トレーニング後の量子化による TFLite へのモデル変換 (float32 → int8)
- カメラ モジュール 3 と完全な Python 推論を使用した Raspberry Pi 5 ハードウェア セットアップ
- 統合パイプライン: キャプチャ → 前処理 → 推論 → MQTT → ダッシュボード
- Pi 4、Pi 5、Coral TPU、Jetson Nano の遅延と消費電力のベンチマーク
- イタリアの主な植物病害と優先対象作物
- 導入のベスト プラクティスと一般的なアンチパターン
フードテックシリーズにおける位置づけ
これはシリーズの 2 番目の記事です フードテック、デジタルテクノロジーに特化 農産物のサプライチェーンのために。このシリーズでは、農業 IoT、コンピューター ビジョン、ブロックチェーンを取り上げます。 トレーサビリティ、需要予測など。
フードテックシリーズ - すべての記事
| # | タイトル | レベル | Stato |
|---|---|---|---|
| 1 | 農業 IoT パイプライン: InfluxDB 上のセンサー、MQTT、時系列 | 中級 | 利用可能 |
| 2 | 作物の病気を検出するための ML Edge: Raspberry Pi 上の TFLite — あなたはここにいる | 高度な | 現在 |
| 3 | サテライト API と NDVI: Sentinel-2 による作物モニタリング | 中級 | 利用可能 |
| 4 | コンピュータビジョンによる食品品質: 欠陥、重量、等級分け | 高度な | 利用可能 |
| 5 | ブロックチェーン食品トレーサビリティ: Hyperledger Fabric と GS1 | 高度な | 利用可能 |
| 6 | FSMA 204 準拠: Python と API を使用したデジタル トレーサビリティ | 中級 | 利用可能 |
| 7 | 垂直農法の自動化: PLC と ML による気候制御 | 高度な | 利用可能 |
| 8 | 食料需要予測: Prophet、LSTM、特徴エンジニアリング | 中級 | 利用可能 |
| 9 | ファーム ダッシュボード リアルタイム: Angular、WebSocket、InfluxDB | 中級 | 利用可能 |
| 10 | サプライチェーンの回復力: OR ツールと AI による最適化 | 高度な | 利用可能 |
エッジ ML の基礎: 農業向けのクラウドとエッジ
コード行を記述する前に、エッジ アプローチが唯一の方法である理由を理解することが重要です。 オープンフィールドでの病気の検出に実用的です。これはスタイル上の選択ではありません。 そして農業の物理的状況によって課せられた建築上の必然性。
農業におけるクラウドの限界
クラウドファーストの疾病検出アプリケーションは次のように動作します。 カメラのクリック音 写真をネットワーク経由でクラウド サーバーに送信し、モデルがサーバー上で実行され、結果が返されます。 デバイス。光ファイバーを備えたオフィスでは、このサイクルは 200 ミリ秒未満続きます。ブドウ畑で トスカーナやカラブリアの柑橘類の果樹園では、変数が根本的に変化します。
クラウド vs エッジ: 農業コンテキスト別の比較
| 特性 | クラウドアプローチ | エッジアプローチ |
|---|---|---|
| 接続が必要です | 連続、1+Mbps帯域 | オプション、データ同期のみ |
| 推論レイテンシ | 200ms ～ 5s (ネットワークによって異なります) | 50～500ミリ秒（ローカル） |
| 運営コスト | 高 (API 呼び出し、ストレージ、帯域幅) | 低 (1 回限りのハードウェア) |
| プライバシーの提供 | クラウド上の企業イメージ | データはオンプレミスに残ります |
| オフライン操作 | 不可能 | 完了 |
| モデルのスケーラビリティ | 無制限 | RAM/CPUによる制限 |
| モデルの更新 | すぐに | 物理的またはOTA展開が必要です |
| エネルギー消費量 | 低 (デバイス) + 高 (データセンター) | デバイス上のすべて (3 ～ 15 W) |
TensorFlow と TensorFlow Lite: 主な違い
TensorFlow (TF) は、GPU とサーバー向けに最適化された完全なトレーニングおよび推論フレームワークです。 TensorFlow Lite (TFLite) は、組み込みデバイスおよびモバイル デバイス向けに圧縮され最適化されたバージョンです。 実際的な違いは次のとおりです。
- 実行時サイズ: TFLite ランタイムと C++ で約 1MB (完全な TF の場合は 400MB+)。 Raspberry Pi では、この違いは RAM 占有量が 2 ～ 4 GB ではなく 50 ～ 100 MB に変わります。
- サポートされている演算子: TFLite は、TF 操作のサブセットをサポートします。モデル カスタム操作または非標準レイヤーでは、CPU へのデリゲートまたはフォールバックが必要です。
- .tflite 形式: FlatBuffers スキーム、ダイレクト メモリ アクセス用に最適化 解析せずに。モデルは逆シリアル化せずにメモリ マップされます。
- ハードウェアアクセラレーション: TFLite は XNNPACK デリゲート (ARM 上の SIMD) をサポートします。 Android 上の GPU デリゲート、Coral Edge TPU デリゲート、および NNAPI。
量子化: float32 から int8
エッジの最適化と 量子化: の 重みの数値精度が 32 ビット浮動小数点から 8 ビット整数に低下します。効果 実用的で、次の 3 つのメリットがあります。
MobileNetV3Small における int8 量子化の影響
| メトリック | float32 | int8 (PTQ) | デルタ |
|---|---|---|---|
| モデルサイズ | 9.8MB | 2.6MB | -73% |
| RAM 推論 (Pi 5) | ～180MB | ~52MB | -71% |
| 推論レイテンシー (Pi 5) | ～180ミリ秒 | ～65ミリ秒 | -64% |
| PlantVillage の精度 | 97.2% | 96.8% | -0.4% |
| エネルギー消費量 | ～5.2W | ～3.8W | -27% |
Raspberry Pi 5 (2.4GHz、8GB RAM) でのベンチマーク。 38 個の PlantVillage クラスの微調整モデルで測定された値。
0.4% の精度の低下は、現場では実際には無関係です: 自然変動 照明条件、角度、病気の段階により不確実性が生じます はるかに大きい。ただし、遅延とサイズの増加は、組み込みの展開にとって決定的です。
PlantVillage データセット: 準備と拡張
葉の病気を検出するための参照データセットと プラントヴィレッジ、 Hughes と Salathé (ペンシルバニア州立大学) によって作成され、2016 年に出版されました。 54,309枚の画像 管理された実験室条件下で撮影され、整理された で 38クラス 14 種類の植物と関連する病気をカバーしています。
データセットの構造
PlantVillage クラス: 種と主な病気
| Specie | 病気 | 原因物質 | 画像 |
|---|---|---|---|
| つる（ブドウ） | ブラックロット | ギグナルディア・ビドウェリー | 1,180 |
| つる（ブドウ） | 餌・黒麻疹 | ファエオアクレモニウム属 | 1,383 |
| つる（ブドウ） | 葉枯れ病 | 偽セルコスポラ・ヴィティス | 1,076 |
| つる（ブドウ） | 健康 | — | 423 |
| トマト（トマト） | 初期疫病 | アルタナリア ソラニ | 1,000 |
| トマト（トマト） | 疫病 | 疫病菌 | 1,909 |
| トマト（トマト） | 腐葉土 | 黄褐色のパサローラ | 952 |
| トマト（トマト） | セプトリア リーフ スポット | セプトリア・リコペルシチ | 1,771 |
| リンゴの木（リンゴ） | リンゴのかさぶた（かさぶた） | ベンチュリア・イナエクアリス | 630 |
| リンゴの木（リンゴ） | ブラックロット | ボトリオスフェリア・オブトゥサ | 621 |
| リンゴの木（リンゴ） | シダーアップルラスト | 裸子胞子嚢ジュニペリ | 275 |
| トウモロコシ (トウモロコシ) | 一般的な錆 | プッチニア・ソルギ | 1,192 |
| ジャガイモ（ジャガイモ） | 初期疫病 | アルタナリア ソラニ | 1,000 |
| ジャガイモ（ジャガイモ） | 疫病 | 疫病菌 | 1,000 |
| ... | + 他の 24 クラス | — | — |
データセットのダウンロードと準備
このデータセットは、Kaggle および Hugging Face で無料で入手できます。以下はダウンロードスクリプトです。 フォルダーへの構造化とトレイン/検証/テストの分割:
# 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}")葉画像のデータ拡張技術
PlantVillage 画像は、実験室の均一な背景で取得されます。一般化すると 実際のフィールド条件（光の変化、自然な背景、さまざまな角度）に合わせて、 増強技術は基本です。幾何学的変換と測光変換に加えて 上記のコードには標準がすでに含まれていますが、農業分野に特化したテクニックもあります。
- ランダム消去/切り抜き： 画像のランダムな四角形を暗くします (シミュレートします) 重なり合う葉、影、作業機械によるオクルージョン）。オーバーフィッティングを軽減します 局所的な特徴を抽出し、実際の画像でモデルの堅牢性を 2 ～ 3% 向上させます。
- ミックスアップとカットミックス: 2 つの画像をランダムな重みで結合 (MixUp) するか、置き換えます 別の画像からのパッチを含む領域 (CutMix)。特に授業に効果的 視覚的には似ています（例：トマトの初期疫病と晩疫病）。
- 激しいカラージッター: 葉の病気は色の症状を示します ステージ、日当たり、気候条件によって異なります。 色相/彩度の変化範囲を増やすと、一般化が容易になります。
- 背景の置換: 白い背景を高度な技術で置き換える 実際のフィールド画像を使用した研究室の様子。予備的なセマンティック セグメンテーションが必要 ただし、フィールド内画像のパフォーマンスは大幅に向上します。
モデルのトレーニング: MobileNetV3 を使用した転移学習
MobileNetV3Small は、この使用例に最適なアーキテクチャです。 モバイルおよび組み込みデバイスでの推論は、精度間の優れたバランスを提供します。 そして計算の複雑さ。 ResNet50 または EfficientNetB4 と比較して、次のことが必要です。 操作が 100 分の 1 に減少 競合する精度を維持しながら推論によって。
アーキテクチャとデザインの選択
事前トレーニングされた ImageNet と勝利戦略からの転移学習: の最初の層 ネットワークは、テクスチャ、エッジ、および一般的な色のパターンを認識することをすでに学習しています。 葉の病気の場合、これらの低レベルの機能は直接再利用可能です。 最後の層 (および必要に応じて最後のブロック) を微調整してモデルを適応させます。 特定のドメインに。
# 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()トレーニングから期待される成果
完全な PlantVillage データセット (38 クラス) 上で、MobileNetV3Small モデルが微調整されました 通常、次のことを達成します。
- フェーズ 1 (15 エポック): Val 精度 91 ～ 93%、T4 GPU でのトレーニング ~45 分
- フェーズ 2 (20 エポック): Val 精度 96 ～ 97%、T4 GPU でのトレーニングは約 2 時間
- テストセット: 精度 96.5 ～ 97.5%、平均 F1 スコア 0.964
- 難しいクラス: トマトの初期疫病と晩疫病 (F1 ~0.91)
- 簡単なクラス: 健康な植物と病気の植物 (F1 ~0.99)
トレーニング後の量子化による TFLite への変換
トレーニングされた Keras モデルは、実行のために .tflite 形式に変換する必要があります ラズベリーパイで。変換プロセスにはオプションで量子化が含まれます。 これにより、サイズが大幅に削減され、推論速度が向上します。
3 つの量子化モード
TFLite は、精度とパフォーマンスのトレードオフが異なる 3 つのレベルの量子化をサポートしています。
- ダイナミックレンジ量子化: 重みのみが量子化されます (float32 → int8)。 アクティベーションは実行時に float32 のままです。キャリブレーション データセットは必要ありません。削減 サイズは 75% 削減され、レイテンシーは 20 ～ 30% 改善されました。おすすめのスタート地点。
- 完全な整数量子化 (int8): int8 で量子化された重みとアクティベーション。 代表的なキャリブレーション データセット (100 ～ 500 枚の画像) が必要です。サイズダウン 75%、遅延が 50 ～ 70% 改善されました。ハードウェア アクセラレーション (Coral TPU) に必要です。
- Float16 量子化: float16 の重み。主に GPU に役立ちます。 ARM CPU では、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.")Raspberry Pi 5 への展開: ハードウェアとソフトウェアのセットアップ
推奨ハードウェア
Raspberry Pi 5 (2023 年 10 月発売、2024 年に 16GB バージョンに更新) および この使用例に推奨されるエッジ プラットフォーム。 Pi 4と比較して、プロセッサ 2.4GHz の Cortex-A76 は、TFLite に近い推論パフォーマンスを提供します 5倍高い、 int8 演算用に最適化された ARM v8.2-A ドット積命令のおかげで。
システム全体のハードウェア リスト
| 成分 | 推奨モデル | 参考コスト | 注意事項 |
|---|---|---|---|
| メインSBC | ラズベリーパイ5（8GB） | ～80ユーロ | RAMヘッドルーム用の8GBバージョン |
| 電源 | 公式 RPi 5 PSU 27W USB-C | ～12ユーロ | 電力の安定供給に必要 |
| ストレージ | MicroSD A2 64GB (Samsung Pro Endurance) | ～15ユーロ | A2 より優れたランダム I/O |
| カメラ | Raspberry Pi カメラ モジュール 3 (12MP) | ～25ユーロ | オートフォーカス、CSI-2 |
| 光学 | 120°広角レンズ | ～8ユーロ | ドローンまたはポールからの取得用 |
| 住宅 | RPi 5 アクティブクーラー + IP65 ケース | ～20ユーロ | 屋外使用向けのIP65保護 |
| ネット | 4G LTE HAT (ウェーブシェア SIM7600G) | ～45ユーロ | WiFiのない地域の場合 |
| フィールド電源 | 20W ソーラーパネル + 10000mAh LiPo バッテリー | ～35ユーロ | 太陽が当たらない場合は約 3 日間自律走行可能 |
| トータルシステム | ～240ユーロ | 現場でのスタンドアロン設置用 |
Raspberry Pi OS でのソフトウェアのセットアップ
# Eseguire sul Raspberry Pi dopo aver installato Raspberry Pi OS 64-bit (Bookworm)
# 1. Aggiornamento sistema
sudo apt update && sudo apt full-upgrade -y
# 2. Dipendenze sistema
sudo apt install -y \
python3-pip python3-venv python3-dev \
libatlas-base-dev libjpeg-dev libopenjp2-7 \
libcamera-dev python3-picamera2 \
mosquitto mosquitto-clients \
git vim htop
# 3. Crea virtual environment
python3 -m venv /home/pi/cropai-env
source /home/pi/cropai-env/bin/activate
# 4. Installa TFLite Runtime (versione ottimizzata per ARM64)
# TFLite Runtime e molto più leggero del TF completo (~1MB vs ~400MB)
pip install tflite-runtime
# Alternativa: installa TensorFlow completo (necessario solo per retraining locale)
# pip install tensorflow # ~400MB, non consigliato per Pi se non necessario
# 5. Dipendenze progetto
pip install \
numpy pillow opencv-python-headless \
paho-mqtt \
requests \
RPi.GPIO # per LED status indicator
# 6. Verifica installazione TFLite
python3 -c "import tflite_runtime.interpreter as tflite; print('TFLite OK')"
# 7. Verifica camera
libcamera-hello --timeout 2000
# 8. Test fps camera
libcamera-vid -t 5000 --framerate 30 -o /dev/null --nopreview
# Output atteso: ~30 FPS senza elaborazione完全なフィールド推論パイプライン
モデルが変換され、ハードウェアが構成されたら、パイプラインを組み立てます。 完了: 画像取得 → 前処理 → TFLite 推論 → 後処理 → MQTT → ダッシュボード経由で結果を送信します。これは実稼働アプリケーションの核心です。
# 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()ハードウェア エッジ ベンチマーク: プラットフォームの比較
ハードウェア プラットフォームの選択は、予算、必要なパフォーマンス、消費量によって決まります。 エネルギーと統合の複雑さ。以下は 2024 年から 2025 年の最新の比較です。 公表されているベンチマークと疾患分類モデルに関する独自の測定に基づいています。
農業機械学習向けエッジプラットフォームの比較
| プラットフォーム | CPU | ラム | MobileNetV3S int8 レイテンシ | 消費 | 料金 | 注意事項 |
|---|---|---|---|---|---|---|
| ラズベリーパイ 4B (4GB) | Cortex-A72 1.8GHz | 4GB LPDDR4 | ～320ミリ秒 | ～6W | ～55ユーロ | 経済ベースライン |
| ラズベリーパイ5（8GB） | Cortex-A76 2.4GHz | 8GB LPDDR4X | ～65ミリ秒 | ～8W | ～80ユーロ | 推奨 |
| RPi 4 + コーラル USB TPU | Cortex-A72 + エッジ TPU | 4ギガバイト | ～15ミリ秒 | ～8W | ～95ユーロ | int8 完全モデルが必要です |
| Google Coral 開発委員会 | Cortex-A53 + エッジ TPU | 1GB | ～12ミリ秒 | ～4W | ～120ユーロ | TFLite + Coral コンパイルのみ |
| NVIDIA Jetson Nano (4GB) | Cortex-A57 + 128 CUDA コア | 4GB LPDDR4 | ～8ミリ秒 | ～10W | ~149 米ドル | 単純な分類にはやりすぎ |
| Arduino ポルテンタ H7 | Cortex-M7 480MHz | 8MB SDRAM | ～2000ミリ秒 | ～0.5W | ～100ユーロ | 小型モデルのみ (TFLite Micro) |
| Sipeed M1s ドック | BL808 RV64 480MHz | 768KB SRAM | ~800 ミリ秒 (マイクロ) | ～0.3W | ~7 米ドル | 超低消費電力、小型モデル |
遅延は、単一の 224x224 画像に対して MobileNetV3Small int8 モデル (2.6MB) で測定されました。 Hackster.io 2024 ベンチマークの Pi 5 値、ジョージア サザン大学 2024 ベンチマークの Coral。
Raspberry Pi 5 のサーマルスロットリング
アクティブな冷却を行わない Raspberry Pi 5 では、5 ～ 10 分後にサーマル スロットルが発生します 継続的推論のパフォーマンスが低下します。 20-30%。 現場導入用（夏期の屋外温度は最大40°C）、冷却 アクティブな電子 義務的な。アクティブクーラーを搭載した公式ケースにCPUを収納 継続的な負荷がかかった場合でも 70°C 未満。
次の方法で体温を監視します。 vcgencmd measure_temp
イタリアの植物病害: Edge ML の優先ターゲット
イタリアの農業環境には、展開の選択を導く特有の特徴があります。 イタリアは（フランス、スペインに次ぐ）世界第3位のワイン生産国であり、世界第1位のワイン生産国です。 EU からのワインとオリーブの生産量が多く、作物の生物多様性はヨーロッパで最も高いものの 1 つです。植物病 イタリアの農業にとって最も経済的な影響を与えるものは次のとおりです。
イタリアの主な植物病害と経済的影響
| 病気 | エージェント | 影響を受ける作物 | 潜在的な損失 | 治療期間 |
|---|---|---|---|---|
| ブドウのべと病 | マラリア原虫 | すべてのDOC/DOCGブドウ | 20～100%の生産 | 出現から48～72時間 |
| ブドウのうどんこ病 | エリシフェ・ネカトル | ブドウ、ウリ科 | 生産量の15～60％ | 5～7日 |
| リンゴの木のかさぶた | ベンチュリア・イナエクアリス | アップルですが、 | 生産量の30～80％ | 3～5日 |
| トマトべと病 | 疫病菌 | トマト、ジャガイモ | 50～100%の生産 | 24～48時間 |
| 灰色カビ病（灰色かび病） | ボトリチス・シネレア | つる、イチゴ、トマト | 生産量の10～40％ | 7～10日 |
| キシレラ・ファスティディオーサ | キシレラ・ファスティディオーサ | オリーブの木（プーリア州） | 全草 | 回復不能 |
ブドウべと病の最大の課題は、PlantVillage に画像が含まれていないことです。 イタリアの DOC/DOCG ブドウ (サンジョヴェーゼ、ネッビオーロ、プリミティーヴォ、ネロ ダーヴォラ) に特化しています。 これにより必要になります ドメイン適応: データセットを収集する イタリアのブドウ畑からの実際の画像の補足とモデルの微調整 PlantVillage はローカル データセットで事前トレーニングされています。
イタリアのデジタル農業に対する資金と奨励金
Il CAP 戦略計画 2023 ～ 2027 年 （共通農業政策）配分 エコスキームと介入を通じた農業デジタル化のための特定の資金 セクター別。 SRA22 (「リスク管理」) および SRA29 (「精密農業」) 介入 デジタル植物検疫監視システムを導入する企業に報奨金を提供します。 国家レベルでは、PNRR は M2C4 措置を通じて次の目的に資金を割り当てています。 グリーンおよびデジタル移行の一環としての農業のイノベーション。
農業中小企業 (売上高 200 万ユーロ未満の企業) の場合、エッジ ML システムのコスト この記事で説明されているものと同様 (ハードウェア + 開発費は約 240 ユーロ)、減価償却が可能です。 べと病の早期発見を考慮すると、ワンシーズンで 5 ヘクタールのブドウ園では、15,000 ～ 30,000 ユーロ相当の生産量を節約できます。
パフォーマンスとモニタリングのメトリクス
監視なしで運用中の ML システムは、劣化する運命にあるシステムです 静かに。追跡されるメトリクスは、次の 2 つのカテゴリに分類されます。 モデル (精度) とシステム メトリクス (遅延、稼働時間)。
# 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
}ベストプラクティスとアンチパターン
ベストプラクティス
アグリカルチュラル エッジ ML の 8 つのゴールデン ルール
- デプロイする前に必ず量子化してください。 float32 と int8 の違い ARM での遅延に関しては 50 ～ 70%、精度の低下は 1% 未満です。 Pi 5 の実稼働環境で float32 を使用する理由はありません。
- 実際のフィールドデータを使用して校正します。 PlantVillage データセット e 研究室で入手したもの。対象フィールドから少なくとも 500 枚の実際の画像を収集する (同じブドウ品種、同じ照明、同じ撮影角度) キャリブレーション用 int8 および微調整用。フィールド内精度の違いは次のとおりです。 この手順を行わないと 10 ～ 20% 減少します。
- フィードバック ループを処理します。 なぜそうなるのか仕組みを実装する 農学者は検出を確認または修正できます。すべての修正は金です: 次のサイクルのトレーニング データになります。
- 入力分布を監視します。 モデルが見たら トレーニング中はサンジョヴェーゼのみですが、実稼働環境ではプリミティーヴォに適合し、精度が向上します。 崩壊する。入力エンベディングの分布をプロットしてデータのドリフトを検出します。
- OTA アップデートのスケジュールを設定します。 テンプレートは更新可能である必要があります デバイスに物理的にアクセスする必要はありません。ダウンロード用の MQTT メカニズムを実装する エラーが発生した場合は、.tflite ファイルを自動的にロールバックしてスワップします。
- 冗長性とオフライン モード: システムは正しく動作する必要があります 接続がなくても。検出にはローカル バッファー (SQLite または JSON ファイル) を使用します 接続が戻ったら同期します。
- 文化の信頼しきい値を調整します。 しきい値は 70% べと病用 (破壊力が高く、経済的な治療法) とは異なります うどんこ病の閾値（緊急性は低く、高価な治療法）。しきい値を次のように調整します。 偽陽性と偽陰性の非対称コスト。
- 取得コンテキストを文書化します。 すべての画像は次のとおりである必要があります メタデータ: 時刻、気象条件、生物季節段階、GPS 位置。 このメタデータは、デバッグと将来のトレーニングにとって重要です。
避けるべきアンチパターン
農業エッジ ML プロジェクトで最もよくある 5 つの間違い
- PlantVillage = 生産準備完了であると仮定します。 データセット e 理想的な条件（均一な光、白い背景、孤立した葉）の下で取得されました。フィールドでは、 画像には影、緑の背景、水滴、昆虫が重ねられています。モデル 現場での微調整を行わずに PlantVillage のみでトレーニングされた場合、通常は精度が高くなります 実際のフィールドでは 60 ～ 70%、臨床検査セットでは 97% でした。このギャップと として知られている ドメインシフト そしてそれが主な現実的な問題です。
-
クラスバランスを無視すると、 PlantVillageにはたくさんのクラスがあります
バランスが取れていない (トマト疫病: 1909 枚の画像; シダーアップルさび病: 275 枚の画像)。
クラスの重み付けやオーバーサンプリングがないと、モデルは豊富なクラスに偏ることになります。
使用
class_weightKeras fit() またはWeightedRandomSampler. - Pi を Web サーバーとして使用する: 同期HTTPリクエストを処理します。 Pi に関する推論はアーキテクチャ上の選択としては間違っています。 Pi はプロデューサーでなければなりません リクエストを待機するサーバーではなく、スタンドアロンの MQTT。接続管理 および HTTP 解析により、オーバーヘッドと不必要な複雑さが追加されます。
- 極端な光条件を扱わないでください。 8月の12時に プーリア州では、直射光が画像を飽和させ、葉が完全に白く見えます。 湿度の高い明け方は葉が濡れて映り方が違います。実装する 事前に画質（平均輝度、彩度）をチェック 推論を開始し、範囲外の画像を破棄/再試行します。
- 精度を検証せずに量子化する: int8 の完全量子化 代表的なキャリブレーション データセットが必要です。キャリブレーション データセットを使用する 小さすぎる（画像が 100 枚未満）、または代表的ではない場合、損失が発生する可能性があります 予想される 0.4% ではなく 3 ～ 5% の精度です。常にモデルを検証する 導入前にテストセットで量子化されます。
無線によるモデル更新
数十または数百のデバイスが現場に導入されている場合は、 モデルと非実用的です。 MQTTによるOTA（Over-the-Air）システムにより配信が可能 デバイスに物理的にアクセスできない新しいモデル。
# 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()結論と次のステップ
私たちは、Edge ML に基づいた完全な作物病気検出システムを構築しました。 PlantVillage での MobileNetV3Small モデルの微調整によるトレーニングから変換まで int8 量子化を備えた TFLite、統合された Python パイプラインを備えた Raspberry Pi 5 にデプロイ MQTT のキャプチャ、推論、送信用。システムのレイテンシは次のとおりです。 65ミリ秒あたり 推論、総電力消費量は 8W 未満で、完全にオフラインで動作します。
農業起業家にとって本当に重要な数字は異なります: 病気による損失 作物は高価です 全世界で2,200億ドル 毎年（FAO）、 早期発見により、タイムリーな治療により損失を 60 ～ 80% 削減できます。システム 説明されているものと同様、ハードウェアのコストは 250 ユーロ未満で、1 シーズンで減価償却が可能です。 あらゆる貴重な作物に。農業における AI 市場は、 2025 年には 59 億ドル、2035 年には 613 億ドルとなり、CAGR は 26.3% となります。
埋めるべき最大のギャップは依然として ドメインシフト: プラントヴィレッジ それだけでは現場での展開には十分ではありません。載せたい方優先 このシステムを本番環境に導入し、作物から実際の画像を収集し、ラベルを付けます 信頼できる農学者と一緒に微調整を行ってください。 500枚の実際の画像も正しく使用されており、 研究室では 97% の精度を持つシステムと、現場では 65% の精度を持つシステムとの間に違いを生むことができます。 対して、どちらの状況でも 95% 安定したシステム。
概要: 完全なテクノロジースタック
| レイヤー | テクノロジー | 2025年版 |
|---|---|---|
| トレーニング | TensorFlow / ケラス | TF2.17+ |
| データセット | PlantVillage (Kaggle/HuggingFace) | 54,309枚の画像 |
| 建築 | MobileNetV3Small | 転移学習 ImageNet |
| デプロイ形式 | TensorFlow Lite (.tflite) | int8 量子化 |
| ハードウェア | ラズベリーパイ5（8GB） | ARM コーテックス-A76 |
| 部屋 | カメラモジュール 3 (12MP) | Picamera2 + libcamera |
| メッセージング | MQTT (パホ-mqtt) | Eclipse モスキート 2.x |
| ダイエット | ソーラー + LiPo | 自主性 3 日以上 |
| 太田 | MQTT + SHA256 検証 | ゼロタッチ導入 |
FoodTech シリーズの今後の記事
- 第 3 条 — サテライト API および NDVI: Sentinel-2 API の使用方法 (Copernicus) データを使用して圃場規模で作物の健康状態を監視する 無料の衛星、Python での NDVI 計算、IoT パイプラインとの統合。
- 第 4 条 — 食品品質のためのコンピューター ビジョン: ディープラーニング 生産ラインの欠陥の自動分類、グレーディング用 重量/寸法/色、自動仕分けシステムとの統合。
- 第 5 条 — ブロックチェーン食品トレーサビリティ: 実装 Hyperledger Fabric と GS1 標準を備えた現場からフォークまでのトレーサビリティ システムの 食品トレーサビリティに関する EU 法を遵守するためのデジタル リンク。
リソースと参考資料
- PlantVillage データセット: ヒューズ、D.P.、サラテ、M. (2016)。 モバイル アプリケーションの開発を可能にする、植物の健康に関する画像のオープン アクセス リポジトリ 病気の診断。 ArXiv プレプリント arXiv:1511.08060。 で利用可能 カグル プラントヴィレッジ.
- FAOの作物損失: FAO 植物の生産と保護 — 害虫や病気による年間損失
- Raspberry Pi 5 上の TFLite: Hackster.io — Raspberry Pi 5 での TFLite のベンチマーク
- Edge AI 2024 ベンチマーク: ジョージア・サザン大学（2024年）。 エッジ AI プラットフォームのベンチマーク: Coral TPU を搭載した NVIDIA Jetson および Raspberry Pi 5 のパフォーマンス分析。 IEEE 会議議事録。 DOI: 10.1109/10971592。
- MobileNetV3 に最適化された PlantVillage: Scientific Reports — 軽量の植物葉の病気検出のために最適化された MobileNet (2025)
- イタリアの精密農業: Rural Hack — 農業 4.0 イタリア 2025: 市場の成長とデジタル成熟度