8687903335

Committed 15 Apr 2024 10:57AM UTC coverage: 74.79% (+7.3%) from 67.499%

Build # 8687903335

Build Type

Pull #60

github

Committed by

web-flow

Commit Message

Merge c97711765 into fc4314d58

Pull Request Pull Request #60: PPS 2021 refactor

Run Details

219 of 371 branches covered (59.03%)

Branch coverage included in aggregate %.

338 of 371 new or added lines in 15 files covered. (91.11%)

8 existing lines in 5 files now uncovered.

1383 of 1771 relevant lines covered (78.09%)

3.12 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/nwcsafpps_runner/metno_update_nwp.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# Copyright (c) 2016 - 2021 Trygve Aspenes

# Author(s):

#   trygveas@met.no

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.

# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""Metno version of preparing the ECMWF nwp data for PPS
"""

import logging
import os
import tempfile
from datetime import datetime
from glob import glob

import eccodes as ecc
import numpy as np

LOG = logging.getLogger(__name__)


class WrongLengthError:
    pass


class NoOptionError:
    pass


def product(*args, **kwds):
    # product('ABCD', 'xy') --> Ax Ay Bx By Cx Cy Dx Dy
    # product(range(2), repeat=3) --> 000 001 010 011 100 101 110 111
    pools = list(map(tuple, args)) * kwds.get('repeat', 1)
    result = [[]]
    for pool in pools:
        result = [x + [y] for x in result for y in pool]
    for prod in result:
        yield tuple(prod)


def copy_needed_field(gid, fout):
    """Copy the needed field"""

    nx = ecc.codes_get(gid, 'Ni')
    ny = ecc.codes_get(gid, 'Nj')
    first_lat = ecc.codes_get(gid, 'latitudeOfFirstGridPointInDegrees')
    north_south_step = ecc.codes_get(gid, 'jDirectionIncrementInDegrees')

    filter_north = 0
    new_ny = int((first_lat - filter_north)/north_south_step) + 1

    values = ecc.codes_get_values(gid)
    values_r = np.reshape(values, (ny, nx))

    new_values = values_r[:new_ny, :]

    clone_id = ecc.codes_clone(gid)

    ecc.codes_set(clone_id, 'latitudeOfLastGridPointInDegrees', (filter_north))
    ecc.codes_set(clone_id, 'Nj', new_ny)

    ecc.codes_set_values(clone_id, new_values.flatten())

    ecc.codes_write(clone_id, fout)
    ecc.codes_release(clone_id)


def update_nwp(params):
    LOG.info("METNO update nwp")

    tempfile.tempdir = params['options']['nwp_outdir']

    ecmwf_path = params['options']['ecmwf_path']
    if not os.path.exists(ecmwf_path):
        ecmwf_path = ecmwf_path.replace("storeB", "storeA")
        LOG.warning("Need to replace storeB with storeA for ecmwf_path: {}".format(str(ecmwf_path)))

    filelist = glob(os.path.join(ecmwf_path,  params['options']['ecmwf_prefix'] + "*"))

    if len(filelist) == 0:
        LOG.info("Found no input files! dir = " +
                 str(os.path.join(ecmwf_path,  params['options']['ecmwf_prefix'] + "*")))
        return

    from trollsift import Parser, compose
    filelist.sort()
    for filename in filelist:
        if os.path.basename(filename).endswith('md5'):
            continue
        if params['options']['ecmwf_file_name_sift'] is not None:
            try:
                parser = Parser(params['options']['ecmwf_file_name_sift'])
            except NoOptionError as noe:
                LOG.error("NoOptionError {}".format(noe))
                continue
            if not parser.validate(os.path.basename(filename)):
                LOG.error("Parser validate on filename: {} failed.".format(filename))
                continue
            res = parser.parse("{}".format(os.path.basename(filename)))

            filename_n1s = filename.replace('N2D', 'N1S')
            if not os.path.exists(filename_n1s):
                LOG.error("Could not find needed file %s. Skip.", filename_n1s)
                continue

            # Do the static fields
            # Note: field not in the filename variable, but a configured filename for static fields
            static_filename = params['options']['ecmwf_static_surface']
            if not os.path.exists(static_filename):
                static_filename = static_filename.replace("storeB", "storeA")
                LOG.warning("Need to replace storeB with storeA")
                if not os.path.exists(static_filename):
                    LOG.error("Could not find needed file %s", static_filename)
                    LOG.error("Skip preparing nwp file.")
                    continue

            time_now = datetime.utcnow()
            if 'analysis_time' in res:
                if res['analysis_time'].year == 1900:
                    # This is tricky. Filename is missing year in name
                    # Need to guess the year from a compination of year now
                    # and month now and month of the analysis time taken from the filename
                    # If the month now is 1(January) and the analysis month is 12,
                    # then the time has passed New Year, but the NWP analysis time is previous year.
                    if time_now.month == 1 and res['analysis_time'].month == 12:
                        analysis_year = time_now.year-1
                    else:
                        analysis_year = time_now.year

                    res['analysis_time'] = res['analysis_time'].replace(year=analysis_year)
            else:
                LOG.error("Can not parse analysis_time in file name. Check config and filename timestamp")

            if 'forecast_time' in res:
                if res['forecast_time'].year == 1900:
                    # See above for explanation
                    if res['analysis_time'].month == 12 and res['forecast_time'].month == 1:
                        forecast_year = res['analysis_time'].year+1
                    else:
                        forecast_year = res['analysis_time'].year

                    res['forecast_time'] = res['forecast_time'].replace(year=forecast_year)
            else:
                LOG.error("Can not parse forecast_time in file name. Check config and filename timestamp")

            forecast_time = res['forecast_time']
            analysis_time = res['analysis_time']
            step_delta = forecast_time - analysis_time
            step = "{:03d}H{:02d}M".format(int(step_delta.days*24 + step_delta.seconds/3600), 0)
        else:
            LOG.error("Not sift pattern given. Can not parse input NWP files")

        if analysis_time < params['starttime']:
            # LOG.debug("skip analysis time {} older than search time {}".format(analysis_time, params['starttime']))
            continue

        if int(step[:3]) not in params['nlengths']:
            # LOG.debug("Skip step {}, not in {}".format(int(step[:3]), params['nlengths']))
            continue

        output_parameters = {}
        output_parameters['analysis_time'] = analysis_time
        output_parameters['step_hour'] = int(step_delta.days*24 + step_delta.seconds/3600)
        output_parameters['step_min'] = 0
        try:
            if not os.path.exists(params['options']['nwp_outdir']):
                os.makedirs(params['options']['nwp_outdir'])
        except OSError as e:
            LOG.error("Failed to create directory: %s", e)
        result_file = ""
        try:
            result_file = os.path.join(params['options']['nwp_outdir'], compose(
                params['options']['nwp_output'], output_parameters))
            _result_file = os.path.join(params['options']['nwp_outdir'], compose(
                "."+params['options']['nwp_output'], output_parameters))
            _result_file_lock = os.path.join(params['options']['nwp_outdir'], compose(
                "."+params['options']['nwp_output']+".lock", output_parameters))
        except Exception as e:
            LOG.error("Joining outdir with output for nwp failed with: {}".format(e))

        LOG.info("Result file: {}".format(result_file))
        if os.path.exists(result_file):
            LOG.info("File: " + str(result_file) + " already there...")
            continue

        import errno
        import fcntl
        import time
        rfl = open(_result_file_lock, 'w+')
        # do some locking
        while True:
            try:
                fcntl.flock(rfl, fcntl.LOCK_EX | fcntl.LOCK_NB)
                LOG.debug("1Got lock for NWP outfile: {}".format(result_file))
                break
            except IOError as e:
                if e.errno != errno.EAGAIN:
                    raise
                else:
                    LOG.debug("Waiting for lock ... {}".format(result_file))
                    time.sleep(1)

        if os.path.exists(result_file):
            LOG.info("File: " + str(result_file) + " already there...")
            # Need to release the lock
            fcntl.flock(rfl, fcntl.LOCK_UN)
            rfl.close()
            continue

        fout = open(_result_file, 'wb')
        try:

            index_vals = []
            index_keys = ['paramId', 'level']
            LOG.debug("Start building index")
            LOG.debug("Handeling file: %s", filename)
            iid = ecc.codes_index_new_from_file(filename, index_keys)
            LOG.debug("Add to index %s", filename_n1s)
            ecc.codes_index_add_file(iid, filename_n1s)
            LOG.debug("Add to index %s", static_filename)
            ecc.codes_index_add_file(iid, static_filename)
            LOG.debug("Done index")
            for key in index_keys:
                key_vals = ecc.codes_index_get(iid, key)
                key_vals = tuple(x for x in key_vals if x != 'undef')
                index_vals.append(key_vals)

            for prod in product(*index_vals):
                for i in range(len(index_keys)):
                    ecc.codes_index_select(iid, index_keys[i], prod[i])

                while 1:
                    gid = ecc.codes_new_from_index(iid)
                    if gid is None:
                        break

                    param = ecc.codes_get(gid, index_keys[0])
                    parameters = [172, 129, 235, 165, 166, 167, 168, 137, 130, 131, 132, 133, 134, 157, 141]
                    if param in parameters:
                        LOG.debug("Doing param: %d", param)
                        copy_needed_field(gid, fout)

                    ecc.codes_release(gid)
            ecc.codes_index_release(iid)

            fout.close()
            os.rename(_result_file, result_file)

        except WrongLengthError as wle:
            LOG.error("Something wrong with the data: %s", wle)
            raise

        # In the end release the lock
        fcntl.flock(rfl, fcntl.LOCK_UN)
        rfl.close()
        os.remove(_result_file_lock)
    return

1	#!/usr/bin/env python
2	# -- coding: utf-8 --
3
4	# Copyright (c) 2016 - 2021 Trygve Aspenes
5
6	# Author(s):
7
8	# trygveas@met.no
9
10	# This program is free software: you can redistribute it and/or modify
11	# it under the terms of the GNU General Public License as published by
12	# the Free Software Foundation, either version 3 of the License, or
13	# (at your option) any later version.
14
15	# This program is distributed in the hope that it will be useful,
16	# but WITHOUT ANY WARRANTY; without even the implied warranty of
17	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18	# GNU General Public License for more details.
19
20	# You should have received a copy of the GNU General Public License
21	# along with this program. If not, see <http://www.gnu.org/licenses/>.
22
23	"""Metno version of preparing the ECMWF nwp data for PPS	×
24	"""
25
26	import logging	×
27	import os	×
NEW 28	import tempfile	×
29	from datetime import datetime	×
NEW 30	from glob import glob	×
31
UNCOV 32	import eccodes as ecc	×
NEW 33	import numpy as np	×
34
35	LOG = logging.getLogger(__name__)	×
36
37
38	class WrongLengthError:	×
39	pass	×
40
41
42	class NoOptionError:	×
43	pass	×
44
45
46	def product(args, *kwds):	×
47	# product('ABCD', 'xy') --> Ax Ay Bx By Cx Cy Dx Dy
48	# product(range(2), repeat=3) --> 000 001 010 011 100 101 110 111
49	pools = list(map(tuple, args)) * kwds.get('repeat', 1)	×
50	result = [[]]	×
51	for pool in pools:	×
52	result = [x + [y] for x in result for y in pool]	×
53	for prod in result:	×
54	yield tuple(prod)	×
55
56
57	def copy_needed_field(gid, fout):	×
58	"""Copy the needed field"""
59
60	nx = ecc.codes_get(gid, 'Ni')	×
61	ny = ecc.codes_get(gid, 'Nj')	×
62	first_lat = ecc.codes_get(gid, 'latitudeOfFirstGridPointInDegrees')	×
63	north_south_step = ecc.codes_get(gid, 'jDirectionIncrementInDegrees')	×
64
65	filter_north = 0	×
66	new_ny = int((first_lat - filter_north)/north_south_step) + 1	×
67
68	values = ecc.codes_get_values(gid)	×
69	values_r = np.reshape(values, (ny, nx))	×
70
71	new_values = values_r[:new_ny, :]	×
72
73	clone_id = ecc.codes_clone(gid)	×
74
75	ecc.codes_set(clone_id, 'latitudeOfLastGridPointInDegrees', (filter_north))	×
76	ecc.codes_set(clone_id, 'Nj', new_ny)	×
77
78	ecc.codes_set_values(clone_id, new_values.flatten())	×
79
80	ecc.codes_write(clone_id, fout)	×
81	ecc.codes_release(clone_id)	×
82
83
84	def update_nwp(params):	×
85	LOG.info("METNO update nwp")	×
86
87	tempfile.tempdir = params['options']['nwp_outdir']	×
88
89	ecmwf_path = params['options']['ecmwf_path']	×
90	if not os.path.exists(ecmwf_path):	×
91	ecmwf_path = ecmwf_path.replace("storeB", "storeA")	×
92	LOG.warning("Need to replace storeB with storeA for ecmwf_path: {}".format(str(ecmwf_path)))	×
93
94	filelist = glob(os.path.join(ecmwf_path, params['options']['ecmwf_prefix'] + "*"))	×
95
96	if len(filelist) == 0:	×
97	LOG.info("Found no input files! dir = " +	×
98	str(os.path.join(ecmwf_path, params['options']['ecmwf_prefix'] + "*")))
99	return	×
100
101	from trollsift import Parser, compose	×
102	filelist.sort()	×
103	for filename in filelist:	×
104	if os.path.basename(filename).endswith('md5'):	×
105	continue	×
106	if params['options']['ecmwf_file_name_sift'] is not None:	×
107	try:	×
108	parser = Parser(params['options']['ecmwf_file_name_sift'])	×
109	except NoOptionError as noe:	×
110	LOG.error("NoOptionError {}".format(noe))	×
111	continue	×
112	if not parser.validate(os.path.basename(filename)):	×
113	LOG.error("Parser validate on filename: {} failed.".format(filename))	×
114	continue	×
115	res = parser.parse("{}".format(os.path.basename(filename)))	×
116
117	filename_n1s = filename.replace('N2D', 'N1S')	×
118	if not os.path.exists(filename_n1s):	×
119	LOG.error("Could not find needed file %s. Skip.", filename_n1s)	×
120	continue	×
121
122	# Do the static fields
123	# Note: field not in the filename variable, but a configured filename for static fields
124	static_filename = params['options']['ecmwf_static_surface']	×
125	if not os.path.exists(static_filename):	×
126	static_filename = static_filename.replace("storeB", "storeA")	×
127	LOG.warning("Need to replace storeB with storeA")	×
128	if not os.path.exists(static_filename):	×
129	LOG.error("Could not find needed file %s", static_filename)	×
130	LOG.error("Skip preparing nwp file.")	×
131	continue	×
132
133	time_now = datetime.utcnow()	×
134	if 'analysis_time' in res:	×
135	if res['analysis_time'].year == 1900:	×
136	# This is tricky. Filename is missing year in name
137	# Need to guess the year from a compination of year now
138	# and month now and month of the analysis time taken from the filename
139	# If the month now is 1(January) and the analysis month is 12,
140	# then the time has passed New Year, but the NWP analysis time is previous year.
141	if time_now.month == 1 and res['analysis_time'].month == 12:	×
142	analysis_year = time_now.year-1	×
143	else:
144	analysis_year = time_now.year	×
145
146	res['analysis_time'] = res['analysis_time'].replace(year=analysis_year)	×
147	else:
148	LOG.error("Can not parse analysis_time in file name. Check config and filename timestamp")	×
149
150	if 'forecast_time' in res:	×
151	if res['forecast_time'].year == 1900:	×
152	# See above for explanation
153	if res['analysis_time'].month == 12 and res['forecast_time'].month == 1:	×
154	forecast_year = res['analysis_time'].year+1	×
155	else:
156	forecast_year = res['analysis_time'].year	×
157
158	res['forecast_time'] = res['forecast_time'].replace(year=forecast_year)	×
159	else:
160	LOG.error("Can not parse forecast_time in file name. Check config and filename timestamp")	×
161
162	forecast_time = res['forecast_time']	×
163	analysis_time = res['analysis_time']	×
164	step_delta = forecast_time - analysis_time	×
165	step = "{:03d}H{:02d}M".format(int(step_delta.days*24 + step_delta.seconds/3600), 0)	×
166	else:
167	LOG.error("Not sift pattern given. Can not parse input NWP files")	×
168
169	if analysis_time < params['starttime']:	×
170	# LOG.debug("skip analysis time {} older than search time {}".format(analysis_time, params['starttime']))
171	continue	×
172
173	if int(step[:3]) not in params['nlengths']:	×
174	# LOG.debug("Skip step {}, not in {}".format(int(step[:3]), params['nlengths']))
175	continue	×
176
177	output_parameters = {}	×
178	output_parameters['analysis_time'] = analysis_time	×
179	output_parameters['step_hour'] = int(step_delta.days*24 + step_delta.seconds/3600)	×
180	output_parameters['step_min'] = 0	×
181	try:	×
182	if not os.path.exists(params['options']['nwp_outdir']):	×
183	os.makedirs(params['options']['nwp_outdir'])	×
184	except OSError as e:	×
185	LOG.error("Failed to create directory: %s", e)	×
186	result_file = ""	×
187	try:	×
188	result_file = os.path.join(params['options']['nwp_outdir'], compose(	×
189	params['options']['nwp_output'], output_parameters))
190	_result_file = os.path.join(params['options']['nwp_outdir'], compose(	×
191	"."+params['options']['nwp_output'], output_parameters))
192	_result_file_lock = os.path.join(params['options']['nwp_outdir'], compose(	×
193	"."+params['options']['nwp_output']+".lock", output_parameters))
194	except Exception as e:	×
195	LOG.error("Joining outdir with output for nwp failed with: {}".format(e))	×
196
197	LOG.info("Result file: {}".format(result_file))	×
198	if os.path.exists(result_file):	×
199	LOG.info("File: " + str(result_file) + " already there...")	×
200	continue	×
201
202	import errno	×
NEW 203	import fcntl	×
204	import time	×
205	rfl = open(_result_file_lock, 'w+')	×
206	# do some locking
207	while True:	×
208	try:	×
209	fcntl.flock(rfl, fcntl.LOCK_EX \| fcntl.LOCK_NB)	×
210	LOG.debug("1Got lock for NWP outfile: {}".format(result_file))	×
211	break	×
212	except IOError as e:	×
213	if e.errno != errno.EAGAIN:	×
214	raise	×
215	else:
216	LOG.debug("Waiting for lock ... {}".format(result_file))	×
217	time.sleep(1)	×
218
219	if os.path.exists(result_file):	×
220	LOG.info("File: " + str(result_file) + " already there...")	×
221	# Need to release the lock
222	fcntl.flock(rfl, fcntl.LOCK_UN)	×
223	rfl.close()	×
224	continue	×
225
226	fout = open(_result_file, 'wb')	×
227	try:	×
228
229	index_vals = []	×
230	index_keys = ['paramId', 'level']	×
231	LOG.debug("Start building index")	×
232	LOG.debug("Handeling file: %s", filename)	×
233	iid = ecc.codes_index_new_from_file(filename, index_keys)	×
234	LOG.debug("Add to index %s", filename_n1s)	×
235	ecc.codes_index_add_file(iid, filename_n1s)	×
236	LOG.debug("Add to index %s", static_filename)	×
237	ecc.codes_index_add_file(iid, static_filename)	×
238	LOG.debug("Done index")	×
239	for key in index_keys:	×
240	key_vals = ecc.codes_index_get(iid, key)	×
241	key_vals = tuple(x for x in key_vals if x != 'undef')	×
242	index_vals.append(key_vals)	×
243
244	for prod in product(*index_vals):	×
245	for i in range(len(index_keys)):	×
246	ecc.codes_index_select(iid, index_keys[i], prod[i])	×
247
248	while 1:	×
249	gid = ecc.codes_new_from_index(iid)	×
250	if gid is None:	×
251	break	×
252
253	param = ecc.codes_get(gid, index_keys[0])	×
254	parameters = [172, 129, 235, 165, 166, 167, 168, 137, 130, 131, 132, 133, 134, 157, 141]	×
255	if param in parameters:	×
256	LOG.debug("Doing param: %d", param)	×
257	copy_needed_field(gid, fout)	×
258
259	ecc.codes_release(gid)	×
260	ecc.codes_index_release(iid)	×
261
262	fout.close()	×
263	os.rename(_result_file, result_file)	×
264
265	except WrongLengthError as wle:	×
266	LOG.error("Something wrong with the data: %s", wle)	×
267	raise	×
268
269	# In the end release the lock
270	fcntl.flock(rfl, fcntl.LOCK_UN)	×
271	rfl.close()	×
272	os.remove(_result_file_lock)	×
273	return	×

pytroll / pytroll-pps-runner / 8687903335

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous