농작물 질병 감지를 위한 ML Edge: Raspberry Pi의 TensorFlow Lite
매년 식물 질병과 농작물 해충으로 인해 생산량의 40% 세계 농업, 경제적 손실이 더 크다. 2,200억 달러 FAO 데이터에 따르면. 이탈리아에서는 와인 재배 국가, 과수원 및 채소밭이 다음과 같은 풍토병 식물 질병에 취약합니다. 포도나무 노균병(플라스모파라 비티콜라), 흰가루병 (에리시페 네카토르) 그리고 사과 딱지 (벤투리아 이나이퀄리스), 각각 1년 전체를 지울 수 있음 제 시간에 가로채지 않으면 생산적입니다.
조기 진단은 역사적으로 농부의 시각적 경험에 맡겨져 왔으며, 가장 좋은 경우에는 농업경제학자의 정기 검사를 받습니다. 두 가지 접근 방식 모두 동일한 제한 사항인 대기 시간이 있습니다. 질병의 발병과 식물위생 개입 사이에는 간격이 있으며 종종 며칠 또는 몇 주가 소요됩니다. 그 동안 곰팡이나 박테리아가 산불처럼 퍼집니다. 에 대한 시간 창 효과적이고 최소한의 치료: 예를 들어 포도나무 노균병의 경우 햇빛에 노출됩니다. 48~72시간 첫 번째 병변의 출현부터.
컴퓨터 비전에 적용된 인공지능은 구체적인 답을 제시합니다. PlantVillage 데이터 세트(54,309개 이미지, 38개 클래스)에서 훈련된 딥 러닝으로 정확도 달성 보다 높은 97% 잎 질병의 인식. 실제 병목 현상 더 이상 모델의 정확성이 아니라 현장 배포. 아펜니노 산맥의 포도원에서 토스카나-에밀리안이나 시칠리아의 감귤나무 숲에서는 인터넷 연결이 없거나 불안정한 경우가 많습니다. 추론을 위해 이미지를 클라우드로 보내는 것은 비현실적입니다. 솔루션과엣지 ML: 모델은 연결 없이 현장에서 장치에서 직접 실행되며 대기 시간은 다음보다 짧습니다. 두 번째.
이 기사에서는 Raspberry Pi 5에서 처음부터 완전한 질병 감지 파이프라인을 구축합니다. TensorFlow Lite 사용: MobileNetV3에서 전이 학습을 통한 기본 모델 훈련부터 int8 양자화, 실시간 Python 추론을 통한 배포, 결과 전송 원격 대시보드에 대한 MQTT입니다. 각 단계에는 작업 코드, 벤치마크 번호가 함께 제공됩니다. 이탈리아 농업 상황에 대한 검증되고 실용적인 고려 사항.
이 기사에서 배울 내용
- Edge ML이 노지 농업 애플리케이션에서 클라우드보다 성능이 뛰어난 이유
- 효과적인 훈련을 위해 PlantVillage 데이터세트를 준비하고 강화하는 방법
- TensorFlow/Keras에서 MobileNetV3Small을 사용한 전이 학습(단계별)
- 학습 후 양자화(float32 → int8)를 사용하여 TFLite로 모델 변환
- 카메라 모듈 3 및 전체 Python 추론을 사용한 Raspberry Pi 5 하드웨어 설정
- 통합 파이프라인: 캡처 → 전처리 → 추론 → MQTT → 대시보드
- Pi 4, Pi 5, Coral TPU, Jetson Nano의 지연 시간 및 전력 소비 벤치마크
- 이탈리아의 주요 식물병 및 우선대상작물
- 배포 모범 사례 및 일반적인 안티 패턴
FoodTech 시리즈에서의 위치
이 시리즈의 두 번째 기사입니다. 푸드테크, 디지털 기술에 전념 농식품 공급망을 위한 것입니다. 이 시리즈는 농업용 IoT, 컴퓨터 비전, 블록체인을 다루고 있습니다. 추적성, 수요 예측 등.
FoodTech 시리즈 - 모든 기사
| # | 제목 | 수준 | 상태 |
|---|---|---|---|
| 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를 통한 최적화 | 고급의 | 사용 가능 |
Edge ML 기본 사항: 클라우드와 농업용 엣지
코드 한 줄을 작성하기 전에 엣지 접근 방식이 유일한 방법인 이유를 이해하는 것이 중요합니다. 개방형 현장에서 질병 탐지에 실용적입니다. 이는 문체적인 선택이 아닙니다. 그리고 농업의 물리적 맥락에 의해 부과된 건축적 필요성.
농업 분야 클라우드의 한계
클라우드 우선 질병 감지 애플리케이션은 다음과 같이 작동합니다. 카메라가 클릭합니다. 사진을 네트워크를 통해 클라우드 서버로 전송하면 모델이 서버에서 실행되고 결과가 장치. 광섬유가 있는 사무실에서는 이 주기가 200ms 미만으로 지속됩니다. 포도원에서 투스카니나 칼라브리아의 감귤나무 숲에서는 변수가 급격하게 변합니다.
클라우드와 엣지: 농업적 맥락별 비교
| 특성 | 클라우드 접근 방식 | 엣지 접근 |
|---|---|---|
| 연결 필요 | 연속, 1+Mbps 대역 | 선택 사항, 데이터 동기화에만 해당 |
| 추론 대기 시간 | 200ms - 5s(네트워크에 따라 다름) | 50-500ms(로컬) |
| 운영 비용 | 높음(API 호출, 스토리지, 대역폭) | 낮음(일회성 하드웨어) |
| 개인 정보 보호 제공 | 클라우드의 기업 이미지 | 데이터는 온프레미스에 남아 있습니다. |
| 오프라인 운영 | 불가능한 | 완벽한 |
| 모델 확장성 | 제한 없는 | RAM/CPU에 의해 제한됨 |
| 모델 업데이트 | 즉각적인 | 물리적 또는 OTA 배포가 필요합니다. |
| 에너지 소비 | 낮음(장치) + 높음(데이터 센터) | 장치의 모든 것(3-15W) |
TensorFlow와 TensorFlow Lite: 주요 차이점
TensorFlow(TF)는 GPU 및 서버에 최적화된 완전한 훈련 및 추론 프레임워크입니다. TensorFlow Lite(TFLite)는 임베디드 및 모바일 장치용으로 압축되고 최적화된 버전입니다. 실질적인 차이점은 다음과 같습니다.
- 런타임 크기: TFLite 런타임 및 C++의 경우 약 1MB(전체 TF의 경우 400MB 이상). Raspberry Pi에서 차이는 2~4GB가 아닌 50~100MB의 RAM 점유로 해석됩니다.
- 지원되는 연산자: TFLite는 TF 작업의 하위 집합을 지원합니다. 모델 사용자 정의 작업 또는 비표준 레이어를 사용하려면 CPU에 대한 위임 또는 대체가 필요합니다.
- .tflite 형식: 직접 메모리 액세스에 최적화된 FlatBuffers 체계 파싱하지 않고. 모델은 역직렬화 없이 메모리 매핑됩니다.
- 하드웨어 가속: TFLite는 XNNPACK 대리자(ARM의 SIMD)를 지원합니다. Android의 GPU 대리자, Coral Edge TPU 대리자 및 NNAPI.
양자화: float32 ~ int8
엣지 최적화를 위한 가장 영향력 있는 기술과 양자화: 32비트 부동 소수점에서 8비트 정수로 가중치의 수치 정밀도가 감소합니다. 효과 실용적이고 세 가지:
MobileNetV3Small에서 int8 양자화의 효과
| 미터법 | float32 | int8(PTQ) | 델타 |
|---|---|---|---|
| 모델 크기 | 9.8MB | 2.6MB | -73% |
| RAM 추론(Pi 5) | ~180MB | ~52MB | -71% |
| 추론 지연 시간(Pi 5) | ~180ms | ~65ms | -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é(Penn State University)가 제작하고 2016년에 출판했습니다. 더 많은 내용이 포함되어 있습니다. 이미지 54,309개 통제된 실험실 조건에서 사진을 촬영하고 정리했습니다. 안으로 38개 수업 14종의 식물종과 관련 질병을 다루고 있습니다.
데이터세트 구조
PlantVillage 클래스: 종 및 주요 질병
| Specie | Malattia | 원인 대리인 | 이미지 |
|---|---|---|---|
| 포도나무(포도) | 검은 부패 | 기냐르디아 비드웰리이 | 1,180 |
| 포도나무(포도) | 미끼 / 검은 홍역 | Phaeoacremonium 종. | 1,383 |
| 포도나무(포도) | 잎마름병 | 가성세르코스포라 vitis | 1,076 |
| 포도나무(포도) | 건강한 | — | 423 |
| 토마토 (토마토) | 초기 역병 | 알터나리아 솔라니 | 1,000 |
| 토마토 (토마토) | 역병 역병 | Phytophthora infestans | 1,909 |
| 토마토 (토마토) | 잎 곰팡이 | 토니 파살로라 | 952 |
| 토마토 (토마토) | 셉토리아 잎 지점 | 셉토리아 리코페르시시 | 1,771 |
| 사과나무(사과) | 사과 딱지 (딱지) | 벤투리아 이나이퀄리스 | 630 |
| 사과나무(사과) | 검은 부패 | Botryosphaeria obtusa | 621 |
| 사과나무(사과) | 삼나무 사과 녹 | Gymnosporangium juniper | 275 |
| 옥수수(옥수수) | 일반적인 녹 | 푸치니아 소르기 | 1,192 |
| 감자 (감자) | 초기 역병 | 알터나리아 솔라니 | 1,000 |
| 감자 (감자) | 역병 역병 | Phytophthora infestans | 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% 향상시킵니다.
- MixUp 및 CutMix: 두 개의 이미지를 임의의 가중치(MixUp)로 결합하거나 대체 다른 이미지의 패치가 있는 영역(CutMix) 특히 수업에 효과적입니다. 시각적으로 유사합니다(예: 토마토의 초기 마름병과 늦은 마름병).
- 공격적인 색상 지터: 잎 질병은 색 증상을 나타냅니다 무대, 태양 노출 및 기후 조건에 따라 다릅니다. 색상/채도 변화 범위를 늘리면 일반화에 도움이 됩니다.
- 배경 대체: 고급 기술이 흰색 배경을 대체합니다. 실제 현장 이미지를 활용한 실험실 모습입니다. 예비 의미론적 분할이 필요합니다. 하지만 현장 이미지의 성능이 대폭 향상됩니다.
모델 훈련: MobileNetV3를 사용한 전이 학습
MobileNetV3Small은 이 사용 사례에 이상적인 아키텍처입니다. 모바일 및 임베디드 장치의 추론은 정확성 간의 탁월한 균형을 제공합니다. 그리고 계산상의 복잡성. ResNet50 또는 EfficientNetB4와 비교하여 다음이 필요합니다. 100배 더 적은 작업 경쟁적 정확성을 유지하면서 추론을 통해.
아키텍처 및 디자인 선택
사전 학습된 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 형식으로 변환해야 합니다. 라즈베리 파이에서. 변환 프로세스에는 선택적으로 양자화가 포함됩니다. 이는 크기를 획기적으로 줄이고 추론 속도를 향상시킵니다.
세 가지 양자화 모드
TFLite는 다양한 정확도/성능 균형을 통해 세 가지 수준의 양자화를 지원합니다.
- 동적 범위 양자화: 가중치만 양자화됩니다(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(삼성 프로 내구성) | ~15유로 | 더 나은 무작위 I/O를 위한 A2 |
| 카메라 | 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년에 대한 업데이트된 비교입니다. 발표된 벤치마크와 질병 분류 모델에 대한 자체 측정을 기반으로 합니다.
농업 ML용 엣지 플랫폼 비교
| 플랫폼 | CPU | 숫양 | MobileNetV3S int8 대기 시간 | 소비 | 비용 | 메모 |
|---|---|---|---|---|---|---|
| 라즈베리 파이 4B(4GB) | Cortex-A72 1.8GHz | 4GB LPDDR4 | ~320ms | ~6W | ~55유로 | 경제적 기준 |
| 라즈베리 파이 5(8GB) | Cortex-A76 2.4GHz | 8GB LPDDR4X | ~65ms | ~8W | ~80유로 | 추천 |
| RPi 4 + 코럴 USB TPU | Cortex-A72 + 엣지 TPU | 4GB | ~15ms | ~8W | ~95유로 | int8 전체 모델이 필요합니다. |
| Google Coral 개발 보드 | Cortex-A53 + 엣지 TPU | 1GB | ~12ms | ~4W | ~120유로 | TFLite + Coral 컴파일 전용 |
| 엔비디아 젯슨 나노(4GB) | Cortex-A57 + 128 CUDA 코어 | 4GB LPDDR4 | ~8ms | ~10W | ~149달러 | 단순 분류를 위한 과잉 |
| 아두이노 포르텐타 H7 | Cortex-M7 480MHz | 8MB SDRAM | ~2000ms | ~0.5W | ~100유로 | 소형 모델만 해당(TFLite Micro) |
| Sipeed M1s 독 | BL808 RV64 480MHz | 768KB SRAM | ~800ms(마이크로) | ~0.3W | ~7달러 | 초저전력, 소형 모델 |
단일 224x224 이미지에 대해 MobileNetV3Small int8 모델(2.6MB)에서 측정된 지연 시간입니다. Hackster.io 2024 벤치마크의 Pi 5 값, Georgia Southern University 2024 벤치마크의 Coral.
Raspberry Pi 5의 열 조절
활성 냉각 기능이 없는 Raspberry Pi 5에서는 5~10분 후에 열 조절이 발생합니다. 지속적인 추론으로 인해 성능이 저하됩니다. 20-30%. 현장 배치용(여름철 실외 온도 최대 40°C), 냉각 활동적인 전자 의무적인. Active Cooler의 공식 케이스에는 CPU가 들어있습니다. 지속적인 부하에도 불구하고 70°C 미만.
다음을 통해 온도를 모니터링하세요. vcgencmd measure_temp
이탈리아 식물 질병: Edge ML의 우선순위 목표
이탈리아 농업 환경에는 배포 선택을 안내하는 특정 특성이 있습니다. 이탈리아는 세계에서 세 번째로 큰 와인 생산국(프랑스와 스페인에 이어)이며, 첫 번째 와인 생산국입니다. EU산 와인과 올리브를 생산하며 유럽에서 가장 높은 작물 생물 다양성을 보유하고 있습니다. 식물병증 이탈리아 농업에 가장 경제적으로 영향을 미치는 요소는 다음과 같습니다.
이탈리아의 주요 식물병 및 경제적 영향
| Malattia | 대리인 | 영향을 받은 작물 | 잠재적 손실 | 치료 창구 |
|---|---|---|---|---|
| 포도나무 노균병 | 플라스모파라 비티콜라 | 모든 DOC/DOCG 포도 | 20-100% 생산 | 출현 후 48~72시간 |
| 포도나무 흰가루병 | 에리시페 네카토르 | 포도나무, 박과 | 15-60% 생산 | 5~7일 |
| 사과나무 딱지 | 벤투리아 이나이퀄리스 | 애플이지만 | 30-80% 생산 | 3~5일 |
| 토마토 노균병 | Phytophthora infestans | 토마토, 감자 | 50-100% 생산 | 24-48시간 |
| Botrytis (회색 곰팡이) | 보트리티스 시네레아 | 포도나무, 딸기, 토마토 | 10-40% 생산 | 7~10일 |
| Xylella fastidiosa | Xylella fastidiosa | 올리브 나무(풀리아) | 전체 식물 | 복구 불가능 |
포도 노균병의 경우 가장 큰 문제점은 PlantVillage에 이미지가 포함되어 있지 않다는 점입니다. 이탈리아 DOC/DOCG 포도나무(Sangiovese, Nebbiolo, Primitivo, Nero d'Avola)에만 해당됩니다. 이것이 필요하게 만든다 도메인 적응: 데이터 세트 수집 이탈리아 포도원의 실제 이미지 보완 및 모델 미세 조정 로컬 데이터 세트에 대해 PlantVillage가 사전 훈련되었습니다.
이탈리아 디지털 농업을 위한 자금 및 인센티브
Il CAP 전략 계획 2023-2027 (공통농업정책) 할당 생태계와 개입을 통한 농업 디지털화를 위한 특정 자금 부문별. SRA22("위험 관리") 및 SRA29("정밀 농업") 개입 디지털 식물위생 모니터링 시스템을 채택한 기업에 보상을 제공합니다. 국가 차원에서 M2C4 법안을 통한 PNRR은 다음에 대한 자금을 할당했습니다. 녹색 및 디지털 전환의 일환으로 농업 혁신.
농업 중소기업(매출액이 200만 유로 미만인 회사)의 경우 엣지 ML 시스템 비용 이 기사에 설명된 것과 유사하며(~240 EUR 하드웨어 + 개발) 감가상각 가능 한 계절에 노균병을 조기에 발견할 수 있다고 생각한다면 5헥타르의 포도원은 EUR 15,000-30,000 상당의 생산을 절약할 수 있습니다.
성능 및 모니터링 지표
모니터링 없이 생산 중인 ML 시스템은 성능이 저하될 수 있는 시스템입니다. 조용히. 추적할 측정항목은 두 가지 범주로 나뉩니다. 모델(정확도) 및 시스템 지표(대기 시간, 가동 시간).
# 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
}모범 사례 및 안티 패턴
모범 사례
농업용 Edge ML을 위한 8가지 황금률
- 배포하기 전에 항상 양자화하십시오. float32와 int8의 차이점 ARM의 지연 시간은 50~70%이고 정확도 손실은 1% 미만입니다. Pi 5의 프로덕션에서는 float32를 사용할 이유가 없습니다.
- 실제 현장 데이터로 교정: PlantVillage 데이터 세트 e 실험실에서 획득했습니다. 대상 분야에서 최소 500개의 실제 이미지를 수집하세요. (동일한 포도 품종, 동일한 조명, 동일한 촬영 각도) 교정용 int8 및 미세 조정용입니다. 현장 정확도의 차이는 다음과 같습니다. 이 단계 없이 10-20% 정도.
- 피드백 루프를 처리합니다. 메커니즘을 구현하는 이유 농업경제학자들은 탐지 내용을 확인하거나 수정할 수 있습니다. 모든 수정 사항은 금입니다. 다음 주기의 훈련 데이터가 됩니다.
- 모니터 입력 분포: 모델이 봤다면 훈련 중에는 Sangiovese만 사용했지만 생산에서는 정확성인 Primitivo를 충족합니다. 무너진다. 데이터 드리프트를 감지하기 위해 입력 임베딩의 분포를 플로팅합니다.
- OTA 업데이트 예약: 템플릿은 업데이트 가능해야 합니다. 장치에 물리적으로 접근하지 않고. 다운로드를 위한 MQTT 메커니즘 구현 오류 발생 시 자동 롤백으로 .tflite 파일을 교체합니다.
- 중복성 및 오프라인 모드: 시스템이 제대로 작동해야 합니다 연결이 없어도. 탐지를 위해 로컬 버퍼(SQLite 또는 JSON 파일) 사용 연결이 돌아오면 동기화합니다.
- 문화에 대한 신뢰도 임계값을 보정합니다. 70%의 임계값 노균병(매우 파괴적이고 경제적인 처리)에 대한 것으로, 기존의 것과는 다릅니다. 흰가루병에 대한 임계값(덜 긴급하고 값비싼 치료). 임계값을 다음에 맞게 조정합니다. 거짓 긍정과 거짓 부정의 비대칭 비용.
- 획득 상황을 문서화합니다. 모든 이미지는 다음과 같아야 합니다. 시간, 기상 조건, 계절 단계, GPS 위치 등이 포함된 메타데이터. 이 메타데이터는 디버깅 및 향후 교육에 중요합니다.
피해야 할 안티패턴
농업용 Edge ML 프로젝트에서 가장 흔히 저지르는 5가지 실수
- PlantVillage = 생산 준비가 완료되었다고 가정합니다. 데이터 세트 e 이상적인 조건(균일한 빛, 흰색 배경, 고립된 나뭇잎)에서 획득됩니다. 현장에서는 이미지에는 그림자, 녹색 배경, 물방울, 곤충이 겹쳐져 있습니다. 모델 현장 미세 조정 없이 PlantVillage에서만 교육을 받은 경우 일반적으로 정확도가 높습니다. 실제 현장에서는 60~70%, 실험실 테스트 세트에서는 97%입니다. 이 격차와 으로 알려진 도메인 이동 이것이 주요 실제 문제입니다.
-
클래스 밸런스 무시: PlantVillage에는 많은 수업이 있습니다.
불균형(토마토 역병: 1909개 이미지, 삼나무 사과 녹: 275개 이미지).
클래스 가중치나 오버샘플링이 없으면 모델은 풍부한 클래스에 편향됩니다.
사용
class_weightKeras fit() 또는WeightedRandomSampler. - Pi를 웹 서버로 사용: 다음에 대한 동기 HTTP 요청을 처리합니다. Pi에 대한 추론은 나쁜 아키텍처 선택입니다. Pi는 프로듀서여야 합니다. 요청을 기다리는 서버가 아닌 독립형 MQTT입니다. 연결 관리 HTTP 구문 분석은 오버헤드와 불필요한 복잡성을 추가합니다.
- 극단적인 조명 조건을 처리하지 마십시오. 8월 12시 Puglia, 직사광선은 이미지를 포화시키고 잎은 완전히 흰색으로 나타납니다. 습도가 높은 새벽에는 나뭇잎이 젖어 다르게 반사됩니다. 구현 화질(평균 밝기, 채도)을 확인하기 전에 추론을 실행하고 범위를 벗어난 이미지를 삭제/재시도합니다.
- 정확성을 검증하지 않고 양자화: int8 전체 양자화 대표적인 교정 데이터 세트가 필요합니다. 교정 데이터 세트 사용 너무 작거나(이미지 100개 미만) 대표성이 없으면 손실이 발생할 수 있습니다. 예상 정확도는 0.4%가 아닌 3~5%입니다. 항상 모델 검증 배포 전에 테스트 세트에서 양자화되었습니다.
OTA 모델 업데이트
현장에 수십 또는 수백 개의 장치가 배포되어 있으면 수동으로 업데이트하십시오. 모델이고 비실용적입니다. 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 캡처, 추론 및 전송을 위한 것입니다. 시스템은 다음의 지연 시간을 달성합니다. 당 65ms 추론, 총 전력 소비량은 8W 미만이며 완전히 오프라인으로 작동합니다.
농업 기업가에게 실제로 중요한 숫자는 다릅니다: 질병으로 인한 손실 농작물의 가격이 비싸다 전 세계적으로 2,200억 달러 매년(FAO), 조기 발견 시 적시에 치료하면 손실을 60~80% 줄일 수 있습니다. 시스템 설명된 것과 같은 하드웨어 비용은 250유로 미만이며 한 시즌에 감가상각될 수 있습니다. 귀중한 작물에. 농업 분야 AI 시장은 다음과 같이 성장할 것으로 예상됩니다. CAGR 26.3%로 2025년 59억 달러, 2035년 613억 달러로 성장할 것입니다.
채워야 할 가장 큰 격차는 도메인 이동: 플랜트빌리지 현장 배포에는 그 자체로는 충분하지 않습니다. 넣고 싶은 분들을 위한 우선순위 이 시스템을 생산에 투입하고 작물에서 실제 이미지를 수집하여 라벨을 붙입니다. 믿을 수 있는 농업경제학자와 함께 미세 조정을 해보세요. 또한 500개의 실제 이미지를 올바르게 사용하면 실험실에서는 97%의 정확도와 현장에서는 65%의 정확도를 보이는 시스템 간의 차이를 만들 수 있습니다. 두 가지 상황 모두에서 95% 안정적인 시스템과 비교됩니다.
요약: 완전한 기술 스택
| 레이어 | 기술 | 2025년 버전 |
|---|---|---|
| 훈련 | 텐서플로우/케라스 | TF 2.17+ |
| 데이터세트 | PlantVillage(캐글/HuggingFace) | 이미지 54,309개 |
| 건축학 | MobileNetV3Small | 전이 학습 ImageNet |
| 배포 형식 | TensorFlow Lite(.tflite) | int8 양자화됨 |
| 하드웨어 | 라즈베리 파이 5(8GB) | ARM Cortex-A76 |
| Camera | 카메라 모듈 3(12MP) | Picamera2 + libcamera |
| 메시징 | MQTT(파호-mqtt) | 이클립스 모기 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., & Salathé, M. (2016). 모바일 개발을 가능하게 하는 식물 건강에 대한 이미지의 개방형 액세스 저장소입니다. 질병 진단. ArXiv 사전 인쇄 arXiv:1511.08060. 사용 가능 날짜 캐글 플랜트빌리지.
- FAO 작물 손실: FAO 식물 생산 및 보호 — 해충 및 질병으로 인한 연간 손실
- Raspberry Pi 5의 TFLite: Hackster.io — Raspberry Pi 5에서 TFLite 벤치마킹
- Edge AI 2024 벤치마크: 조지아 서던 대학교(2024). Edge AI 플랫폼 벤치마킹: Coral TPU를 사용한 NVIDIA Jetson 및 Raspberry Pi 5의 성능 분석. IEEE 회의 절차. DOI: 10.1109/10971592.
- MobileNetV3 최적화 PlantVillage: 과학 보고서 — 경량 식물 잎 질병 탐지를 위해 최적화된 MobileNet(2025)
- 이탈리아 정밀 농업: Rural Hack — 농업 4.0 이탈리아 2025: 시장 성장 및 디지털 성숙도