File size: 15,062 Bytes

f1d3656

####################################################################################################
#
#  Copyright (C) 2022
#
####################################################################################################
#
#  project     : atmorep
#
#  author      : atmorep collaboration
# 
#  description :
#
#  license     :
#
####################################################################################################

import datetime
import json
import os
from pathlib import Path
from enum import Enum
import wandb
import code
from calendar import monthrange
#import properscoring as ps
import numpy as np

import torch.distributed as dist
import torch.utils.data.distributed

import pandas as pd

import atmorep.config.config as config
from atmorep.utils.logger import logger

####################################################################################################
class NetMode( Enum) :
  indeterminate = 0
  train = 1
  test = 2

####################################################################################################
# Helper function to be able to json dump configs with classes 
# in which case the class name is dumped
# Note that loading a config then will lead to problems/not be possible
def json_default(o):
  if type(o) == type :
    return o.__name__
  else :
    return o.to_json()

####################################################################################################
class Config :

  def __init__( self) :
    pass

  def add_to_wandb( self, wandb) :
    wandb.config.update( self.__dict__)

  def get_self_dict( self) :
    return self.__dict__

  def print( self) :
    self_dict = self.__dict__
    for key, value in self_dict.items() : 
        print("{} : {}".format( key, value))

  def create_dirs( self, wandb) :
    dirname = Path( config.path_results, 'models/id{}'.format( wandb.run.id))
    if not os.path.exists(dirname):
      os.makedirs( dirname)
      
    dirname = Path( config.path_results, 'id{}'.format( wandb.run.id))
    if not os.path.exists(dirname):
      os.makedirs( dirname)
      
  def write_json( self, wandb) :

    if not hasattr( wandb.run, 'id') :
      return

    json_str = json.dumps(self.__dict__ )

    # save in directory with model files
    dirname = Path( config.path_results, 'models/id{}'.format( wandb.run.id))
    if not os.path.exists(dirname):
      os.makedirs( dirname)
    fname =Path(config.path_results,'models/id{}/model_id{}.json'.format(wandb.run.id,wandb.run.id))
    with open(fname, 'w') as f :
      f.write( json_str)

    # also save in results directory
    dirname = Path( config.path_results,'id{}'.format( wandb.run.id))
    if not os.path.exists(dirname):
      os.makedirs( dirname)
    fname = Path( dirname, 'model_id{}.json'.format( wandb.run.id))
    with open(fname, 'w') as f :
      f.write( json_str)

  def load_json( self, wandb_id) :
    if '/' in wandb_id :   # assumed to be full path instead of just id
      fname = wandb_id
    else :
      fname = Path( config.path_models, 'id{}/model_id{}.json'.format( wandb_id, wandb_id))
    try :
      with open(fname, 'r') as f :
        json_str = f.readlines() 
    except (OSError, IOError) as e:
      # try path used for logging training results and checkpoints
      try :
        fname = Path( config.path_results, '/models/id{}/model_id{}.json'.format(wandb_id,wandb_id))
        with open(fname, 'r') as f :
          json_str = f.readlines()
      except (OSError, IOError) as e:
        print( f'Could not find fname due to {e}. Aborting.')
        quit()

    self.__dict__ = json.loads( json_str[0])

    # fix for backward compatibility
    if not hasattr( self, 'model_id') :
      self.model_id = self.wandb_id

    return self

####################################################################################################
def identity( func, *args) :
  return func( *args)

####################################################################################################

def str_to_tensor(modelid):
    return torch.tensor([ord(c) for c in modelid], dtype=torch.int32)

def tensor_to_str(tensor):
    return ''.join([chr(x) for x in tensor])

####################################################################################################
def init_torch() :
  
  torch.set_printoptions( linewidth=120)

  use_cuda = torch.cuda.is_available()
  if not use_cuda :
    return torch.device( 'cpu')

  num_accs_per_task = torch.cuda.device_count()
  if num_accs_per_task == '1' :
    devices = ['cuda']
  else :
    devices = [f'cuda:{i}' for i in range(num_accs_per_task)]
  logger.info( 'Using devices : {}'.format( devices) )

  torch.backends.cuda.matmul.allow_tf32 = True

  return devices 

####################################################################################################
def setup_ddp( with_ddp = True) :

  rank = 0
  size = 1

  master_node = os.environ.get('MASTER_ADDR', '-1')

  if with_ddp and (master_node != '-1'):

    local_rank = int(os.environ.get("SLURM_LOCALID"))
    ranks_per_node = int( os.environ.get('SLURM_TASKS_PER_NODE', '1')[0] )
    rank = int(os.environ.get("SLURM_NODEID")) * ranks_per_node + local_rank
    size = int(os.environ.get("SLURM_NTASKS"))

    master_node = os.environ.get('MASTER_ADDR', '-1')
    dist.init_process_group( backend='nccl', init_method='tcp://' + master_node + ':1345',
                              timeout=datetime.timedelta(seconds=10*8192),
                              world_size = size, rank = rank) 
    logger.info( f'Using DDP with MASTER_ADDR={master_node}.' )
  else :
    logger.info( 'DDP is not used.' )

  return rank, size

