1408365101

Build Type

github

Committed by GitHub

Commit Message

Merge 260baf68f into a622f504d

Pull Request Pull Request #29: Add pps2021 support

Run Details

100 of 424 branches covered (23.58%)

Branch coverage included in aggregate %.

157 of 210 new or added lines in 8 files covered. (74.76%)

3 existing lines in 2 files now uncovered.

943 of 1733 relevant lines covered (54.41%)

2.16 hits per line

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

0.0

/nwcsafpps_runner/prepare_nwp.py

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

# Copyright (c) 2015 - 2021 Pytroll

# Author(s):

#   Adam.Dybbroe <adam.dybbroe@smhi.se>

# 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/>.

"""Prepare NWP data for PPS
"""

from glob import glob
import os
from datetime import datetime
import time
import tempfile
from trollsift import Parser
import pygrib  # @UnresolvedImport
from six.moves.configparser import NoOptionError

from nwcsafpps_runner.config import get_config
from nwcsafpps_runner.config import CONFIG_FILE
from nwcsafpps_runner.config import CONFIG_PATH  # @UnresolvedImport
from nwcsafpps_runner.utils import run_command
from nwcsafpps_runner.utils import NwpPrepareError

import logging
LOG = logging.getLogger(__name__)


LOG.debug("Path to prepare_nwp config file = %s", str(CONFIG_PATH))
LOG.debug("Prepare_nwp config file = %s", str(CONFIG_FILE))
OPTIONS = get_config(CONFIG_FILE)

try:
    nhsp_path = OPTIONS['nhsp_path']
except KeyError:
    LOG.exception('Parameter not set in config file: ' + 'nhsp_path')
try:
    nhsp_prefix = OPTIONS['nhsp_prefix']
except KeyError:
    LOG.exception('Parameter not set in config file: ' + 'nhsp_prefix')

nhsf_file_name_sift = OPTIONS.get('nhsf_file_name_sift')

nhsf_path = OPTIONS.get('nhsf_path', None)
nhsf_prefix = OPTIONS.get('nhsf_prefix', None)
nwp_outdir = OPTIONS.get('nwp_outdir', None)
nwp_lsmz_filename = OPTIONS.get('nwp_static_surface', None)
nwp_output_prefix = OPTIONS.get('nwp_output_prefix', None)
nwp_req_filename = OPTIONS.get('pps_nwp_requirements', None)


def logreader(stream, log_func):
    while True:
        s = stream.readline()
        if not s:
            break
        log_func(s.strip())
    stream.close()


def make_temp_filename(*args, **kwargs):
    tmp_filename_handle, tmp_filename = tempfile.mkstemp(*args, **kwargs)
    os.close(tmp_filename_handle)
    return tmp_filename


