diffusion-model small example

diffusion-model for MNIST

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
import torch
import torch.nn as nn
import torch.nn.functional as F
from torchvision import datasets, transforms, utils
from torch.utils.data import DataLoader
import matplotlib.pyplot as plt

# 超参数设置
batch_size = 128
num_epochs = 20
timesteps = 1000  # 扩散总步数
beta_start = 1e-4
beta_end = 0.02
learning_rate = 1e-3

print(f"CUDA是否可用: {torch.cuda.is_available()}")
if torch.cuda.is_available():
    print(f"当前GPU设备: {torch.cuda.get_device_name(0)}")
    print(f"GPU内存使用: {torch.cuda.memory_allocated()/1024**2:.2f} MB")

# 在设备初始化后添加
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"当前使用的设备: {device}")

# 定义噪声调度（线性调度）
def linear_beta_schedule(timesteps, beta_start, beta_end):
    return torch.linspace(beta_start, beta_end, timesteps, device=device)

# 预计算扩散过程的关键参数
betas = linear_beta_schedule(timesteps, beta_start, beta_end)
alphas = 1. - betas
alphas_cumprod = torch.cumprod(alphas, dim=0)
sqrt_alphas_cumprod = torch.sqrt(alphas_cumprod)
sqrt_one_minus_alphas_cumprod = torch.sqrt(1. - alphas_cumprod)

