4354892438

Build Type

github

Committed by GitHub

Commit Message

Merge 9c511a95b into 5deea6f73

Pull Request Pull Request #57: Fix avhrr

Run Details

294 of 590 branches covered (49.83%)

Branch coverage included in aggregate %.

34 of 42 new or added lines in 7 files covered. (80.95%)

3 existing lines in 1 file now uncovered.

1591 of 2198 relevant lines covered (72.38%)

2.89 hits per line

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

71.88

/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 six.moves.urllib.parse import urlparse
from multiprocessing import cpu_count
from multiprocessing import Process, Manager
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', 'metop-b', 'metop-a', 'metop-c'],
                        '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 MessageTypeNotSupported(Exception):
    pass


class MessageContentMissing(Exception):
    pass


class PlatformNameInconsistentWithService(Exception):
    pass


class DatasetIsEmpty(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)

        self.initialize(service_name)
        self._l1c_processor_call_kwargs = options.get('l1cprocess_call_arguments', {'engine': 'h5netcdf'})
        self.time_limit_seconds = options.get('time_limit_seconds', 60)

        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))

        self.sensor = "unknown"
        self.orbit_number = 99999  # Initialized orbit number
        self.platform_name = 'unknown'

        self.result_home = options.get('output_dir', '/tmp')
        self.publish_topic = options.get('publish_topic')
        self.message_data = None
        self.nameservers = options.get('nameservers')
        if self.nameservers is not None and not isinstance(self.nameservers, list):
            self.nameservers = [self.nameservers]
        self.orbit_number_from_msg = options.get('orbit_number_from_msg', False)

    def initialize(self, service):
        """Initialize the processor."""
        check_service_is_supported(service)
        self.l1c_result = None
        self.pass_start_time = None
        self.l1cfile = None
        self.level1_files = []
        self.service = service

    def run_inner(self, l1c_proc_func, result_dict):
        """Start the L1c processing using the relevant sensor specific function from level1c4pps."""
        result_dict["l1cfile"] = l1c_proc_func(self.level1_files,
                                               self.result_home,
                                               **self._l1c_processor_call_kwargs)

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

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

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

        if self.orbit_number_from_msg:
            if 'orbit_number' in self.message_data:
                self.orbit_number = int(self.message_data.get('orbit_number'))
            else:
                LOG.warning("You asked for orbit_number from the message, but its not there. Keep init orbit.")

        if msg.type == 'dataset':
            level1_dataset = self.message_data.get('dataset')
            self.get_level1_files_from_dataset(level1_dataset)
        else:
            # Just one file; e.g. NOAA-POES or Metop AVHRR level-1 data
            self.level1_files = [urlparse(self.message_data.get('uri')).path]

        if len(self.level1_files) < 1:
            raise DatasetIsEmpty('No level-1 data in 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)
        LOG.debug(
            "Starting level1c processing in separate process.")
        manager = Manager()
        result_dict = manager.dict()
        process1 = Process(
            name="Process one level1c file in a separate process.",
            target=self.run_inner,
            args=(l1c_proc, result_dict))
        process1.start()
        process1.join(self.time_limit_seconds)  # Normally takes 3-8s for VIIRS
        self.l1cfile = result_dict.get("l1cfile", None)
        try:
            process1.close()
        except ValueError:
            # if process1.is_alive():
            LOG.warning(
                "Processing level1c file not terminated after {:d}s.".format(int(self.time_limit_seconds)))
            process1.terminate()

    def _get_message_data(self, message):
        """Return the data dict in the Posttroll message."""
        return message.data

    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(urlparse(level1['uri']).path)

    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, []):
            errmsg = ("%s: Platform name not supported for this service: %s",
                      str(self.platform_name), self.service)
            raise PlatformNameInconsistentWithService(errmsg)


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 msg.type != 'dataset' and msg.type != 'file':
        raise MessageTypeNotSupported("Not a dataset or a file, don't do anything...")

    if ('platform_name' not in msg.data):
        raise MessageContentMissing("Message is lacking crucial fields: platform_name")

    if ('start_time' not in msg.data):
        raise MessageContentMissing("Message is lacking crucial fields: start_time")


