| 1360 | |
| 1361 | |
| 1362 | class LatentUpscaleDiffusion(LatentDiffusion): |
| 1363 | def __init__(self, *args, low_scale_config, low_scale_key="LR", noise_level_key=None, **kwargs): |
| 1364 | super().__init__(*args, **kwargs) |
| 1365 | # assumes that neither the cond_stage nor the low_scale_model contain trainable params |
| 1366 | assert not self.cond_stage_trainable |
| 1367 | self.instantiate_low_stage(low_scale_config) |
| 1368 | self.low_scale_key = low_scale_key |
| 1369 | self.noise_level_key = noise_level_key |
| 1370 | |
| 1371 | def instantiate_low_stage(self, config): |
| 1372 | model = instantiate_from_config(config) |
| 1373 | self.low_scale_model = model.eval() |
| 1374 | self.low_scale_model.train = disabled_train |
| 1375 | for param in self.low_scale_model.parameters(): |
| 1376 | param.requires_grad = False |
| 1377 | |
| 1378 | @torch.no_grad() |
| 1379 | def get_input(self, batch, k, cond_key=None, bs=None, log_mode=False): |
| 1380 | if not log_mode: |
| 1381 | z, c = super().get_input(batch, k, force_c_encode=True, bs=bs) |
| 1382 | else: |
| 1383 | z, c, x, xrec, xc = super().get_input(batch, self.first_stage_key, return_first_stage_outputs=True, |
| 1384 | force_c_encode=True, return_original_cond=True, bs=bs) |
| 1385 | x_low = batch[self.low_scale_key][:bs] |
| 1386 | x_low = rearrange(x_low, 'b h w c -> b c h w') |
| 1387 | x_low = x_low.to(memory_format=torch.contiguous_format).float() |
| 1388 | zx, noise_level = self.low_scale_model(x_low) |
| 1389 | if self.noise_level_key is not None: |
| 1390 | # get noise level from batch instead, e.g. when extracting a custom noise level for bsr |
| 1391 | raise NotImplementedError('TODO') |
| 1392 | |
| 1393 | all_conds = {"c_concat": [zx], "c_crossattn": [c], "c_adm": noise_level} |
| 1394 | if log_mode: |
| 1395 | # TODO: maybe disable if too expensive |
| 1396 | x_low_rec = self.low_scale_model.decode(zx) |
| 1397 | return z, all_conds, x, xrec, xc, x_low, x_low_rec, noise_level |
| 1398 | return z, all_conds |
| 1399 | |
| 1400 | @torch.no_grad() |
| 1401 | def log_images(self, batch, N=8, n_row=4, sample=True, ddim_steps=200, ddim_eta=1., return_keys=None, |
| 1402 | plot_denoise_rows=False, plot_progressive_rows=True, plot_diffusion_rows=True, |
| 1403 | unconditional_guidance_scale=1., unconditional_guidance_label=None, use_ema_scope=True, |
| 1404 | **kwargs): |
| 1405 | ema_scope = self.ema_scope if use_ema_scope else nullcontext |
| 1406 | use_ddim = ddim_steps is not None |
| 1407 | |
| 1408 | log = dict() |
| 1409 | z, c, x, xrec, xc, x_low, x_low_rec, noise_level = self.get_input(batch, self.first_stage_key, bs=N, |
| 1410 | log_mode=True) |
| 1411 | N = min(x.shape[0], N) |
| 1412 | n_row = min(x.shape[0], n_row) |
| 1413 | log["inputs"] = x |
| 1414 | log["reconstruction"] = xrec |
| 1415 | log["x_lr"] = x_low |
| 1416 | log[f"x_lr_rec_@noise_levels{'-'.join(map(lambda x: str(x), list(noise_level.cpu().numpy())))}"] = x_low_rec |
| 1417 | if self.model.conditioning_key is not None: |
| 1418 | if hasattr(self.cond_stage_model, "decode"): |
| 1419 | xc = self.cond_stage_model.decode(c) |
nothing calls this directly
no outgoing calls
no test coverage detected