1# 建議使用 uv 建立虛擬環境
2uv venv monai-gen-env
3source monai-gen-env/bin/activate
4
5# 安裝 monai-generative
6uv pip install monai-generative
7
8# 安裝完整 MONAI（含醫學影像 I/O 支援）
9uv pip install "monai[all]"

 1uv venv monai-gen-dev
 2source monai-gen-dev/bin/activate
 3
 4# 從 GitHub main branch 安裝
 5uv pip install git+https://github.com/Project-MONAI/GenerativeModels.git
 6
 7# 或 clone 後以可編輯模式安裝
 8git clone https://github.com/Project-MONAI/GenerativeModels.git
 9cd GenerativeModels
10uv pip install -e "."

1# 使用 MONAI 官方 Docker 映像（已含 CUDA + PyTorch）
2docker run --gpus all -it \
3  -v $(pwd)/data:/workspace/data \
4  projectmonai/monai:latest \
5  bash -c "pip install monai-generative && bash"

1import generative
2print(f"monai-generative version: {generative.__version__}")
3
4from generative.networks.nets import AutoencoderKL, DiffusionModelUNet, VQVAE
5from generative.networks.schedulers import DDPMScheduler, DDIMScheduler
6from generative.inferers import DiffusionInferer, LatentDiffusionInferer
7print("All core modules imported successfully.")

1# 安裝開發依賴（用於貢獻程式碼）
2git clone https://github.com/Project-MONAI/GenerativeModels.git
3cd GenerativeModels
4uv pip install -r requirements-dev.txt
5uv pip install -e "."
6
7# 執行測試
8python -m pytest tests/ -v --tb=short

  1import torch
  2import torch.nn.functional as F
  3from monai.config import print_config
  4from monai.data import DataLoader
  5from monai.utils import set_determinism
  6from monai.apps import MedNISTDataset
  7from monai.transforms import (
  8    Compose,
  9    EnsureChannelFirstd,
 10    Lambdad,
 11    LoadImaged,
 12    ScaleIntensityRanged,
 13)
 14from generative.networks.nets import DiffusionModelUNet
 15from generative.networks.schedulers import DDPMScheduler
 16from generative.inferers import DiffusionInferer
 17
 18# ── 1. 設定與資料載入 ──────────────────────────────
 19set_determinism(42)
 20device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
 21
 22# 使用 MONAI 內建的 MedNIST 資料集（自動下載）
 23train_transforms = Compose([
 24    LoadImaged(keys=["image"]),
 25    EnsureChannelFirstd(keys=["image"]),
 26    ScaleIntensityRanged(keys=["image"], a_min=0.0, a_max=255.0,
 27                         b_min=0.0, b_max=1.0, clip=True),
 28    Lambdad(keys=["image"],
 29            func=lambda x: F.pad(x, (2, 2, 2, 2))),  # 64→68 padding
 30])
 31
 32train_ds = MedNISTDataset(
 33    root_dir="./data", section="training",
 34    transform=train_transforms, download=True,
 35    filter_class="Hand"  # 只取手部 X 光
 36)
 37train_loader = DataLoader(train_ds, batch_size=32, shuffle=True,
 38                          num_workers=4, persistent_workers=True)
 39
 40# ── 2. 建立模型元件 ──────────────────────────────
 41# DiffusionModelUNet：預測噪聲的 U-Net
 42model = DiffusionModelUNet(
 43    spatial_dims=2,
 44    in_channels=1,           # 灰階醫學影像
 45    out_channels=1,
 46    num_channels=(64, 128, 256),
 47    attention_levels=(False, True, True),
 48    num_res_blocks=1,
 49    num_head_channels=64,
 50).to(device)
 51
 52# DDPM Scheduler：控制前向加噪與反向去噪
 53scheduler = DDPMScheduler(
 54    num_train_timesteps=1000,
 55    beta_start=0.0015,
 56    beta_end=0.0195,
 57    schedule="scaled_linear_beta",
 58)
 59
 60# DiffusionInferer：封裝訓練與取樣邏輯
 61inferer = DiffusionInferer(scheduler=scheduler)
 62
 63optimizer = torch.optim.Adam(model.parameters(), lr=2.5e-5)
 64
 65# ── 3. 訓練迴圈 ──────────────────────────────────
 66num_epochs = 50
 67for epoch in range(num_epochs):
 68    model.train()
 69    epoch_loss = 0
 70    for batch in train_loader:
 71        images = batch["image"].to(device)
 72
 73        # 隨機取樣時間步
 74        timesteps = torch.randint(0, scheduler.num_train_timesteps,
 75                                  (images.shape[0],),
 76                                  device=device).long()
 77
 78        # 透過 inferer 計算訓練損失（內部執行加噪 + 預測噪聲）
 79        noise_pred = inferer(
 80            inputs=images,
 81            diffusion_model=model,
 82            noise=torch.randn_like(images),
 83            timesteps=timesteps,
 84        )
 85
 86        # 計算 MSE 損失（預測噪聲 vs 真實噪聲）
 87        noise = torch.randn_like(images)
 88        noisy_images = scheduler.add_noise(
 89            original_samples=images, noise=noise, timesteps=timesteps
 90        )
 91        pred = model(noisy_images, timesteps=timesteps)
 92        loss = F.mse_loss(pred, noise)
 93
 94        optimizer.zero_grad()
 95        loss.backward()
 96        optimizer.step()
 97
 98        epoch_loss += loss.item()
 99