def check_service_is_supported(service_name):
    """Check that the service 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
24	"""
25
26	import logging
27	from six.moves.urllib.parse import urlparse
28	from multiprocessing import cpu_count
29	from multiprocessing import Process, Manager
30	from level1c4pps.seviri2pps_lib import process_one_scan as process_seviri
31	from level1c4pps.viirs2pps_lib import process_one_scene as process_viirs
32	from level1c4pps.modis2pps_lib import process_one_scene as process_modis
33	from level1c4pps.avhrr2pps_lib import process_one_scene as process_avhrr
34
35	from nwcsafpps_runner.config import get_config_from_yamlfile as get_config
36
37	LOG = logging.getLogger(__name__)
38
39	SUPPORTED_SERVICE_NAMES = ['seviri-l1c', 'viirs-l1c', 'avhrr-l1c', 'modis-l1c']
40
41	SUPPORTED_SATELLITES = {'seviri-l1c':
42	['meteosat-8', 'meteosat-9', 'meteosat-10', 'meteosat-11'],
43	'viirs-l1c': ['suomi-npp', 'noaa-20', 'noaa-21'],
44	'avhrr-l1c': ['noaa-19', 'noaa-18', 'metop-b', 'metop-a', 'metop-c'],
45	'modis-l1c': ['eos-terra', 'eos-aqua']
46	}
47
48	LVL1C_PROCESSOR_MAPPING = {'seviri-l1c': process_seviri,
49	'viirs-l1c': process_viirs,
50	'modis-l1c': process_modis,
51	'avhrr-l1c': process_avhrr}
52
53
54	class ServiceNameNotSupported(Exception):
55	pass
56
57
58	class MessageTypeNotSupported(Exception):
59	pass
60
61
62	class MessageContentMissing(Exception):
63	pass
64
65
66	class PlatformNameInconsistentWithService(Exception):
67	pass
68
69
70	class DatasetIsEmpty(Exception):
71	pass
72
73
74	class L1cProcessor(object):
75	"""Container for the NWCSAF/PPS Level-c processing."""
76
77	def __init__(self, config_filename, service_name):
78
79	options = get_config(config_filename, service_name)
80
81	self.initialize(service_name)
82	self._l1c_processor_call_kwargs = options.get('l1cprocess_call_arguments', {'engine': 'h5netcdf'})
83	self.time_limit_seconds = options.get('time_limit_seconds', 60)
84
85	self.subscribe_topics = options['message_types']
86	LOG.debug("Listens for messages of type: %s", str(self.subscribe_topics))
87
88	ncpus_available = cpu_count()
89	LOG.info("Number of CPUs available = %s", str(ncpus_available))
90
91	self.sensor = "unknown"
92	self.orbit_number = 99999 # Initialized orbit number
93	self.platform_name = 'unknown'
94
95	self.result_home = options.get('output_dir', '/tmp')
96	self.publish_topic = options.get('publish_topic')
97	self.message_data = None
98	self.nameservers = options.get('nameservers')
99	if self.nameservers is not None and not isinstance(self.nameservers, list):	4!
100	self.nameservers = [self.nameservers]
101	self.orbit_number_from_msg = options.get('orbit_number_from_msg', False)
102
103	def initialize(self, service):
104	"""Initialize the processor."""
105	check_service_is_supported(service)
106	self.l1c_result = None
107	self.pass_start_time = None
108	self.l1cfile = None
109	self.level1_files = []
110	self.service = service
111
112	def run_inner(self, l1c_proc_func, result_dict):
113	"""Start the L1c processing using the relevant sensor specific function from level1c4pps."""
114	result_dict["l1cfile"] = l1c_proc_func(self.level1_files,
115	self.result_home,
116	**self._l1c_processor_call_kwargs)
117
118	def run(self, msg):
119	"""Start the L1c processing using the relevant sensor specific function from level1c4pps."""
120	check_message_okay(msg)
121
122	self.platform_name = str(msg.data['platform_name'])
123	self.check_platform_name_consistent_with_service()
124
125	self.sensor = str(msg.data['sensor'])
126	self.message_data = self._get_message_data(msg)
127
128	if self.orbit_number_from_msg:	4!
129	if 'orbit_number' in self.message_data:	8✔
130	self.orbit_number = int(self.message_data.get('orbit_number'))
131	else:
132	LOG.warning("You asked for orbit_number from the message, but its not there. Keep init orbit.")
133
134	if msg.type == 'dataset':	4!
135	level1_dataset = self.message_data.get('dataset')
136	self.get_level1_files_from_dataset(level1_dataset)
137	else:
138	# Just one file; e.g. NOAA-POES or Metop AVHRR level-1 data
NEW 139	self.level1_files = [urlparse(self.message_data.get('uri')).path]
140
141	if len(self.level1_files) < 1:	4!
142	raise DatasetIsEmpty('No level-1 data in dataset!')
143
144	l1c_proc = LVL1C_PROCESSOR_MAPPING.get(self.service)
145	if not l1c_proc:	4!
146	raise AttributeError("Could not find suitable level-1c processor! Service = %s" % self.service)
147	LOG.debug(
148	"Starting level1c processing in separate process.")
149	manager = Manager()
150	result_dict = manager.dict()
151	process1 = Process(
152	name="Process one level1c file in a separate process.",
153	target=self.run_inner,
154	args=(l1c_proc, result_dict))
155	process1.start()
156	process1.join(self.time_limit_seconds) # Normally takes 3-8s for VIIRS
157	self.l1cfile = result_dict.get("l1cfile", None)
158	try:
159	process1.close()
UNCOV 160	except ValueError:
161	# if process1.is_alive():
UNCOV 162	LOG.warning(
163	"Processing level1c file not terminated after {:d}s.".format(int(self.time_limit_seconds)))
UNCOV 164	process1.terminate()
165
166	def _get_message_data(self, message):
167	"""Return the data dict in the Posttroll message."""
168	return message.data
169
170	def get_level1_files_from_dataset(self, level1_dataset):
171	"""Get the level-1 files from the dataset."""
172
173	if self.service in ['seviri-l1c']:	4!
174	self.level1_files = get_seviri_level1_files_from_dataset(level1_dataset)
175	else:
176	for level1 in level1_dataset:	8✔
177	self.level1_files.append(urlparse(level1['uri']).path)
178
179	def check_platform_name_consistent_with_service(self):
180	"""Check that the platform name is consistent with the service name."""
181
182	if self.platform_name.lower() not in SUPPORTED_SATELLITES.get(self.service, []):	4!
183	errmsg = ("%s: Platform name not supported for this service: %s",
184	str(self.platform_name), self.service)
185	raise PlatformNameInconsistentWithService(errmsg)
186
187
188	def get_seviri_level1_files_from_dataset(level1_dataset):
189	"""Get the seviri level-1 filenames from the dataset and return as list."""
190
191	pro_files = False
192	epi_files = False
193	level1_files = []
194	for level1 in level1_dataset:	×
195	level1_filename = level1['uri']
196	level1_files.append(level1_filename)
197	if '-PRO' in level1_filename:	×
198	pro_files = True
199	if '-EPI' in level1_filename:	×
200	epi_files = True
201
202	if not pro_files:	×
203	LOG.warning("PRO file is missing...")
204	return []
205	if not epi_files:	×
206	LOG.warning("EPI file is missing...")
207	return []
208
209	return level1_files
210
211
212	def check_message_okay(msg):
213	"""Check that the message is okay and has the necessary fields."""
214	if msg.type != 'dataset' and msg.type != 'file':	8✔
215	raise MessageTypeNotSupported("Not a dataset or a file, don't do anything...")
216
217	if ('platform_name' not in msg.data):	8✔
218	raise MessageContentMissing("Message is lacking crucial fields: platform_name")
219
220	if ('start_time' not in msg.data):	8✔
221	raise MessageContentMissing("Message is lacking crucial fields: start_time")
222
223
224	def check_service_is_supported(service_name):
225	"""Check that the service is supported."""
226	if service_name not in SUPPORTED_SERVICE_NAMES:	8✔
227	errmsg = "Service name %s is not yet supported" % service_name
228	raise ServiceNameNotSupported(errmsg)

pytroll / pytroll-pps-runner / 4354892438

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