import os import torch import numpy as np import lpips as lp import pandas as pd import torchmetrics import matplotlib.pyplot as plt from bisect import bisect_right import torchvision.transforms as T from torch import nn from matplotlib.colors import ListedColormap, BoundaryNorm from matplotlib.lines import Line2D from data import dutils # ======================================================================= # Scheduler Helper Function # ======================================================================= class SequentialLR(torch.optim.lr_scheduler._LRScheduler): """Receives the list of schedulers that is expected to be called sequentially during optimization process and milestone points that provides exact intervals to reflect which scheduler is supposed to be called at a given epoch. Args: schedulers (list): List of chained schedulers. milestones (list): List of integers that reflects milestone points. Example: >>> # Assuming optimizer uses lr = 1. for all groups >>> # lr = 0.1 if epoch == 0 >>> # lr = 0.1 if epoch == 1 >>> # lr = 0.9 if epoch == 2 >>> # lr = 0.81 if epoch == 3 >>> # lr = 0.729 if epoch == 4 >>> scheduler1 = ConstantLR(self.opt, factor=0.1, total_iters=2) >>> scheduler2 = ExponentialLR(self.opt, gamma=0.9) >>> scheduler = SequentialLR(self.opt, schedulers=[scheduler1, scheduler2], milestones=[2]) >>> for epoch in range(100): >>> train(...) >>> validate(...) >>> scheduler.step() """ def __init__(self, optimizer, schedulers, milestones, last_epoch=-1, verbose=False): for scheduler_idx in range(1, len(schedulers)): if (schedulers[scheduler_idx].optimizer != schedulers[0].optimizer): raise ValueError( "Sequential Schedulers expects all schedulers to belong to the same optimizer, but " "got schedulers at index {} and {} to be different".format(0, scheduler_idx) ) if (len(milestones) != len(schedulers) - 1): raise ValueError( "Sequential Schedulers expects number of schedulers provided to be one more " "than the number of milestone points, but got number of schedulers {} and the " "number of milestones to be equal to {}".format(len(schedulers), len(milestones)) ) self.optimizer = optimizer self._schedulers = schedulers self._milestones = milestones self.last_epoch = last_epoch + 1 def step(self, ref=None): self.last_epoch += 1 idx = bisect_right(self._milestones, self.last_epoch) if idx > 0 and self._milestones[idx - 1] == self.last_epoch: self._schedulers[idx].step(0) else: # Check HERE if isinstance(self._schedulers[idx], torch.optim.lr_scheduler.ReduceLROnPlateau): self._schedulers[idx].step(ref) else: self._schedulers[idx].step() def state_dict(self): """Returns the state of the scheduler as a :class:`dict`. It contains an entry for every variable in self.__dict__ which is not the optimizer. The wrapped scheduler states will also be saved. """ state_dict = {key: value for key, value in self.__dict__.items() if key not in ('optimizer', '_schedulers')} state_dict['_schedulers'] = [None] * len(self._schedulers) for idx, s in enumerate(self._schedulers): state_dict['_schedulers'][idx] = s.state_dict() return state_dict def load_state_dict(self, state_dict): """Loads the schedulers state. Args: state_dict (dict): scheduler state. Should be an object returned from a call to :meth:`state_dict`. """ _schedulers = state_dict.pop('_schedulers') self.__dict__.update(state_dict) # Restore state_dict keys in order to prevent side effects # https://github.com/pytorch/pytorch/issues/32756 state_dict['_schedulers'] = _schedulers for idx, s in enumerate(_schedulers): self._schedulers[idx].load_state_dict(s) def warmup_lambda(warmup_steps, min_lr_ratio=0.1): def ret_lambda(epoch): if epoch <= warmup_steps: return min_lr_ratio + (1.0 - min_lr_ratio) * epoch / warmup_steps else: return 1.0 return ret_lambda # ======================================================================= # Utils in utils :) # ======================================================================= def to_cpu_tensor(*args): ''' Input arbitrary number of array/tensors, each will be converted to CPU torch.Tensor ''' out = [] for tensor in args: if type(tensor) is np.ndarray: tensor = torch.Tensor(tensor) if type(tensor) is torch.Tensor: tensor = tensor.cpu() out.append(tensor) # single value input: return single value output if len(out) == 1: return out[0] return out def merge_leading_dims(tensor, n=2): ''' Merge the first N dimension of a tensor ''' return tensor.reshape((-1, *tensor.shape[n:])) # ======================================================================= # Model Preparation, saving & loading (copied from utils.py) # ======================================================================= def build_model_name(model_type, model_config): ''' Build the model name (without extension) ''' model_name = model_type + '_' for k, v in model_config.items(): model_name += k if type(v) is list or type(v) is tuple: model_name += '-' for i, item in enumerate(v): model_name += (str(item) if type(item) is not bool else '') + ('-' if i < len(v)-1 else '') else: model_name += (('-' + str(v)) if type(v) is not bool else '') model_name += '_' return model_name[:-1] def build_model_path(base_dir, dataset_type, model_type, timestamp=None): if timestamp is None: return os.path.join(base_dir, dataset_type, model_type) elif timestamp == True: return os.path.join(base_dir, dataset_type, model_type, pd.Timestamp.now().strftime('%Y%m%d%H%M%S')) return os.path.join(base_dir, dataset_type, model_type, timestamp) # ======================================================================= # Preprocess Function for Loading HKO-7 dataset # ======================================================================= def hko7_preprocess(x_seq, x_mask, dt_clip, args): resize = args.resize if 'resize' in args else x_seq.shape[-1] seq_len = args.seq_len if 'seq_len' in args else 5 # post-process on HKO-10 x_seq = x_seq.transpose((1, 0, 2, 3, 4)) / 255. # => (batch_size, seq_length, 1, 480, 480) if 'scale' in args and args.scale == 'non-linear': x_seq = dutils.linear_to_nonlinear_batched(x_seq, dt_clip) else: x_seq = dutils.nonlinear_to_linear_batched(x_seq, dt_clip) b, t, c, h, w = x_seq.shape assert c == 1, f'# channels ({c}) != 1' # resize (downsample) the images if necessary x_seq = torch.Tensor(x_seq).float().reshape((b*t, c, h, w)) if resize != h: tform = T.Compose([ T.ToPILImage(), T.Resize(resize), T.ToTensor(), ]) else: tform = T.Compose([]) x_seq = torch.stack([tform(x_frame) for x_frame in x_seq], dim=0) x_seq = x_seq.reshape((b, t, c, resize, resize)) x, y = x_seq[:, :seq_len], x_seq[:, seq_len:] return x, y # ======================================================================= # Evaluation Metrics-Related # ======================================================================= mae = lambda *args: torch.nn.functional.l1_loss(*args).cpu().detach().numpy() mse = lambda *args: torch.nn.functional.mse_loss(*args).cpu().detach().numpy() def ssim(y_pred, y): y, y_pred = to_cpu_tensor(y, y_pred) b, t, c, h, w = y.shape y = y.reshape((b*t, c, h, w)) y_pred = y_pred.reshape((b*t, c, h, w)) # to further ensure any of the input is not negative y = torch.clamp(y, 0, 1) y_pred = torch.clamp(y_pred, 0, 1) return torchmetrics.image.StructuralSimilarityIndexMeasure(data_range=1.0)(y_pred, y) def psnr(y_pred, y): y, y_pred = to_cpu_tensor(y, y_pred) b, t, c, h, w = y.shape y = y.reshape((b*t, c, h, w)) y_pred = y_pred.reshape((b*t, c, h, w)) acc_score = 0 for i in range(b*t): acc_score += torchmetrics.image.PeakSignalNoiseRatio(data_range=1.0)(y_pred[i], y[i]) / (b*t) return acc_score GLOBAL_LPIPS_OBJ = None # a static variable def lpips64(y_pred, y, net='vgg'): # convert the image range into [-1, 1], assuming the input range to be [0, 1] y = merge_leading_dims(y) y_pred = merge_leading_dims(y_pred) y = torch.nn.functional.interpolate(y, (64, 64), mode='bicubic').clamp(0,1) y_pred = torch.nn.functional.interpolate(y_pred, (64, 64), mode='bicubic').clamp(0,1) y = (2 * y - 1) y_pred = (2 * y_pred - 1) global GLOBAL_LPIPS_OBJ if GLOBAL_LPIPS_OBJ is None: GLOBAL_LPIPS_OBJ = lp.LPIPS(net=net).to(y.device) return GLOBAL_LPIPS_OBJ(y_pred, y).mean() def tfpn(y_pred, y, threshold, radius=1): ''' convert to cpu, and merge the first two dimensions ''' y = merge_leading_dims(y) y_pred = merge_leading_dims(y_pred) with torch.no_grad(): if radius > 1: pool = nn.MaxPool2d(radius) y = pool(y) y_pred = pool(y_pred) y = torch.where(y >= threshold, 1, 0) y_pred = torch.where(y_pred >= threshold, 1, 0) mat = torchmetrics.functional.confusion_matrix(y_pred, y, task='binary', threshold=threshold) (tn, fp), (fn, tp) = to_cpu_tensor(mat) return tp, tn, fp, fn def tfpn_pool(y_pred, y, threshold, radius): y_pred = merge_leading_dims(y_pred) y = merge_leading_dims(y) pool = nn.MaxPool2d(radius, stride=radius//4 if radius//4 > 0 else radius) with torch.no_grad(): y = torch.where(y>=threshold, 1, 0).float() y_pred = torch.where(y_pred>=threshold, 1, 0).float() y = pool(y) y_pred = pool(y_pred) mat = torchmetrics.functional.confusion_matrix(y_pred, y, task='binary', threshold=threshold) (tn, fp), (fn, tp) = to_cpu_tensor(mat) return tp, tn, fp, fn def csi(tp, tn, fp, fn): '''Critical Success Index. The larger the better.''' if (tp + fn + fp) < 1e-7: return 0. return tp / (tp + fn + fp) def hss(tp, tn, fp, fn): '''Heidke Skill Score. (-inf, 1]. Larger better.''' if (tp+fn)*(fn+tn) + (tp+fp)*(fp+tn) == 0: return 0. return 2 * (tp*tn - fp*fn) / ((tp+fn)*(fn+tn) + (tp+fp)*(fp+tn)) # ======================================================================= # Data Visualization # ======================================================================= def torch_visualize(sequences, savedir=None, horizontal=10, vmin=0, vmax=1): ''' input: sequences, a list/dict of numpy/torch arrays with shape (B, T, C, H, W) C is assumed to be 1 and squeezed If batch > 1, only the first sequence will be printed ''' # First pass: compute the vertical height and convert to proper format vertical = 0 display_texts = [] if (type(sequences) is dict): temp = [] for k, v in sequences.items(): vertical += int(np.ceil(v.shape[1] / horizontal)) temp.append(v) display_texts.append(k) sequences = temp else: for i, sequence in enumerate(sequences): vertical += int(np.ceil(sequence.shape[1] / horizontal)) display_texts.append(f'Item {i+1}') sequences = to_cpu_tensor(*sequences) # Plot the sequences j = 0 fig, axes = plt.subplots(vertical, horizontal, figsize=(2*horizontal, 2*vertical), tight_layout=True) plt.setp(axes, xticks=[], yticks=[]) for k, sequence in enumerate(sequences): # only take the first batch, now seq[0] is the temporal dim sequence = sequence[0].squeeze() # (T, H, W) axes[j, 0].set_ylabel(display_texts[k]) for i, frame in enumerate(sequence): j_shift = j + i // horizontal i_shift = i % horizontal axes[j_shift, i_shift].imshow(frame, vmin=vmin, vmax=vmax, cmap='gray') j += int(np.ceil(sequence.shape[0] / horizontal)) if savedir: plt.savefig(savedir + '' if savedir.endswith('.png') else '.png') plt.close() else: plt.show() """ Visualize function with colorbar and a line seprate input and output """ def color_visualize(sequences, savedir='', horizontal=5, skip=1, ypos=0): ''' input: sequences, a list/dict of numpy/torch arrays with shape (B, T, C, H, W) C is assumed to be 1 and squeezed If batch > 1, only the first sequence will be printed ''' plt.style.use(['science', 'no-latex']) VIL_COLORS = [[0, 0, 0], [0.30196078431372547, 0.30196078431372547, 0.30196078431372547], [0.1568627450980392, 0.7450980392156863, 0.1568627450980392], [0.09803921568627451, 0.5882352941176471, 0.09803921568627451], [0.0392156862745098, 0.4117647058823529, 0.0392156862745098], [0.0392156862745098, 0.29411764705882354, 0.0392156862745098], [0.9607843137254902, 0.9607843137254902, 0.0], [0.9294117647058824, 0.6745098039215687, 0.0], [0.9411764705882353, 0.43137254901960786, 0.0], [0.6274509803921569, 0.0, 0.0], [0.9058823529411765, 0.0, 1.0]] VIL_LEVELS = [0.0, 16.0, 31.0, 59.0, 74.0, 100.0, 133.0, 160.0, 181.0, 219.0, 255.0] # First pass: compute the vertical height and convert to proper format vertical = 0 display_texts = [] if (type(sequences) is dict): temp = [] for k, v in sequences.items(): vertical += int(np.ceil(v.shape[1] / horizontal)) temp.append(v) display_texts.append(k) sequences = temp else: for i, sequence in enumerate(sequences): vertical += int(np.ceil(sequence.shape[1] / horizontal)) display_texts.append(f'Item {i+1}') sequences = to_cpu_tensor(*sequences) # Plot the sequences j = 0 fig, axes = plt.subplots(vertical, horizontal, figsize=(2*horizontal, 2*vertical), tight_layout=True) plt.subplots_adjust(hspace=0.0, wspace=0.0) # tight layout plt.setp(axes, xticks=[], yticks=[]) for k, sequence in enumerate(sequences): # only take the first batch, now seq[0] is the temporal dim sequence = sequence[0].squeeze() # (T, H, W) ## ================= # = labels of time = if k == 0: for i in range(len(sequence)): axes[j, i].set_xlabel(f'$t-{(len(sequence)-i)-1}$', fontsize=16) axes[j, i].xaxis.set_label_position('top') elif k == len(sequences)-1: for i in range(len(sequence)): axes[j, i].set_xlabel(f'$t+{skip*i+1}$', fontsize=16) axes[j, i].xaxis.set_label_position('bottom') ## ================= axes[j, 0].set_ylabel(display_texts[k], fontsize=16) for i, frame in enumerate(sequence): j_shift = j + i // horizontal i_shift = i % horizontal im = axes[j_shift, i_shift].imshow(frame*255, cmap=ListedColormap(VIL_COLORS), \ norm=BoundaryNorm(VIL_LEVELS, ListedColormap(VIL_COLORS).N)) j += int(np.ceil(sequence.shape[0] / horizontal)) ## = plot splittin line = if ypos == 0: ypos = 1 - 1 / len(sequences) - 0.017 fig.lines.append(Line2D((0, 1), (ypos, ypos), transform=fig.transFigure, ls='--', linewidth=2, color='#444')) # color bar cax = fig.add_axes([1, 0.05, 0.02, 0.5]) fig.colorbar(im, cax=cax) ## ================= if savedir: plt.savefig(savedir + '' if len(savedir)>0 else 'out.png') plt.close() else: plt.show()