Xóa các đám mây khỏi hình ảnh vệ tinh bằng GAN (Mạng đối thủ tạo) | của Aleksei Rozanov | Tháng 6 năm 2024

Vì bây giờ chúng tôi đã có bộ dữ liệu sẵn sàng để sử dụng, hãy nói về GAN.

Để minh họa rõ hơn ý tưởng này, hãy tưởng tượng rằng bạn đang đi du lịch vòng quanh Đông Nam Á và thấy mình đang cần gấp một chiếc áo hoodie vì bên ngoài trời quá lạnh. Đến khu chợ đường phố gần nhất, bạn tìm thấy một cửa hàng nhỏ bán quần áo hàng hiệu. Người bán mang đến cho bạn một chiếc áo hoodie đẹp để thử nói rằng đó là thương hiệu nổi tiếng ExpensiveButNotWorthIt. Bạn xem xét kỹ hơn và kết luận rằng đó rõ ràng là hàng giả. Người bán nói: ‘Đợi chút, tôi có hàng THẬT. Anh ta quay lại với một chiếc áo hoodie khác, trông giống chiếc áo hàng hiệu hơn, nhưng vẫn là hàng giả. Sau nhiều lần lặp lại như thế này, người bán mang đến một bản sao không thể phân biệt được của ExpensiveButNotWorthIt huyền thoại và bạn sẵn sàng mua nó. Về cơ bản đó là cách GAN hoạt động!

Trong trường hợp GAN, bạn được gọi là người phân biệt đối xử (D). Mục tiêu của người phân biệt đối xử là phân biệt giữa đối tượng thật và đối tượng giả hoặc giải quyết nhiệm vụ phân loại nhị phân. Người bán được gọi là người tạo ra (G), vì anh ta đang cố gắng tạo ra hàng giả chất lượng cao. Người phân biệt đối xử và người tạo ra được đào tạo độc lập để vượt trội hơn lẫn nhau. Do đó, cuối cùng chúng ta nhận được hàng giả chất lượng cao.

Quá trình đào tạo ban đầu trông như thế này:

1. Tiếng ồn đầu vào mẫu (trong trường hợp của chúng tôi là hình ảnh có mây).
2. Đưa tiếng ồn vào G và thu thập dự đoán.
3. Tính tổn thất D bằng cách nhận 2 dự đoán, một cho đầu ra của G và một cho dữ liệu thực.
4. Cập nhật trọng số của D.
5. Lấy mẫu lại tiếng ồn đầu vào.
6. Đưa tiếng ồn vào G và thu thập dự đoán.
7. Tính tổn thất G bằng cách đưa dự đoán của nó cho D.
8. Cập nhật trọng số của G.

Nói cách khác, chúng ta có thể định nghĩa hàm giá trị V(G,D):

nơi chúng tôi muốn giảm thiểu thời hạn log(1-D(G(z))) để huấn luyện G và tối đa hóa log D(x) để huấn luyện D (trong ký hiệu này x – mẫu dữ liệu thực và z – nhiễu).

Bây giờ hãy thử triển khai nó trong pytorch!

Trong bài báo gốc, các tác giả nói về việc sử dụng Perceptron đa lớp (MLP); nó cũng thường được gọi đơn giản là ANN, nhưng tôi muốn thử cách tiếp cận phức tạp hơn một chút – tôi muốn sử dụng kiến ​​trúc UNet (5) làm Trình tạo và ResNet (6) làm Trình phân biệt đối xử. Đây đều là những kiến ​​trúc CNN nổi tiếng, vì vậy tôi sẽ không giải thích chúng ở đây (hãy cho tôi biết liệu tôi có nên viết một bài viết riêng trong phần bình luận hay không).

Hãy xây dựng chúng. Người phân biệt đối xử:

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.useful as F
from torch.utils.information import Dataset, DataLoader
from torchvision import transforms
from torch.utils.information import Subset

