Changeset f20af73 in flex_extract.git

Timestamp:

Mar 8, 2019, 10:05:20 AM (5 years ago)

Author:

Anne Philipp <anne.philipp@…>

Branches:

master, ctbto, dev

Children:

82c2959

Parents:

b4a4777

Message:

added CDS API support for ERA5 instead of ECMWFAPI / refactored setup of CONTROL parameter because for local version it wanted to use not installed ECMWF_ENV file.

Location:

source/python

Files:

• _config.py (modified) (1 diff)
• classes/ControlFile.py (modified) (3 diffs)
• classes/EcFlexpart.py (modified) (15 diffs)
• classes/MarsRetrieval.py (modified) (7 diffs)
• mods/checks.py (modified) (2 diffs)
• mods/get_mars_data.py (modified) (7 diffs)
• mods/prepare_flexpart.py (modified) (3 diffs)
• mods/tools.py (modified) (6 diffs)
• submit.py (modified) (4 diffs)

source/python/_config.py

@@ -28 +28 @@
 import sys
 import inspect
+import socket
 
 _VERSION_STR = '7.1'
+
+FLAG_ON_ECMWFSERVER = 'ecmwf' in socket.gethostname()
 
 QUEUES_LIST = ['ecgate', 'cca', 'ccb']
 
 INSTALL_TARGETS = ['local', 'ecgate', 'cca']
+
+CDS_DATASET = 'reanalysis-era5-complete'
 
 # up-to-date available maximum level numbers at ECMWF, 05.10.2018

source/python/classes/ControlFile.py

@@ -318 +318 @@
         Default value is 0.
 
-    ecapi : boolean
-        Tells wether the ECMWF Web APi was able to load or not.
+    ec_api : boolean
+        Tells wether the ECMWF Web API was able to load or not.
+        Default value is None.
+
+    cds_api : boolean
+        Tells wether the CDS API was able to load or not.
         Default value is None.
 
@@ -417 +421 @@
         self.request = 0
         self.public = 0
-        self.ecapi = None
+        self.ec_api = None
+        self.cds_api = None
         self.purefc = 0
         self.rrint = 0
@@ -481 +486 @@
                                 data[1] = data[1][:i] + var + data[1][k+1:]
                             else:
-                                my_error(self.mailfail,
-                                         'Could not find variable '
+                                my_error('Could not find variable '
                                          + data[1][j+1:k] + ' while reading ' +
                                          self.controlfile)

source/python/classes/EcFlexpart.py

@@ -73 +73 @@
 sys.path.append('../')
 import _config
-from GribUtil import GribUtil
+from .GribUtil import GribUtil
 from mods.tools import (init128, to_param_id, silent_remove, product,
                         my_error, make_dir, get_informations, get_dimensions,
-                        execute_subprocess)
-from MarsRetrieval import MarsRetrieval
-from UioFiles import UioFiles
+                        execute_subprocess, to_param_id_with_tablenumber)
+from .MarsRetrieval import MarsRetrieval
+from .UioFiles import UioFiles
 import mods.disaggregation as disaggregation
 
@@ -403 +403 @@
         elif not gauss and not eta:
             self.params['OG__ML'][0] += '/U/V'
