Wis2 integration - MQTT publishing using sr3 stack

sarracenia/config.py

@@ -1077,7 +1077,7 @@ def add_option(self, option, kind='list', default_value=None, all_values=None ):
 
         if kind not in [ 'list', 'set' ] and type(v) == list:
             v=v[-1]
-            logger.warning( f"{self.files}{self.lineno} multiple declarations of {option}={getattr(self,option)} choosing last one: {v}" )
+            logger.warning( f"{self.files}{self.lineno} multiple declarations of {kind} {option}={getattr(self,option)} choosing last one: {v}" )
 
 
         if kind == 'count':
@@ -1490,7 +1490,6 @@ def parse_file(self, cfg, component=None):
                     setattr(self, k, True)
                 else:
                     setattr(self, k, isTrue(v))
-
                 if k in ['logReject'] and self.logReject:
                     self.logEvents = self.logEvents | set(['reject'])
                 continue
@@ -1975,7 +1974,7 @@ def finalize(self, component=None, config=None):
         if self.messageCountMax > 0:
             if self.batch > self.messageCountMax:
                 self.batch = self.messageCountMax
-                logger.info( 'overriding batch for consistency with messageCountMax: {self.batch}' )
+                logger.info( f'overriding batch for consistency with messageCountMax: {self.batch}' )
 
         if (component not in ['poll' ]):
             self.path = list(map( os.path.expanduser, self.path ))

sarracenia/moth/amqp.py

@@ -201,7 +201,6 @@ def __init__(self, props, is_subscriber) -> None:
                 logger.setLevel(self.o['logLevel'].upper())
 
         self.connection = None
-
     def __connect(self, broker) -> bool:
         """
           connect to broker. 

sarracenia/moth/mqtt.py

@@ -738,6 +738,10 @@ def putNewMessage(self,
                     logger.info( f"UserProperty:{props.UserProperty}" )
 
 
+            if not topic:
+                logger.error(f"message without topic will not be received - publish aborted")
+                return False
+
             info = self.client.publish(topic=topic, payload=raw_body, qos=self.o['qos'], properties=props)
 
             if info.rc == paho.mqtt.client.MQTT_ERR_SUCCESS:

sarracenia/postformat/v02.py

@@ -63,6 +63,9 @@ def importMine(body, headers, options) -> sarracenia.Message:
         msg['relPath'] = relPath
         msg['subtopic'] = relPath.split('/')
         msg['to_clusters'] = 'ALL'
+        if 'integrity' in msg:
+            msg['identity'] = msg['integrity']
+            del msg['integrity']
         msg['_deleteOnPost'] |= set(['subtopic'])
 
         for t in ['atime', 'mtime']:

sarracenia/postformat/wis.py

@@ -125,7 +125,9 @@ def exportMine(body, options) -> (str, dict, str):
                 if literal in body:
                     GeoJSONBody[literal] = body[literal]
 
-            if 'topic' in options:
+            if 'topic' in body:
+                topic = body['topic'].split('/')
+            elif 'topic' in options:
                 topic=options['topic'].split('/')
             else:
                 topic= []
@@ -136,7 +138,7 @@ def exportMine(body, options) -> (str, dict, str):
 
             """
             for h in body:
-                if h not in [ 'geometry', 'properties', 'size', 'baseUrl', 'relPath', 'retrievePath', 'subtopic', 'pubTime' ]:
+                if h not in [ 'contentType', 'geometry', 'properties', 'size', 'baseUrl', 'relPath', 'retrievePath', 'subtopic', 'pubTime', 'to_clusters', 'from_cluster', 'filename', 'sundew_extension', 'mtime', 'atime', 'mode', 'identity', 'topic' ]:
                     GeoJSONBody['properties'][h] = body[h]
 
             t=body['pubTime']
@@ -145,13 +147,13 @@ def exportMine(body, options) -> (str, dict, str):
             if 'geometry' in body:
                 GeoJSONBody['geometry'] = body['geometry']
 
-            if 'data_id' not in body:
-                GeoJSONBody['properties']['data_id'] =  str(uuid.uuid4())
+            if 'id' not in body:
+                GeoJSONBody['id'] =  str(uuid.uuid4())
 
             if 'retrievePath' in body :
-                url = body['baseUrl'] + body['retrievePath']
+                url = body['baseUrl'] + "/" + body['retrievePath']
             else:
-                url = body['baseUrl'] + body['relPath']
+                url = body['baseUrl'] + "/" + body['relPath']
 
             if not 'links' in GeoJSONBody: