source: flex_extract.git/python/tools.py @ b1d07c9

ctbtodev
Last change on this file since b1d07c9 was b1d07c9, checked in by Anne Philipp <anne.philipp@…>, 6 years ago

bugfix: removed an item from the function list of submit_job_to_ecserver

  • Property mode set to 100644
File size: 16.1 KB
Line 
1#!/usr/bin/env python
2# -*- coding: utf-8 -*-
3#*******************************************************************************
4# @Author: Anne Philipp (University of Vienna)
5#
6# @Date: May 2018
7#
8# @Change History:
9#    October 2014 - Anne Fouilloux (University of Oslo)
10#        - created functions silent_remove and product (taken from ECMWF)
11#
12#    November 2015 - Leopold Haimberger (University of Vienna)
13#        - created functions: interpret_args_and_control, clean_up
14#          my_error, normal_exit, init128, to_param_id
15#
16#    April 2018 - Anne Philipp (University of Vienna):
17#        - applied PEP8 style guide
18#        - added documentation
19#        - moved all functions from file Flexparttools to this file tools
20#        - added function get_list_as_string
21#        - seperated args and control interpretation
22#
23# @License:
24#    (C) Copyright 2014-2018.
25#
26#    This software is licensed under the terms of the Apache Licence Version 2.0
27#    which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
28#
29# @Modul Description:
30#    This module contains a couple of helpful functions which are
31#    used in different places in flex_extract.
32#
33# @Module Content:
34#    - get_cmdline_arguments
35#    - clean_up
36#    - my_error
37#    - normal_exit
38#    - product
39#    - silent_remove
40#    - init128
41#    - to_param_id
42#    - get_list_as_string
43#    - make_dir
44#
45#*******************************************************************************
46
47# ------------------------------------------------------------------------------
48# MODULES
49# ------------------------------------------------------------------------------
50import os
51import errno
52import sys
53import glob
54import subprocess
55import traceback
56from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
57
58# ------------------------------------------------------------------------------
59# FUNCTIONS
60# ------------------------------------------------------------------------------
61
62def get_cmdline_arguments():
63    '''
64    @Description:
65        Decomposes the command line arguments and assigns them to variables.
66        Apply default values for non mentioned arguments.
67
68    @Input:
69        <nothing>
70
71    @Return:
72        args: instance of ArgumentParser
73            Contains the commandline arguments from script/program call.
74    '''
75
76    parser = ArgumentParser(description='Retrieve FLEXPART input from \
77                                ECMWF MARS archive',
78                            formatter_class=ArgumentDefaultsHelpFormatter)
79
80    # the most important arguments
81    parser.add_argument("--start_date", dest="start_date", default=None,
82                        help="start date YYYYMMDD")
83    parser.add_argument("--end_date", dest="end_date", default=None,
84                        help="end_date YYYYMMDD")
85    parser.add_argument("--date_chunk", dest="date_chunk", default=None,
86                        help="# of days to be retrieved at once")
87
88    # some arguments that override the default in the CONTROL file
89    parser.add_argument("--basetime", dest="basetime", default=None,
90                        help="base such as 00/12 (for half day retrievals)")
91    parser.add_argument("--step", dest="step", default=None,
92                        help="steps such as 00/to/48")
93    parser.add_argument("--levelist", dest="levelist", default=None,
94                        help="Vertical levels to be retrieved, e.g. 30/to/60")
95    parser.add_argument("--area", dest="area", default=None,
96                        help="area defined as north/west/south/east")
97
98    # set the working directories
99    parser.add_argument("--inputdir", dest="inputdir", default=None,
100                        help="root directory for storing intermediate files")
101    parser.add_argument("--outputdir", dest="outputdir", default=None,
102                        help="root directory for storing output files")
103    parser.add_argument("--flexpart_root_scripts", dest="flexpart_root_scripts",
104                        default=None,
105                        help="FLEXPART root directory (to find grib2flexpart \
106                        and COMMAND file)\n Normally flex_extract resides in \
107                        the scripts directory of the FLEXPART distribution")
108
109    # this is only used by prepare_flexpart.py to rerun a postprocessing step
110    parser.add_argument("--ppid", dest="ppid", default=None,
111                        help="specify parent process id for \
112                        rerun of prepare_flexpart")
113
114    # arguments for job submission to ECMWF, only needed by submit.py
115    parser.add_argument("--job_template", dest='job_template',
116                        default="job.temp",
117                        help="job template file for submission to ECMWF")
118    parser.add_argument("--queue", dest="queue", default=None,
119                        help="queue for submission to ECMWF \
120                        (e.g. ecgate or cca )")
121    parser.add_argument("--controlfile", dest="controlfile",
122                        default='CONTROL.temp',
123                        help="file with CONTROL parameters")
124    parser.add_argument("--debug", dest="debug", default=None,
125                        help="debug mode - leave temporary files intact")
126
127    args = parser.parse_args()
128
129    return args
130
131def read_ecenv(filename):
132    '''
133    @Description:
134        Reads the file into a dictionary where the key values are the parameter
135        names.
136
137    @Input:
138        filename: string
139            Name of file where the ECMWV environment parameters are stored.
140
141    @Return:
142        envs: dict
143    '''
144    envs= {}
145    print filename
146    with open(filename, 'r') as f:
147        for line in f:
148            data = line.strip().split()
149            envs[str(data[0])] = str(data[1])
150
151    return envs
152
153def clean_up(c):
154    '''
155    @Description:
156        Remove all files from intermediate directory
157        (inputdir from CONTROL file).
158
159    @Input:
160        c: instance of class ControlFile
161            Contains all the parameters of CONTROL file, which are e.g.:
162            DAY1(start_date), DAY2(end_date), DTIME, MAXSTEP, TYPE, TIME,
163            STEP, CLASS(marsclass), STREAM, NUMBER, EXPVER, GRID, LEFT,
164            LOWER, UPPER, RIGHT, LEVEL, LEVELIST, RESOL, GAUSS, ACCURACY,
165            OMEGA, OMEGADIFF, ETA, ETADIFF, DPDETA, SMOOTH, FORMAT,
166            ADDPAR, WRF, CWC, PREFIX, ECSTORAGE, ECTRANS, ECFSDIR,
167            MAILOPS, MAILFAIL, GRIB2FLEXPART, FLEXPARTDIR, BASETIME
168            DATE_CHUNK, DEBUG, INPUTDIR, OUTPUTDIR, FLEXPART_ROOT_SCRIPTS
169
170            For more information about format and content of the parameter
171            see documentation.
172
173    @Return:
174        <nothing>
175    '''
176
177    print "clean_up"
178
179    cleanlist = glob.glob(c.inputdir + "/*")
180    for clist in cleanlist:
181        if c.prefix not in clist:
182            silent_remove(clist)
183        if c.ecapi is False and (c.ectrans == '1' or c.ecstorage == '1'):
184            silent_remove(clist)
185
186    print "Done"
187
188    return
189
190
191def my_error(users, message='ERROR'):
192    '''
193    @Description:
194        Prints a specified error message which can be passed to the function
195        before exiting the program.
196
197    @Input:
198        user: list of strings
199            Contains all email addresses which should be notified.
200            It might also contain just the ecmwf user name which wil trigger
201            mailing to the associated email address for this user.
202
203        message: string, optional
204            Error message. Default value is "ERROR".
205
206    @Return:
207        <nothing>
208    '''
209
210    print message
211
212    # comment if user does not want email notification directly from python
213    for user in users:
214        if '${USER}' in user:
215            user = os.getenv('USER')
216        try:
217            p = subprocess.Popen(['mail', '-s flex_extract_v7.1 ERROR',
218                                  os.path.expandvars(user)],
219                                 stdin=subprocess.PIPE,
220                                 stdout=subprocess.PIPE,
221                                 stderr=subprocess.PIPE,
222                                 bufsize=1)
223            trace = '\n'.join(traceback.format_stack())
224            pout = p.communicate(input=message + '\n\n' + trace)[0]
225        except ValueError as e:
226            print 'ERROR: ', e
227            sys.exit('Email could not be sent!')
228        else:
229            print 'Email sent to ' + os.path.expandvars(user) + ' ' + \
230                  pout.decode()
231
232    sys.exit(1)
233
234    return
235
236
237def normal_exit(users, message='Done!'):
238    '''
239    @Description:
240        Prints a specific exit message which can be passed to the function.
241
242    @Input:
243        user: list of strings
244            Contains all email addresses which should be notified.
245            It might also contain just the ecmwf user name which wil trigger
246            mailing to the associated email address for this user.
247
248        message: string, optional
249            Message for exiting program. Default value is "Done!".
250
251    @Return:
252        <nothing>
253
254    '''
255    print message
256
257    # comment if user does not want notification directly from python
258    for user in users:
259        if '${USER}' in user:
260            user = os.getenv('USER')
261        try:
262            p = subprocess.Popen(['mail', '-s flex_extract_v7.1 normal exit',
263                                  os.path.expandvars(user)],
264                                 stdin=subprocess.PIPE,
265                                 stdout=subprocess.PIPE,
266                                 stderr=subprocess.PIPE,
267                                 bufsize=1)
268            pout = p.communicate(input=message+'\n\n')[0]
269        except ValueError as e:
270            print 'ERROR: ', e
271            print 'Email could not be sent!'
272        else:
273            print 'Email sent to ' + os.path.expandvars(user) + ' ' + \
274                  pout.decode()
275
276    return
277
278
279def product(*args, **kwds):
280    '''
281    @Description:
282        This method is taken from an example at the ECMWF wiki website.
283        https://software.ecmwf.int/wiki/display/GRIB/index.py; 2018-03-16
284
285        This method combines the single characters of the passed arguments
286        with each other. So that each character of each argument value
287        will be combined with each character of the other arguments as a tuple.
288
289        Example:
290        product('ABCD', 'xy') --> Ax Ay Bx By Cx Cy Dx Dy
291        product(range(2), repeat = 3) --> 000 001 010 011 100 101 110 111
292
293    @Input:
294        *args: tuple
295            Positional arguments (arbitrary number).
296
297        **kwds: dictionary
298            Contains all the keyword arguments from *args.
299
300    @Return:
301        prod: tuple
302            Return will be done with "yield". A tuple of combined arguments.
303            See example in description above.
304    '''
305    pools = map(tuple, args) * kwds.get('repeat', 1)
306    result = [[]]
307    for pool in pools:
308        result = [x + [y] for x in result for y in pool]
309    for prod in result:
310        yield tuple(prod)
311
312    return
313
314
315def silent_remove(filename):
316    '''
317    @Description:
318        If "filename" exists , it is removed.
319        The function does not fail if the file does not exist.
320
321    @Input:
322        filename: string
323            The name of the file to be removed without notification.
324
325    @Return:
326        <nothing>
327    '''
328    try:
329        os.remove(filename)
330    except OSError as e:
331        # this would be "except OSError, e:" before Python 2.6
332        if e.errno != errno.ENOENT:
333            # errno.ENOENT  =  no such file or directory
334            raise  # re-raise exception if a different error occured
335
336    return
337
338
339def init128(filepath):
340    '''
341    @Description:
342        Opens and reads the grib file with table 128 information.
343
344    @Input:
345        filepath: string
346            Path to file of ECMWF grib table number 128.
347
348    @Return:
349        table128: dictionary
350            Contains the ECMWF grib table 128 information.
351            The key is the parameter number and the value is the
352            short name of the parameter.
353    '''
354    table128 = dict()
355    with open(filepath) as f:
356        fdata = f.read().split('\n')
357    for data in fdata:
358        if data[0] != '!':
359            table128[data[0:3]] = data[59:64].strip()
360
361    return table128
362
363
364def to_param_id(pars, table):
365    '''
366    @Description:
367        Transform parameter names to parameter ids
368        with ECMWF grib table 128.
369
370    @Input:
371        pars: string
372            Addpar argument from CONTROL file in the format of
373            parameter names instead of ids. The parameter short
374            names are sepearted with "/" and they are passed as
375            one single string.
376
377        table: dictionary
378            Contains the ECMWF grib table 128 information.
379            The key is the parameter number and the value is the
380            short name of the parameter.
381
382    @Return:
383        ipar: list of integer
384            List of addpar parameters from CONTROL file transformed to
385            parameter ids in the format of integer.
386    '''
387    cpar = pars.upper().split('/')
388    ipar = []
389    for par in cpar:
390        for k, v in table.iteritems():
391            if par == k or par == v:
392                ipar.append(int(k))
393                break
394        else:
395            print 'Warning: par ' + par + ' not found in table 128'
396
397    return ipar
398
399def get_list_as_string(list_obj, concatenate_sign=', '):
400    '''
401    @Description:
402        Converts a list of arbitrary content into a single string.
403
404    @Input:
405        list_obj: list
406            A list with arbitrary content.
407
408        concatenate_sign: string, optional
409            A string which is used to concatenate the single
410            list elements. Default value is ", ".
411
412    @Return:
413        str_of_list: string
414            The content of the list as a single string.
415    '''
416
417    str_of_list = concatenate_sign.join(str(l) for l in list_obj)
418
419    return str_of_list
420
421def make_dir(directory):
422    '''
423    @Description:
424        Creates a directory and gives a warning if the directory
425        already exists. The program stops only if there is another problem.
426
427    @Input:
428        directory: string
429            The directory path which should be created.
430
431    @Return:
432        <nothing>
433    '''
434    try:
435        os.makedirs(directory)
436    except OSError as e:
437        if e.errno != errno.EEXIST:
438            # errno.EEXIST = directory already exists
439            raise # re-raise exception if a different error occured
440        else:
441            print 'WARNING: Directory {0} already exists!'.format(directory)
442
443    return
444
445def put_file_to_ecserver(ecd, filename, target, ecuid, ecgid):
446    '''
447    @Description:
448        Uses the ecaccess command to send a file to the ECMWF servers.
449        Catches and prints the error if it failed.
450
451    @Input:
452        ecd: string
453            The path were the file is to be stored.
454
455        filename: string
456            The name of the file to send to the ECMWF server.
457
458        target: string
459            The target where the file should be sent to, e.g. the queue.
460
461        ecuid: string
462            The user id on ECMWF server.
463
464        ecgid: string
465            The group id on ECMWF server.
466
467    @Return:
468        <nothing>
469    '''
470
471    try:
472        subprocess.check_call(['ecaccess-file-put',
473                               ecd + '../' + filename,
474                               target + ':/home/ms/' +
475                               ecgid + '/' + ecuid +
476                               '/' + filename])
477    except subprocess.CalledProcessError as e:
478        print 'ERROR:'
479        print e
480        sys.exit('ecaccess-file-put failed!\n' + \
481                 'Probably the eccert key has expired.')
482
483    return
484
485def submit_job_to_ecserver(target, jobname):
486    '''
487    @Description:
488        Uses ecaccess to submit a job to the ECMWF server.
489        Catches and prints the error if one arise.
490
491    @Input:
492        target: string
493            The target where the file should be sent to, e.g. the queue.
494
495        jobname: string
496            The name of the jobfile to be submitted to the ECMWF server.
497
498    @Return:
499        rcode: integer
500            Resulting code of subprocess.check_call.
501    '''
502
503    try:
504        rcode = subprocess.check_call(['ecaccess-job-submit',
505                                       '-queueName', target,
506                                       jobname])
507    except subprocess.CalledProcessError as e:
508        print '... ERROR CODE: ', e.returncode
509        sys.exit('... ECACCESS-JOB-SUBMIT FAILED!')
510
511    return rcode
Note: See TracBrowser for help on using the repository browser.
hosted by ZAMG