class ResidualBlock(nn.Module):
def __init__(self, in_channels, out_channels, stride = 1, downsample = None):
tremendous(ResidualBlock, self).__init__()
self.conv1 = nn.Sequential(
nn.Conv2d(in_channels, out_channels, kernel_size = 3, stride = stride, padding = 1),
nn.BatchNorm2d(out_channels),
nn.ReLU())
self.conv2 = nn.Sequential(
nn.Conv2d(out_channels, out_channels, kernel_size = 3, stride = 1, padding = 1),
nn.BatchNorm2d(out_channels))
self.downsample = downsample
self.relu = nn.ReLU()
self.out_channels = out_channels
def ahead(self, x):
residual = x
out = self.conv1(x)
out = self.conv2(out)
if self.downsample:
residual = self.downsample(x)
out += residual
out = self.relu(out)
return out
class ResNet(nn.Module):
def __init__(self, block=ResidualBlock, all_connections=(3,4,6,3)):
tremendous(ResNet, self).__init__()
self.inputs = 16
self.conv1 = nn.Sequential(
nn.Conv2d(3, 16, kernel_size = 3, stride = 1, padding = 1),
nn.BatchNorm2d(16),
nn.ReLU()) #16x64x64
self.maxpool = nn.MaxPool2d(kernel_size = 2, stride = 2) #16x32x32
self.layer0 = self.makeLayer(block, 16, all_connections(0), stride = 1) #connections = 3, form: 16x32x32
self.layer1 = self.makeLayer(block, 32, all_connections(1), stride = 2)#connections = 4, form: 32x16x16
self.layer2 = self.makeLayer(block, 128, all_connections(2), stride = 2)#connections = 6, form: 1281x8x8
self.layer3 = self.makeLayer(block, 256, all_connections(3), stride = 2)#connections = 3, form: 256x4x4
self.avgpool = nn.AvgPool2d(4, stride=1)
self.fc = nn.Linear(256, 1)
def makeLayer(self, block, outputs, connections, stride=1):
downsample = None
if stride != 1 or self.inputs != outputs:
downsample = nn.Sequential(
nn.Conv2d(self.inputs, outputs, kernel_size=1, stride=stride),
nn.BatchNorm2d(outputs),
)
layers = ()
layers.append(block(self.inputs, outputs, stride, downsample))
self.inputs = outputs
for i in vary(1, connections):
layers.append(block(self.inputs, outputs))
return nn.Sequential(*layers)
def ahead(self, x):
x = self.conv1(x)
x = self.maxpool(x)
x = self.layer0(x)
x = self.layer1(x)
x = self.layer2(x)
x = self.layer3(x)
x = self.avgpool(x)
x = x.view(-1, 256)
x = self.fc(x).flatten()
return F.sigmoid(x)

Máy phát điện:

class DoubleConv(nn.Module):
def __init__(self, in_channels, out_channels):
tremendous(DoubleConv, self).__init__()
self.double_conv = nn.Sequential(
nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=1),
nn.BatchNorm2d(out_channels),
nn.ReLU(inplace=True),
nn.Conv2d(out_channels, out_channels, kernel_size=3, padding=1),
nn.BatchNorm2d(out_channels),
nn.ReLU(inplace=True)
)
def ahead(self, x):
return self.double_conv(x)
class UNet(nn.Module):
def __init__(self):
tremendous().__init__()
self.conv_1 = DoubleConv(3, 32) # 32x64x64
self.pool_1 = nn.MaxPool2d(kernel_size=2, stride=2) # 32x32x32
self.conv_2 = DoubleConv(32, 64)  #64x32x32
self.pool_2 = nn.MaxPool2d(kernel_size=2, stride=2) #64x16x16
self.conv_3 = DoubleConv(64, 128)  #128x16x16
self.pool_3 = nn.MaxPool2d(kernel_size=2, stride=2) #128x8x8
self.conv_4 = DoubleConv(128, 256)  #256x8x8
self.pool_4 = nn.MaxPool2d(kernel_size=2, stride=2) #256x4x4
self.conv_5 = DoubleConv(256, 512)  #512x2x2
#DECODER
self.upconv_1 = nn.ConvTranspose2d(512, 256, kernel_size=2, stride=2) #256x4x4
self.conv_6 = DoubleConv(512, 256) #256x4x4
self.upconv_2 = nn.ConvTranspose2d(256, 128, kernel_size=2, stride=2) #128x8x8
self.conv_7 = DoubleConv(256, 128)  #128x8x8
self.upconv_3 = nn.ConvTranspose2d(128, 64, kernel_size=2, stride=2) #64x16x16
self.conv_8 = DoubleConv(128, 64)  #64x16x16
self.upconv_4 = nn.ConvTranspose2d(64, 32, kernel_size=2, stride=2) #32x32x32
self.conv_9 = DoubleConv(64, 32)  #32x32x32
self.output = nn.Conv2d(32, 3, kernel_size = 3, stride = 1, padding = 1) #3x64x64
def ahead(self, batch):
conv_1_out = self.conv_1(batch)
conv_2_out = self.conv_2(self.pool_1(conv_1_out))
conv_3_out = self.conv_3(self.pool_2(conv_2_out))
conv_4_out = self.conv_4(self.pool_3(conv_3_out))
conv_5_out = self.conv_5(self.pool_4(conv_4_out))
conv_6_out = self.conv_6(torch.cat((self.upconv_1(conv_5_out), conv_4_out), dim=1))
conv_7_out = self.conv_7(torch.cat((self.upconv_2(conv_6_out), conv_3_out), dim=1))
conv_8_out = self.conv_8(torch.cat((self.upconv_3(conv_7_out), conv_2_out), dim=1))
conv_9_out = self.conv_9(torch.cat((self.upconv_4(conv_8_out), conv_1_out), dim=1))
output = self.output(conv_9_out)
return F.sigmoid(output)