def update_nwp(starttime, nlengths):
    """Prepare NWP grib files for PPS. Consider only analysis times newer than
    *starttime*. And consider only the forecast lead times in hours given by
    the list *nlengths* of integers

    """

    LOG.info("Path to prepare_nwp config file = %s", str(CONFIG_PATH))
    LOG.info("Prepare_nwp config file = %s", str(CONFIG_FILE))
    LOG.info("Path to nhsf files: %s", str(nhsf_path))
    LOG.info("Path to nhsp files: %s", str(nhsp_path))

    tempfile.tempdir = nwp_outdir
    filelist = glob(os.path.join(nhsf_path, nhsf_prefix + "*"))
    if len(filelist) == 0:
        LOG.info("No input files! dir = %s", str(nhsf_path))
        return

    LOG.debug('NHSF NWP files found = %s', str(filelist))
    nfiles_error = 0
    for filename in filelist:
        if nhsf_file_name_sift is None:
            raise NwpPrepareError()

        try:
            parser = Parser(nhsf_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)))
        if 'analysis_time' in res:
            if res['analysis_time'].year == 1900:
                res['analysis_time'] = res['analysis_time'].replace(year=datetime.utcnow().year)

            analysis_time = res['analysis_time']
            timestamp = analysis_time.strftime("%Y%m%d%H%M")
        else:
            raise NwpPrepareError("Can not parse analysis_time in file name. Check config and filename timestamp")

        if 'forecast_time' in res:
            if res['forecast_time'].year == 1900:
                res['forecast_time'] = res['forecast_time'].replace(year=datetime.utcnow().year)
            forecast_time = res['forecast_time']
            forecast_step = forecast_time - analysis_time
            forecast_step = "{:03d}H{:02d}M".format(forecast_step.days*24 + forecast_step.seconds/3600, 0)
            timeinfo = "{:s}{:s}{:s}".format(analysis_time.strftime(
                "%m%d%H%M"), forecast_time.strftime("%m%d%H%M"), res['end'])
        else:
            # This needs to be done more solid using the sift pattern! FIXME!
            timeinfo = filename.rsplit("_", 1)[-1]
            # Forecast step in hours:
            if 'forecast_step' in res:
                forecast_step = res['forecast_step']
            else:
                raise NwpPrepareError(
                    'Failed parsing forecast_step in file name. Check config and filename timestamp.')

        if analysis_time < starttime:
            continue
        if forecast_step not in nlengths:
            LOG.debug("Skip step. Forecast step and nlengths: %s %s", str(forecast_step), str(nlengths))
            continue

        LOG.debug("Analysis time and start time: %s %s", str(analysis_time), str(starttime))
        LOG.info("timestamp, step: %s %s", str(timestamp), str(forecast_step))
        result_file = os.path.join(
            nwp_outdir, nwp_output_prefix + timestamp + "+" + '%.3dH00M' % forecast_step)
        if os.path.exists(result_file):
            LOG.info("File: " + str(result_file) + " already there...")
            continue

        tmp_filename = make_temp_filename(suffix="_" + timestamp + "+" +
                                          '%.3dH00M' % forecast_step, dir=nwp_outdir)

        LOG.info("result and tmp files: " + str(result_file) + " " + str(tmp_filename))
        nhsp_file = os.path.join(nhsp_path, nhsp_prefix + timeinfo)
        if not os.path.exists(nhsp_file):
            LOG.warning("Corresponding nhsp-file not there: " + str(nhsp_file))
            continue

        cmd = ("grib_copy -w gridType=regular_ll " + nhsp_file + " " + tmp_filename)
        retv = run_command(cmd)
        LOG.debug("Returncode = " + str(retv))
        if retv != 0:
            LOG.error(
                "Failed doing the grib-copy! Will continue with the next file")
            nfiles_error = nfiles_error + 1
            if nfiles_error > len(filelist) / 2:
                LOG.error(
                    "More than half of the Grib files failed upon grib_copy!")
                raise IOError('Failed running grib_copy on many Grib files')

        if not os.path.exists(nwp_lsmz_filename):
            LOG.error("No static grib file with land-sea mask and " +
                      "topography available. Can't prepare NWP data")
            raise IOError('Failed getting static land-sea mask and topography')

        tmp_result_filename = make_temp_filename()
        cmd = ('cat ' + tmp_filename + " " +
               os.path.join(nhsf_path, nhsf_prefix + timeinfo) +
               " " + nwp_lsmz_filename + " > " + tmp_result_filename)
        LOG.debug("Add topography and land-sea mask to data:")
        LOG.debug("Command = " + str(cmd))
        _start = time.time()
        retv = os.system(cmd)
        _end = time.time()
        LOG.debug("os.system call took: %f seconds", _end - _start)
        LOG.debug("Returncode = " + str(retv))
        if retv != 0:
            LOG.warning("Failed generating nwp file %s ...", result_file)
            if os.path.exists(tmp_result_filename):
                os.remove(tmp_result_filename)
            raise IOError("Failed adding topography and land-sea " +
                          "mask data to grib file")

        if os.path.exists(tmp_filename):
            os.remove(tmp_filename)
        else:
            LOG.warning("tmp file %s gone! Cannot clean it...", tmp_filename)

        if check_nwp_content(tmp_result_filename):
            LOG.info('A check of the NWP file content has been attempted: %s',
                     result_file)
            _start = time.time()
            os.rename(tmp_result_filename, result_file)
            _end = time.time()
            LOG.debug("Rename file %s to %s: This took %f seconds",
                      tmp_result_filename, result_file, _end - _start)
        else:
            LOG.warning("Missing important fields. No nwp file %s written to disk",
                        result_file)
            if os.path.exists(tmp_result_filename):
                os.remove(tmp_result_filename)

    return


