File size: 13,365 Bytes
ec0fdfd
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
import torch
import torch.nn.functional as F
from .base_model import BaseModel
from . import networks, losses


class TC(BaseModel):
    """This class implements the transformer for image completion"""
    def name(self):
        return "Transformer Image Completion"

    @staticmethod
    def modify_options(parser, is_train=True):
        """Add new options and rewrite default values for existing options"""

        parser.add_argument('--coarse_or_refine', type=str, default='refine', help='train the transform or refined network')
        parser.add_argument('--down_layers', type=int, default=4, help='# times down sampling for refine generator')
        parser.add_argument('--mid_layers', type=int, default=6, help='# times middle layers for refine generator')
        if is_train:
            parser.add_argument('--lambda_rec', type=float, default=10.0, help='weight for image reconstruction loss')
            parser.add_argument('--lambda_g', type=float, default=1.0, help='weight for discriminator loss')
            parser.add_argument('--lambda_lp', type=float, default=10.0, help='weight for the perceptual loss')
            parser.add_argument('--lambda_gradient', type=float, default=0.0, help='weight for the gradient penalty')

        return parser

    def __init__(self, opt):
        """inital the Transformer model"""
        BaseModel.__init__(self, opt)
        self.visual_names = ['img_org', 'img_m', 'img_g', 'img_out']
        self.model_names = ['E', 'G', 'D', 'T']
        self.loss_names = ['G_rec', 'G_lp', 'G_GAN', 'D_real', 'D_fake']

        self.netE = networks.define_E(opt)
        self.netT = networks.define_T(opt)
        self.netG = networks.define_G(opt)
        self.netD = networks.define_D(opt, opt.fixed_size)

        if 'refine' in self.opt.coarse_or_refine:
            opt = self._refine_opt(opt)
            self.netG_Ref = networks.define_G(opt)
            self.netD_Ref = networks.define_D(opt, opt.fine_size)
            self.visual_names += ['img_ref', 'img_ref_out']
            self.model_names += ['G_Ref', 'D_Ref']

        if self.isTrain:
            # define the loss function
            self.L1loss = torch.nn.L1Loss()
            self.GANloss = losses.GANLoss(opt.gan_mode).to(self.device)
            self.NormalVGG = losses.Normalization(self.device)
            self.LPIPSloss = losses.LPIPSLoss(ckpt_path=opt.lipip_path).to(self.device)
            if len(self.opt.gpu_ids) > 0:
                self.LPIPSloss = torch.nn.parallel.DataParallel(self.LPIPSloss, self.opt.gpu_ids)
            # define the optimizer
            if 'coarse' in self.opt.coarse_or_refine:
                self.optimizerG = torch.optim.Adam(list(self.netE.parameters()) + list(self.netG.parameters())
                                                   + list(self.netT.parameters()), lr=opt.lr, betas=(opt.beta1, opt.beta2))
                self.optimizerD = torch.optim.Adam(self.netD.parameters(), lr=opt.lr * 4, betas=(opt.beta1, opt.beta2))
                self.optimizers.append(self.optimizerG)
                self.optimizers.append(self.optimizerD)
            if 'refine' in self.opt.coarse_or_refine:
                self.optimizerGRef = torch.optim.Adam(self.netG_Ref.parameters(), lr=opt.lr, betas=(opt.beta1, opt.beta2))
                self.optimizerDRef = torch.optim.Adam(self.netD_Ref.parameters(), lr=opt.lr * 4, betas=(opt.beta1, opt.beta2))
                self.optimizers.append(self.optimizerGRef)
                self.optimizers.append(self.optimizerDRef)
        else:
            self.visual_names = ['img_org', 'img_m', 'img_out']
            if 'refine' in self.opt.coarse_or_refine:
                self.visual_names += ['img_ref_out']

    def set_input(self, input):
        """Unpack input data from the data loader and perform necessary pre-process steps"""
        self.input = input

        self.image_paths = self.input['img_path']
        self.img_org = input['img_org'].to(self.device) * 2 - 1
        self.img = input['img'].to(self.device) * 2 - 1
        self.mask = input['mask'].to(self.device)

        # get I_m and I_c for image with mask and complement regions for training
        self.img_m = self.mask * self.img_org

    @torch.no_grad()
    def test(self):
        """Run forward processing for testing"""
        fixed_img = F.interpolate(self.img_m, size=[self.opt.fixed_size, self.opt.fixed_size], mode='bicubic', align_corners=True).clamp(-1, 1)
        fixed_mask = (F.interpolate(self.mask, size=[self.opt.fixed_size, self.opt.fixed_size], mode='bicubic', align_corners=True) > 0.9).type_as(fixed_img)
        out, mask = self.netE(fixed_img, mask=fixed_mask, return_mask=True)
        out = self.netT(out, mask, bool_mask=False)

        # sample result
        for i in range(self.opt.nsampling):
            img_g = self.netG(out, mask=self.mask)
            img_g_org = F.interpolate(img_g, size=self.img_org.size()[2:], mode='bicubic', align_corners=True).clamp(-1, 1)
            self.img_out = self.mask * self.img_org + (1 - self.mask) * img_g_org
            # save for multiple results
            self.save_results(self.img_out, path=self.opt.save_dir + '/img_out', data_name=i)
            if 'refine' in self.opt.coarse_or_refine:
                img_ref = self.netG_Ref(self.img_out, mask=self.mask)
                self.img_ref_out = self.mask * self.img_org + (1 - self.mask) * img_ref
                # save for multiple results
                self.save_results(self.img_ref_out, path=self.opt.save_dir + '/img_ref_out', data_name=i)

    def forward(self):
        """Run forward processing to get the outputs"""
        fixed_img = F.interpolate(self.img_m, size=[self.opt.fixed_size, self.opt.fixed_size], mode='bicubic', align_corners=True).clamp(-1, 1)
        self.fixed_mask = (F.interpolate(self.mask, size=[self.opt.fixed_size, self.opt.fixed_size], mode='bicubic', align_corners=True) > 0.9).type_as(fixed_img)
        out, mask = self.netE(fixed_img, mask=self.fixed_mask, return_mask=True)
        out = self.netT(out, mask, bool_mask=False)
        self.img_g = self.netG(out, mask=self.mask)
        img_g_org = F.interpolate(self.img_g, size=self.img_org.size()[2:], mode='bicubic', align_corners=True).clamp(-1, 1)
        self.img_out = self.mask * self.img_org + (1 - self.mask) * img_g_org

        if 'refine' in self.opt.coarse_or_refine:
            self.img_ref = self.netG_Ref(self.img_out, self.mask)
            self.img_ref_out = self.mask * self.img_org + (1 - self.mask) * self.img_ref

    def backward_D_basic(self, netD, real, fake):
        """
        Calculate GAN loss for the discriminator
        :param netD: the discriminator D
        :param real: real examples
        :param fake: examples generated by a generator
        :return: discriminator loss
        """
        self.loss_D_real = self.GANloss(netD(real), True, is_dis=True)
        self.loss_D_fake = self.GANloss(netD(fake), False, is_dis=True)
        loss_D = self.loss_D_real + self.loss_D_fake
        if self.opt.lambda_gradient > 0:
            self.loss_D_Gradient, _ = losses.cal_gradient_penalty(netD, real, fake, real.device, lambda_gp=self.opt.lambda_gradient)
            loss_D += self.loss_D_Gradient
        loss_D.backward()
        return loss_D

    def backward_D(self):
        """Calculate the GAN loss for discriminator"""
        self.loss_D = 0
        if 'coarse' in self.opt.coarse_or_refine:
            self.set_requires_grad([self.netD], True)
            self.optimizerD.zero_grad()
            real = self.img.detach()
            fake = self.img_g.detach()
            self.loss_D += self.backward_D_basic(self.netD, real, fake) if self.opt.lambda_g > 0 else 0
        if 'refine' in self.opt.coarse_or_refine:
            self.set_requires_grad([self.netD_Ref], True)
            self.optimizerDRef.zero_grad()
            real = self.img_org.detach()
            fake = self.img_ref.detach()
            self.loss_D += self.backward_D_basic(self.netD_Ref, real, fake) if self.opt.lambda_g > 0 else 0

    def backward_G(self):
        """Calculate the loss for generator"""
        self.loss_G_GAN = 0
        self.loss_G_rec = 0
        self.loss_G_lp =0
        if 'coarse' in self.opt.coarse_or_refine:
            self.set_requires_grad([self.netD], False)
            self.optimizerG.zero_grad()
            self.loss_G_GAN += self.GANloss(self.netD(self.img_g), True) * self.opt.lambda_g if self.opt.lambda_g > 0 else 0
            self.loss_G_rec += (self.L1loss(self.img_g * (1 - self.fixed_mask), self.img * (1 - self.fixed_mask)) * 3 +
                                self.L1loss(self.img_g * self.fixed_mask, self.img_g * self.fixed_mask)) * self.opt.lambda_rec
            norm_real = self.NormalVGG((self.img + 1) * 0.5)
            norm_fake = self.NormalVGG((self.img_g + 1) * 0.5)
            self.loss_G_lp += (self.LPIPSloss(norm_real, norm_fake).mean()) * self.opt.lambda_lp if self.opt.lambda_lp > 0 else 0
        if 'refine' in self.opt.coarse_or_refine:
            self.set_requires_grad([self.netD_Ref], False)
            self.optimizerGRef.zero_grad()
            self.loss_G_GAN += self.GANloss(self.netD_Ref(self.img_ref), True) * self.opt.lambda_g if self.opt.lambda_g > 0 else 0
            self.loss_G_rec += (self.L1loss(self.img_ref * (1 - self.mask), self.img_org * (1 - self.mask)) * 3 +
                                self.L1loss(self.img_ref * self.mask, self.img_org * self.mask)) * self.opt.lambda_rec
            norm_real = self.NormalVGG((self.img_org + 1) * 0.5)
            norm_fake = self.NormalVGG((self.img_ref + 1) * 0.5)
            self.loss_G_lp += (self.LPIPSloss(norm_real, norm_fake).mean()) * self.opt.lambda_lp if self.opt.lambda_lp > 0 else 0

        self.loss_G = self.loss_G_GAN + self.loss_G_rec + self.loss_G_lp

        self.loss_G.backward()

    def optimize_parameters(self):
        """update network weights"""
        # forward
        self.set_requires_grad([self.netE, self.netT, self.netG], 'coarse' in self.opt.coarse_or_refine)
        self.forward()
        # update D
        self.backward_D()
        if 'coarse' in self.opt.coarse_or_refine:
            self.optimizerD.step()
        if 'refine' in self.opt.coarse_or_refine:
            self.optimizerDRef.step()
        # update G
        self.backward_G()
        if 'coarse' in self.opt.coarse_or_refine:
            self.optimizerG.step()
        if 'refine' in self.opt.coarse_or_refine:
            self.optimizerGRef.step()

    def configure_optimizers(self):
        """
        Following minGPT:
        This long function is unfortunately doing something very simple and is being very defensive:
        We are separating out all parameters of the model into two buckets: those that will experience
        weight decay for regularization and those that won't (biases, and layernorm/embedding weights).
        We are then returning the PyTorch optimizer object.
        """
        # separate out all parameters to those that will and won't experience regularizing weight decay
        decay = set()
        no_decay = set()
        whitelist_weight_modules = (torch.nn.Linear, torch.nn.Conv2d)
        blacklist_weight_modules = (torch.nn.LayerNorm, torch.nn.Embedding)
        for mn, m in self.netT.named_modules():
            for pn, p in m.named_parameters():
                fpn = '%s.%s' % (mn, pn) if mn else pn  # full param name

                if pn.endswith('bias') or pn.endswith('alpha'):
                    # all biases will not be decayed
                    no_decay.add(fpn)
                elif pn.endswith('weight') and isinstance(m, whitelist_weight_modules):
                    # weights of whitelist modules will be weight decayed
                    decay.add(fpn)
                elif pn.endswith('weight') and isinstance(m, blacklist_weight_modules):
                    # weights of blacklist modules will NOT be weight decayed
                    no_decay.add(fpn)

        # validate that we considered every parameter
        param_dict = {pn: p for pn, p in self.netT.named_parameters()}
        inter_params = decay & no_decay
        union_params = decay | no_decay
        assert len(inter_params) == 0, "parameters %s made it into both decay/no_decay sets!" % (str(inter_params),)
        assert len(param_dict.keys() - union_params) == 0, "parameters %s were not separated into either decay/no_decay set!" \
                                                                % (str(param_dict.keys() - union_params),)

        # create the pytorch optimizer object
        optim_groups = [
            {"params": [param_dict[pn] for pn in sorted(list(decay))], "weight_decay": 0.01, "betas":(0.9, 0.95)},
            {"params": [param_dict[pn] for pn in sorted(list(no_decay))], "weight_decay": 0.0, "betas":(0.9, 0.95)},
            {"params": list(filter(lambda p: p.requires_grad, self.netE.parameters()))},
            {"params": list(filter(lambda p: p.requires_grad, self.netG.parameters()))}
        ]
        optimizer = torch.optim.Adam(optim_groups, lr=self.opt.lr, betas=(self.opt.beta1, self.opt.beta2))
        return optimizer

    def _refine_opt(self, opt):
        """modify the opt for refine generator and discriminator"""
        opt.netG = 'refine'
        opt.netD = 'style'

        return opt