863906473

Build Type

github

Committed by GitHub

Commit Message

Merge 833b85bc1 into b917f0a72

Pull Request Pull Request #28: Add level1c runner

Run Details

80 of 438 branches covered (18.26%)

Branch coverage included in aggregate %.

156 of 255 new or added lines in 8 files covered. (61.18%)

16 existing lines in 1 file now uncovered.

710 of 1567 relevant lines covered (45.31%)

1.8 hits per line

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

31.11

/nwcsafpps_runner/l1c_processing.py

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

# Copyright (c) 2021 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/>.

"""The level-1c processing tools
"""

import logging
from multiprocessing import cpu_count
from multiprocessing.pool import ThreadPool
from level1c4pps.seviri2pps_lib import process_one_scan as process_seviri
from level1c4pps.viirs2pps_lib import process_one_scene as process_viirs
from level1c4pps.modis2pps_lib import process_one_scene as process_modis
from level1c4pps.avhrr2pps_lib import process_one_scene as process_avhrr

from nwcsafpps_runner.config import get_config_from_yamlfile as get_config

LOG = logging.getLogger(__name__)

SUPPORTED_SERVICE_NAMES = ['seviri-l1c', 'viirs-l1c', 'avhrr-l1c', 'modis-l1c']

SUPPORTED_SATELLITES = {'seviri-l1c':
                        ['meteosat-8', 'meteosat-9', 'meteosat-10', 'meteosat-11'],
                        'viirs-l1c': ['suomi-npp', 'noaa-20', 'noaa-21'],
                        'avhrr-l1c': ['noaa-19', 'noaa-18'],
                        'modis-l1c': ['eos-terra', 'eos-aqua']
                        }

LVL1C_PROCESSOR_MAPPING = {'seviri-l1c': process_seviri,
                           'viirs-l1c': process_viirs,
                           'modis-l1c': process_modis,
                           'avhrr-l1c': process_avhrr}


class ServiceNameNotSupported(Exception):
    pass


class L1cProcessor(object):
    """Container for the NWCSAF/PPS Level-c processing."""

    def __init__(self, config_filename, service_name):

        options = get_config(config_filename, service_name)
        check_service_name_okay(service_name)

        self.service = service_name

        self.subscribe_topics = options['message_types']
        LOG.debug("Listens for messages of type: %s", str(self.subscribe_topics))

        ncpus_available = cpu_count()
        LOG.info("Number of CPUs available = %s", str(ncpus_available))
        ncpus = int(options.get('num_of_cpus', 1))
        LOG.info("Will use %d CPUs when running the level-1c processing instances", ncpus)

        self.pool = ThreadPool(ncpus)

        self.sensor = "unknown"
        self.orbit_number = 99999  # Initialized orbit number
        self.platform_name = 'unknown'
        self.l1c_result = None
        self.pass_start_time = None
        self.l1cfile = None
        self.level1_files = []

        self.result_home = options.get('output_dir', '/tmp')
        self.publish_topic = options.get('publish_topic')
        self.message_data = None

    def initialise(self, service):
        """Initialise the processor."""
        self.check_service_name_okay(service)
        self.l1c_result = None
        self.pass_start_time = None
        self.l1cfile = None
        self.level1_files = []
        self.service = service

    def run(self, msg):
        """Start the L1c processing using the relevant sensor specific function from level1c4pps."""

        if not check_message_okay(msg):
            return False

        self.platform_name = str(msg.data['platform_name'])

        if not self.check_platform_name_consistent_with_service():
            return False

        self.sensor = str(msg.data['sensor'])
        self.message_data = msg.data

        level1_dataset = msg.data['dataset']

        if len(level1_dataset) < 1:
            return False

        self.get_level1_files_from_dataset(level1_dataset)

        l1c_proc = LVL1C_PROCESSOR_MAPPING.get(self.service)
        if not l1c_proc:
            raise AttributeError("Could not find suitable level-1c processor! Service = %s" % self.service)

        self.l1c_result = self.pool.apply_async(l1c_proc, (self.level1_files,
                                                           self.result_home))
        return True

    def get_level1_files_from_dataset(self, level1_dataset):
        """Get the level-1 files from the dataset."""

        if self.service in ['seviri-l1c']:
            self.level1_files = get_seviri_level1_files_from_dataset(level1_dataset)
        else:
            for level1 in level1_dataset:
                self.level1_files.append(level1['uri'])

    def check_platform_name_consistent_with_service(self):
        """Check that the platform name is consistent with the service name."""

        if self.platform_name.lower() not in SUPPORTED_SATELLITES.get(self.service, []):
            LOG.warning("%s: Platform name not supported for this service: %s",
                        str(self.platform_name), self.service)
            return False

        return True