def check_nwp_content(gribfile):
    """Check the content of the NWP file. If all fields required for PPS is
    available, then return True

    """

    with pygrib.open(gribfile) as grbs:
        entries = []
        for grb in grbs:
            entries.append("%s %s %s %s" % (grb['paramId'],
                                            grb['name'],
                                            grb['level'],
                                            grb['typeOfLevel']))
        entries.sort()

    try:
        with open(nwp_req_filename, 'r') as fpt:
            lines = fpt.readlines()
    except IOError:
        LOG.exception(
            "Failed reading nwp-requirements file: %s", nwp_req_filename)
        LOG.warning("Cannot check if NWP files is ok!")
        return True

    srplines = [ll.strip('M ').strip('\n')
                for ll in lines if str(ll).startswith('M')]

    file_ok = True
    for item in srplines:
        if item not in entries:
            LOG.warning("Mandatory field missing in NWP file: %s", str(item))
            file_ok = False

    if file_ok:
        LOG.info("NWP file has all required fields for PPS: %s", gribfile)

    return file_ok


if __name__ == "__main__":

    #: Default time format
    _DEFAULT_TIME_FORMAT = '%Y-%m-%d %H:%M:%S'
    #: Default log format
    _DEFAULT_LOG_FORMAT = '[%(levelname)s: %(asctime)s : %(name)s] %(message)s'

    import sys
    handler = logging.StreamHandler(sys.stderr)
    handler.setLevel(logging.DEBUG)
    formatter = logging.Formatter(fmt=_DEFAULT_LOG_FORMAT,
                                  datefmt=_DEFAULT_TIME_FORMAT)
    handler.setFormatter(formatter)
    logging.getLogger('').addHandler(handler)
    logging.getLogger('').setLevel(logging.DEBUG)
    logging.getLogger('posttroll').setLevel(logging.INFO)

    LOG = logging.getLogger('test_update_nwp')

    from datetime import timedelta
    now = datetime.utcnow()
    update_nwp(now - timedelta(days=1), [9])

