# coding=utf-8 import glob from datetime import datetime import os from autosubmit.config.log import Log from autosubmit.date.chunk_date_lib import parse_date, chunk_start_date, chunk_end_date, previous_day from earthdiagnostics.cmorizer import Cmorizer from earthdiagnostics.datamanager import DataManager, NetCDFFile from earthdiagnostics.utils import TempFile, Utils from earthdiagnostics.variable import Variable class CMORManager(DataManager): """ Data manager class for CMORized experiments """ def get_file(self, domain, var, startdate, member, chunk, grid=None, box=None, frequency=None): """ Copies a given file from the CMOR repository to the scratch folder and returns the path to the scratch's copy :param domain: CMOR domain :type domain: str :param var: variable name :type var: str :param startdate: file's startdate :type startdate: str :param member: file's member :type member: int :param chunk: file's chunk :type chunk: int :param grid: file's grid (only needed if it is not the original) :type grid: str|NoneType :param box: file's box (only needed to retrieve sections or averages) :type box: Box :param frequency: file's frequency (only needed if it is different from the default) :type frequency: str :return: path to the copy created on the scratch folder :rtype: str """ filepath = self.get_file_path(startdate, member, domain, var, chunk, frequency, box, grid, None, None) temp_path = TempFile.get() Utils.copy_file(filepath, temp_path) return temp_path def get_file_path(self, startdate, member, domain, var, chunk, frequency, box=None, grid=None, year=None, date_str=None): """ Returns the path to a concrete file :param startdate: file's startdate :type startdate: str :param member: file's member :type member: int :param domain: file's domain :type domain: str :param var: file's var :type var: str :param chunk: file's chunk :type chunk: int :param frequency: file's frequency :type frequency: str :param box: file's box :type box: Box :param grid: file's grid :type grid: str :param year: file's year :type year: int|str :param date_str: date string to add directly. Overrides year or chunk configurations :type date_str: str :return: path to the file :rtype: str """ if not frequency: frequency = self.config.frequency var = self._get_final_var_name(box, var) folder_path = self._get_full_cmor_folder_path(startdate, member, domain, var, frequency, grid) file_name = self._get_cmor_file_name(startdate, member, domain, var, frequency, chunk, year, date_str) filepath = os.path.join(folder_path, file_name) return filepath def _get_cmor_file_name(self, startdate, member, domain, var, frequency, chunk, year, date_str): domain_abreviattion = Variable.get_table_name(domain, frequency) if chunk is not None: time_bound = self._get_chunk_time_bounds(startdate, chunk) elif year: if frequency is not 'yr': raise ValueError('Year may be provided instead of chunk only if frequency is "yr"') time_bound = str(year) elif date_str: time_bound = date_str else: raise ValueError('Chunk, year and date_str can not be None at the same time') file_name = '{0}_{1}_{2}_{3}_S{4}_r{5}i1p1_{6}.nc'.format(var, domain_abreviattion, self.experiment.model, self.experiment.experiment_name, startdate, member + 1, time_bound) return file_name def _get_full_cmor_folder_path(self, startdate, member, domain, var, frequency, grid): folder_path = os.path.join(self._get_startdate_path(startdate), frequency, domain, var) if grid: folder_path = os.path.join(folder_path, grid) folder_path = os.path.join(folder_path, 'r{0}i1p1'.format(member + 1)) return folder_path def _get_chunk_time_bounds(self, startdate, chunk): start = parse_date(startdate) chunk_start = chunk_start_date(start, chunk, self.experiment.chunk_size, 'month', 'standard') chunk_end = chunk_end_date(chunk_start, self.experiment.chunk_size, 'month', 'standard') chunk_end = previous_day(chunk_end, 'standard') time_bound = "{0:04}{1:02}-{2:04}{3:02}".format(chunk_start.year, chunk_start.month, chunk_end.year, chunk_end.month) return time_bound def link_file(self, domain, var, startdate, member, chunk=None, grid=None, box=None, frequency=None, year=None, date_str=None, move_old=False): """ Creates the link of a given file from the CMOR repository. :param move_old: :param date_str: :param year: if frequency is yearly, this parameter is used to give the corresponding year :type year: int :param domain: CMOR domain :type domain: str :param var: variable name :type var: str :param startdate: file's startdate :type startdate: str :param member: file's member :type member: int :param chunk: file's chunk :type chunk: int :param grid: file's grid (only needed if it is not the original) :type grid: str :param box: file's box (only needed to retrieve sections or averages) :type box: Box :param frequency: file's frequency (only needed if it is different from the default) :type frequency: str :return: path to the copy created on the scratch folder :rtype: str """ var = self._get_final_var_name(box, var) if not frequency: frequency = self.config.frequency domain = DataManager.correct_domain(domain) filepath = self.get_file_path(startdate, member, domain, var, chunk, frequency, grid, year, date_str) self._create_link(domain, filepath, frequency, var, grid, move_old) def send_file(self, filetosend, domain, var, startdate, member, chunk=None, grid=None, region=None, box=None, rename_var=None, frequency=None, year=None, date_str=None, move_old=False): """ Copies a given file to the CMOR repository. It also automatically converts to netCDF 4 if needed and can merge with already existing ones as needed :param move_old: if true, moves files following older conventions that may be found on the links folder :type move_old: bool :param date_str: exact date_str to use in the cmorized file :type: str :param year: if frequency is yearly, this parameter is used to give the corresponding year :type year: int :param rename_var: if exists, the given variable will be renamed to the one given by var :type rename_var: str :param filetosend: path to the file to send to the CMOR repository :type filetosend: str :param region: specifies the region represented by the file. If it is defined, the data will be appended to the CMOR repository as a new region in the file or will overwrite if region was already present :type region: str :param domain: CMOR domain :type domain: str :param var: variable name :type var: str :param startdate: file's startdate :type startdate: str :param member: file's member :type member: int :param chunk: file's chunk :type chunk: int :param grid: file's grid (only needed if it is not the original) :type grid: str :param box: file's box (only needed to retrieve sections or averages) :type box: Box :param frequency: file's frequency (only needed if it is different from the default) :type frequency: str :return: path to the copy created on the scratch folder :rtype: str """ original_var = var cmor_var = Variable.get_variable(var) var = self._get_final_var_name(box, var) if rename_var and rename_var != var: Utils.rename_variable(filetosend, rename_var, var) elif original_var != var: Utils.rename_variable(filetosend, original_var, var) if not frequency: frequency = self.config.frequency domain = DataManager.correct_domain(domain) filepath = self.get_file_path(startdate, member, domain, var, chunk, frequency, box, grid, year, date_str) netcdf_file = NetCDFFile(filepath, filetosend, domain, var, cmor_var) netcdf_file.send() self._create_link(domain, filepath, frequency, var, grid, move_old) def get_year(self, domain, var, startdate, member, year, grid=None, box=None): """ Ge a file containing all the data for one year for one variable :param domain: variable's domain :type domain: str :param var: variable's name :type var: str :param startdate: startdate to retrieve :type startdate: str :param member: member to retrieve :type member: int :param year: year to retrieve :type year: int :param grid: variable's grid :type grid: str :param box: variable's box :type box: Box :return: """ chunk_files = list() for chunk in self.experiment.get_year_chunks(startdate, year): chunk_files.append(self.get_file(domain, var, startdate, member, chunk, grid=grid, box=box)) if len(chunk_files) > 1: temp = self._merge_chunk_files(chunk_files) else: temp = chunk_files[0] temp2 = self._select_data_of_given_year(temp, year) os.remove(temp) return temp2 @staticmethod def _select_data_of_given_year(data_file, year): temp2 = TempFile.get() Utils.cdo.selyear(str(year), input=data_file, output=temp2) return temp2 @staticmethod def _merge_chunk_files(chunk_files): temp = TempFile.get() Utils.nco.ncrcat(input=' '.join(chunk_files), output=temp) for chunk_file in chunk_files: os.remove(chunk_file) return temp # noinspection PyPep8Naming def prepare_CMOR_files(self): """ Prepares the data to be used by the diagnostic. If CMOR data is not created, it show a warning and closes. In the future, an automatic cmorization procedure will be launched If CMOR data is available but packed, the procedure will unpack it. :return: """ # Check if cmorized and convert if not for startdate, member in self.experiment.get_member_list(): if self._is_cmorized(startdate, member) and not self.config.cmor.force: continue if not self._unpack_cmor_files(member): self._cmorize_member(startdate, member) def _is_cmorized(self, startdate, member): startdate_path = self._get_startdate_path(startdate) if not os.path.exists(startdate_path): return False for freq in os.listdir(startdate_path): freq_path = os.path.join(startdate_path, freq) for domain in os.listdir(freq_path): domain_path = os.path.join(freq_path, domain) for var in os.listdir(domain_path): member_path = os.path.join(domain_path, var, 'r{0}i1p1'.format(member + 1)) if os.path.exists(member_path): return True return False def _cmorize_member(self, startdate, member): start_time = datetime.now() member_str = self.experiment.get_member_str(member) Log.info('CMORizing startdate {0} member {1}. Starting at {0}', startdate, member_str, start_time) cmorizer = Cmorizer(self, startdate, member) cmorizer.cmorize_ocean() cmorizer.cmorize_atmos() Log.result('CMORized startdate {0} member {1}!\n\n', startdate, member_str, datetime.now() - start_time) def _unpack_cmor_files(self, startdate): if self.config.cmor.force: return False filepaths = self._get_transferred_cmor_data_filepaths(startdate, 'tar.gz') if len(filepaths) > 0: Log.info('Unzipping cmorized data...') Utils.unzip(filepaths, True) if not os.path.exists(self.cmor_path): os.mkdir(self.cmor_path) filepaths = self._get_transferred_cmor_data_filepaths(startdate, 'tar') if len(filepaths) > 0: Log.info('Unpacking cmorized data...') Utils.untar(filepaths, self.cmor_path) self._correct_paths(startdate) self._create_links(startdate) return True return False def _get_transferred_cmor_data_filepaths(self, startdate, extension): tar_path = os.path.join(self.config.data_dir, self.experiment.expid, 'original_files', 'cmorfiles') tar_original_files = os.path.join(self.config.data_dir, 'original_files', self.experiment.expid, 'cmorfiles') file_name = 'CMOR?_{0}_{1}_*.{2}'.format(self.experiment.expid, startdate, extension) filepaths = glob.glob(os.path.join(tar_path, file_name)) filepaths += glob.glob(os.path.join(tar_path, 'outputs', file_name)) filepaths += glob.glob(os.path.join(tar_original_files, file_name)) filepaths += glob.glob(os.path.join(tar_original_files, 'outputs', file_name)) return filepaths def _correct_paths(self, startdate): self._remove_extra_output_folder() self._fix_model_as_experiment_error(startdate) def _fix_model_as_experiment_error(self, startdate): if self.experiment.experiment_name != self.experiment.model: bad_path = os.path.join(self.cmor_path, self.experiment.institute, self.experiment.model, self.experiment.model) Log.debug('Correcting double model appearance') for (dirpath, dirnames, filenames) in os.walk(bad_path, False): for filename in filenames: filepath = os.path.join(dirpath, filename) good = filepath.replace('_{0}_output_'.format(self.experiment.model), '_{0}_{1}_S{2}_'.format(self.experiment.model, self.experiment.experiment_name, startdate)) good = good.replace('/{0}/{0}'.format(self.experiment.model), '/{0}/{1}'.format(self.experiment.model, self.experiment.experiment_name)) Utils.move_file(filepath, good) os.rmdir(dirpath) Log.debug('Done') def _remove_extra_output_folder(self): bad_path = os.path.join(self.cmor_path, 'output', self.experiment.institute) if os.path.exists(bad_path): Log.debug('Moving CMOR files out of the output folder') Utils.execute_shell_command(['mv', bad_path, os.path.join(bad_path, '..', '..')]) os.rmdir(os.path.join(self.cmor_path, 'output')) Log.debug('Done') def _create_links(self, startdate): Log.info('Creating links for CMOR files ()') path = self._get_startdate_path(startdate) for freq in os.listdir(path): for domain in os.listdir(os.path.join(path, freq)): for var in os.listdir(os.path.join(path, freq, domain)): for member in os.listdir(os.path.join(path, freq, domain, var)): for name in os.listdir(os.path.join(path, freq, domain, var, member)): filepath = os.path.join(path, freq, domain, var, member, name) if os.path.isfile(filepath): self._create_link(domain, filepath, freq, var, "", False) else: for filename in os.listdir(filepath): self._create_link(domain, os.path.join(filepath, filename), freq, var, "", False) Log.info('Creating lings for CMOR files') def _get_startdate_path(self, startdate): """ Returns the path to the startdate's CMOR folder :param startdate: target startdate :type startdate: str :return: path to the startdate's CMOR folder :rtype: str """ return os.path.join(self.config.data_dir, self.experiment.expid, 'cmorfiles', self.experiment.institute, self.experiment.model, self.experiment.experiment_name, 'S' + startdate)