-        else:
-            print('Warning: Collecting etadot and parameters for gaussian grid \
-                            is a very costly parameter combination, \
-                            use this combination only for debugging!')
-            self.params['GG__SL'] = ['Q', 'ML', '1', \
+        else:  # GAUSS and ETA
+            print('Warning: Collecting etadot and parameters for gaussian grid '
+                           'is a very costly parameter combination, '
+                           'use this combination only for debugging!')
+            self.params['GG__SL'] = ['Q', 'ML', '1',
                                      '{}'.format((int(self.resol) + 1) / 2)]
-            self.params['GG__ML'] = ['U/V/D/77', 'ML', self.glevelist, \
+            self.params['GG__ML'] = ['U/V/D/ETADOT', 'ML', self.glevelist,
                                      '{}'.format((int(self.resol) + 1) / 2)]
 
@@ -419 +419 @@
 
         # ADDITIONAL FIELDS FOR FLEXPART-WRF MODEL (IF QUESTIONED)
-        #-----------------------------------------------------------------------
+        # -----------------------------------------------------------------------
         if wrf:
             self.params['OG__ML'][0] += '/Z/VO'
             if '/D' not in self.params['OG__ML'][0]:
                 self.params['OG__ML'][0] += '/D'
+
             wrf_sfc = ['SP','SKT','SST','CI','STL1','STL2', 'STL3','STL4',
                        'SWVL1','SWVL2','SWVL3','SWVL4']
@@ -452 +453 @@
 
         '''
-        self.params['OG_acc_SL'] = ["LSP/CP/SSHF/EWSS/NSSS/SSR", \
+        self.params['OG_acc_SL'] = ["LSP/CP/SSHF/EWSS/NSSS/SSR",
                                     'SFC', '1', self.grid]
         return
@@ -684 +685 @@
                                         pk,
                                         retr_param_dict['date'].split('/')[0])
-                retr_param_dict['param'] = pv[0]
+                table128 = init128(_config.PATH_GRIBTABLE)
+                ids = to_param_id_with_tablenumber(pv[0], table128)
+                retr_param_dict['param'] = ids
                 retr_param_dict['levtype'] = pv[1]
                 retr_param_dict['levelist'] = pv[2]
@@ -1238 +1241 @@
 
         '''
-        print('... disaggregation or precipitation with new method.')
+        print('... disaggregation of precipitation with new method.')
         lsp_new_np = np.zeros((ni * nj, nt * 3), dtype=np.float64)
         cp_new_np = np.zeros((ni * nj, nt * 3), dtype=np.float64)
@@ -1268 +1271 @@
                     filename1 = c.prefix + date.strftime('%y%m%d%H') + '_1'
                     filename2 = c.prefix + date.strftime('%y%m%d%H') + '_2'
-
-                # collect for final processing
-                self.outputfilelist.append(os.path.basename(fluxfilename))
-                self.outputfilelist.append(os.path.basename(filename1))
-                self.outputfilelist.append(os.path.basename(filename2))
 
                 # write original time step to flux file as usual
@@ -1449 +1447 @@
             cdate_hour = datetime.strftime(timestamp, '%Y%m%d%H')
 
-            # eliminate all temporary times
+            # skip all temporary times
             # which are outside the retrieval period
             if timestamp < start_period or \
@@ -1545 +1543 @@
             os.chdir(c.inputdir)
             if os.stat('fort.21').st_size == 0 and c.eta:
-                print('Parameter 77 (etadot) is missing, most likely it is \
-                       not available for this type or date/time\n')
+                print('Parameter 77 (etadot) is missing, most likely it is '
+                      'not available for this type or date / time\n')
                 print('Check parameters CLASS, TYPE, STREAM, START_DATE\n')
-                my_error(c.mailfail, 'fort.21 is empty while parameter eta \
-                         is set to 1 in CONTROL file')
-#============================================================================================
+                my_error('fort.21 is empty while parameter eta '
+                         'is set to 1 in CONTROL file')
+# ============================================================================================
             # write out all output to log file before starting fortran programm
             sys.stdout.flush()
@@ -1560 +1558 @@
 
             os.chdir(pwd)
-#============================================================================================
+# ============================================================================================
             # create name of final output file, e.g. EN13040500 (ENYYMMDDHH)
             if c.purefc:
@@ -1577 +1575 @@
             # collect for final processing
             self.outputfilelist.append(os.path.basename(fnout))
-#============================================================================================
+            # get additional precipitation subgrid data if available
+            if c.rrint:
+                self.outputfilelist.append(os.path.basename(fnout + '_1'))
+                self.outputfilelist.append(os.path.basename(fnout + '_2'))
+# ============================================================================================
             # create outputfile and copy all data from intermediate files
             # to the outputfile (final GRIB input files for FLEXPART)
@@ -1597 +1599 @@
                     shutil.copyfileobj(open(os.path.join(c.inputdir, 'fort.25'),
                                             'rb'), fout)
-#============================================================================================
+# ============================================================================================
         if c.wrf:
             fwrf.close()
@@ -1692 +1694 @@
         print('\n\nPostprocessing:\n Format: {}\n'.format(c.format))
 
-        if not c.ecapi:
+        if _config.FLAG_ON_ECMWFSERVER:
             print('ecstorage: {}\n ecfsdir: {}\n'.
                   format(c.ecstorage, c.ecfsdir))
@@ -1714 +1716 @@
                                    'FILES FAILED!')
 
-            if c.ectrans and not c.ecapi:
+            if c.ectrans and _config.FLAG_ON_ECMWFSERVER:
                 execute_subprocess(['ectrans', '-overwrite', '-gateway',
                                     c.gateway, '-remote', c.destination,
@@ -1720 +1722 @@
                                    error_msg='TRANSFER TO LOCAL SERVER FAILED!')
 
-            if c.ecstorage and not c.ecapi:
+            if c.ecstorage and _config.FLAG_ON_ECMWFSERVER:
                 execute_subprocess(['ecp', '-o', ofile,
                                     os.path.expandvars(c.ecfsdir)],

source/python/classes/MarsRetrieval.py

@@ -35 +35 @@
 import sys
 import subprocess
+import traceback
 
 # software specific classes and modules from flex_extract
 sys.path.append('../')
 import _config
+try:
+    ec_api = True
+    import ecmwfapi
+except ImportError:
+    ec_api = False
+
+try:
+    cds_api = True
+    import cdsapi
+except ImportError:
+    cds_api = False
 # ------------------------------------------------------------------------------
 # CLASS
@@ -130 +142 @@
     '''
 
-    def __init__(self, server, public, marsclass="ei", dataset="", type="",
+    def __init__(self, server, public, marsclass="EA", dataset="", type="",
                  levtype="", levelist="", repres="", date="", resol="",
                  stream="", area="", time="", step="", expver="1",
@@ -158 +170 @@
         marsclass : str, optional
             Characterisation of dataset. E.g. EI (ERA-Interim),
-            E4 (ERA40), OD (Operational archive), ea (ERA5).
-            Default is the ERA-Interim dataset "ei".
+            E4 (ERA40), OD (Operational archive), EA (ERA5).
+            Default is the ERA5 dataset "EA".
 
         dataset : str, optional
@@ -429 +441 @@
     def data_retrieve(self):
         '''Submits a MARS retrieval. Depending on the existence of
-        ECMWF Web-API it is submitted via Python or a
+        ECMWF Web-API or CDS API it is submitted via Python or a
         subprocess in the Shell. The parameter for the mars retrieval
         are taken from the defined class attributes.
@@ -452 +464 @@
         attrs['class'] = mclass
 
-        # prepare target variable as needed for the Web API mode
+        # prepare target variable as needed for the Web API or CDS API mode
         # within the dictionary for full access
         # as a single variable for public access
@@ -475 +487 @@
         if self.server:
             try:
-                if self.public:
-                    print('RETRIEVE PUBLIC DATA!')
+                if cds_api and isinstance(self.server, cdsapi.Client):
+                    print('RETRIEVE ERA5 WITH CDS API!')
+                    self.server.retrieve(_config.CDS_DATASET,
+                                         attrs, target)
+                elif ec_api and isinstance(self.server, ecmwfapi.ECMWFDataServer):
+                    print('RETRIEVE PUBLIC DATA (NOT ERA5)!')
                     self.server.retrieve(attrs)
+                elif ec_api and isinstance(self.server, ecmwfapi.ECMWFService):
+                    print('EXECUTE NON-PUBLIC RETRIEVAL (NOT ERA5)!')
+                    self.server.execute(attrs, target)
                 else:
-                    print('EXECUTE NON-PUBLIC RETRIEVAL!')
-                    self.server.execute(attrs, target)
-            except:
-                e = sys.exc_info()[0]
-                print("ERROR: ", e)
-                print('MARS Request failed!')
-                if not self.public and os.stat(target).st_size == 0:
-                    print('MARS Request returned no data - '
-                          'please check request')
+                    print('ERROR:')
+                    print('No match for Web API instance!')
                     raise IOError
-                elif self.public and os.stat(target).st_size == 0:
-                    print('Public MARS Request returned no data - '
-                          'please check request')
-                    raise IOError
-                else:
-                    raise IOError
-        # MARS request via extra process in shell
+            except Exception as e:
+                print('\n\nMARS Request failed!')
+                print(e)
+                tb = sys.exc_info()[2]
+                print(traceback.format_exc())
+                sys.exit()
+
+        # MARS request via call in shell
         else:
             request_str = 'ret'
@@ -501 +514 @@
                 request_str = request_str + ',' + key + '=' + str(value)
             request_str += ',target="' + target + '"'
-            p = subprocess.Popen(['mars'],
+            p = subprocess.Popen(['mars', '-p'],
                                  stdin=subprocess.PIPE,
                                  stdout=subprocess.PIPE,

source/python/mods/checks.py

@@ -27 +27 @@
 import sys
 import _config
-import exceptions
-from tools import my_error, silent_remove
+try:
+    import exceptions
+except ImportError:
+    import builtins as exceptions
+from .tools import my_error, silent_remove
 from datetime import datetime
 import numpy as np
@@ -296 +299 @@
             step = ['{:0>3}'.format(i) for i in ilist]
         elif 'to' in step.lower() and 'by' not in step.lower():
-            my_error(mailfail, step + ':\n' +
+            my_error(step + ':\n' +
                      'if "to" is used in steps parameter, '
                      'please use "by" as well')

source/python/mods/get_mars_data.py

@@ -67 +67 @@
     inspect.getfile(inspect.currentframe()))) + '/../')
 import _config
-from tools import (my_error, normal_exit, get_cmdline_args,
+from .tools import (setup_controldata, my_error, normal_exit, get_cmdline_args,
                    read_ecenv, make_dir)
 from classes.EcFlexpart import EcFlexpart
@@ -74 +74 @@
 
 try:
-    ecapi = True
+    ec_api = True
     import ecmwfapi
 except ImportError:
-    ecapi = False
+    ec_api = False
+
+try:
+    cds_api = True
+    import cdsapi
+except ImportError:
+    cds_api = False
 # ------------------------------------------------------------------------------
 # FUNCTION
@@ -95 +101 @@
     '''
 
-    args = get_cmdline_args()
-    c = ControlFile(args.controlfile)
-
-    env_parameter = read_ecenv(_config.PATH_ECMWF_ENV)
-    c.assign_args_to_control(args)
-    c.assign_envs_to_control(env_parameter)
-    c.check_conditions(args.queue)
-
+    c, _, _, _ = setup_controldata()
     get_mars_data(c)
-    normal_exit(c.mailops, c.queue, 'Done!')
+    normal_exit('Retrieving MARS data: Done!')
 
     return
@@ -125 +124 @@
 
     '''
-    c.ecapi = ecapi
+    c.ec_api = ec_api
+    c.cds_api = cds_api
 
     if not os.path.exists(c.inputdir):
@@ -182 +182 @@
 
 def mk_server(c):
-    '''Creates server connection if ECMWF WebAPI is available.
+    '''Creates a server connection with available python API.
+
+    Which API is used depends on availability and the dataset to be retrieved.
+    The CDS API is used for ERA5 dataset no matter if the user is a member or
+    a public user. ECMWF WebAPI is used for all other available datasets.
 
     Parameters
@@ -192 +196 @@
     Return
     ------
-    server : ECMWFDataServer or ECMWFService
-        Connection to ECMWF server via python interface ECMWF WebAPI.
-
-    '''
-    if c.ecapi:
+    server : ECMWFDataServer, ECMWFService or Client
+        Connection to ECMWF server via python interface ECMWF WebAPI or CDS API.
+
+    '''
+    if cds_api and (c.marsclass.upper() == 'EA'):
+        server = cdsapi.Client()
+        c.ec_api = False
+    elif c.ec_api:
         if c.public:
             server = ecmwfapi.ECMWFDataServer()
         else:
             server = ecmwfapi.ECMWFService("mars")
+        c.cds_api = False
     else:
         server = False
 
-    print('Using ECMWF WebAPI: ' + str(c.ecapi))
+    print('Using ECMWF WebAPI: ' + str(c.ec_api))
+    print('Using CDS API: ' + str(c.cds_api))
 
     return server
@@ -353 +362 @@
             flexpart.retrieve(server, dates, c.public, c.request, c.inputdir)
         except IOError:
-            my_error(c.mailfail, 'MARS request failed')
+            my_error('MARS request failed')
 
         day += delta_t

source/python/mods/prepare_flexpart.py

@@ -71 +71 @@
     inspect.getfile(inspect.currentframe()))) + '/../')
 import _config
-from checks import check_ppid
+from .checks import check_ppid
 from classes.UioFiles import UioFiles
 from classes.ControlFile import ControlFile
-from tools import clean_up, get_cmdline_args, read_ecenv, make_dir
+from .tools import (setup_controldata, clean_up, get_cmdline_args,
+                   read_ecenv, make_dir)
 from classes.EcFlexpart import EcFlexpart
-
-ecapi = 'ecmwf' not in socket.gethostname()
-try:
-    if ecapi:
-        import ecmwfapi
-except ImportError:
-    ecapi = False
 
 # ------------------------------------------------------------------------------
@@ -101 +95 @@
     '''
 
-    args = get_cmdline_args()
-    c = ControlFile(args.controlfile)
-
-    env_parameter = read_ecenv(_config.PATH_ECMWF_ENV)
-    c.assign_args_to_control(args)
-    c.assign_envs_to_control(env_parameter)
-    c.check_conditions(args.queue)
-
-    prepare_flexpart(args.ppid, c)
+    c, ppid, _, _ = setup_controldata()
+    prepare_flexpart(ppid, c)
+    normal_exit('Preparing FLEXPART output files: Done!')
 
     return
@@ -136 +124 @@
     '''
     check_ppid(c, ppid)
-
-    c.ecapi = ecapi
 
     # create the start and end date

source/python/mods/tools.py

@@ -65 +65 @@
 import subprocess
 import traceback
-import exceptions
+try:
+    import exceptions
+except ImportError:
+    import builtins as exceptions
 from datetime import datetime
 from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
+
+
 
 # ------------------------------------------------------------------------------
 # METHODS
 # ------------------------------------------------------------------------------
+
+def setup_controldata():
+    '''Collects, stores and checks controlling arguments from command line,
+    CONTROL file and ECMWF_ENV file.
+
+    Parameters
+    ----------
+
+    Return
+    ------
+    c : ControlFile
+        Contains all the parameters of CONTROL file and
+        command line.
+
+    ppid : str
+        Parent process id.
+
+    queue : str
+        Name of queue for submission to ECMWF (e.g. ecgate or cca )
+
+    job_template : str
+        Name of the job template file for submission to ECMWF server.
+    '''
+    import _config
+    from classes.ControlFile import ControlFile
+
+    args = get_cmdline_args()
+    c = ControlFile(args.controlfile)
+    c.assign_args_to_control(args)
+    if os.path.isfile(_config.PATH_ECMWF_ENV):
+        env_parameter = read_ecenv(_config.PATH_ECMWF_ENV)
+        c.assign_envs_to_control(env_parameter)
+    c.check_conditions(args.queue)
+
+    return c, args.ppid, args.queue, args.job_template
 
 def none_or_str(value):
@@ -259 +299 @@
     print("... clean inputdir!")
 
-    cleanlist = glob.glob(os.path.join(c.inputdir, "*"))
+    cleanlist = [file for file in glob.glob(os.path.join(c.inputdir, "*"))
+                 if not os.path.basename(file).startswith(c.prefix)]
 
     if cleanlist:
         for element in cleanlist:
-            if c.prefix not in element:
-                silent_remove(element)
-            if c.ecapi is False and (c.ectrans == 1 or c.ecstorage == 1):
-                silent_remove(element)
+            silent_remove(element)
         print("... done!")
     else:
@@ -274 +312 @@
 
 
-def my_error(users, message='ERROR'):
+def my_error(message='ERROR'):
     '''Prints a specified error message which can be passed to the function
     before exiting the program.
@@ -280 +318 @@
     Parameters
     ----------
-    user : list of str
-        Contains all email addresses which should be notified.
-        It might also contain just the ecmwf user name which wil trigger
-        mailing to the associated email address for this user.
-
     message : str, optional
         Error message. Default value is "ERROR".
@@ -464 +497 @@
     else:
         for data in fdata:
-            if data[0] != '!':
-                table128[data[0:3]] = data[59:64].strip()
+            if data != '' and data[0] != '!':
+                table128[data[0:3]] = data[59:65].strip()
 
     return table128
@@ -508 +541 @@
 
     return ipar
+
+def to_param_id_with_tablenumber(pars, table):
+    '''Transform parameter names to parameter ids and add table id.
+
+    Conversion with ECMWF grib table 128.
+
+    Parameters
+    ----------
+    pars : str
+        Addpar argument from CONTROL file in the format of
+        parameter names instead of ids. The parameter short
+        names are sepearted with "/" and they are passed as
+        one single string.
+
+    table : dict
+        Contains the ECMWF grib table 128 information.
+        The key is the parameter number and the value is the
+        short name of the parameter.
+
+    Return
+    ------
+    spar : str
+        List of addpar parameters from CONTROL file transformed to
+        parameter ids in the format of integer.
+    '''
+    if not pars:
+        return []
+    if not isinstance(pars, str):
+        pars=str(pars)
+
+    cpar = pars.upper().split('/')
+    spar = []
+    for par in cpar:
+        for k, v in table.iteritems():
+            if par == k or par == v:
+                spar.append(k + '.128')
+                break
+        else:
+            print('\n\n\t\tWarning: par ' + par + ' not found in table 128\n\n')
+
+    return '/'.join(spar)
 
 def get_list_as_string(list_obj, concatenate_sign=', '):

source/python/submit.py

@@ -67 +67 @@
 # software specific classes and modules from flex_extract
 import _config
-from mods.tools import (normal_exit, get_cmdline_args,
+from mods.tools import (setup_controldata, normal_exit, get_cmdline_args,
                         submit_job_to_ecserver, read_ecenv)
 from mods.get_mars_data import get_mars_data
@@ -91 +91 @@
     '''
 
-    args = get_cmdline_args()
-    c = ControlFile(args.controlfile)
-
-    env_parameter = read_ecenv(_config.PATH_ECMWF_ENV)
-    c.assign_args_to_control(args)
-    c.assign_envs_to_control(env_parameter)
-    c.check_conditions(args.queue)
+    c, ppid, queue, job_template = setup_controldata()
 
     # on local side
-    # on ECMWF server this would also be the local side
+    # starting from an ECMWF server this would also be the local side
     called_from_dir = os.getcwd()
-    if args.queue is None:
+    if queue is None:
         if c.inputdir[0] != '/':
             c.inputdir = os.path.join(called_from_dir, c.inputdir)
@@ -109 +103 @@
         get_mars_data(c)
         if c.request == 0 or c.request == 2:
-            prepare_flexpart(args.ppid, c)
+            prepare_flexpart(ppid, c)
             exit_message = 'FLEX_EXTRACT IS DONE!'
         else:
@@ -115 +109 @@
     # send files to ECMWF server
     else:
-        submit(args.job_template, c, args.queue)
+        submit(job_template, c, queue)
         exit_message = 'FLEX_EXTRACT JOB SCRIPT IS SUBMITED!'