initial weather story loader

Dukestep · Dec 19, 2024 · 277831b · 277831b
1 parent 814204c
commit 277831b
Show file tree

Hide file tree

Showing 19 changed files with 229 additions and 88 deletions.
diff --git a/deploy/default/sarracenia/aqhi-realtime.conf b/deploy/default/sarracenia/aqhi-realtime.conf
@@ -10,4 +10,4 @@ discard True
 report False
 directory ${MSC_PYGEOAPI_CACHEDIR}
 logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
-callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+callback msc_pygeoapi.event.EventAfterWork
diff --git a/deploy/default/sarracenia/bulletins-realtime.conf b/deploy/default/sarracenia/bulletins-realtime.conf
@@ -7,6 +7,6 @@ subtopic *.WXO-DD.bulletins.alphanumeric.#
 
 mirror True
 directory /tmp/bulletins
-callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+callback msc_pygeoapi.event.EventAfterAccept
 logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
 report False
diff --git a/deploy/default/sarracenia/cap-alerts.conf b/deploy/default/sarracenia/cap-alerts.conf
@@ -5,7 +5,7 @@ instances 2
 
 subtopic *.WXO-DD.alerts.cap.#
 
-callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+callback msc_pygeoapi.event.EventAfterWork
 directory /data/geomet/feeds/hpfx
 mirror True
 discard True

diff --git a/deploy/default/sarracenia/citypageweather.conf b/deploy/default/sarracenia/citypageweather.conf
@@ -6,7 +6,7 @@ instances 2
 subtopic *.WXO-DD.citypage_weather.xml.#
 
 directory ${MSC_PYGEOAPI_CACHEDIR}
-callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+callback msc_pygeoapi.event.EventAfterWork
 mirror True
 discard True
 strip 3
diff --git a/deploy/default/sarracenia/coastal-flood-risk-index.conf b/deploy/default/sarracenia/coastal-flood-risk-index.conf
@@ -7,7 +7,7 @@ instances 2
 subtopic coastal-flooding.risk-index.#
 
 directory ${MSC_PYGEOAPI_CACHEDIR}
-callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+callback msc_pygeoapi.event.EventAfterWork
 mirror True
 discard True
 logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}

diff --git a/deploy/default/sarracenia/hurricanes.conf b/deploy/default/sarracenia/hurricanes.conf
@@ -6,7 +6,7 @@ instances 2
 subtopic *.WXO-DD.hurricanes.#
 
 directory ${MSC_PYGEOAPI_CACHEDIR}
-callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+callback msc_pygeoapi.event.EventAfterWork
 mirror True
 discard True
 slip 3
diff --git a/deploy/default/sarracenia/hydrometric-realtime.conf b/deploy/default/sarracenia/hydrometric-realtime.conf
@@ -6,7 +6,7 @@ instances 4
 subtopic *.WXO-DD.hydrometric.#
 
 directory ${MSC_PYGEOAPI_CACHEDIR}
-callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+callback msc_pygeoapi.event.EventAfterWork
 logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
 discard True
 report False

diff --git a/deploy/default/sarracenia/marine_weather.conf b/deploy/default/sarracenia/marine_weather.conf
@@ -6,7 +6,7 @@ instances 2
 subtopic *.WXO-DD.marine_weather.xml.#
 
 directory /data/geomet/feeds/hpfx
-callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+callback msc_pygeoapi.event.EventAfterWork
 mirror True
 discard True
 skip 3

diff --git a/deploy/default/sarracenia/metnotes.conf b/deploy/default/sarracenia/metnotes.conf
@@ -6,7 +6,7 @@ instances 2
 subtopic *.WXO-DD.metnotes.#
 
 directory ${MSC_PYGEOAPI_CACHEDIR}
-callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+callback msc_pygeoapi.event.EventAfterWork
 mirror True
 discard True
 report False

diff --git a/deploy/default/sarracenia/swob-realtime.conf b/deploy/default/sarracenia/swob-realtime.conf
@@ -6,7 +6,7 @@ instances 4
 subtopic *.WXO-DD.observations.swob-ml.#
 
 directory ${MSC_PYGEOAPI_CACHEDIR}
-callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+callback msc_pygeoapi.event.EventAfterWork
 mirror True
 discard True
 logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}

diff --git a/deploy/default/sarracenia/thunderstorm-outlook.conf b/deploy/default/sarracenia/thunderstorm-outlook.conf
@@ -7,7 +7,7 @@ instances 2
 subtopic thunderstorm-outlooks.#
 
 directory ${MSC_PYGEOAPI_CACHEDIR}
-callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+callback msc_pygeoapi.event.EventAfterWork
 mirror True
 discard True
 logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}

diff --git a/deploy/default/sarracenia/umos-realtime.conf b/deploy/default/sarracenia/umos-realtime.conf
@@ -7,7 +7,7 @@ subtopic *.WXO-DD.model_gem_global.stat-post-processing.#
 subtopic *.WXO-DD.model_gem_regional.stat-post-processing.#
 
 directory ${MSC_PYGEOAPI_CACHEDIR}
-callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+callback msc_pygeoapi.event.EventAfterWork
 mirror True
 discard True
 logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}

diff --git a/deploy/default/sarracenia/weather-story.conf b/deploy/default/sarracenia/weather-story.conf
@@ -0,0 +1,13 @@
+broker amqps://[email protected]
+exchange xs_snlw001
+queue_name q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
+topic_prefix v03.post
+instances 2
+
+subtopic #
+
+strip 2
+directory ${MSC_PYGEOAPI_CACHEDIR}
+callback msc_pygeoapi.event.EventAfterWork
+logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
+report False
diff --git a/msc-pygeoapi.env b/msc-pygeoapi.env
@@ -8,7 +8,6 @@ export MSC_PYGEOAPI_ES_URL=http://${MSC_PYGEOAPI_ES_USERNAME}:${MSC_PYGEOAPI_ES_
 export MSC_PYGEOAPI_CACHEDIR=/tmp
 export MSC_PYGEOAPI_OGC_API_URL=https://api.wxod-dev.cmc.ec.gc.ca
 export MSC_PYGEOAPI_OGC_API_URL_BASEPATH=/
-export MSC_PYGEOAPI_METPX_FLOW_CALLBACK=msc_pygeoapi.event.event_plugin.Event
 export MSC_PYGEOAPI_TEMPLATES=theme/templates
 export MSC_PYGEOAPI_STATIC=theme/static
 export MSC_PYGEOAPI_LOCALE=locale

diff --git a/msc_pygeoapi/event/__init__.py b/msc_pygeoapi/event/__init__.py
@@ -1,8 +1,13 @@
 # =================================================================
 #
 # Author: Tom Kralidis <[email protected]>
+#         Louis-Philippe Rousseau-Lambert
+#             <[email protected]>
+#         Etienne Pelletier <[email protected]>
 #
 # Copyright (c) 2021 Tom Kralidis
+# Copyright (c) 2024 Louis-Philippe Rousseau-Lambert
+# Copyright (c) 2024 Etienne Pelletier
 #
 # Permission is hereby granted, free of charge, to any person
 # obtaining a copy of this software and associated documentation
@@ -26,3 +31,63 @@
 # OTHER DEALINGS IN THE SOFTWARE.
 #
 # =================================================================
+import logging
+
+from sarracenia.flowcb import FlowCB
+
+LOGGER = logging.getLogger(__name__)
+
+
+class EventBase(FlowCB):
+
+    def process_messages(self, worklist) -> bool:
+        """
+        Process messages from the worklist
+
+        :param worklist: `sarracenia.flow.worklist`
+
+        :returns: `bool`
+        """
+
+        for msg in worklist.incoming:
+
+            try:
+                from msc_pygeoapi.handler.core import CoreHandler
+
+                filepath = f"{msg['new_dir']}/{msg['new_file']}"
+                LOGGER.debug(f'Filepath: {filepath}')
+                handler = CoreHandler(filepath)
+                result = handler.handle()
+                LOGGER.debug(f'Result: {result}')
+            except Exception as err:
+                LOGGER.error(f'Error handling message: {err}')
+                worklist.failed.append(msg)
+                return False
+
+        return True
+
+
+class EventAfterWork(EventBase):
+
+    def after_work(self, worklist) -> None:
+        """
+        sarracenia after_work dispatcher
+
+        :param worklist: `sarracenia.flow.worklist`
+
+        :returns: `bool`
+        """
+        return self.process_messages(worklist)
+
+
+class EventAfterAccept(EventBase):
+
+    def after_accept(self, worklist) -> None:
+        """
+        sarracenia after_accept dispatcher
+
+        :param worklist: `sarracenia.flow.worklist`
+
+        :returns: `bool`
+        """
+        return self.process_messages(worklist)
diff --git a/msc_pygeoapi/event/event_plugin.py b/msc_pygeoapi/event/event_plugin.py
diff --git a/msc_pygeoapi/loader/__init__.py b/msc_pygeoapi/loader/__init__.py
@@ -72,8 +72,11 @@ def metadata():
     ('msc_pygeoapi.loader.nwp_dataset_footprints', 'nwp_dataset_footprints'),
     ('msc_pygeoapi.loader.umos_realtime', 'umos_realtime'),
     ('msc_pygeoapi.loader.thunderstorm_outlook', 'thunderstorm_outlook'),
-    ('msc_pygeoapi.loader.coastal_flood_risk_index',
-     'coastal_flood_risk_index')
+    (
+        'msc_pygeoapi.loader.coastal_flood_risk_index',
+        'coastal_flood_risk_index',
+    ),
+    ('msc_pygeoapi.loader.weather_story_realtime', 'weather_story_realtime'),
 )
 
 for module, name in commands: