農業現場における病害虫の早期発見は、収量と品質を維持する上で不可欠です。しかし、圃場やビニールハウスなどの環境では通信インフラが脆弱であり、クラウド依存の画像解析APIはレイテンシや接続断のリスクを抱えています。本稿では、PyTorchとDockerを活用し、ネットワーク接続を前提としないエッジデバイス上で動作する農作物病害検出システムの実装と最適化手法について解説します。

オフライン推論アーキテクチャの要件

圃場でのリアルタイム診断を実現するため、システムには以下の特性が求められます。

• 低レイテンシ: エッジデバイス上でのミリ秒単位の推論処理。
• 通信独立: 4G/5Gの電波状態や悪天候によるネットワーク障害の影響を受けない自律性。
• リソース効率: 限られたVRAM（例：4GB〜8GB）での安定稼働。

サポート対象とモデル構成

ベースラインとなるConvolutional Neural Network（CNN）としてResNet50を採用し、以下の主要農作物の病変特徴量を抽出するようにファインチューニングを行います。

• イネ: いもち病、紋枯病、白葉枯病
• コムギ: さび病、うどんこ病、赤かび病
• トウモロコシ: ごま葉枯病、さび病
• 野菜類: べと病、炭そ病

Dockerによるコンテナ環境の構築

依存関係の解消とポータビリティを確保するため、NVIDIA Container Toolkitを利用したDockerイメージをビルド・実行します。

# Dockerイメージのビルド
docker build -t crop-pathology-detector:latest -f Dockerfile.gpu .

# コンテナの起動（GPUアクセスとポートマッピング）
docker run --rm -d \
  --gpus '"device=0"' \
  --shm-size=2g \
  -p 8080:80 \
  -v $(pwd)/model_weights:/app/weights \
  crop-pathology-detector:latest

サービス起動後、http://<エッジデバイスのIP>:8080 にアクセスすることで、推論用のWebインターフェースを利用できます。

推論パイプラインのカスタマイズと精度向上

特定の圃場環境や作物の品種に合わせて、前処理と推論パラメータを動的に調整することで検出精度を向上させます。

import torch
import torchvision.transforms as transforms
from typing import Dict

class PathologyInferenceEngine:
    def __init__(self, weights_path: str, device: str = "cuda"):
        self.device = torch.device(device if torch.cuda.is_available() else "cpu")
        self.model = self._load_model(weights_path)
        self.model.eval()
        
        # 入力解像度の拡張と詳細な病斑抽出のための前処理
        self.transform = transforms.Compose([
            transforms.Resize((384, 384)),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406], 
                                 std=[0.229, 0.224, 0.225])
        ])
        self.confidence_cutoff = 0.65

    def _load_model(self, path: str):
        checkpoint = torch.load(path, map_location=self.device)
        net = checkpoint["architecture"]
        net.load_state_dict(checkpoint["state_dict"])
        return net.to(self.device)

    def predict_with_tta(self, image_tensor: torch.Tensor) -> Dict:
        # 水平反転によるTest-Time Augmentation (TTA) アンサンブル推論
        flipped_tensor = torch.flip(image_tensor, dims=[-1])
        inputs = torch.cat([image_tensor, flipped_tensor], dim=0).to(self.device)
        
        with torch.no_grad():
            logits = self.model(inputs)
            probs = torch.softmax(logits, dim=1)
            
        # 元画像と反転画像の確率を平均化
        averaged_probs = probs.mean(dim=0)
        max_prob, predicted_class = torch.max(averaged_probs, dim=0)
        
        return {
            "class_id": predicted_class.item(),
            "confidence": max_prob.item(),
            "is_valid": max_prob.item() >= self.confidence_cutoff
        }

REST APIを用いたバッチ処理と自動化

ドローンや圃場監視カメラから収集した大量の画像を、夜間などのオフピーク時に自動処理するためのAPIリクエストとスケジューリングの実装例です。

import requests
import json
from concurrent.futures import ThreadPoolExecutor
from typing import List

ENDPOINT = "http://127.0.0.1:8080/api/v1/analyze"

def process_image_batch(file_paths: List[str]) -> List[Dict]:
    def send_request(path: str) -> Dict:
        with open(path, "rb") as img_file:
            payload = {
                "include_visualization": True,
                "export_format": "json"
            }
            files = {"media": (path.split("/")[-1], img_file, "image/jpeg")}
            response = requests.post(ENDPOINT, data=payload, files=files, timeout=10)
            return response.json()

    # スレッドプールによる並列リクエスト
    with ThreadPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(send_request, file_paths))
        
    return results

if __name__ == "__main__":
    from pathlib import Path
    target_images = [str(p) for p in Path("/data/daily_captures").glob("*.jpg")]
    analysis_report = process_image_batch(target_images)
    
    with open("/data/reports/daily_diagnosis.json", "w") as f:
        json.dump(analysis_report, f, indent=2)

リソース制約下でのメモリ最適化

エッジデバイスではVRAMが限られるため、推論時のメモリフットプリントを厳密に制御する必要があります。

import torch

def optimize_memory_footprint(model: torch.nn.Module, batch_size: int = 2):
    # 不要な勾配計算の無効化とキャッシュのクリア
    torch.set_grad_enabled(False)
    if torch.cuda.is_available():
        torch.cuda.empty_cache()
        
    # メモリ使用量を抑えたDataLoaderの設定
    dataloader_kwargs = {
        "batch_size": batch_size,
        "num_workers": 2,
        "pin_memory": True,
        "prefetch_factor": 2
    }
    return dataloader_kwargs

# 推論ループ内での自動混合精度（AMP）適用例
def run_inference_loop(model, dataloader):
    model.eval()
    with torch.no_grad():
        for batch in dataloader:
            images = batch["images"].cuda(non_blocking=True)
            with torch.cuda.amp.autocast(dtype=torch.float16):
                outputs = model(images)
            # 後処理...

検出精度の低下に対するアプローチ

モデルの誤検知や未検出が発生した場合、以下の要因を切り分けて対応します。

• 光学特性のノイズ: 強い日差しによる白飛びや、葉の重なりによる影の影響を排除するため、偏光フィルターの使用や複数アングルからの撮影をルール化します。
• ドメインシフト: 栽培品種や生育ステージが学習データと異なる場合、少量のローカルデータを用いて最終層の重みを更新します。

from torch.optim import AdamW
from torch.nn import CrossEntropyLoss

def adapt_to_local_environment(model, local_dataloader, epochs=5):
    # 最終層（分類ヘッド）のみを学習対象とする
    for param in model.parameters():
        param.requires_grad = False
    for param in model.classifier.parameters():
        param.requires_grad = True
        
    optimizer = AdamW(model.classifier.parameters(), lr=1e-4, weight_decay=1e-2)
    criterion = CrossEntropyLoss(label_smoothing=0.1)
    
    model.train()
    for epoch in range(epochs):
        for batch in local_dataloader:
            imgs, labels = batch["image"].cuda(), batch["label"].cuda()
            optimizer.zero_grad()
            
            with torch.cuda.amp.autocast():
                preds = model(imgs)
                loss = criterion(preds, labels)
                
            loss.backward()
            optimizer.step()
            
    return model