8685139248

Committed 15 Apr 2024 07:29AM UTC coverage: 74.712% (+7.2%) from 67.499%

Build # 8685139248

Build Type

Pull #60

github

Committed by

web-flow

Commit Message

Merge 1bde12c98 into fc4314d58

Pull Request Pull Request #60: PPS 2021 refactor

Run Details

219 of 373 branches covered (58.71%)

Branch coverage included in aggregate %.

283 of 309 new or added lines in 10 files covered. (91.59%)

1 existing line in 1 file now uncovered.

1400 of 1794 relevant lines covered (78.04%)

3.12 hits per line

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

77.31

/nwcsafpps_runner/utils.py

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

# Copyright (c) 2018 - 2022 Pytroll Developers

# Author(s):

#   Adam.Dybbroe <Firstname.Lastname at 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/>.

"""Utility functions for NWCSAF/pps runner(s).
"""
import threading
from trollsift.parser import parse  # @UnresolvedImport
# from trollsift import Parser
from posttroll.message import Message  # @UnresolvedImport
from subprocess import Popen, PIPE
import os
import shlex
from glob import glob
import socket

from urllib.parse import urlparse

from posttroll.address_receiver import get_local_ips

import logging
LOG = logging.getLogger(__name__)


class NwpPrepareError(Exception):
    pass


class FindTimeControlFileError(Exception):
    pass


PPS_OUT_PATTERN = ("S_NWC_{segment}_{orig_platform_name}_{orbit_number:05d}_" +
                   "{start_time:%Y%m%dT%H%M%S%f}Z_{end_time:%Y%m%dT%H%M%S%f}Z.{extention}")
PPS_OUT_PATTERN_MULTIPLE = ("S_NWC_{segment1}_{segment2}_{orig_platform_name}_{orbit_number:05d}_" +
                            "{start_time:%Y%m%dT%H%M%S%f}Z_{end_time:%Y%m%dT%H%M%S%f}Z.{extention}")
PPS_STAT_PATTERN = ("S_NWC_{segment}_{orig_platform_name}_{orbit_number:05d}_" +
                    "{start_time:%Y%m%dT%H%M%S%f}Z_{end_time:%Y%m%dT%H%M%S%f}Z_statistics.xml")

SUPPORTED_AVHRR_SATELLITES = ['NOAA-15', 'NOAA-18', 'NOAA-19',
                              'Metop-B', 'Metop-A', 'Metop-C']
SUPPORTED_EARS_AVHRR_SATELLITES = ['Metop-B', 'Metop-C']
SUPPORTED_MODIS_SATELLITES = ['EOS-Terra', 'EOS-Aqua']
SUPPORTED_VIIRS_SATELLITES = ['Suomi-NPP', 'NOAA-20', 'NOAA-21', 'NOAA-22', 'NOAA-23']
SUPPORTED_SEVIRI_SATELLITES = ['Meteosat-09', 'Meteosat-10', 'Meteosat-11']
SUPPORTED_METIMAGE_SATELLITES = ['Metop-SG-A1', 'Metop-SG-A2', 'Metop-SG-A3']

SUPPORTED_PPS_SATELLITES = (SUPPORTED_AVHRR_SATELLITES +
                            SUPPORTED_MODIS_SATELLITES +
                            SUPPORTED_SEVIRI_SATELLITES +
                            SUPPORTED_METIMAGE_SATELLITES +
                            SUPPORTED_VIIRS_SATELLITES)

GEOLOC_PREFIX = {'EOS-Aqua': 'MYD03', 'EOS-Terra': 'MOD03'}
DATA1KM_PREFIX = {'EOS-Aqua': 'MYD021km', 'EOS-Terra': 'MOD021km'}

PPS_SENSORS = ['amsu-a', 'amsu-b', 'mhs', 'avhrr/3', 'viirs', 'modis', 'seviri', 'metimage']
NOAA_METOP_PPS_SENSORNAMES = ['avhrr/3', 'amsu-a', 'amsu-b', 'mhs']

METOP_NAME_LETTER = {'metop01': 'metopb', 'metop02': 'metopa', 'metop03': 'metopc'}
METOP_NAME = {'metop01': 'Metop-B', 'metop02': 'Metop-A', 'metop03': 'Metop-C'}
METOP_NAME_INV = {'metopb': 'metop01', 'metopa': 'metop02', 'metopc': 'metop03'}