Bây giờ chúng ta cần chia dữ liệu của mình thành huấn luyện/kiểm tra và gói chúng vào tập dữ liệu ngọn đuốc:

class dataset(Dataset):
def __init__(self, batch_size, images_paths, targets, img_size = 64):
self.batch_size = batch_size
self.img_size = img_size
self.images_paths = images_paths
self.targets = targets
self.len = len(self.images_paths) // batch_size
self.remodel = transforms.Compose((
transforms.ToTensor(),
))
self.batch_im = (self.images_paths(idx * self.batch_size:(idx + 1) * self.batch_size) for idx in vary(self.len))
self.batch_t = (self.targets(idx * self.batch_size:(idx + 1) * self.batch_size) for idx in vary(self.len))
def __getitem__(self, idx):
pred = torch.stack((
self.remodel(Picture.open(be part of(path_input,file_name)))
for file_name in self.batch_im(idx)
))
goal = torch.stack((
self.remodel(Picture.open(be part of(path_target,file_name)))
for file_name in self.batch_im(idx)
))
return pred, goal
def __len__(self):
return self.len

Hoàn hảo. Đã đến lúc viết vòng lặp đào tạo. Trước khi làm như vậy, hãy xác định các hàm mất mát và trình tối ưu hóa của chúng ta:

system = torch.system("cuda" if torch.cuda.is_available() else "cpu")
batch_size = 64
num_epochs = 15
learning_rate_D = 1e-5
learning_rate_G = 1e-4
discriminator = ResNet()
generator = UNet()
bce = nn.BCEWithLogitsLoss()
l1loss = nn.L1Loss()
optimizer_D = optim.Adam(discriminator.parameters(), lr=learning_rate_D)
optimizer_G = optim.Adam(generator.parameters(), lr=learning_rate_G)
scheduler_D = optim.lr_scheduler.StepLR(optimizer_D, step_size=10, gamma=0.1)
scheduler_G = optim.lr_scheduler.StepLR(optimizer_G, step_size=10, gamma=0.1)

Như bạn có thể thấy, những tổn thất này khác với hình ảnh của thuật toán GAN. Đặc biệt, tôi đã thêm L1Loss. Ý tưởng là chúng tôi không chỉ đơn giản tạo ra một hình ảnh ngẫu nhiên từ nhiễu, mà chúng tôi muốn giữ lại hầu hết thông tin từ đầu vào và chỉ loại bỏ nhiễu. Vậy tổn thất G sẽ là:

G_loss = log(1 − D(G(z))) + 𝝀 |G(z)-y|

Thay vì chỉ

