1307488438

Build Type

github

Committed by GitHub

Commit Message

Merge e3a6dd755 into a622f504d

Pull Request Pull Request #29: Add pps2021 support

Run Details

105 of 442 branches covered (23.76%)

Branch coverage included in aggregate %.

153 of 202 new or added lines in 6 files covered. (75.74%)

2 existing lines in 1 file now uncovered.

950 of 1765 relevant lines covered (53.82%)

2.14 hits per line

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

0.0

/nwcsafpps_runner/publish_and_listen.py

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

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

"""Publisher and Listener classes for the PPS runners.
"""

import posttroll.subscriber
from posttroll.publisher import Publish
import threading
from nwcsafpps_runner.utils import (SUPPORTED_PPS_SATELLITES,
                                    SUPPORTED_SEVIRI_SATELLITES)

import logging
LOG = logging.getLogger(__name__)


class FileListener(threading.Thread):

    def __init__(self, queue, subscribe_topics):
        threading.Thread.__init__(self)
        self.loop = True
        self.queue = queue
        self.subscribe_topics = subscribe_topics

    def stop(self):
        """Stops the file listener."""
        self.loop = False
        self.queue.put(None)

    def run(self):

        LOG.debug("Subscribe topics = %s", str(self.subscribe_topics))
        with posttroll.subscriber.Subscribe("", self.subscribe_topics, True) as subscr:

            for msg in subscr.recv(timeout=90):
                if not self.loop:
                    break

                # Check if it is a relevant message:
                if self.check_message(msg):
                    LOG.info("Put the message on the queue...")
                    LOG.debug("Message = " + str(msg))
                    self.queue.put(msg)

    def check_message(self, msg):

        if not msg:
            return False

        if ('platform_name' not in msg.data or
                'start_time' not in msg.data):
            LOG.warning("Message is lacking crucial fields...")
            return False
        #: Orbit_number not needed for seviri
        if (msg.data['platform_name'] not in SUPPORTED_SEVIRI_SATELLITES):
            if ('orbit_number' not in msg.data):
                LOG.warning("Message is lacking crucial fields...")
                return False
        if (msg.data['platform_name'] not in SUPPORTED_PPS_SATELLITES):
            LOG.info(str(msg.data['platform_name']) + ": " +
                     "Not a NOAA/Metop/S-NPP/Terra/Aqua scene. Continue...")
            return False

        return True


class FilePublisher(threading.Thread):
    """A publisher for the PPS result files.

    Picks up the return value from the
    pps_worker when ready, and publishes the files via posttroll
    """

    def __init__(self, queue, publish_topic, **kwargs):
        threading.Thread.__init__(self)
        self.loop = True
        self.queue = queue
        self.jobs = {}
        self.publish_topic = publish_topic
        self.runner_name = kwargs.get('runner_name', 'pps_runner')

    def stop(self):
        """Stops the file publisher."""
        self.loop = False
        self.queue.put(None)

    def run(self):

        with Publish(self.runner_name, 0, self.publish_topic) as publisher:

            while self.loop:
                retv = self.queue.get()

                if retv is not None:
                    LOG.info("Publish the files...")
                    publisher.send(retv)

1	#!/usr/bin/env python
2	# -- coding: utf-8 --
3
4	# Copyright (c) 2018 - 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	"""Publisher and Listener classes for the PPS runners.	×
24	"""
25
26	import posttroll.subscriber	×
27	from posttroll.publisher import Publish	×
28	import threading	×
29	from nwcsafpps_runner.utils import (SUPPORTED_PPS_SATELLITES,	×
30	SUPPORTED_SEVIRI_SATELLITES)
31
32	import logging	×
33	LOG = logging.getLogger(__name__)	×
34
35
36	class FileListener(threading.Thread):	×
37
38	def __init__(self, queue, subscribe_topics):	×
39	threading.Thread.__init__(self)	×
40	self.loop = True	×
41	self.queue = queue	×
42	self.subscribe_topics = subscribe_topics	×
43
44	def stop(self):	×
45	"""Stops the file listener."""
46	self.loop = False	×
47	self.queue.put(None)	×
48
49	def run(self):	×
50
51	LOG.debug("Subscribe topics = %s", str(self.subscribe_topics))	×
52	with posttroll.subscriber.Subscribe("", self.subscribe_topics, True) as subscr:	×
53
54	for msg in subscr.recv(timeout=90):	×
55	if not self.loop:	×
56	break	×
57
58	# Check if it is a relevant message:
59	if self.check_message(msg):	×
60	LOG.info("Put the message on the queue...")	×
61	LOG.debug("Message = " + str(msg))	×
62	self.queue.put(msg)	×
63
64	def check_message(self, msg):	×
65
66	if not msg:	×
67	return False	×
68
69	if ('platform_name' not in msg.data or	×
70	'start_time' not in msg.data):
71	LOG.warning("Message is lacking crucial fields...")	×
72	return False	×
73	#: Orbit_number not needed for seviri
NEW 74	if (msg.data['platform_name'] not in SUPPORTED_SEVIRI_SATELLITES):	×
75	if ('orbit_number' not in msg.data):	×
76	LOG.warning("Message is lacking crucial fields...")	×
77	return False	×
78	if (msg.data['platform_name'] not in SUPPORTED_PPS_SATELLITES):	×
79	LOG.info(str(msg.data['platform_name']) + ": " +	×
80	"Not a NOAA/Metop/S-NPP/Terra/Aqua scene. Continue...")
81	return False	×
82
83	return True	×
84
85
86	class FilePublisher(threading.Thread):	×
87	"""A publisher for the PPS result files.
88
89	Picks up the return value from the
90	pps_worker when ready, and publishes the files via posttroll
91	"""
92
93	def __init__(self, queue, publish_topic, **kwargs):	×
94	threading.Thread.__init__(self)	×
95	self.loop = True	×
96	self.queue = queue	×
97	self.jobs = {}	×
98	self.publish_topic = publish_topic	×
99	self.runner_name = kwargs.get('runner_name', 'pps_runner')	×
100
101	def stop(self):	×
102	"""Stops the file publisher."""
103	self.loop = False	×
104	self.queue.put(None)	×
105
106	def run(self):	×
107
108	with Publish(self.runner_name, 0, self.publish_topic) as publisher:	×
109
110	while self.loop:	×
111	retv = self.queue.get()	×
112
113	if retv is not None:	×
114	LOG.info("Publish the files...")	×
115	publisher.send(retv)	×

pytroll / pytroll-pps-runner / 1307488438

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