# SATELLITE_NAME = {}
# for sat in SUPPORTED_PPS_SATELLITES:
#     SATELLITE_NAME[sat] = sat.lower().replace('-', '')
# historic exceptions
# SATELLITE_NAME['Suomi-NPP'] = 'npp'
# SATELLITE_NAME['EOS-Aqua'] = 'eos2'
# SATELLITE_NAME['EOS-Terra'] = 'eos1'
# SATELLITE_NAME['Metop-A']= 'metop02'
# SATELLITE_NAME['Metop-B']= 'metop01'
# SATELLITE_NAME['Metop-C']= 'metop03'

SENSOR_LIST = {}
for sat in SUPPORTED_PPS_SATELLITES:
    if sat in SUPPORTED_AVHRR_SATELLITES:
        SENSOR_LIST[sat] = ['avhrr/3']
    elif sat in SUPPORTED_MODIS_SATELLITES:
        SENSOR_LIST[sat] = ['modis']
    elif sat in SUPPORTED_VIIRS_SATELLITES:
        SENSOR_LIST[sat] = ['viirs']
    elif sat in SUPPORTED_SEVIRI_SATELLITES:
        SENSOR_LIST[sat] = ['seviri']
    elif sat in SUPPORTED_METIMAGE_SATELLITES:
        SENSOR_LIST[sat] = ['metimage']

METOP_SENSOR = {'amsu-a': 'amsua', 'avhrr/3': 'avhrr',
                'amsu-b': 'amsub', 'hirs/4': 'hirs'}


def run_command(cmdstr):
    """Run system command."""
    myargs = shlex.split(str(cmdstr))

    LOG.debug("Command: " + str(cmdstr))
    LOG.debug('Command sequence= ' + str(myargs))
    #: TODO: What is this
    try:
        proc = Popen(myargs, shell=False, stderr=PIPE, stdout=PIPE)
    except NwpPrepareError:
        LOG.exception("Failed when preparing NWP data for PPS...")

    out_reader = threading.Thread(
        target=logreader, args=(proc.stdout, LOG.info))
    err_reader = threading.Thread(
        target=logreader, args=(proc.stderr, LOG.info))
    out_reader.start()
    err_reader.start()
    out_reader.join()
    err_reader.join()

    return proc.wait()


def check_uri(uri):
    """Check that the provided *uri* is on the local host and return the
    file path.
    """
    if isinstance(uri, (list, set, tuple)):
        paths = [check_uri(ressource) for ressource in uri]
        return paths
    url = urlparse(uri)
    try:
        if url.hostname:
            url_ip = socket.gethostbyname(url.hostname)

            if url_ip not in get_local_ips():
                try:
                    os.stat(url.path)
                except OSError:
                    raise IOError(
                        "Data file %s unaccessible from this host" % uri)

    except socket.gaierror:
        LOG.warning("Couldn't check file location, running anyway")

    return url.path


class PpsRunError(Exception):
    pass


def get_lvl1c_file_from_msg(msg):
    """Get level1c file from msg."""
    destination = msg.data.get('destination')

    uris = []

    if msg.type == 'file':
        if destination is None:
            uris = [(msg.data['uri'])]
        else:
            uris = [os.path.join(destination, msg.data['uid'])]
    else:
        LOG.debug(
            "Ignoring this type of message data: type = " + str(msg.type))
        return None

    try:
        level1c_files = check_uri(uris)
    except IOError:
        LOG.info('One or more files not present on this host!')
        return None

    LOG.debug("files4pps: %s", str(level1c_files))
    return level1c_files[0]


def check_host_ok(msg):
    """Check that host is ok."""
    try:
        url_ip = socket.gethostbyname(msg.host)
        if url_ip not in get_local_ips():
            LOG.warning("Server %s not the current one: %s", str(url_ip), socket.gethostname())
            return False
    except (AttributeError, socket.gaierror) as err:
        LOG.error("Failed checking host! Hostname = %s", socket.gethostname())
        LOG.exception(err)
    return True


def ready2run(msg, scene, **kwargs):
    """Check whether pps is ready to run or not."""

    LOG.info("Got message: " + str(msg))
    if not check_host_ok(msg):
        return False

    if scene['file4pps'] is None:
        return False

    if msg.data['platform_name'] in SUPPORTED_PPS_SATELLITES:
        LOG.info(
            "This is a PPS supported scene. Start the PPS lvl2 processing!")
        LOG.info("Process the file = %s" +
                 os.path.basename(scene['file4pps']))

        LOG.debug("Ready to run...")
        return True


