add GTS headers support

Co-authored-by: Copilot <copilot@github.com>

Author: maaikelimper

wis2box_api/plugins/process/synop2bufr.py

@@ -132,16 +132,25 @@ def execute(self, data):
         """
 
         LOGGER.info('Executing process {}'.format(self.name))
-
+        gts_headers = None
         try:
             notify = data['notify']
             metadata_id = data.get('metadata_id', None)
             channel = data.get('channel', None)
+            gts_ttaaii = data.get('gts_ttaaii', None)
+            gts_cccc = data.get('gts_cccc', None)
             if metadata_id is None and notify:
                 raise Exception('metadata_id must be provided if notify is True') # noqa
         except Exception as err:
             return handle_error({err})
 
+        gts_headers = None
+        if gts_ttaaii is not None and gts_cccc is not None:
+            gts_headers = {
+                'ttaaii': gts_ttaaii,
+                'cccc': gts_cccc
+            }
+
         # get the channel from the metadata
         if channel is None:
             try:
@@ -211,7 +220,7 @@ def execute(self, data):
 
         LOGGER.debug(f'synop2bufr-transform returned {len(output_items)} items') # noqa
 
-        return data_handler.process_items(output_items)
+        return data_handler.process_items(output_items, gts_headers=gts_headers) # noqa
 
     def __repr__(self):
         return '<submit> {}'.format(self.name)

wis2box_api/wis2box/handle.py

@@ -79,10 +79,11 @@ def __init__(self, channel, notify, metadata_id=None):
         self._channel = channel.replace('origin/a/wis2/', '')
         self.metadata_id = metadata_id
 
-    def process_items(self, output_items: []):
+    def process_items(self, output_items: [], gts_headers: dict = None):
         """Process output_items, store and publish them
 
         :param output_items: list of output-items from the transform
+        :param gts_headers: optional dict of GTS headers for notification
 
         :returns: 'application/json'
         """
@@ -163,7 +164,7 @@ def process_items(self, output_items: []):
                     })
                 if self._notify:
                     # send the last entry in the data list as a notification
-                    result = self.send_data_publish_request(data[-1])
+                    result = self.send_data_publish_request(data[-1], gts_headers=gts_headers) # noqa
                     if result != 'success':
                         errors.append(f'{result}')
                     else:
@@ -188,7 +189,7 @@ def process_items(self, output_items: []):
 
         return mimetype, outputs
 
-    def send_data_publish_request(self, data_item: dict):
+    def send_data_publish_request(self, data_item: dict, gts_headers: dict = None): # noqa
         """Send DataPublishRequest
 
         :param data: data_item
@@ -205,6 +206,8 @@ def send_data_publish_request(self, data_item: dict):
                 'filename': data_item['filename'],
                 '_meta': data_item['_meta']
             }
+            if gts_headers is not None:
+                msg['gts'] = gts_headers
             # publish notification on internal broker
             private_auth = {
                 'username': BROKER_USERNAME,