# 定义UNet模型（简化版）
class SimpleUNet(nn.Module):
    def __init__(self):
        super().__init__()
        # 增加通道数和层数
        self.encoder = nn.Sequential(
            nn.Conv2d(1, 64, 3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(inplace=False),
            nn.Conv2d(64, 128, 3, stride=2, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(inplace=False)
        )
        self.mid = nn.Sequential(
            nn.Conv2d(128, 128, 3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(inplace=False),
            nn.Conv2d(128, 128, 3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(inplace=False)
        )
        self.decoder = nn.Sequential(
            nn.ConvTranspose2d(128, 64, 3, stride=2, padding=1, output_padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(inplace=False),
            nn.Conv2d(64, 1, 3, padding=1)
        )
        self.time_embed = nn.Embedding(timesteps, 128)  # 增加嵌入维度

    def forward(self, x, t):
        # 时间嵌入
        t_emb = self.time_embed(t).unsqueeze(-1).unsqueeze(-1)
        
        # 编码器
        x = self.encoder(x)
        # 添加时间信息 - 避免使用原地操作
        x = x + t_emb  # 将 x += t_emb 改为 x = x + t_emb
        # 中间层
        x = self.mid(x)
        # 解码器
        x = self.decoder(x)
        return x

# 加载MNIST数据集
transform = transforms.Compose([transforms.ToTensor(), transforms.Normalize((0.5,), (0.5,))])
dataset = datasets.MNIST(root='./data', train=True, download=True, transform=transform)
dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)

# 初始化模型和优化器
model = SimpleUNet().to(device)
# 检查模型所在设备
print(f"模型所在设备: {next(model.parameters()).device}")

optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)

# 在训练循环之前添加这行代码
torch.autograd.set_detect_anomaly(True)

# 训练循环
for epoch in range(num_epochs):
    model.train()
    total_loss = 0
    for step, (images, _) in enumerate(dataloader):
        images = images.to(device)
        batch_size = images.shape[0]
        
        # 随机采样时间步
        t = torch.randint(0, timesteps, (batch_size,), device=device).long()
        
        # 前向扩散过程（加噪）
        sqrt_alpha_cumprod_t = sqrt_alphas_cumprod[t].view(batch_size, 1, 1, 1)
        sqrt_one_minus_alpha_cumprod_t = sqrt_one_minus_alphas_cumprod[t].view(batch_size, 1, 1, 1)
        noise = torch.randn_like(images)
        noisy_images = sqrt_alpha_cumprod_t * images + sqrt_one_minus_alpha_cumprod_t * noise
        
        # 预测噪声
        predicted_noise = model(noisy_images, t)
        
        # 计算损失
        loss = F.mse_loss(noise, predicted_noise)
        
        # 反向传播
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        
        total_loss += loss.item()
        if (step + 1) % 50 == 0:
            print(f"Epoch [{epoch+1}/{num_epochs}], Step [{step+1}/{len(dataloader)}], Loss: {loss.item():.4f}")
        
    avg_loss = total_loss / len(dataloader)
    print(f"Epoch {epoch+1}/{num_epochs}, Average Loss: {avg_loss:.4f}")

# 保存模型权重
model_save_path = "diffusion_model.pth"
torch.save(model.state_dict(), model_save_path)
print(f"模型已保存到: {model_save_path}")

# 生成新样本（反向过程）
@torch.no_grad()
def sample(model, image_size=(1, 28, 28), num_samples=16, temperature=0.8):
    model.eval()
    x = torch.randn(num_samples, *image_size).to(device)
    
    for t in reversed(range(timesteps)):
        t_tensor = torch.full((num_samples,), t, device=device, dtype=torch.long)
        predicted_noise = model(x, t_tensor)
        
        alpha_t = alphas[t]
        beta_t = betas[t]
        sqrt_alpha_t = torch.sqrt(alpha_t)
        sqrt_one_minus_alpha_cumprod_t = sqrt_one_minus_alphas_cumprod[t]
        
        if t > 0:
            noise = torch.randn_like(x) * temperature  # 添加温度参数控制噪声强度
        else:
            noise = torch.zeros_like(x)
            
        x = (1 / sqrt_alpha_t) * (x - beta_t / sqrt_one_minus_alpha_cumprod_t * predicted_noise) + torch.sqrt(beta_t) * noise
    
    x = (x.clamp(-1, 1) + 1) / 2
    return x.cpu()

def post_process_image(image, threshold=0.5, enhance_contrast=True):
    """增强的后处理函数"""
    processed = image.clone()
    
    # 归一化到 [0,1] 范围
    processed = (processed - processed.min()) / (processed.max() - processed.min())
    
    if enhance_contrast:
        # 对比度增强
        mean = processed.mean()
        processed = (processed - mean) * 1.5 + mean  # 增加对比度
        processed = processed.clamp(0, 1)
    
    # 自适应阈值
    local_threshold = processed.mean() + 0.1
    threshold = min(max(threshold, local_threshold), 0.7)
    
    # 二值化
    processed = torch.where(processed > threshold,
                          torch.ones_like(processed),
                          torch.zeros_like(processed))
    
    return processed

# 生成并保存原始样本和处理后的样本
generated_images = sample(model, temperature=0.6)

# 创建一个函数来保存图像网格
def save_image_grid(images, filename, title=None):
    """保存图像网格到文件
    Args:
        images: 图像张量 [N, C, H, W]
        filename: 保存的文件名
        title: 图像标题
    """
    fig, axes = plt.subplots(4, 4, figsize=(8,8))
    if title:
        fig.suptitle(title)
    
    for i, ax in enumerate(axes.flatten()):
        ax.imshow(images[i].squeeze(), cmap='gray')
        ax.axis('off')
    
    plt.savefig(filename, dpi=300, bbox_inches='tight')
    plt.close()

# 保存原始生成图像
save_image_grid(generated_images, 'generated_images_raw.png', 'original generated images')

# 对生成的图像进行后处理
processed_images = torch.stack([post_process_image(img, threshold=0.5) for img in generated_images])

# 保存处理后的图像
save_image_grid(processed_images, 'generated_images_processed.png', 'processed generated images')

print("已保存原始图像到 generated_images_raw.png")
print("已保存处理后图像到 generated_images_processed.png")

# 可选：创建对比图
fig, axes = plt.subplots(2, 1, figsize=(8, 16))
axes[0].imshow(utils.make_grid(generated_images, nrow=4).permute(1, 2, 0), cmap='gray')
axes[0].set_title('before post-processing')
axes[0].axis('off')

axes[1].imshow(utils.make_grid(processed_images, nrow=4).permute(1, 2, 0), cmap='gray')
axes[1].set_title('after post-processing')
axes[1].axis('off')

plt.savefig('comparison.png', dpi=300, bbox_inches='tight')
plt.close()
print("已保存对比图到 comparison.png")

具体解释待补充…