画像分類の古典的モデルであるLeNetを、実際の眼疾患データセット（iChallenge-PM）に適用し、病的な近視の識別を行うプロセスを解説します。近視は進行すると網膜変性などの病理的変化を引き起こす可能性があるため、初期段階での自動識別は非常に重要です。

データセットの概要

iChallenge-PMデータセットは、以下の命名規則に基づいて分類されています。

• Pから始まるファイル名：病理性近視（ラベル：1 / 正解サンプル）
• HまたはNから始まるファイル名：高度近視または正常（ラベル：0 / 負例サンプル）

ここでは、画像を224x224にリサイズし、LeNetを用いて特徴抽出と二値分類を行います。

データ処理の実装

モデルに供給する前に、画像を適切なフォーマットに変換するプリプロセッサを定義します。

import cv2
import numpy as np
import os
import paddle

def prepare_image(img_path):
    # 画像の読み込みとリサイズ
    raw_img = cv2.imread(img_path)
    resized_img = cv2.resize(raw_img, (224, 224))
    
    # HWCからCHW形式へ変換
    transposed_img = np.transpose(resized_img, (2, 0, 1))
    
    # ピクセル値を[-1.0, 1.0]の範囲に正規化
    normalized_img = (transposed_img.astype('float32') / 255.0) * 2.0 - 1.0
    return normalized_img

def create_data_generator(data_dir, batch_size=10, is_training=True):
    file_list = os.listdir(data_dir)
    
    def generator():
        if is_training:
            np.random.shuffle(file_list)
        
        imgs_buffer = []
        labels_buffer = []
        
        for fname in file_list:
            path = os.path.join(data_dir, fname)
            img_data = prepare_image(path)
            
            # 命名規則によるラベル付け
            label = 1 if fname.startswith('P') else 0
            
            imgs_buffer.append(img_data)
            labels_buffer.append(label)
            
            if len(imgs_buffer) == batch_size:
                yield np.array(imgs_buffer).astype('float32'), \
                      np.array(labels_buffer).astype('float32').reshape(-1, 1)
                imgs_buffer, labels_buffer = [], []
                
        if imgs_buffer:
            yield np.array(imgs_buffer).astype('float32'), \
                  np.array(labels_buffer).astype('float32').reshape(-1, 1)
                  
    return generator

LeNetモデルの定義

PaddlePaddleを使用してLeNetの構造を定義します。入力解像度が224x224であるため、全結合層（Linear）の入力次元数はそれに応じて計算されています。

import paddle.nn as nn
import paddle.nn.functional as F

class EyeDiseaseClassifier(nn.Layer):
    def __init__(self, num_outputs=1):
        super(EyeDiseaseClassifier, self).__init__()
        
        # 畳み込み層とプーリング層の構成
        self.conv_block1 = nn.Conv2D(in_channels=3, out_channels=6, kernel_size=5)
        self.pool1 = nn.MaxPool2D(kernel_size=2, stride=2)
        
        self.conv_block2 = nn.Conv2D(in_channels=6, out_channels=16, kernel_size=5)
        self.pool2 = nn.MaxPool2D(kernel_size=2, stride=2)
        
        self.conv_block3 = nn.Conv2D(in_channels=16, out_channels=120, kernel_size=4)
        
        # 全結合層
        # 入力画像224x224の場合、最終的な特徴マップサイズから計算
        self.fc_layer1 = nn.Linear(in_features=300000, out_features=64)
        self.fc_layer2 = nn.Linear(in_features=64, out_features=num_outputs)

    def forward(self, inputs):
        x = F.sigmoid(self.conv_block1(inputs))
        x = self.pool1(x)
        
        x = F.sigmoid(self.conv_block2(x))
        x = self.pool2(x)
        
        x = F.sigmoid(self.conv_block3(x))
        
        # フラット化
        x = paddle.flatten(x, start_axis=1)
        
        x = F.sigmoid(self.fc_layer1(x))
        logits = self.fc_layer2(x)
        return logits

学習と検証の実行

モデルの重みを更新する訓練ループと、各エポック終了時の検証プロセスを記述します。損失関数にはbinary_cross_entropy_with_logitsを使用します。

def execute_training(model, train_path, val_path, val_csv):
    optimizer = paddle.optimizer.Adam(learning_rate=0.001, parameters=model.parameters())
    train_loader = create_data_generator(train_path, batch_size=16)
    
    epochs = 5
    for epoch in range(epochs):
        model.train()
        for batch_id, (images, labels) in enumerate(train_loader()):
            images_tensor = paddle.to_tensor(images)
            labels_tensor = paddle.to_tensor(labels)
            
            outputs = model(images_tensor)
            loss = F.binary_cross_entropy_with_logits(outputs, labels_tensor)
            avg_loss = paddle.mean(loss)
            
            avg_loss.backward()
            optimizer.step()
            optimizer.clear_grad()
            
            if batch_id % 20 == 0:
                print(f"Epoch {epoch}, Batch {batch_id}, Loss: {avg_loss.numpy()[0]:.4f}")
        
        # バリデーション処理
        model.eval()
        accuracies = []
        with paddle.no_checkpoint():
            # 検証データの読み込み処理（ここでは簡易的に評価指標のみ算出）
            # 実装に合わせて検証用ジェネレータを使用
            pass
        
        paddle.save(model.state_dict(), f'eye_lenet_epoch_{epoch}.pdparams')

# インスタンス化と実行
network = EyeDiseaseClassifier()
# execute_training(network, 'path/to/train', 'path/to/val', 'labels.csv')

結果の考察

画像分類の基礎であるLeNetを用いて眼疾患データの分類を行いましたが、単純なLeNet構造では複雑な医療画像の微細な特徴を捉えきれない場合があります。精度の向上のためには、ResNetやVGGのようなより深いネットワークの採用、またはデータオーギュメンテーション（画像拡張）の導入を検討する必要があります。