100    avg_loss = epoch_loss / len(train_loader)
101    if (epoch + 1) % 10 == 0:
102        print(f"Epoch {epoch+1}/{num_epochs}, Loss: {avg_loss:.4f}")
103
104# ── 4. 生成合成影像 ──────────────────────────────
105model.eval()
106with torch.no_grad():
107    # 從純噪聲開始，逐步去噪
108    noise = torch.randn(8, 1, 68, 68, device=device)
109    synthetic_images = inferer.sample(
110        input_noise=noise,
111        diffusion_model=model,
112        scheduler=scheduler,
113    )
114    print(f"Generated images shape: {synthetic_images.shape}")
115    # → torch.Size([8, 1, 68, 68])

 1import torch
 2from generative.networks.nets import AutoencoderKL, DiffusionModelUNet
 3from generative.networks.schedulers import DDIMScheduler
 4from generative.inferers import LatentDiffusionInferer
 5
 6device = torch.device("cuda")
 7
 8# ── Stage 1: AutoencoderKL（潛空間編碼器-解碼器）─────
 9autoencoder = AutoencoderKL(
10    spatial_dims=3,           # 3D 體積資料
11    in_channels=1,            # 單通道 MRI
12    out_channels=1,
13    num_channels=(64, 128, 256),
14    latent_channels=3,        # 潛空間通道數
15    num_res_blocks=2,
16    attention_levels=(False, False, True),
17    with_encoder_nonlocal_attn=False,
18    with_decoder_nonlocal_attn=False,
19).to(device)
20
21# ── Stage 2: DiffusionModelUNet（潛空間去噪網路）──────
22unet = DiffusionModelUNet(
23    spatial_dims=3,
24    in_channels=3,            # 匹配 latent_channels
25    out_channels=3,
26    num_channels=(64, 128, 256),
27    attention_levels=(False, True, True),
28    num_res_blocks=1,
29    num_head_channels=64,
30).to(device)
31
32# ── DDIM Scheduler（加速取樣）──────────────────────
33scheduler = DDIMScheduler(
34    num_train_timesteps=1000,
35    beta_start=0.0015,
36    beta_end=0.0195,
37    schedule="scaled_linear_beta",
38    clip_sample=False,        # 潛空間不需裁剪
39)
40
41# ── LatentDiffusionInferer（整合推論）───────────────
42inferer = LatentDiffusionInferer(
43    scheduler=scheduler,
44    scale_factor=1.0,         # 潛空間縮放因子
45)
46
47# ── 推論：生成 3D 合成腦部 MRI ──────────────────────
48# 假設 autoencoder 和 unet 已訓練完成並載入權重
49autoencoder.eval()
50unet.eval()
51
52with torch.no_grad():
53    # 設定 DDIM 取樣步數（遠少於 DDPM 的 1000 步）
54    scheduler.set_timesteps(num_inference_steps=50)
55
56    # 計算潛空間大小
57    # 若原始影像為 [1, 1, 96, 96, 96]
58    # AutoencoderKL 下採樣 4x → 潛空間 [1, 3, 24, 24, 24]
59    latent_shape = (1, 3, 24, 24, 24)
60    noise = torch.randn(latent_shape, device=device)
61
62    # 在潛空間中逐步去噪
63    synthetic_latent = inferer.sample(
64        input_noise=noise,
65        autoencoder_model=autoencoder,
66        diffusion_model=unet,
67        scheduler=scheduler,
68    )
69    print(f"Synthetic 3D MRI shape: {synthetic_latent.shape}")
70    # → torch.Size([1, 1, 96, 96, 96])

  1import torch
  2from generative.networks.nets import DiffusionModelUNet
  3from generative.networks.schedulers import DDPMScheduler, DDIMScheduler
  4from generative.inferers import DiffusionInferer
  5
  6device = torch.device("cuda")
  7
  8# ── 建立帶 class conditioning 的 DiffusionModelUNet ──
  9model = DiffusionModelUNet(
 10    spatial_dims=2,
 11    in_channels=1,
 12    out_channels=1,
 13    num_channels=(128, 256, 512),
 14    attention_levels=(False, True, True),
 15    num_res_blocks=2,
 16    num_head_channels=64,
 17    num_class_embeds=10,     # 支援 10 個類別的條件嵌入
 18).to(device)
 19
 20scheduler = DDIMScheduler(
 21    num_train_timesteps=1000,
 22    beta_start=0.0015,
 23    beta_end=0.0195,
 24    schedule="scaled_linear_beta",
 25)
 26inferer = DiffusionInferer(scheduler=scheduler)
 27
 28# ── 訓練：Classifier-Free Guidance 策略 ─────────────
 29# 訓練時以一定機率（通常 10-20%）丟棄類別標籤
 30# 使模型同時學習條件與無條件生成
 31optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)
 32uncond_prob = 0.1  # 10% 的時間用無條件訓練
 33
 34model.train()
 35for batch_images, batch_labels in train_loader:  # 假設已定義
 36    images = batch_images.to(device)
 37    labels = batch_labels.to(device)
 38
 39    timesteps = torch.randint(
 40        0, 1000, (images.shape[0],), device=device
 41    ).long()
 42    noise = torch.randn_like(images)
 43
 44    # 隨機丟棄類別標籤以訓練無條件分支
 45    mask = torch.rand(labels.shape[0]) < uncond_prob
 46    class_labels = labels.clone()
 47    class_labels[mask] = -1  # -1 表示無條件
 48
 49    noisy_images = scheduler.add_noise(images, noise, timesteps)
 50    pred = model(
 51        noisy_images,
 52        timesteps=timesteps,
 53        class_labels=class_labels,
 54    )
 55
 56    loss = torch.nn.functional.mse_loss(pred, noise)
 57    optimizer.zero_grad()
 58    loss.backward()
 59    optimizer.step()
 60
 61# ── 推論：Classifier-Free Guidance 取樣 ─────────────
 62model.eval()
 63guidance_scale = 7.5  # CFG 強度
 64
 65with torch.no_grad():
 66    scheduler.set_timesteps(num_inference_steps=50)
 67    noise = torch.randn(4, 1, 64, 64, device=device)
 68    target_class = torch.tensor([3, 3, 3, 3], device=device)  # 目標類別
 69
 70    sample = noise
 71    for t in scheduler.timesteps:
 72        t_batch = t.expand(4).to(device)
 73
 74        # 條件預測
 75        noise_pred_cond = model(
 76            sample, timesteps=t_batch, class_labels=target_class
 77        )
 78        # 無條件預測
 79        noise_pred_uncond = model(
 80            sample, timesteps=t_batch,
 81            class_labels=torch.tensor([-1, -1, -1, -1], device=device)
 82        )
 83
 84        # CFG 組合：增強條件方向
 85        noise_pred = noise_pred_uncond + guidance_scale * (
 86            noise_pred_cond - noise_pred_uncond
 87        )
 88
 89        # 去噪一步
 90        sample, _ = scheduler.step(noise_pred, t, sample)
 91
 92    print(f"Conditional generated images: {sample.shape}")
 93
 94# ── 異常偵測應用 ────────────────────────────────────
 95# 概念：正常影像的重建誤差低，異常影像的重建誤差高
 96def compute_anomaly_score(model, scheduler, inferer, image, num_steps=50):
 97    """
 98    計算影像的異常分數 (Anomaly Score)。
 99
100    原理：對輸入影像加噪到特定時間步 t，
101    再用訓練好的模型去噪回來，
102    比較原始影像與重建影像的差異。
103    """
104    scheduler.set_timesteps(num_inference_steps=num_steps)
105
106    # 選擇中等強度的噪聲水平
107    t_start = torch.tensor([250], device=image.device)
108    noise = torch.randn_like(image)
109    noisy = scheduler.add_noise(image, noise, t_start)
110
111    # 從加噪影像去噪回來
112    sample = noisy
113    start_idx = (scheduler.timesteps >= t_start.item()).sum().item()
114    for t in scheduler.timesteps[start_idx:]:
115        t_batch = t.expand(image.shape[0]).to(image.device)
116        pred = model(sample, timesteps=t_batch)
117        sample, _ = scheduler.step(pred, t, sample)
118
119    # 異常分數 = 像素級重建誤差
120    anomaly_map = (image - sample).abs()
121    anomaly_score = anomaly_map.mean().item()
122
123    return anomaly_score, anomaly_map

 1from monai.data import CacheDataset, DataLoader
 2from monai.transforms import (
 3    Compose, LoadImaged, EnsureChannelFirstd,
 4    Orientationd, Spacingd, ScaleIntensityd,
 5    CropForegroundd, RandSpatialCropd,
 6)
 7
 8# 標準 MONAI 3D 醫學影像前處理管線
 9train_transforms = Compose([
10    LoadImaged(keys=["image"]),
11    EnsureChannelFirstd(keys=["image"]),
12    Orientationd(keys=["image"], axcodes="RAS"),
13    Spacingd(keys=["image"], pixdim=(1.0, 1.0, 1.0),
14             mode="bilinear"),
15    ScaleIntensityd(keys=["image"]),
16    CropForegroundd(keys=["image"], source_key="image"),
17    RandSpatialCropd(keys=["image"],
18                     roi_size=(96, 96, 96),
19                     random_size=False),
20])
21
22# 直接用於 generative models 的訓練
23dataset = CacheDataset(
24    data=data_dicts, transform=train_transforms, cache_rate=1.0
25)
26loader = DataLoader(dataset, batch_size=2, shuffle=True)

 1from monai.bundle import ConfigParser, download
 2
 3# 下載腦部影像合成 LDM bundle
 4download(
 5    name="brain_image_synthesis_latent_diffusion_model",
 6    source="Project-MONAI/GenerativeModels",
 7    bundle_dir="./bundles"
 8)
 9
10# 載入推論配置
11parser = ConfigParser()
12parser.read_config(
13    "./bundles/brain_image_synthesis_latent_diffusion_model/"
14    "configs/inference.json"
15)

 1from medigan import Generators
 2
 3generators = Generators()
 4
 5# 列出所有可用的生成模型（含 MONAI-based 模型）
 6models = generators.find_matching_model_ids(
 7    task="synthesis",
 8    modality="MRI",
 9)
10
11# 透過 medigan 統一介面呼叫
12synthetic_images = generators.generate(
13    model_id=models[0],
14    num_samples=100,
15    output_path="./synthetic_data/",
16)