####################################################################################################
def setup_wandb( with_wandb, cf, rank, project_name = None, entity = 'atmorep', wandb_id = None,
                 mode='offline') :

  if with_wandb :
    wandb.require("service")
  
    if 0 == rank :

      slurm_job_id_node = os.environ.get('SLURM_JOB_ID', '-1')
      if slurm_job_id_node != '-1' :
        cf.slurm_job_id = slurm_job_id_node

      if None == wandb_id : 
        wandb.init( project = project_name, entity = entity,
                    mode = mode,
                    config = cf.get_self_dict() )
      else :
        wandb.init( id=wandb_id, resume='must',
                    mode = mode,
                    config = cf.get_self_dict() )
      wandb.run.log_code( root='./atmorep', include_fn=lambda path : path.endswith('.py'))
      
      # append slurm job id if defined
      if slurm_job_id_node != '-1' :
        wandb.run.name = 'atmorep-{}-{}'.format( wandb.run.id, slurm_job_id_node)
      else :
        wandb.run.name = 'atmorep-{}'.format( wandb.run.id)
      print( 'Wandb run: {}'.format( wandb.run.name))

      cf.wandb_id = wandb.run.id

  # communicate wandb id to all nodes
  wandb_id_int = torch.zeros( 8, dtype=torch.int32).cuda()
  if cf.with_wandb and cf.with_ddp:
    if 0 == rank :
      wandb_id_int = str_to_tensor( cf.wandb_id).cuda()
    dist.all_reduce( wandb_id_int, op=torch.distributed.ReduceOp.SUM )
    cf.wandb_id = tensor_to_str( wandb_id_int)

####################################################################################################
def init_weights_uniform( m, scale=0.01):
    '''Initialization of weights using uniform distribution'''

    classname = m.__class__.__name__

    if classname.find('ModuleList') != -1:
      for mm in m :
        mm.apply( lambda n: init_weights_uniform( n, scale) )

    if classname.find('Linear') != -1:
        # apply a uniform distribution to the weights and a bias=0
        m.weight.data.uniform_(0.0, scale)
        if m.bias is not None :
          m.bias.data.fill_(0)

####################################################################################################
def shape_to_str( shape) :
  ret ='{}'.format( list(shape)).replace(' ', '').replace(',','_').replace('(','s_').replace(')','')
  ret = ret.replace('[','s_').replace(']','')
  return ret

####################################################################################################
def get_model_filename( model = None, model_id = '', epoch=-2, with_model_path = True) :

  if isinstance( model, str) :
    name = model 
  elif model :
    name = model.__class__.__name__
  else : # backward compatibility
    name = 'mod'

  mpath = 'id{}'.format(model_id) if with_model_path else ''

  if epoch > -2 :
    # model_file = Path( config.path_results, 'models/id{}/{}_id{}_epoch{}.mod'.format(
    #                                                        model_id, name, model_id, epoch))
    model_file = Path( config.path_models, mpath, '{}_id{}_epoch{}.mod'.format(
                                                           name, model_id, epoch))
  else :
    model_file = Path( config.path_models, mpath, '{}_id{}.mod'.format( name, model_id))
      
  return model_file

####################################################################################################
def relMSELoss( pred, target = None) :
  val = torch.mean( (pred - target) * (pred - target)) / torch.mean( target * target)
  return val
  
####################################################################################################
def days_in_month( year, month) :
  '''Days in month in specific year'''
  return monthrange( year, month)[1]

def days_until_month_in_year( year, month) :
  '''Days in year until month starts'''

  offset = 0
  for im in range( month - 1) :
    offset += monthrange( year, im+1)[1]
  
  return offset

####################################################################################################
def tokenize( data, token_size = [-1,-1,-1]) :

  data_tokenized = data
  if token_size[0] > -1 :

    data_shape = data.shape
    tok_tot_t = int( data_shape[-3] / token_size[0])
    tok_tot_x = int( data_shape[-2] / token_size[1])
    tok_tot_y = int( data_shape[-1] / token_size[2])

    if 5 == len(data_shape) :
      t2 = torch.reshape( data, (data.shape[0], data.shape[1], tok_tot_t, token_size[0], 
                                tok_tot_x, token_size[1], tok_tot_y, token_size[2]))
      data_tokenized = t2.permute( [0, 1, 2, 4, 6, 3, 5, 7])
    elif 4 == len(data_shape) :
      t2 = torch.reshape( data, (-1, tok_tot_t, token_size[0], 
                                tok_tot_x, token_size[1], tok_tot_y, token_size[2]))
      data_tokenized = t2.permute( [0, 1, 3, 5, 4, 3, 6])
    elif 3 == len(data_shape) :
      t2 = torch.reshape( data, (tok_tot_t, token_size[0], tok_tot_x, token_size[1], tok_tot_y, token_size[2]))
      data_tokenized = torch.transpose(torch.transpose( torch.transpose( t2, 4, 3), 2, 1), 3, 2)
    elif 2 == len(data_shape) :
      t2 = torch.reshape( data, (tok_tot_x, token_size[0], tok_tot_y, token_size[1]))
      data_tokenized = torch.transpose( t2, 1, 2)
    else :
      assert False

  return data_tokenized.contiguous()