1	#!/usr/bin/env python
2	# -- coding: utf-8 --
3
4	# Copyright (c) 2015 - 2021 Pytroll
5
6	# Author(s):
7
8	# Adam.Dybbroe <adam.dybbroe@smhi.se>
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	"""Prepare NWP data for PPS	×
24	"""
25
26	from glob import glob	×
27	import os	×
28	from datetime import datetime	×
29	import time	×
30	import tempfile	×
31	from trollsift import Parser	×
32	import pygrib # @UnresolvedImport	×
33	from six.moves.configparser import NoOptionError	×
34
35	from nwcsafpps_runner.config import get_config	×
36	from nwcsafpps_runner.config import CONFIG_FILE	×
37	from nwcsafpps_runner.config import CONFIG_PATH # @UnresolvedImport	×
38	from nwcsafpps_runner.utils import run_command	×
39	from nwcsafpps_runner.utils import NwpPrepareError	×
40
41	import logging	×
42	LOG = logging.getLogger(__name__)	×
43
44
45	LOG.debug("Path to prepare_nwp config file = %s", str(CONFIG_PATH))	×
46	LOG.debug("Prepare_nwp config file = %s", str(CONFIG_FILE))	×
47	OPTIONS = get_config(CONFIG_FILE)	×
48
49	try:	×
50	nhsp_path = OPTIONS['nhsp_path']	×
51	except KeyError:	×
52	LOG.exception('Parameter not set in config file: ' + 'nhsp_path')	×
53	try:	×
54	nhsp_prefix = OPTIONS['nhsp_prefix']	×
55	except KeyError:	×
56	LOG.exception('Parameter not set in config file: ' + 'nhsp_prefix')	×
57
58	nhsf_file_name_sift = OPTIONS.get('nhsf_file_name_sift')	×
59
60	nhsf_path = OPTIONS.get('nhsf_path', None)	×
61	nhsf_prefix = OPTIONS.get('nhsf_prefix', None)	×
62	nwp_outdir = OPTIONS.get('nwp_outdir', None)	×
63	nwp_lsmz_filename = OPTIONS.get('nwp_static_surface', None)	×
64	nwp_output_prefix = OPTIONS.get('nwp_output_prefix', None)	×
65	nwp_req_filename = OPTIONS.get('pps_nwp_requirements', None)	×
66
67
68	def logreader(stream, log_func):	×
69	while True:
70	s = stream.readline()	×
71	if not s:	×
72	break	×
73	log_func(s.strip())	×
74	stream.close()	×
75
76
77	def make_temp_filename(args, *kwargs):	×
78	tmp_filename_handle, tmp_filename = tempfile.mkstemp(args, *kwargs)	×
79	os.close(tmp_filename_handle)	×
80	return tmp_filename	×
81
82
83	def update_nwp(starttime, nlengths):	×
84	"""Prepare NWP grib files for PPS. Consider only analysis times newer than
85	starttime. And consider only the forecast lead times in hours given by
86	the list nlengths of integers
87
88	"""
89
90	LOG.info("Path to prepare_nwp config file = %s", str(CONFIG_PATH))	×
91	LOG.info("Prepare_nwp config file = %s", str(CONFIG_FILE))	×
92	LOG.info("Path to nhsf files: %s", str(nhsf_path))	×
93	LOG.info("Path to nhsp files: %s", str(nhsp_path))	×
94
95	tempfile.tempdir = nwp_outdir	×
96	filelist = glob(os.path.join(nhsf_path, nhsf_prefix + "*"))	×
97	if len(filelist) == 0:	×
98	LOG.info("No input files! dir = %s", str(nhsf_path))	×
99	return	×
100
101	LOG.debug('NHSF NWP files found = %s', str(filelist))	×
102	nfiles_error = 0	×
103	for filename in filelist:	×
104	if nhsf_file_name_sift is None:	×
105	raise NwpPrepareError()	×
106
107	try:	×
108	parser = Parser(nhsf_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
116	res = parser.parse("{}".format(os.path.basename(filename)))	×
117	if 'analysis_time' in res:	×
118	if res['analysis_time'].year == 1900:	×
119	res['analysis_time'] = res['analysis_time'].replace(year=datetime.utcnow().year)	×
120
121	analysis_time = res['analysis_time']	×
122	timestamp = analysis_time.strftime("%Y%m%d%H%M")	×
123	else:
124	raise NwpPrepareError("Can not parse analysis_time in file name. Check config and filename timestamp")	×
125
126	if 'forecast_time' in res:	×
127	if res['forecast_time'].year == 1900:	×
128	res['forecast_time'] = res['forecast_time'].replace(year=datetime.utcnow().year)	×
129	forecast_time = res['forecast_time']	×
130	forecast_step = forecast_time - analysis_time	×
131	forecast_step = "{:03d}H{:02d}M".format(forecast_step.days*24 + forecast_step.seconds/3600, 0)	×
132	timeinfo = "{:s}{:s}{:s}".format(analysis_time.strftime(	×
133	"%m%d%H%M"), forecast_time.strftime("%m%d%H%M"), res['end'])
134	else:
135	# This needs to be done more solid using the sift pattern! FIXME!
UNCOV 136	timeinfo = filename.rsplit("_", 1)[-1]	×
137	# Forecast step in hours:
138	if 'forecast_step' in res:	×
139	forecast_step = res['forecast_step']	×
140	else:
141	raise NwpPrepareError(	×
142	'Failed parsing forecast_step in file name. Check config and filename timestamp.')
143
144	if analysis_time < starttime:	×
145	continue	×
146	if forecast_step not in nlengths:	×
147	LOG.debug("Skip step. Forecast step and nlengths: %s %s", str(forecast_step), str(nlengths))	×
148	continue	×
149
NEW 150	LOG.debug("Analysis time and start time: %s %s", str(analysis_time), str(starttime))	×
151	LOG.info("timestamp, step: %s %s", str(timestamp), str(forecast_step))	×
152	result_file = os.path.join(	×
153	nwp_outdir, nwp_output_prefix + timestamp + "+" + '%.3dH00M' % forecast_step)
154	if os.path.exists(result_file):	×
155	LOG.info("File: " + str(result_file) + " already there...")	×
156	continue	×
157
158	tmp_filename = make_temp_filename(suffix="_" + timestamp + "+" +	×
159	'%.3dH00M' % forecast_step, dir=nwp_outdir)
160
161	LOG.info("result and tmp files: " + str(result_file) + " " + str(tmp_filename))	×
162	nhsp_file = os.path.join(nhsp_path, nhsp_prefix + timeinfo)	×
163	if not os.path.exists(nhsp_file):	×
164	LOG.warning("Corresponding nhsp-file not there: " + str(nhsp_file))	×
165	continue	×
166
167	cmd = ("grib_copy -w gridType=regular_ll " + nhsp_file + " " + tmp_filename)	×
168	retv = run_command(cmd)	×
169	LOG.debug("Returncode = " + str(retv))	×
170	if retv != 0:	×
171	LOG.error(	×
172	"Failed doing the grib-copy! Will continue with the next file")
173	nfiles_error = nfiles_error + 1	×
174	if nfiles_error > len(filelist) / 2:	×
175	LOG.error(	×
176	"More than half of the Grib files failed upon grib_copy!")
177	raise IOError('Failed running grib_copy on many Grib files')	×
178
179	if not os.path.exists(nwp_lsmz_filename):	×
180	LOG.error("No static grib file with land-sea mask and " +	×
181	"topography available. Can't prepare NWP data")
182	raise IOError('Failed getting static land-sea mask and topography')	×
183
184	tmp_result_filename = make_temp_filename()	×
185	cmd = ('cat ' + tmp_filename + " " +	×
186	os.path.join(nhsf_path, nhsf_prefix + timeinfo) +
187	" " + nwp_lsmz_filename + " > " + tmp_result_filename)
188	LOG.debug("Add topography and land-sea mask to data:")	×
189	LOG.debug("Command = " + str(cmd))	×
190	_start = time.time()	×
191	retv = os.system(cmd)	×
192	_end = time.time()	×
193	LOG.debug("os.system call took: %f seconds", _end - _start)	×
194	LOG.debug("Returncode = " + str(retv))	×
195	if retv != 0:	×
196	LOG.warning("Failed generating nwp file %s ...", result_file)	×
197	if os.path.exists(tmp_result_filename):	×
198	os.remove(tmp_result_filename)	×
199	raise IOError("Failed adding topography and land-sea " +	×
200	"mask data to grib file")
201
202	if os.path.exists(tmp_filename):	×
203	os.remove(tmp_filename)	×
204	else:
205	LOG.warning("tmp file %s gone! Cannot clean it...", tmp_filename)	×
206
207	if check_nwp_content(tmp_result_filename):	×
208	LOG.info('A check of the NWP file content has been attempted: %s',	×
209	result_file)
210	_start = time.time()	×
211	os.rename(tmp_result_filename, result_file)	×
212	_end = time.time()	×
213	LOG.debug("Rename file %s to %s: This took %f seconds",	×
214	tmp_result_filename, result_file, _end - _start)
215	else:
216	LOG.warning("Missing important fields. No nwp file %s written to disk",	×
217	result_file)
218	if os.path.exists(tmp_result_filename):	×
219	os.remove(tmp_result_filename)	×
220
221	return	×
222
223
224	def check_nwp_content(gribfile):	×
225	"""Check the content of the NWP file. If all fields required for PPS is
226	available, then return True
227
228	"""
229
230	with pygrib.open(gribfile) as grbs:	×
231	entries = []	×
232	for grb in grbs:	×
233	entries.append("%s %s %s %s" % (grb['paramId'],	×
234	grb['name'],
235	grb['level'],
236	grb['typeOfLevel']))
237	entries.sort()	×
238
239	try:	×
240	with open(nwp_req_filename, 'r') as fpt:	×
241	lines = fpt.readlines()	×
242	except IOError:	×
243	LOG.exception(	×
244	"Failed reading nwp-requirements file: %s", nwp_req_filename)
245	LOG.warning("Cannot check if NWP files is ok!")	×
246	return True	×
247
248	srplines = [ll.strip('M ').strip('\n')	×
249	for ll in lines if str(ll).startswith('M')]
250
251	file_ok = True	×
252	for item in srplines:	×
253	if item not in entries:	×
254	LOG.warning("Mandatory field missing in NWP file: %s", str(item))	×
255	file_ok = False	×
256
257	if file_ok:	×
258	LOG.info("NWP file has all required fields for PPS: %s", gribfile)	×
259
260	return file_ok	×
261
262
263	if __name__ == "__main__":
264
265	#: Default time format
266	_DEFAULT_TIME_FORMAT = '%Y-%m-%d %H:%M:%S'
267	#: Default log format
268	_DEFAULT_LOG_FORMAT = '[%(levelname)s: %(asctime)s : %(name)s] %(message)s'
269
270	import sys
271	handler = logging.StreamHandler(sys.stderr)
272	handler.setLevel(logging.DEBUG)
273	formatter = logging.Formatter(fmt=_DEFAULT_LOG_FORMAT,
274	datefmt=_DEFAULT_TIME_FORMAT)
275	handler.setFormatter(formatter)
276	logging.getLogger('').addHandler(handler)
277	logging.getLogger('').setLevel(logging.DEBUG)
278	logging.getLogger('posttroll').setLevel(logging.INFO)
279
280	LOG = logging.getLogger('test_update_nwp')
281
282	from datetime import timedelta
283	now = datetime.utcnow()
284	update_nwp(now - timedelta(days=1), [9])

pytroll / pytroll-pps-runner / 1408365101

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