def terminate_process(popen_obj, scene):
    """Terminate a Popen process."""
    if popen_obj.returncode is None:
        popen_obj.kill()
        LOG.info(
            "Process timed out and pre-maturely terminated. Scene: " + str(scene))
    else:
        LOG.info(
            "Process finished before time out - workerScene: " + str(scene))


def create_pps_call_command(python_exec, pps_script_name, scene):
    """Create the pps call command.

    Supports PPSv2021.
    """
    cmdstr = ("%s" % python_exec + " %s " % pps_script_name +
              "-af %s" % scene['file4pps'])
    LOG.debug("PPS call command: %s", str(cmdstr))
    return cmdstr


def create_xml_timestat_from_lvl1c(scene, pps_control_path):
    """From lvl1c file create XML file and return a file list."""
    try:
        txt_time_control = create_pps_file_from_lvl1c(scene['file4pps'], pps_control_path, "timectrl", ".txt")
    except KeyError:
        return []
    if os.path.exists(txt_time_control):
        return create_xml_timestat_from_ascii(txt_time_control, pps_control_path)
    else:
        LOG.warning('No XML Time statistics file created!')
        return []


def find_product_statistics_from_lvl1c(scene, pps_control_path):
    """From lvl1c file find product XML files and return a file list."""
    try:
        glob_pattern = create_pps_file_from_lvl1c(scene['file4pps'], pps_control_path, "*", "_statistics.xml")
        return glob(glob_pattern)
    except KeyError:
        return []


def create_pps_file_from_lvl1c(l1c_file_name, pps_control_path, name_tag, file_type):
    """From lvl1c file create name_tag-file of type file_type."""
    from trollsift import parse, compose
    f_pattern = 'S_NWC_{name_tag}_{platform_id}_{orbit_number}_{start_time}Z_{end_time}Z{file_type}'
    l1c_path, l1c_file = os.path.split(l1c_file_name)
    data = parse(f_pattern, l1c_file)
    data["name_tag"] = name_tag
    data["file_type"] = file_type
    return os.path.join(pps_control_path, compose(f_pattern, data))


def create_xml_timestat_from_ascii(infile, pps_control_path):
    """From ascii file(s) with PPS time statistics create XML file(s) and return a file list."""
    try:
        from pps_time_control import PPSTimeControl
    except ImportError:
        LOG.warning("Failed to import the PPSTimeControl from pps")
        return []
    LOG.info("Time control ascii file: " + str(infile))
    LOG.info("Read time control ascii file and generate XML")
    ppstime_con = PPSTimeControl(infile)
    ppstime_con.sum_up_processing_times()
    try:
        ppstime_con.write_xml()
    except Exception as e:  # TypeError as e:
        LOG.warning('Not able to write time control xml file')
        LOG.warning(e)

    # There should always be only one xml file for each ascii file found above!
    return [infile.replace('.txt', '.xml')]


def publish_pps_files(input_msg, publish_q, scene, result_files, **kwargs):
    """
    Publish messages for the files provided.
    """

    servername = kwargs.get('servername')
    station = kwargs.get('station', 'unknown')

    for result_file in result_files:
        # Get true start and end time from filenames and adjust the end time in
        # the publish message:
        filename = os.path.basename(result_file)
        LOG.info("file to publish = %s", str(filename))
        try:
            try:
                metadata = parse(PPS_OUT_PATTERN, filename)
            except ValueError:
                metadata = parse(PPS_OUT_PATTERN_MULTIPLE, filename)
                metadata['segment'] = '_'.join([metadata['segment1'],
                                                metadata['segment2']])
                del metadata['segment1'], metadata['segment2']
        except ValueError:
            metadata = parse(PPS_STAT_PATTERN, filename)

        endtime = metadata['end_time']
        starttime = metadata['start_time']

        to_send = input_msg.data.copy()
        to_send.pop('dataset', None)
        to_send.pop('collection', None)
        to_send['uri'] = result_file
        to_send['uid'] = filename
        to_send['sensor'] = scene.get('instrument', None)
        if not to_send['sensor']:
            to_send['sensor'] = scene.get('sensor', None)

        to_send['platform_name'] = scene['platform_name']
        to_send['orbit_number'] = scene['orbit_number']
        if result_file.endswith("xml"):
            to_send['format'] = 'PPS-XML'
            to_send['type'] = 'XML'
        if result_file.endswith("nc"):
            to_send['format'] = 'CF'
            to_send['type'] = 'netCDF4'
        to_send['data_processing_level'] = '2'

        to_send['start_time'], to_send['end_time'] = starttime, endtime
        pubmsg = Message('/' + to_send['format'] + '/' +
                         to_send['data_processing_level'] +
                         '/' + station +
                         '/polar/direct_readout/',
                         "file", to_send).encode()
        LOG.info("Sending: %s", str(pubmsg))
        try:
            publish_q.put(pubmsg)
        except Exception:
            LOG.warning("Failed putting message on the queue, will send it now...")
            publish_q.send(pubmsg)


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