G_loss = log(1 − D(G(z)))

𝝀 là hệ số tùy ý, cân bằng hai thành phần tổn thất.

Cuối cùng, hãy phân chia dữ liệu để bắt đầu quá trình đào tạo:

test_ratio, train_ratio = 0.3, 0.7
num_test = int(len(listdir(path_target))*test_ratio)
num_train = int((int(len(listdir(path_target)))-num_test))
img_size = (64, 64)
print("Variety of practice samples:", num_train)
print("Variety of take a look at samples:", num_test)
random.seed(231)
train_idxs = np.array(random.pattern(vary(num_test+num_train), num_train))
masks = np.ones(num_train+num_test, dtype=bool)
masks(train_idxs) = False
pictures = {}
options = random.pattern(listdir(path_input),num_test+num_train)
targets = random.pattern(listdir(path_target),num_test+num_train)
random.Random(231).shuffle(options)
random.Random(231).shuffle(targets)
train_input_img_paths = np.array(options)(train_idxs)
train_target_img_path = np.array(targets)(train_idxs)
test_input_img_paths = np.array(options)(masks)
test_target_img_path = np.array(targets)(masks)
train_loader = dataset(batch_size=batch_size, img_size=img_size, images_paths=train_input_img_paths, targets=train_target_img_path)
test_loader = dataset(batch_size=batch_size, img_size=img_size, images_paths=test_input_img_paths, targets=test_target_img_path)

Bây giờ chúng ta có thể chạy vòng đào tạo của mình:

train_loss_G, train_loss_D, val_loss_G, val_loss_D = (), (), (), ()
all_loss_G, all_loss_D = (), ()
best_generator_epoch_val_loss, best_discriminator_epoch_val_loss = -np.inf, -np.inf
for epoch in vary(num_epochs):
discriminator.practice()
generator.practice()
discriminator_epoch_loss, generator_epoch_loss = 0, 0
for inputs, targets in train_loader:
inputs, true = inputs, targets
'''1. Coaching the Discriminator (ResNet)'''
optimizer_D.zero_grad()
faux = generator(inputs).detach()
pred_fake = discriminator(faux).to(system)
loss_fake = bce(pred_fake, torch.zeros(batch_size, system=system))
pred_real = discriminator(true).to(system)
loss_real = bce(pred_real, torch.ones(batch_size, system=system))
loss_D = (loss_fake+loss_real)/2
loss_D.backward()
optimizer_D.step()
discriminator_epoch_loss += loss_D.merchandise()
all_loss_D.append(loss_D.merchandise())
'''2. Coaching the Generator (UNet)'''
optimizer_G.zero_grad()
faux = generator(inputs)
pred_fake = discriminator(faux).to(system)
loss_G_bce = bce(pred_fake, torch.ones_like(pred_fake, system=system))
loss_G_l1 = l1loss(faux, targets)*100
loss_G = loss_G_bce + loss_G_l1
loss_G.backward()
optimizer_G.step()
generator_epoch_loss += loss_G.merchandise()
all_loss_G.append(loss_G.merchandise())
discriminator_epoch_loss /= len(train_loader)
generator_epoch_loss /= len(train_loader)
train_loss_D.append(discriminator_epoch_loss)
train_loss_G.append(generator_epoch_loss)
discriminator.eval()
generator.eval()
discriminator_epoch_val_loss, generator_epoch_val_loss = 0, 0
with torch.no_grad():
for inputs, targets in test_loader:
inputs, targets = inputs, targets
faux = generator(inputs)
pred = discriminator(faux).to(system)
loss_G_bce = bce(faux, torch.ones_like(faux, system=system))
loss_G_l1 = l1loss(faux, targets)*100
loss_G = loss_G_bce + loss_G_l1
loss_D = bce(pred.to(system), torch.zeros(batch_size, system=system))
discriminator_epoch_val_loss += loss_D.merchandise()
generator_epoch_val_loss += loss_G.merchandise()
discriminator_epoch_val_loss /= len(test_loader)
generator_epoch_val_loss /= len(test_loader)
val_loss_D.append(discriminator_epoch_val_loss)
val_loss_G.append(generator_epoch_val_loss)
print(f"------Epoch ({epoch+1}/{num_epochs})------nTrain Loss D: {discriminator_epoch_loss:.4f}, Val Loss D: {discriminator_epoch_val_loss:.4f}")
print(f'Practice Loss G: {generator_epoch_loss:.4f}, Val Loss G: {generator_epoch_val_loss:.4f}')
if discriminator_epoch_val_loss > best_discriminator_epoch_val_loss:
discriminator_epoch_val_loss = best_discriminator_epoch_val_loss
torch.save(discriminator.state_dict(), "discriminator.pth")
if generator_epoch_val_loss > best_generator_epoch_val_loss:
generator_epoch_val_loss = best_generator_epoch_val_loss
torch.save(generator.state_dict(), "generator.pth")
#scheduler_D.step()
#scheduler_G.step()
fig, ax = plt.subplots(1,3)
ax(0).imshow(np.transpose(inputs.numpy()(7), (1,2,0)))
ax(1).imshow(np.transpose(targets.numpy()(7), (1,2,0)))
ax(2).imshow(np.transpose(faux.detach().numpy()(7), (1,2,0)))
plt.present()

