10369361246

Committed 13 Aug 2024 12:04PM UTC coverage: 75.862% (+1.0%) from 74.884%

Build # 10369361246

Build Type

Pull #62

github

Committed by

web-flow

Commit Message

Merge 92e49dcb6 into 311997dcf

Pull Request Pull Request #62: PPS product and level1c collector

Run Details

386 of 553 branches covered (69.8%)

Branch coverage included in aggregate %.

74 of 107 new or added lines in 4 files covered. (69.16%)

6 existing lines in 1 file now uncovered.

1462 of 1883 relevant lines covered (77.64%)

2.33 hits per line

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

81.73

/nwcsafpps_runner/message_utils.py

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

# Copyright (c) 2021 Adam.Dybbroe

# Author(s):

#   Adam.Dybbroe <a000680@c21856.ad.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/>.

"""Message utilities."""

import logging
import os

from posttroll.message import Message
from nwcsafpps_runner.utils import create_pps_file_from_lvl1c


LOG = logging.getLogger(__name__)


def remove_non_pps_products(msg_data):
    """ Remove non-PPS files from datasetlis.t"""
    msg_data["dataset"] = [item for item in msg_data["dataset"] if "S_NWC" in item["uid"]]


def get_pps_sensor_from_msg(sensor_msg):
    """ Get pps sensor from msg sensor."""
    sensor = None
    if type(sensor_msg) is list and len(sensor_msg) == 1:
        sensor = sensor_msg[0]
    if sensor is None:
        for pps_sensor in ['viirs', 'avhrr', 'modis', 'mersi2', 'metimage', 'slstr']:
            if pps_sensor in sensor_msg:
                sensor = pps_sensor
    if "avhrr/3" in sensor_msg:
        sensor = "avhrr"
    return sensor


def add_lvl1c_to_msg(msg_data, options):
    """Add PPS lvl1c file to a collection of PPS products."""
    level1c_path = os.environ.get('SM_IMAGER_DIR', options.get('pps_lvl1c_dir', './'))
    sensor = options.get('sensor', get_pps_sensor_from_msg(msg_data["sensor"]))
    num_files = len(msg_data['dataset'])
    to_add = {}
    for item in msg_data['dataset']:
        lvl1c_file = create_pps_file_from_lvl1c(item["uri"], level1c_path,
                                                name_tag=sensor, file_type='.nc')
        to_add[lvl1c_file] = {
            "uri": lvl1c_file,
            "uid": os.path.basename(lvl1c_file)}
    msg_data['dataset'].extend(to_add.values())


def flatten_collection(msg_data):
    """Flatten collection msg to dataset msg."""
    if "collection" in msg_data:
        collection = msg_data.pop("collection")
        msg_data["dataset"] = []
        for ind in range(0, len(collection)):
            for item in collection[ind]["dataset"]:
                if type(item) == dict:
                    msg_data["dataset"].append(item)


def prepare_pps_collector_message(msg, options):
    to_send = msg.data.copy()
    flatten_collection(to_send)
    remove_non_pps_products(to_send)
    add_lvl1c_to_msg(to_send, options)
    return to_send


def prepare_nwp_message(result_file, publish_topic):
    """Prepare message for NWP files."""
    to_send = {}
    to_send["uri"] = result_file
    filename = os.path.basename(result_file)
    to_send["uid"] = filename
    to_send['format'] = 'NWP grib'
    to_send['type'] = 'grib'
    return to_send


def prepare_l1c_message(result_file, mda, **kwargs):
    """Prepare the output message for the level-1c file creation."""
    if not result_file:
        return

    # Now publish:
    to_send = mda.copy()
    # Delete the input level-1 dataset from the message:
    try:
        del to_send['dataset']
    except KeyError:
        LOG.warning("Couldn't remove dataset from message")
    try:
        del to_send['filename']
    except KeyError:
        LOG.warning("Couldn't remove filename from message")

    if ('orbit' in kwargs) and ('orbit_number' in to_send.keys()):
        to_send["orig_orbit_number"] = to_send["orbit_number"]
        to_send["orbit_number"] = kwargs['orbit']

    to_send["uri"] = result_file
    filename = os.path.basename(result_file)
    to_send["uid"] = filename

    to_send['format'] = 'PPS-L1C'
    to_send['type'] = 'NETCDF'
    to_send['data_processing_level'] = '1c'

    return to_send


def publish_l1c(publisher, publish_msg, publish_topic, msg_type="file"):
    """Publish the messages that l1c files are ready."""
    LOG.debug('Publish topic = %s', publish_topic)
    for topic in publish_topic:
        msg = Message(topic, msg_type, publish_msg).encode()
        LOG.debug("sending: %s", str(msg))
        publisher.send(msg)