1	#!/usr/bin/env python
2	# -- coding: utf-8 --
3
4	# Copyright (c) 2018 - 2022 Pytroll Developers
5
6	# Author(s):
7
8	# Adam.Dybbroe <Firstname.Lastname at 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	"""Utility functions for NWCSAF/pps runner(s).	4✔
24	"""
25	import threading	4✔
26	from trollsift.parser import parse # @UnresolvedImport	4✔
27	# from trollsift import Parser
28	from posttroll.message import Message # @UnresolvedImport	4✔
29	from subprocess import Popen, PIPE	4✔
30	import os	4✔
31	import shlex	4✔
32	from glob import glob	4✔
33	import socket	4✔
34
35	from urllib.parse import urlparse	4✔
36
37	from posttroll.address_receiver import get_local_ips	4✔
38
39	import logging	4✔
40	LOG = logging.getLogger(__name__)	4✔
41
42
43	class NwpPrepareError(Exception):	4✔
44	pass	4✔
45
46
47	class FindTimeControlFileError(Exception):	4✔
48	pass	4✔
49
50
51	PPS_OUT_PATTERN = ("S_NWC_{segment}_{orig_platform_name}_{orbit_number:05d}_" +	4✔
52	"{start_time:%Y%m%dT%H%M%S%f}Z_{end_time:%Y%m%dT%H%M%S%f}Z.{extention}")
53	PPS_OUT_PATTERN_MULTIPLE = ("S_NWC_{segment1}_{segment2}_{orig_platform_name}_{orbit_number:05d}_" +	4✔
54	"{start_time:%Y%m%dT%H%M%S%f}Z_{end_time:%Y%m%dT%H%M%S%f}Z.{extention}")
55	PPS_STAT_PATTERN = ("S_NWC_{segment}_{orig_platform_name}_{orbit_number:05d}_" +	4✔
56	"{start_time:%Y%m%dT%H%M%S%f}Z_{end_time:%Y%m%dT%H%M%S%f}Z_statistics.xml")
57
58	SUPPORTED_AVHRR_SATELLITES = ['NOAA-15', 'NOAA-18', 'NOAA-19',	4✔
59	'Metop-B', 'Metop-A', 'Metop-C']
60	SUPPORTED_EARS_AVHRR_SATELLITES = ['Metop-B', 'Metop-C']	4✔
61	SUPPORTED_MODIS_SATELLITES = ['EOS-Terra', 'EOS-Aqua']	4✔
62	SUPPORTED_VIIRS_SATELLITES = ['Suomi-NPP', 'NOAA-20', 'NOAA-21', 'NOAA-22', 'NOAA-23']	4✔
63	SUPPORTED_SEVIRI_SATELLITES = ['Meteosat-09', 'Meteosat-10', 'Meteosat-11']	4✔
64	SUPPORTED_METIMAGE_SATELLITES = ['Metop-SG-A1', 'Metop-SG-A2', 'Metop-SG-A3']	4✔
65
66	SUPPORTED_PPS_SATELLITES = (SUPPORTED_AVHRR_SATELLITES +	4✔
67	SUPPORTED_MODIS_SATELLITES +
68	SUPPORTED_SEVIRI_SATELLITES +
69	SUPPORTED_METIMAGE_SATELLITES +
70	SUPPORTED_VIIRS_SATELLITES)
71
72	GEOLOC_PREFIX = {'EOS-Aqua': 'MYD03', 'EOS-Terra': 'MOD03'}	4✔
73	DATA1KM_PREFIX = {'EOS-Aqua': 'MYD021km', 'EOS-Terra': 'MOD021km'}	4✔
74
75	PPS_SENSORS = ['amsu-a', 'amsu-b', 'mhs', 'avhrr/3', 'viirs', 'modis', 'seviri', 'metimage']	4✔
76	NOAA_METOP_PPS_SENSORNAMES = ['avhrr/3', 'amsu-a', 'amsu-b', 'mhs']	4✔
77
78	METOP_NAME_LETTER = {'metop01': 'metopb', 'metop02': 'metopa', 'metop03': 'metopc'}	4✔
79	METOP_NAME = {'metop01': 'Metop-B', 'metop02': 'Metop-A', 'metop03': 'Metop-C'}	4✔
80	METOP_NAME_INV = {'metopb': 'metop01', 'metopa': 'metop02', 'metopc': 'metop03'}	4✔
81
82	# SATELLITE_NAME = {}
83	# for sat in SUPPORTED_PPS_SATELLITES:
84	# SATELLITE_NAME[sat] = sat.lower().replace('-', '')
85	# historic exceptions
86	# SATELLITE_NAME['Suomi-NPP'] = 'npp'
87	# SATELLITE_NAME['EOS-Aqua'] = 'eos2'
88	# SATELLITE_NAME['EOS-Terra'] = 'eos1'
89	# SATELLITE_NAME['Metop-A']= 'metop02'
90	# SATELLITE_NAME['Metop-B']= 'metop01'
91	# SATELLITE_NAME['Metop-C']= 'metop03'
92
93	SENSOR_LIST = {}	4✔
94	for sat in SUPPORTED_PPS_SATELLITES:	4✔
95	if sat in SUPPORTED_AVHRR_SATELLITES:	4✔
96	SENSOR_LIST[sat] = ['avhrr/3']	4✔
97	elif sat in SUPPORTED_MODIS_SATELLITES:	4✔
98	SENSOR_LIST[sat] = ['modis']	4✔
99	elif sat in SUPPORTED_VIIRS_SATELLITES:	4✔
100	SENSOR_LIST[sat] = ['viirs']	4✔
101	elif sat in SUPPORTED_SEVIRI_SATELLITES:	4✔
102	SENSOR_LIST[sat] = ['seviri']	4✔
103	elif sat in SUPPORTED_METIMAGE_SATELLITES:	4!
104	SENSOR_LIST[sat] = ['metimage']	4✔
105
106	METOP_SENSOR = {'amsu-a': 'amsua', 'avhrr/3': 'avhrr',	4✔
107	'amsu-b': 'amsub', 'hirs/4': 'hirs'}
108
109
110	def run_command(cmdstr):	4✔
111	"""Run system command."""
112	myargs = shlex.split(str(cmdstr))	4✔
113
114	LOG.debug("Command: " + str(cmdstr))	4✔
115	LOG.debug('Command sequence= ' + str(myargs))	4✔
116	#: TODO: What is this
117	try:	4✔
118	proc = Popen(myargs, shell=False, stderr=PIPE, stdout=PIPE)	4✔
119	except NwpPrepareError:	×
120	LOG.exception("Failed when preparing NWP data for PPS...")	×
121
122	out_reader = threading.Thread(	4✔
123	target=logreader, args=(proc.stdout, LOG.info))
124	err_reader = threading.Thread(	4✔
125	target=logreader, args=(proc.stderr, LOG.info))
126	out_reader.start()	4✔
127	err_reader.start()	4✔
128	out_reader.join()	4✔
129	err_reader.join()	4✔
130
131	return proc.wait()	4✔
132
133
134	def check_uri(uri):	4✔
135	"""Check that the provided uri is on the local host and return the
136	file path.
137	"""
138	if isinstance(uri, (list, set, tuple)):	4✔
139	paths = [check_uri(ressource) for ressource in uri]	4✔
140	return paths	4✔
141	url = urlparse(uri)	4✔
142	try:	4✔
143	if url.hostname:	4!
144	url_ip = socket.gethostbyname(url.hostname)	×
145
146	if url_ip not in get_local_ips():	×
147	try:	×
148	os.stat(url.path)	×
149	except OSError:	×
150	raise IOError(	×
151	"Data file %s unaccessible from this host" % uri)
152
153	except socket.gaierror:	×
154	LOG.warning("Couldn't check file location, running anyway")	×
155
156	return url.path	4✔
157
158
159	class PpsRunError(Exception):	4✔
160	pass	4✔
161
162
163	def get_lvl1c_file_from_msg(msg):	4✔
164	"""Get level1c file from msg."""
165	destination = msg.data.get('destination')	4✔
166
167	uris = []	4✔
168
169	if msg.type == 'file':	4✔
170	if destination is None:	4✔
171	uris = [(msg.data['uri'])]	4✔
172	else:
173	uris = [os.path.join(destination, msg.data['uid'])]	4✔
174	else:
175	LOG.debug(	4✔
176	"Ignoring this type of message data: type = " + str(msg.type))
177	return None	4✔
178
179	try:	4✔
180	level1c_files = check_uri(uris)	4✔
181	except IOError:	×
182	LOG.info('One or more files not present on this host!')	×
NEW 183	return None	×
184
185	LOG.debug("files4pps: %s", str(level1c_files))	4✔
186	return level1c_files[0]	4✔
187
188
189	def check_host_ok(msg):	4✔
190	"""Check that host is ok."""
191	try:	×
192	url_ip = socket.gethostbyname(msg.host)	×
193	if url_ip not in get_local_ips():	×
194	LOG.warning("Server %s not the current one: %s", str(url_ip), socket.gethostname())	×
195	return False	×
196	except (AttributeError, socket.gaierror) as err:	×
197	LOG.error("Failed checking host! Hostname = %s", socket.gethostname())	×
198	LOG.exception(err)	×
NEW 199	return True	×
200
201
202	def ready2run(msg, scene, **kwargs):	4✔
203	"""Check whether pps is ready to run or not."""
204
205	LOG.info("Got message: " + str(msg))	4✔
206	if not check_host_ok(msg):	4!
NEW 207	return False	×
208
209	if scene['file4pps'] is None:	4✔
210	return False	4✔
211
212	if msg.data['platform_name'] in SUPPORTED_PPS_SATELLITES:	4!
213	LOG.info(	4✔
214	"This is a PPS supported scene. Start the PPS lvl2 processing!")
215	LOG.info("Process the file = %s" +	4✔
216	os.path.basename(scene['file4pps']))
217
218	LOG.debug("Ready to run...")	4✔
219	return True	4✔
220
221
222	def terminate_process(popen_obj, scene):	4✔
223	"""Terminate a Popen process."""
224	if popen_obj.returncode is None:	×
225	popen_obj.kill()	×
226	LOG.info(	×
227	"Process timed out and pre-maturely terminated. Scene: " + str(scene))
228	else:
229	LOG.info(	×
230	"Process finished before time out - workerScene: " + str(scene))
231
232
233	def create_pps_call_command(python_exec, pps_script_name, scene):	4✔
234	"""Create the pps call command.
235
236	Supports PPSv2021.
237	"""
NEW 238	cmdstr = ("%s" % python_exec + " %s " % pps_script_name +	×
239	"-af %s" % scene['file4pps'])
NEW 240	LOG.debug("PPS call command: %s", str(cmdstr))	×
UNCOV 241	return cmdstr	×
242
243
244	def create_xml_timestat_from_lvl1c(scene, pps_control_path):	4✔
245	"""From lvl1c file create XML file and return a file list."""
246	try:	4✔
247	txt_time_control = create_pps_file_from_lvl1c(scene['file4pps'], pps_control_path, "timectrl", ".txt")	4✔
248	except KeyError:	4✔
249	return []	4✔
250	if os.path.exists(txt_time_control):	4✔
251	return create_xml_timestat_from_ascii(txt_time_control, pps_control_path)	4✔
252	else:
253	LOG.warning('No XML Time statistics file created!')	4✔
254	return []	4✔
255
256
257	def find_product_statistics_from_lvl1c(scene, pps_control_path):	4✔
258	"""From lvl1c file find product XML files and return a file list."""
259	try:	4✔
260	glob_pattern = create_pps_file_from_lvl1c(scene['file4pps'], pps_control_path, "*", "_statistics.xml")	4✔
261	return glob(glob_pattern)	4✔
262	except KeyError:	4✔
263	return []	4✔
264
265
266	def create_pps_file_from_lvl1c(l1c_file_name, pps_control_path, name_tag, file_type):	4✔
267	"""From lvl1c file create name_tag-file of type file_type."""
268	from trollsift import parse, compose	4✔
269	f_pattern = 'S_NWC_{name_tag}_{platform_id}_{orbit_number}_{start_time}Z_{end_time}Z{file_type}'	4✔
270	l1c_path, l1c_file = os.path.split(l1c_file_name)	4✔
271	data = parse(f_pattern, l1c_file)	4✔
272	data["name_tag"] = name_tag	4✔
273	data["file_type"] = file_type	4✔
274	return os.path.join(pps_control_path, compose(f_pattern, data))	4✔
275
276
277	def create_xml_timestat_from_ascii(infile, pps_control_path):	4✔
278	"""From ascii file(s) with PPS time statistics create XML file(s) and return a file list."""
279	try:	4✔
280	from pps_time_control import PPSTimeControl	4✔
281	except ImportError:	×
282	LOG.warning("Failed to import the PPSTimeControl from pps")	×
283	return []	×
284	LOG.info("Time control ascii file: " + str(infile))	4✔
285	LOG.info("Read time control ascii file and generate XML")	4✔
286	ppstime_con = PPSTimeControl(infile)	4✔
287	ppstime_con.sum_up_processing_times()	4✔
288	try:	4✔
289	ppstime_con.write_xml()	4✔
290	except Exception as e: # TypeError as e:	×
291	LOG.warning('Not able to write time control xml file')	×
292	LOG.warning(e)	×
293
294	# There should always be only one xml file for each ascii file found above!
295	return [infile.replace('.txt', '.xml')]	4✔
296
297
298	def publish_pps_files(input_msg, publish_q, scene, result_files, **kwargs):	4✔
299	"""
300	Publish messages for the files provided.
301	"""
302
303	servername = kwargs.get('servername')	4✔
304	station = kwargs.get('station', 'unknown')	4✔
305
306	for result_file in result_files:	4✔
307	# Get true start and end time from filenames and adjust the end time in
308	# the publish message:
309	filename = os.path.basename(result_file)	4✔
310	LOG.info("file to publish = %s", str(filename))	4✔
311	try:	4✔
312	try:	4✔
313	metadata = parse(PPS_OUT_PATTERN, filename)	4✔
314	except ValueError:	4✔
315	metadata = parse(PPS_OUT_PATTERN_MULTIPLE, filename)	4✔
316	metadata['segment'] = '_'.join([metadata['segment1'],	×
317	metadata['segment2']])
318	del metadata['segment1'], metadata['segment2']	×
319	except ValueError:	4✔
320	metadata = parse(PPS_STAT_PATTERN, filename)	4✔
321
322	endtime = metadata['end_time']	4✔
323	starttime = metadata['start_time']	4✔
324
325	to_send = input_msg.data.copy()	4✔
326	to_send.pop('dataset', None)	4✔
327	to_send.pop('collection', None)	4✔
328	to_send['uri'] = result_file	4✔
329	to_send['uid'] = filename	4✔
330	to_send['sensor'] = scene.get('instrument', None)	4✔
331	if not to_send['sensor']:	4!
332	to_send['sensor'] = scene.get('sensor', None)	×
333
334	to_send['platform_name'] = scene['platform_name']	4✔
335	to_send['orbit_number'] = scene['orbit_number']	4✔
336	if result_file.endswith("xml"):	4!
337	to_send['format'] = 'PPS-XML'	4✔
338	to_send['type'] = 'XML'	4✔
339	if result_file.endswith("nc"):	4!
340	to_send['format'] = 'CF'	×
341	to_send['type'] = 'netCDF4'	×
342	to_send['data_processing_level'] = '2'	4✔
343
344	to_send['start_time'], to_send['end_time'] = starttime, endtime	4✔
345	pubmsg = Message('/' + to_send['format'] + '/' +	4✔
346	to_send['data_processing_level'] +
347	'/' + station +
348	'/polar/direct_readout/',
349	"file", to_send).encode()
350	LOG.info("Sending: %s", str(pubmsg))	4✔
351	try:	4✔
352	publish_q.put(pubmsg)	4✔
353	except Exception:	×
354	LOG.warning("Failed putting message on the queue, will send it now...")	×
355	publish_q.send(pubmsg)	×
356
357
358	def logreader(stream, log_func):	4✔
359	while True:	2✔
360	mystring = stream.readline()	4✔
361	if not mystring:	4!
362	break	4✔
363	log_func(mystring.strip())	×
364	stream.close()	4✔

pytroll / pytroll-pps-runner / 8685139248

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