# # Copyright (c) 2019 Jonathan Weyn # # See the file LICENSE for your rights. # """ Example of training a DLWP model using a dataset of predictors generated with DLWP.model.Preprocessor. Uses Microsoft Azure resources. Launch this script as an Azure experiment using 'Train on Azure.ipynb' """ import argparse import os import shutil import time import numpy as np import pandas as pd import xarray as xr from datetime import datetime from DLWP.model import DLWPNeuralNet, SeriesDataGenerator from DLWP.util import save_model, train_test_split_ind from DLWP.custom import RNNResetStates, EarlyStoppingMin, latitude_weighted_loss, RunHistory, anomaly_correlation_loss from tensorflow.keras.losses import mean_squared_error from tensorflow.keras.callbacks import TensorBoard from azureml.core import Run import tensorflow as tf #%% Parse user arguments parser = argparse.ArgumentParser() parser.add_argument('--root-directory', type=str, dest='root_directory', default='.', help='Destination root data directory on Azure Blob storage') parser.add_argument('--predictor-file', type=str, dest='predictor_file', help='Path and name of data file in root-directory') parser.add_argument('--model-file', type=str, dest='model_file', help='Path and name of model save file in root-directory') parser.add_argument('--log-directory', type=str, dest='log_directory', default='./logs', help='Destination for log files in root-directory') parser.add_argument('--temp-dir', type=str, dest='temp_dir', default='None', help='If specified, copies the predictor file here for use during training (e.g., fast SSD)') parser.add_argument('--seed', type=int, dest='seed', default=-1, help='Specify random number seed >= 0') args = parser.parse_args() if args.temp_dir != 'None': os.makedirs(args.temp_dir, exist_ok=True) if args.seed >= 0: np.random.seed(args.seed) tf.compat.v1.set_random_seed(args.seed) #%% Parameters root_directory = args.root_directory predictor_file = os.path.join(root_directory, args.predictor_file) model_file = os.path.join(root_directory, args.model_file) log_directory = os.path.join(root_directory, args.log_directory) # NN parameters. Regularization is applied to LSTM layers by default. weight_loss indicates whether to weight the # loss function preferentially in the mid-latitudes. model_is_convolutional = True model_is_recurrent = False min_epochs = 200 max_epochs = 1000 patience = 50 batch_size = 64 lambda_ = 1.e-4 weight_loss = False acc_loss = False shuffle = True # Data parameters. Specify the input variables/levels, output variables/levels, and time steps in/out. Note that for # LSTM layers, the model can only predict effectively if the output time steps is 1 or equal to the input time steps. # Ensure that the selections use LISTS of values (even for only 1) to keep dimensions correct. input_selection = {'varlev': ['HGT/500', 'THICK/300-700']} output_selection = {'varlev': ['HGT/500', 'THICK/300-700']} input_time_steps = 1 output_time_steps = 1 step_interval = 6 # Option to crop the north pole. Necessary for getting an even number of latitudes for up-sampling layers. crop_north_pole = True # Add incoming solar radiation forcing add_solar = False # If system memory permits, loading the predictor data can greatly increase efficiency when training on GPUs, if the # train computation takes less time than the data loading. load_memory = True # Use multiple GPUs, if available n_gpu = 1 # Force use of the keras model.fit() method. May run faster in some instances, but uses (input_time_steps + # output_time_steps) times more memory. use_keras_fit = False # Validation set to use. Either an integer (number of validation samples, taken from the end), or an iterable of # pandas datetime objects. The train set can be set to the first samples, an iterable of dates, or None to # simply use the remaining points. Match the type of validation_set. validation_set = list(pd.date_range(datetime(2003, 1, 1, 0), datetime(2006, 12, 31, 18), freq='6H')) train_set = list(pd.date_range(datetime(1979, 1, 1, 6), datetime(2002, 12, 31, 18), freq='6H')) # validation_set = (list(pd.date_range(datetime(1985, 1, 1, 0), datetime(1986, 1, 1, 12), freq='6H')) + # list(pd.date_range(datetime(1992, 1, 1, 0), datetime(1993, 1, 1, 12), freq='6H')) + # list(pd.date_range(datetime(1999, 1, 1, 0), datetime(2000, 1, 1, 12), freq='6H')) + # list(pd.date_range(datetime(2006, 1, 1, 0), datetime(2007, 1, 1, 12), freq='6H'))) # train_set = (list(pd.date_range(datetime(1979, 1, 6, 0), datetime(1985, 1, 1, 12), freq='6H')) + # list(pd.date_range(datetime(1986, 1, 1, 0), datetime(1992, 1, 1, 12), freq='6H')) + # list(pd.date_range(datetime(1993, 1, 1, 0), datetime(1999, 1, 1, 12), freq='6H')) + # list(pd.date_range(datetime(2000, 1, 1, 0), datetime(2006, 1, 1, 12), freq='6H'))) #%% Open data. If temporary file is specified, copy it there. if args.temp_dir != 'None': new_predictor_file = os.path.join(args.temp_dir, args.predictor_file) print('Copying predictor file to %s...' % new_predictor_file) if os.path.isfile(new_predictor_file): print('File already exists!') else: shutil.copy(predictor_file, new_predictor_file, follow_symlinks=True) data = xr.open_dataset(new_predictor_file, chunks={'sample': batch_size}) else: data = xr.open_dataset(predictor_file, chunks={'sample': batch_size}) if 'time_step' in data.dims: time_dim = data.dims['time_step'] else: time_dim = 1 n_sample = data.dims['sample'] if crop_north_pole: data = data.isel(lat=(data.lat < 90.0)) #%% Build a model and the data generators dlwp = DLWPNeuralNet(is_convolutional=model_is_convolutional, is_recurrent=model_is_recurrent, time_dim=time_dim, scaler_type=None, scale_targets=False) # Find the validation set if isinstance(validation_set, int): n_sample = data.dims['sample'] ts, val_set = train_test_split_ind(n_sample, validation_set, method='last') if train_set is None: train_set = ts elif isinstance(train_set, int): train_set = list(range(train_set)) validation_data = data.isel(sample=val_set) train_data = data.isel(sample=train_set) elif validation_set is None: if train_set is None: train_set = data.sample.values validation_data = None train_data = data.sel(sample=train_set) else: # we must have a list of datetimes if train_set is None: train_set = np.isin(data.sample.values, np.array(validation_set, dtype='datetime64[ns]'), assume_unique=True, invert=True) validation_data = data.sel(sample=validation_set) train_data = data.sel(sample=train_set) # For multiple GPUs, increase the batch size batch_size = n_gpu * batch_size # Build the data generators if load_memory or use_keras_fit: print('Loading data to memory...') generator = SeriesDataGenerator(dlwp, train_data, input_sel=input_selection, output_sel=output_selection, input_time_steps=input_time_steps, output_time_steps=output_time_steps, batch_size=batch_size, add_insolation=add_solar, load=load_memory, shuffle=shuffle, interval=step_interval) if use_keras_fit: p_train, t_train = generator.generate([]) if validation_data is not None: val_generator = SeriesDataGenerator(dlwp, validation_data, input_sel=input_selection, output_sel=output_selection, input_time_steps=input_time_steps, output_time_steps=output_time_steps, batch_size=batch_size, add_insolation=add_solar, load=load_memory, interval=step_interval) if use_keras_fit: val = val_generator.generate([]) else: val_generator = None if use_keras_fit: val = None #%% Compile the model structure with some generator data information # Up-sampling convolutional network with optional LSTM layer cs = generator.convolution_shape cso = generator.output_convolution_shape layers = ( # --- These layers add a convolutional LSTM at the beginning --- # # ('PeriodicPadding3D', ((0, 0, 2),), { # 'data_format': 'channels_first', # 'input_shape': cs # }), # ('ZeroPadding3D', ((0, 2, 0),), {'data_format': 'channels_first'}), # ('ConvLSTM2D', (cs[1], 3), { # 4 * cs[1] # 'dilation_rate': 2, # 'padding': 'valid', # 'data_format': 'channels_first', # 'activation': 'tanh', # 'return_sequences': True, # 'kernel_regularizer': l2(lambda_) # }), # ('Reshape', ((cs[1] * cs[0], cs[2], cs[3]),), None), # 4 * cs[1] * cs[0] # -------------------------------------------------------------- # ('PeriodicPadding2D', ((0, 2),), { 'data_format': 'channels_first', 'input_shape': cs }), ('ZeroPadding2D', ((2, 0),), {'data_format': 'channels_first'}), ('Conv2D', (32, 3), { 'dilation_rate': 2, 'padding': 'valid', 'activation': 'tanh', 'data_format': 'channels_first' }), # ('BatchNormalization', None, {'axis': 1}), ('MaxPooling2D', (2,), {'data_format': 'channels_first'}), ('PeriodicPadding2D', ((0, 1),), {'data_format': 'channels_first'}), ('ZeroPadding2D', ((1, 0),), {'data_format': 'channels_first'}), ('Conv2D', (64, 3), { 'dilation_rate': 1, 'padding': 'valid', 'activation': 'tanh', 'data_format': 'channels_first' }), # ('BatchNormalization', None, {'axis': 1}), ('MaxPooling2D', (2,), {'data_format': 'channels_first'}), ('PeriodicPadding2D', ((0, 1),), {'data_format': 'channels_first'}), ('ZeroPadding2D', ((1, 0),), {'data_format': 'channels_first'}), ('Conv2D', (128, 3), { 'dilation_rate': 1, 'padding': 'valid', 'activation': 'tanh', 'data_format': 'channels_first' }), # ('BatchNormalization', None, {'axis': 1}), ('UpSampling2D', (2,), {'data_format': 'channels_first'}), ('PeriodicPadding2D', ((0, 1),), {'data_format': 'channels_first'}), ('ZeroPadding2D', ((1, 0),), {'data_format': 'channels_first'}), ('Conv2D', (64, 3), { 'dilation_rate': 1, 'padding': 'valid', 'activation': 'tanh', 'data_format': 'channels_first' }), # ('BatchNormalization', None, {'axis': 1}), ('UpSampling2D', (2,), {'data_format': 'channels_first'}), ('PeriodicPadding2D', ((0, 2),), {'data_format': 'channels_first'}), ('ZeroPadding2D', ((2, 0),), {'data_format': 'channels_first'}), ('Conv2D', (32, 3), { 'dilation_rate': 2, 'padding': 'valid', 'activation': 'tanh', 'data_format': 'channels_first' }), # ('BatchNormalization', None, {'axis': 1}), ('PeriodicPadding2D', ((0, 2),), {'data_format': 'channels_first'}), ('ZeroPadding2D', ((2, 0),), {'data_format': 'channels_first'}), # --- Change the number of filters to cso[0] * cso[1] for LSTM model, and uncomment the last Reshape layer --- # ('Conv2D', (cso[0], 5), { 'padding': 'valid', 'activation': 'linear', 'data_format': 'channels_first' }), # ('Reshape', (cso,), None) ) # # Fully-LSTM upsampling convolutional NN # layers = ( # ('PeriodicPadding3D', ((0, 0, 2),), { # 'data_format': 'channels_first', # 'input_shape': cs # }), # ('ZeroPadding3D', ((0, 2, 0),), {'data_format': 'channels_first'}), # ('ConvLSTM2D', (16, 3), { # 'dilation_rate': 2, # 'padding': 'valid', # 'data_format': 'channels_first', # 'activation': 'tanh', # 'return_sequences': True, # 'kernel_regularizer': l2(lambda_) # }), # ('MaxPooling3D', ((1, 2, 2),), {'data_format': 'channels_first'}), # ('PeriodicPadding3D', ((0, 0, 1),), {'data_format': 'channels_first'}), # ('ZeroPadding3D', ((0, 1, 0),), {'data_format': 'channels_first'}), # ('ConvLSTM2D', (32, 3), { # 'dilation_rate': 1, # 'padding': 'valid', # 'data_format': 'channels_first', # 'activation': 'tanh', # 'return_sequences': True, # 'kernel_regularizer': l2(lambda_) # }), # ('MaxPooling3D', ((1, 2, 2),), {'data_format': 'channels_first'}), # ('PeriodicPadding3D', ((0, 0, 1),), {'data_format': 'channels_first'}), # ('ZeroPadding3D', ((0, 1, 0),), {'data_format': 'channels_first'}), # ('ConvLSTM2D', (64, 3), { # 'dilation_rate': 1, # 'padding': 'valid', # 'data_format': 'channels_first', # 'activation': 'tanh', # 'return_sequences': True, # 'kernel_regularizer': l2(lambda_) # }), # ('UpSampling3D', ((1, 2, 2),), {'data_format': 'channels_first'}), # ('PeriodicPadding3D', ((0, 0, 1),), {'data_format': 'channels_first'}), # ('ZeroPadding3D', ((0, 1, 0),), {'data_format': 'channels_first'}), # ('ConvLSTM2D', (32, 3), { # 'dilation_rate': 1, # 'padding': 'valid', # 'data_format': 'channels_first', # 'activation': 'tanh', # 'return_sequences': True, # 'kernel_regularizer': l2(lambda_) # }), # ('UpSampling3D', ((1, 2, 2),), {'data_format': 'channels_first'}), # ('PeriodicPadding3D', ((0, 0, 2),), {'data_format': 'channels_first'}), # ('ZeroPadding3D', ((0, 2, 0),), {'data_format': 'channels_first'}), # ('ConvLSTM2D', (16, 3), { # 'dilation_rate': 2, # 'padding': 'valid', # 'data_format': 'channels_first', # 'activation': 'tanh', # 'return_sequences': True, # 'kernel_regularizer': l2(lambda_) # }), # ('PeriodicPadding3D', ((0, 0, 2),), {'data_format': 'channels_first'}), # ('ZeroPadding3D', ((0, 2, 0),), {'data_format': 'channels_first'}), # ('ConvLSTM2D', (cso[0], 5), { # 'dilation_rate': 1, # 'padding': 'valid', # 'data_format': 'channels_first', # 'activation': 'linear', # 'return_sequences': True, # }), # ) # Example custom loss function: pass to loss= in build_model() if acc_loss: # Generate the data to fit the scaler. The generator will by default apply scaling because it is necessary # to automate its use in the Keras fit_generator method, so disable it when dealing with data to fit the scaler print('Finding climatology for ACC loss...') p_fit, t_fit = generator.generate([], scale_and_impute=False) climo = t_fit.mean(axis=0, keepdims=True) p_fit, t_fit = (None, None) loss_function = anomaly_correlation_loss(climo, regularize_mean='mse', reverse=True) else: loss_function = mean_squared_error if weight_loss: loss_function = latitude_weighted_loss(loss_function, generator.ds.lat.values, generator.convolution_shape, axis=-2, weighting='midlatitude') # Build the model try: dlwp.build_model(layers, loss=loss_function, optimizer='adam', metrics=['mae'], gpus=n_gpu) except (ValueError, IndexError): for layer in dlwp.base_model.layers: print(layer.name, layer.output_shape) raise print(dlwp.base_model.summary()) #%% Train, evaluate, and save the model # Train and evaluate the model start_time = time.time() print('Begin training...') run = Run.get_context() history = RunHistory(run) early = EarlyStoppingMin(min_epochs=min_epochs, monitor='val_loss' if val_generator is not None else 'loss', min_delta=0., patience=patience, restore_best_weights=True, verbose=1) tensorboard = TensorBoard(log_dir=log_directory, batch_size=batch_size, update_freq='epoch') if use_keras_fit: dlwp.fit(p_train, t_train, batch_size=batch_size, epochs=max_epochs, verbose=2, validation_data=val, shuffle=shuffle, callbacks=[history, RNNResetStates(), early]) else: dlwp.fit_generator(generator, epochs=max_epochs, verbose=2, validation_data=val_generator, use_multiprocessing=True, callbacks=[history, RNNResetStates(), early]) end_time = time.time() # Save the model if model_file is not None: os.makedirs(os.path.sep.join(model_file.split(os.path.sep)[:-1]), exist_ok=True) save_model(dlwp, model_file, history=history) print('Wrote model %s' % model_file) # Evaluate the model print("\nTrain time -- %s seconds --" % (end_time - start_time)) try: print('Train loss:', history.history['loss'][-patience - 1]) run.log('TRAIN_LOSS', history.history['loss'][-patience - 1]) print('Train mean absolute error:', history.history['mean_absolute_error'][-patience - 1]) except (KeyError, IndexError): pass if validation_data is not None: score = dlwp.evaluate(*val_generator.generate([]), verbose=0) print('Validation loss:', score[0]) try: print('Validation mean absolute error:', score[1]) except: pass run.log('VAL_LOSS', score[0])