Sau khi mã hoàn tất, chúng ta có thể vẽ biểu đồ tổn thất. Mã này đã được thông qua một phần từ trang web tuyệt vời này:

from matplotlib.font_manager import FontProperties
background_color = '#001219'
font = FontProperties(fname='LexendDeca-VariableFont_wght.ttf')
fig, ax = plt.subplots(1, 2, figsize=(16, 9))
fig.set_facecolor(background_color)
ax(0).set_facecolor(background_color)
ax(1).set_facecolor(background_color)
ax(0).plot(vary(len(all_loss_G)), all_loss_G, shade='#bc6c25', lw=0.5) 
ax(1).plot(vary(len(all_loss_D)), all_loss_D, shade='#00b4d8', lw=0.5)
ax(0).scatter(
(np.array(all_loss_G).argmax(), np.array(all_loss_G).argmin()),
(np.array(all_loss_G).max(), np.array(all_loss_G).min()),
s=30, shade='#bc6c25',
)
ax(1).scatter(
(np.array(all_loss_D).argmax(), np.array(all_loss_D).argmin()),
(np.array(all_loss_D).max(), np.array(all_loss_D).min()),
s=30, shade='#00b4d8',
)
ax_text(
np.array(all_loss_G).argmax()+60, np.array(all_loss_G).max()+0.1,
f'{spherical(np.array(all_loss_G).max(),1)}',
fontsize=13, shade='#bc6c25',
font=font,
ax=ax(0)
)
ax_text(
np.array(all_loss_G).argmin()+60, np.array(all_loss_G).min()-0.1,
f'{spherical(np.array(all_loss_G).min(),1)}',
fontsize=13, shade='#bc6c25',
font=font,
ax=ax(0)
)
ax_text(
np.array(all_loss_D).argmax()+60, np.array(all_loss_D).max()+0.01,
f'{spherical(np.array(all_loss_D).max(),1)}',
fontsize=13, shade='#00b4d8',
font=font,
ax=ax(1)
)
ax_text(
np.array(all_loss_D).argmin()+60, np.array(all_loss_D).min()-0.005,
f'{spherical(np.array(all_loss_D).min(),1)}',
fontsize=13, shade='#00b4d8',
font=font,
ax=ax(1)
)
for i in vary(2):
ax(i).tick_params(axis='x', colours='white')
ax(i).tick_params(axis='y', colours='white')
ax(i).spines('left').set_color('white') 
ax(i).spines('backside').set_color('white') 
ax(i).set_xlabel('Epoch', shade='white', fontproperties=font, fontsize=13)
ax(i).set_ylabel('Loss', shade='white', fontproperties=font, fontsize=13)
ax(0).set_title('Generator', shade='white', fontproperties=font, fontsize=18)
ax(1).set_title('Discriminator', shade='white', fontproperties=font, fontsize=18)
plt.savefig('Loss.jpg')
plt.present()
# ax(0).set_axis_off()
# ax(1).set_axis_off()