Spaces:
Running
Running
| import torch | |
| import math | |
| import torch.nn as nn | |
| import torch.nn.functional as F | |
| from src.model.laplacian import LapLoss | |
| class CharbonnierLoss(nn.Module): | |
| def __init__(self, eps: float = 1e-6): | |
| super(CharbonnierLoss, self).__init__() | |
| self.eps = eps | |
| def forward(self, pred: torch.Tensor, gt: torch.Tensor) -> torch.Tensor: | |
| return torch.mean(torch.sqrt((pred - gt) ** 2 + self.eps ** 2)) | |
| class SSIMLoss(nn.Module): | |
| """Differentiable SSIM for PyTorch training to preserve cloud structures.""" | |
| def __init__(self, window_size: int = 11, size_average: bool = True): | |
| super(SSIMLoss, self).__init__() | |
| self.window_size = window_size | |
| self.size_average = size_average | |
| self.channel = 1 | |
| self.window = self.create_window(window_size, self.channel) | |
| def gaussian(self, window_size, sigma): | |
| gauss = torch.Tensor([math.exp(-(x - window_size//2)**2/float(2*sigma**2)) for x in range(window_size)]) | |
| return gauss/gauss.sum() | |
| def create_window(self, window_size, channel): | |
| import math | |
| _1D_window = self.gaussian(window_size, 1.5).unsqueeze(1) | |
| _2D_window = _1D_window.mm(_1D_window.t()).float().unsqueeze(0).unsqueeze(0) | |
| window = _2D_window.expand(channel, 1, window_size, window_size).contiguous() | |
| return window | |
| def forward(self, img1, img2): | |
| # A simple SSIM computation | |
| device = img1.device | |
| window = self.window.to(device) | |
| mu1 = F.conv2d(img1, window, padding=self.window_size//2, groups=self.channel) | |
| mu2 = F.conv2d(img2, window, padding=self.window_size//2, groups=self.channel) | |
| mu1_sq = mu1.pow(2) | |
| mu2_sq = mu2.pow(2) | |
| mu1_mu2 = mu1 * mu2 | |
| sigma1_sq = F.conv2d(img1*img1, window, padding=self.window_size//2, groups=self.channel) - mu1_sq | |
| sigma2_sq = F.conv2d(img2*img2, window, padding=self.window_size//2, groups=self.channel) - mu2_sq | |
| sigma12 = F.conv2d(img1*img2, window, padding=self.window_size//2, groups=self.channel) - mu1_mu2 | |
| C1 = 0.01**2 | |
| C2 = 0.03**2 | |
| ssim_map = ((2 * mu1_mu2 + C1) * (2 * sigma12 + C2)) / ((mu1_sq + mu2_sq + C1) * (sigma1_sq + sigma2_sq + C2)) | |
| if self.size_average: | |
| return 1 - ssim_map.mean() # Return 1 - SSIM to minimize it as a loss | |
| else: | |
| return 1 - ssim_map.mean(1).mean(1).mean(1) | |
| class CompositeLoss(nn.Module): | |
| """ | |
| SOTA Loss Strategy for Meteorological Data: | |
| L = (alpha * Charbonnier) + (beta * SSIM) + (gamma * Distillation/Laplacian) | |
| """ | |
| def __init__(self, alpha: float = 1.0, beta: float = 0.5, gamma: float = 0.5, channels: int = 1): | |
| super(CompositeLoss, self).__init__() | |
| self.charbonnier = CharbonnierLoss() | |
| self.ssim_loss = SSIMLoss() | |
| self.laplacian = LapLoss(channels=channels) # Acting as our structural distillation/refinement | |
| self.alpha = alpha | |
| self.beta = beta | |
| self.gamma = gamma | |
| def forward(self, pred: torch.Tensor, gt: torch.Tensor) -> tuple[torch.Tensor, dict]: | |
| loss_char = self.charbonnier(pred, gt) | |
| loss_ssim = self.ssim_loss(pred, gt) | |
| loss_lap = self.laplacian(pred, gt).mean() | |
| # Total Equation | |
| total_loss = (self.alpha * loss_char) + (self.beta * loss_ssim) + (self.gamma * loss_lap) | |
| # Returning dictionary of individual losses for easy logging in your training loop | |
| loss_dict = { | |
| "total": total_loss, | |
| "charbonnier": loss_char.item(), | |
| "ssim": loss_ssim.item(), | |
| "laplacian": loss_lap.item() | |
| } | |
| return total_loss, loss_dict |