1	#!/usr/bin/env python
2	# -- coding: utf-8 --
3
4	# Copyright (c) 2021 Adam.Dybbroe
5
6	# Author(s):
7
8	# Adam.Dybbroe <a000680@c21856.ad.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	"""Message utilities."""	3✔
24
25	import logging	3✔
26	import os	3✔
27
28	from posttroll.message import Message	3✔
29	from nwcsafpps_runner.utils import create_pps_file_from_lvl1c	3✔
30
31
32	LOG = logging.getLogger(__name__)	3✔
33
34
35	def remove_non_pps_products(msg_data):	3✔
36	""" Remove non-PPS files from datasetlis.t"""
37	msg_data["dataset"] = [item for item in msg_data["dataset"] if "S_NWC" in item["uid"]]	3✔
38
39
40	def get_pps_sensor_from_msg(sensor_msg):	3✔
41	""" Get pps sensor from msg sensor."""
42	sensor = None	3✔
43	if type(sensor_msg) is list and len(sensor_msg) == 1:	3!
44	sensor = sensor_msg[0]	3✔
45	if sensor is None:	3!
NEW 46	for pps_sensor in ['viirs', 'avhrr', 'modis', 'mersi2', 'metimage', 'slstr']:	×
NEW 47	if pps_sensor in sensor_msg:	×
NEW 48	sensor = pps_sensor	×
49	if "avhrr/3" in sensor_msg:	3!
NEW 50	sensor = "avhrr"	×
51	return sensor	3✔
52
53
54	def add_lvl1c_to_msg(msg_data, options):	3✔
55	"""Add PPS lvl1c file to a collection of PPS products."""
56	level1c_path = os.environ.get('SM_IMAGER_DIR', options.get('pps_lvl1c_dir', './'))	3✔
57	sensor = options.get('sensor', get_pps_sensor_from_msg(msg_data["sensor"]))	3✔
58	num_files = len(msg_data['dataset'])	3✔
59	to_add = {}	3✔
60	for item in msg_data['dataset']:	3✔
61	lvl1c_file = create_pps_file_from_lvl1c(item["uri"], level1c_path,	3✔
62	name_tag=sensor, file_type='.nc')
63	to_add[lvl1c_file] = {	3✔
64	"uri": lvl1c_file,
65	"uid": os.path.basename(lvl1c_file)}
66	msg_data['dataset'].extend(to_add.values())	3✔
67
68
69	def flatten_collection(msg_data):	3✔
70	"""Flatten collection msg to dataset msg."""
71	if "collection" in msg_data:	3✔
72	collection = msg_data.pop("collection")	3✔
73	msg_data["dataset"] = []	3✔
74	for ind in range(0, len(collection)):	3✔
75	for item in collection[ind]["dataset"]:	3✔
76	if type(item) == dict:	3!
77	msg_data["dataset"].append(item)	3✔
78
79
80	def prepare_pps_collector_message(msg, options):	3✔
81	to_send = msg.data.copy()	3✔
82	flatten_collection(to_send)	3✔
83	remove_non_pps_products(to_send)	3✔
84	add_lvl1c_to_msg(to_send, options)	3✔
85	return to_send	3✔
86
87
88	def prepare_nwp_message(result_file, publish_topic):	3✔
89	"""Prepare message for NWP files."""
90	to_send = {}	3✔
91	to_send["uri"] = result_file	3✔
92	filename = os.path.basename(result_file)	3✔
93	to_send["uid"] = filename	3✔
94	to_send['format'] = 'NWP grib'	3✔
95	to_send['type'] = 'grib'	3✔
96	return to_send	3✔
97
98
99	def prepare_l1c_message(result_file, mda, **kwargs):	3✔
100	"""Prepare the output message for the level-1c file creation."""
101	if not result_file:	3!
102	return	×
103
104	# Now publish:
105	to_send = mda.copy()	3✔
106	# Delete the input level-1 dataset from the message:
107	try:	3✔
108	del to_send['dataset']	3✔
109	except KeyError:	×
110	LOG.warning("Couldn't remove dataset from message")	×
111	try:	3✔
112	del to_send['filename']	3✔
113	except KeyError:	3✔
114	LOG.warning("Couldn't remove filename from message")	3✔
115
116	if ('orbit' in kwargs) and ('orbit_number' in to_send.keys()):	3!
117	to_send["orig_orbit_number"] = to_send["orbit_number"]	×
118	to_send["orbit_number"] = kwargs['orbit']	×
119
120	to_send["uri"] = result_file	3✔
121	filename = os.path.basename(result_file)	3✔
122	to_send["uid"] = filename	3✔
123
124	to_send['format'] = 'PPS-L1C'	3✔
125	to_send['type'] = 'NETCDF'	3✔
126	to_send['data_processing_level'] = '1c'	3✔
127
128	return to_send	3✔
129
130
131	def publish_l1c(publisher, publish_msg, publish_topic, msg_type="file"):	3✔
132	"""Publish the messages that l1c files are ready."""
133	LOG.debug('Publish topic = %s', publish_topic)	3✔
134	for topic in publish_topic:	3✔
135	msg = Message(topic, msg_type, publish_msg).encode()	3✔
136	LOG.debug("sending: %s", str(msg))	3✔
137	publisher.send(msg)	3✔

pytroll / pytroll-pps-runner / 10369361246

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