モデル③ CT画像分類CNN（読み解き）

import numpy as np
import pydicom
import torch
import torch.nn.functional as F
from torch.utils.data import Dataset

SIZE = 128   # モデルに入れる画像サイズ（装置ごとに違う元サイズをそろえる）

def preprocess(path):
    # レッスン7の前処理：読込 → HU → ウィンドウ → 0〜1正規化
    ds = pydicom.dcmread(path)
    hu = ds.pixel_array * ds.RescaleSlope + ds.RescaleIntercept
    lo, hi = -600 - 750, -600 + 750   # 肺野条件 WL=-600, WW=1500
    img = np.clip(hu, lo, hi)
    img = (img - lo) / (hi - lo)          # 0〜1 に正規化
    return img.astype(np.float32)

class CTDataset(Dataset):
    def __init__(self, paths, labels):
        self.paths = paths        # DICOMファイルのパス一覧
        self.labels = labels      # 各画像のラベル（0/1）

    def __len__(self):
        return len(self.paths)

    def __getitem__(self, i):
        img = preprocess(self.paths[i])                    # (H, W)
        x = torch.from_numpy(img).unsqueeze(0)             # (1, H, W) チャネル次元を足す
        x = F.interpolate(x.unsqueeze(0), size=(SIZE, SIZE),
                          mode="bilinear", align_corners=False).squeeze(0)  # (1, SIZE, SIZE) にリサイズ
        y = torch.tensor(self.labels[i])
        return x, y

from pathlib import Path

# フォルダ構成の例：data/<患者ID>/<ラベル>/<スライス>.dcm
#   data/P001/normal/0001.dcm  data/P001/normal/0002.dcm ...
#   data/P006/lesion/0001.dcm  ...
paths, labels, patient_ids = [], [], []
for p in sorted(Path("data").glob("*/*/*.dcm")):
    paths.append(str(p))
    labels.append(1 if p.parent.name == "lesion" else 0)   # 病変あり=1
    patient_ids.append(p.parts[-3])                          # 患者ID＝フォルダ名

print("画像数:", len(paths), " 患者数:", len(set(patient_ids)))
print("病変あり:", sum(labels), " 病変なし:", len(labels) - sum(labels))

画像数: 200  患者数: 20
病変あり: 100  病変なし: 100

from sklearn.model_selection import GroupShuffleSplit

# paths[i] の画像が「どの患者のものか」を patient_ids に持たせておく
# 例: patient_ids = ["P001", "P001", "P002", ...]（スライスごとに患者ID）

gss = GroupShuffleSplit(n_splits=1, test_size=0.2, random_state=42)
train_idx, test_idx = next(gss.split(paths, labels, groups=patient_ids))

# 同じ患者が train と test の両方に出ないことを確認
train_patients = set(patient_ids[i] for i in train_idx)
test_patients  = set(patient_ids[i] for i in test_idx)
print("患者の重複:", len(train_patients & test_patients))

患者の重複: 0

from torch.utils.data import DataLoader

torch.manual_seed(0)   # 再現性のため乱数を固定

train_ds = CTDataset([paths[i] for i in train_idx], [labels[i] for i in train_idx])
test_ds  = CTDataset([paths[i] for i in test_idx],  [labels[i] for i in test_idx])

train_loader = DataLoader(train_ds, batch_size=16, shuffle=True)   # 学習：順序をシャッフル
test_loader  = DataLoader(test_ds,  batch_size=16, shuffle=False)  # 評価：順序は固定

x, y = next(iter(train_loader))
print("1バッチの形:", x.shape, " ラベル:", y.shape)

1バッチの形: torch.Size([16, 1, 128, 128])  ラベル: torch.Size([16])

import torch.nn as nn

class SmallCNN(nn.Module):
    def __init__(self):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(1, 16, 3, padding=1), nn.ReLU(), nn.MaxPool2d(2),  # (N,16,H/2,W/2)
            nn.Conv2d(16, 32, 3, padding=1), nn.ReLU(), nn.MaxPool2d(2),  # (N,32,H/4,W/4)
        )
        self.head = nn.Sequential(
            nn.AdaptiveAvgPool2d(1), nn.Flatten(),   # (N, 32)
            nn.Linear(32, 2),                        # 2クラス（正常/病変）
        )

    def forward(self, x):
        return self.head(self.features(x))

model = SmallCNN()
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)

model.train()
for epoch in range(5):
    for x, y in train_loader:              # x:(N,1,H,W)  y:(N,)
        optimizer.zero_grad()             # ① 勾配リセット
        out = model(x)                   # ② 順伝播
        loss = criterion(out, y)         # ③ 損失
        loss.backward()                 # ④ 逆伝播
        optimizer.step()                  # ⑤ 更新
    print(f"epoch {epoch}: loss={loss.item():.3f}")

epoch 0: loss=0.702
epoch 1: loss=0.696
epoch 2: loss=0.695
epoch 3: loss=0.694
epoch 4: loss=0.693   # ← 上の合成データ（200枚）では、5エポックではほとんど下がらない

model.eval()                            # 評価モード（dropout等を止める）
probs, trues = [], []
with torch.no_grad():                   # 勾配計算を止める（評価では不要）
    for x, y in test_loader:
        p = torch.softmax(model(x), dim=1)[:, 1]  # 病変クラスの確率
        probs.extend(p.tolist())
        trues.extend(y.tolist())

# あとはレッスン5と同じ：閾値で感度・特異度、roc_auc_score で AUC
from sklearn.metrics import roc_auc_score
print("AUC:", roc_auc_score(trues, probs))