####################################################################################################
def detokenize( data) :

  data = data.transpose( [*np.arange( len(data.shape)-5), -3, -5, -2, -4, -1])
  data = data.reshape( [*data.shape[:-6], np.prod( data.shape[-6:-4]), # time
                                          np.prod( data.shape[-4:-2]), # lat
                                          np.prod( data.shape[-2:])])  #lon
  return data 

####################################################################################################
def sgn_exp( x ) :
  '''exponential preserving sign'''
  return x.sign() * (torch.exp( x.abs() ) - 1.)

####################################################################################################
def token_info_to_time( token_info, return_pd = True) :
  str = f'{int(token_info[0])}-{int(np.floor(token_info[1]))+1}-{int(token_info[2])}'
  # correct for 1 day since %j strangely starts from 1
  date = pd.to_datetime( str, format='%Y-%j-%H')
  return date if return_pd else (date.year, date.month, date.day, date.hour)

####################################################################################################
def list_replace_rec( list, idxs, val) :
  if len(idxs) == 1 :
    list.__setitem__( idxs[0], val)
  else :
    list_replace_rec( list.__getitem__( idxs[0]), idxs[1:], val)
    list.__setitem__( idxs[0], list.__getitem__( idxs[0]) )

####################################################################################################
def Gaussian( x, mu=0., std_dev=1.) :
  # return (1 / (std_dev*np.sqrt(2.*np.pi))) * torch.exp( -0.5 * (x-mu)*(x-mu) / (std_dev*std_dev))
  # unnormalized Gaussian where maximum is one
  return torch.exp( -0.5 * (x-mu)*(x-mu) / (std_dev*std_dev))

def erf( x, mu=0., std_dev=1.) :
  c1 = torch.sqrt( torch.tensor(0.5 * np.pi) )
  c2 = torch.sqrt( 1. / torch.tensor(std_dev * std_dev))
  c3 = torch.sqrt( torch.tensor( 2.) )
  val = c1 * ( 1./c2 - std_dev * torch.special.erf( (mu - x) / (c3 * std_dev) ) )
  return val

########################################
# def CRPS_ps( y, mu, std_dev) :
#   val = ps.crps_gaussian(y.cpu().detach().numpy(), mu=mu.cpu().detach().numpy(), sig=std_dev.cpu().detach().numpy())
#   return torch.tensor(val)

def CRPS( y, mu, std_dev) :
   
  # see Eq. A2 in S. Rasp and S. Lerch. Neural networks for postprocessing ensemble weather forecasts. Monthly Weather Review, 146(11):3885 – 3900, 2018.
  c1 = np.sqrt(1./np.pi)
  t1 = 2. * erf( (y-mu) / std_dev) - 1.
  t2 = 2. * Gaussian( (y-mu) / std_dev)
  val = std_dev * ( (y-mu)/std_dev * t1 + t2 - c1 )
  return val

########################################
# def kernel_crps_ps( target, ens) :
#   val = ps.crps_ensemble(target.cpu().detach().numpy(), ens.permute([1,2,0]).cpu().detach().numpy())
#   return torch.tensor(val)

def kernel_crps( target, ens, fair = True) :
  ens_size = ens.shape[0]
  mae = torch.cat( [(target - mem).abs().mean().unsqueeze(0) for mem in ens], 0).mean()

  if ens_size == 1:
    return mae

  coef = -1.0 / (2.0 * ens_size * (ens_size - 1)) if fair else -1.0 / (2.0 * ens_size**2)
  temp = [(p1 - p2).abs().sum() for p1 in ens for p2 in ens]

  ens_var = coef * torch.tensor( [(p1 - p2).abs().sum() for p1 in ens for p2 in ens]).sum()
  ens_var /= (ens.shape[1]*ens.shape[2])

  return mae + ens_var

########################################

def get_weights(lats_idx, lat_min = -90., lat_max = 90., reso = 0.25):
  lat_range = lat_max - lat_min 
  bins = lat_range/reso+1

  theta_weight = np.array([np.cos(w) for w in np.arange( lat_max * np.pi/lat_range , lat_min * np.pi/lat_range, -np.pi/bins)], dtype = np.float32)
  
  return theta_weight[lats_idx]

########################################

def weighted_mse(x, target, weights):
        return torch.sum(weights * (x - target) **2 )/torch.sum(weights)

########################################

def check_num_samples(num_samples_validate, batch_size):
  assert num_samples_validate // batch_size > 0, f"Num samples validate: {num_samples_validate} is smaller than batch size: {batch_size}. Please increase it."