def get_seviri_level1_files_from_dataset(level1_dataset):
    """Get the seviri level-1 filenames from the dataset and return as list."""

    pro_files = False
    epi_files = False
    level1_files = []
    for level1 in level1_dataset:
        level1_filename = level1['uri']
        level1_files.append(level1_filename)
        if '-PRO' in level1_filename:
            pro_files = True
        if '-EPI' in level1_filename:
            epi_files = True

    if not pro_files:
        LOG.warning("PRO file is missing...")
        return []
    if not epi_files:
        LOG.warning("EPI file is missing...")
        return []

    return level1_files


def check_message_okay(msg):
    """Check that the message is okay and has the necessary fields."""

    if not msg:
        return False

    if msg.type != 'dataset':
        LOG.info("Not a dataset, don't do anything...")
        return False

    if ('platform_name' not in msg.data or 'start_time' not in msg.data):
        LOG.error("Message is lacking crucial fields...")
        return False

    return True


def check_service_name_okay(service_name):
    """Check that the service name is supported."""
    if service_name not in SUPPORTED_SERVICE_NAMES:
        errmsg = "Service name %s is not yet supported" % service_name
        raise ServiceNameNotSupported(errmsg)

1	#!/usr/bin/env python
2	# -- coding: utf-8 --
3
4	# Copyright (c) 2021 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	"""The level-1c processing tools	4✔
24	"""
25
26	import logging	4✔
27	from multiprocessing import cpu_count	4✔
28	from multiprocessing.pool import ThreadPool	4✔
29	from level1c4pps.seviri2pps_lib import process_one_scan as process_seviri	4✔
30	from level1c4pps.viirs2pps_lib import process_one_scene as process_viirs	4✔
31	from level1c4pps.modis2pps_lib import process_one_scene as process_modis	4✔
32	from level1c4pps.avhrr2pps_lib import process_one_scene as process_avhrr	4✔
33
34	from nwcsafpps_runner.config import get_config_from_yamlfile as get_config	4✔
35
36	LOG = logging.getLogger(__name__)	4✔
37
38	SUPPORTED_SERVICE_NAMES = ['seviri-l1c', 'viirs-l1c', 'avhrr-l1c', 'modis-l1c']	4✔
39
40	SUPPORTED_SATELLITES = {'seviri-l1c':	4✔
41	['meteosat-8', 'meteosat-9', 'meteosat-10', 'meteosat-11'],
42	'viirs-l1c': ['suomi-npp', 'noaa-20', 'noaa-21'],
43	'avhrr-l1c': ['noaa-19', 'noaa-18'],
44	'modis-l1c': ['eos-terra', 'eos-aqua']
45	}
46
47	LVL1C_PROCESSOR_MAPPING = {'seviri-l1c': process_seviri,	4✔
48	'viirs-l1c': process_viirs,
49	'modis-l1c': process_modis,
50	'avhrr-l1c': process_avhrr}
51
52
53	class ServiceNameNotSupported(Exception):	4✔
54	pass	4✔
55
56
57	class L1cProcessor(object):	4✔
58	"""Container for the NWCSAF/PPS Level-c processing."""
59
60	def __init__(self, config_filename, service_name):	4✔
61
NEW 62	options = get_config(config_filename, service_name)	×
NEW 63	check_service_name_okay(service_name)	×
64
NEW 65	self.service = service_name	×
66
NEW 67	self.subscribe_topics = options['message_types']	×
NEW 68	LOG.debug("Listens for messages of type: %s", str(self.subscribe_topics))	×
69
NEW 70	ncpus_available = cpu_count()	×
NEW 71	LOG.info("Number of CPUs available = %s", str(ncpus_available))	×
NEW 72	ncpus = int(options.get('num_of_cpus', 1))	×
NEW 73	LOG.info("Will use %d CPUs when running the level-1c processing instances", ncpus)	×
74
NEW 75	self.pool = ThreadPool(ncpus)	×
76
NEW 77	self.sensor = "unknown"	×
NEW 78	self.orbit_number = 99999 # Initialized orbit number	×
NEW 79	self.platform_name = 'unknown'	×
NEW 80	self.l1c_result = None	×
NEW 81	self.pass_start_time = None	×
NEW 82	self.l1cfile = None	×
NEW 83	self.level1_files = []	×
84
NEW 85	self.result_home = options.get('output_dir', '/tmp')	×
NEW 86	self.publish_topic = options.get('publish_topic')	×
NEW 87	self.message_data = None	×
88
89	def initialise(self, service):	4✔
90	"""Initialise the processor."""
NEW 91	self.check_service_name_okay(service)	×
NEW 92	self.l1c_result = None	×
NEW 93	self.pass_start_time = None	×
NEW 94	self.l1cfile = None	×
NEW 95	self.level1_files = []	×
NEW 96	self.service = service	×
97
98	def run(self, msg):	4✔
99	"""Start the L1c processing using the relevant sensor specific function from level1c4pps."""
100
NEW 101	if not check_message_okay(msg):	×
NEW 102	return False	×
103
NEW 104	self.platform_name = str(msg.data['platform_name'])	×
105
NEW 106	if not self.check_platform_name_consistent_with_service():	×
NEW 107	return False	×
108
NEW 109	self.sensor = str(msg.data['sensor'])	×
NEW 110	self.message_data = msg.data	×
111
NEW 112	level1_dataset = msg.data['dataset']	×
113
NEW 114	if len(level1_dataset) < 1:	×
NEW 115	return False	×
116
NEW 117	self.get_level1_files_from_dataset(level1_dataset)	×
118
NEW 119	l1c_proc = LVL1C_PROCESSOR_MAPPING.get(self.service)	×
NEW 120	if not l1c_proc:	×
NEW 121	raise AttributeError("Could not find suitable level-1c processor! Service = %s" % self.service)	×
122
NEW 123	self.l1c_result = self.pool.apply_async(l1c_proc, (self.level1_files,	×
124	self.result_home))
NEW 125	return True	×
126
127	def get_level1_files_from_dataset(self, level1_dataset):	4✔
128	"""Get the level-1 files from the dataset."""
129
NEW 130	if self.service in ['seviri-l1c']:	×
NEW 131	self.level1_files = get_seviri_level1_files_from_dataset(level1_dataset)	×
132	else:
NEW 133	for level1 in level1_dataset:	×
NEW 134	self.level1_files.append(level1['uri'])	×
135
136	def check_platform_name_consistent_with_service(self):	4✔
137	"""Check that the platform name is consistent with the service name."""
138
NEW 139	if self.platform_name.lower() not in SUPPORTED_SATELLITES.get(self.service, []):	×
NEW 140	LOG.warning("%s: Platform name not supported for this service: %s",	×
141	str(self.platform_name), self.service)
NEW 142	return False	×
143
NEW 144	return True	×
145
146
147	def get_seviri_level1_files_from_dataset(level1_dataset):	4✔
148	"""Get the seviri level-1 filenames from the dataset and return as list."""
149
NEW 150	pro_files = False	×
NEW 151	epi_files = False	×
NEW 152	level1_files = []	×
NEW 153	for level1 in level1_dataset:	×
NEW 154	level1_filename = level1['uri']	×
NEW 155	level1_files.append(level1_filename)	×
NEW 156	if '-PRO' in level1_filename:	×
NEW 157	pro_files = True	×
NEW 158	if '-EPI' in level1_filename:	×
NEW 159	epi_files = True	×
160
NEW 161	if not pro_files:	×
NEW 162	LOG.warning("PRO file is missing...")	×
NEW 163	return []	×
NEW 164	if not epi_files:	×
NEW 165	LOG.warning("EPI file is missing...")	×
NEW 166	return []	×
167
NEW 168	return level1_files	×
169
170
171	def check_message_okay(msg):	4✔
172	"""Check that the message is okay and has the necessary fields."""
173
174	if not msg:	4!
NEW 175	return False	×
176
177	if msg.type != 'dataset':	4✔
178	LOG.info("Not a dataset, don't do anything...")	4✔
179	return False	4✔
180
181	if ('platform_name' not in msg.data or 'start_time' not in msg.data):	4✔
182	LOG.error("Message is lacking crucial fields...")	4✔
183	return False	4✔
184
185	return True	4✔
186
187
188	def check_service_name_okay(service_name):	4✔
189	"""Check that the service name is supported."""
190	if service_name not in SUPPORTED_SERVICE_NAMES:	4✔
191	errmsg = "Service name %s is not yet supported" % service_name	4✔
192	raise ServiceNameNotSupported(errmsg)	4✔

pytroll / pytroll-pps-runner / 863906473

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