Changed type for bytes converter to parse bytes as ascii numbers, not utf8 string, added hexMode to bytes mqtt uplink converter

Author: imbeacon

thingsboard_gateway/connectors/mqtt/bytes_mqtt_uplink_converter.py

@@ -14,6 +14,7 @@
 
 
 class BytesMqttUplinkConverter(MqttUplinkConverter):
+    SUPPORTS_BYTES_PAYLOAD = True
     def __init__(self, config, logger):
         self.__config = config.get('converter')
         self._log = logger
@@ -48,8 +49,8 @@ def convert(self, topic, data):
         try:
             for datatype in datatypes:
                 for datatype_config in self.__config.get(datatype, []):
-                    key = self.parse_data(datatype_config['key'], data)
-                    value = self.parse_data(datatype_config['value'], data)
+                    key = self.parse_data(datatype_config['key'], data, hex_mode=datatype_config.get('hexMode', False))
+                    value = self.parse_data(datatype_config['value'], data, hex_mode=datatype_config.get('hexMode', False))
                     datapoint_key = TBUtility.convert_key_to_datapoint_key(key,
                                                                            device_report_strategy,
                                                                            datatype_config,
@@ -74,23 +75,25 @@ def convert(self, topic, data):
         return converted_data
 
     @staticmethod
-    def parse_data(expression, data):
+    def parse_data(expression: str, data: list, hex_mode: bool = False) -> str:
         expression_arr = findall(r'\[\S[0-9:]*]', expression)
         converted_data = expression
 
         for exp in expression_arr:
             indexes = exp[1:-1].split(':')
-
             data_to_replace = ''
+
             if len(indexes) == 2:
-                from_index, to_index = indexes
-                concat_arr = data[
-                             int(from_index) if from_index != '' else None:int(
-                                 to_index) if to_index != '' else None]
-                for sub_item in concat_arr:
-                    data_to_replace += str(sub_item)
+                from_index = int(indexes[0]) if indexes[0] else None
+                to_index = int(indexes[1]) if indexes[1] else None
+                slice_data = data[from_index:to_index]
+            else:
+                slice_data = [data[int(indexes[0])]]
+
+            if hex_mode:
+                data_to_replace = ''.join(f'{int(b):02x}' for b in slice_data)
             else:
-                data_to_replace += str(data[int(indexes[0])])
+                data_to_replace = ''.join(str(b) for b in slice_data)
 
             converted_data = converted_data.replace(exp, data_to_replace)
 

thingsboard_gateway/connectors/mqtt/mqtt_connector.py

@@ -544,6 +544,8 @@ def _on_subscribe(self, _, __, mid, granted_qos, *args):
 
     def put_data_to_convert(self, converter, message, content) -> bool:
         if not self.__msg_queue.full():
+            if not hasattr(converter, 'SUPPORTS_BYTES_PAYLOAD'):
+                content = TBUtility.decode(content)
             self.__msg_queue.put((converter.convert, message.topic, content), True, 100)
             return True
         return False
@@ -615,7 +617,7 @@ def _process_on_message(self):
                 client, userdata, message = self._on_message_queue.get_nowait()
 
                 self.statistics['MessagesReceived'] += 1
-                content = TBUtility.decode(message)
+                content = None
 
                 # Check if message topic exists in mappings "i.e., I'm posting telemetry/attributes" -------------------
                 topic_handlers = [regex for regex in self.__mapping_sub_topics if fullmatch(regex, message.topic)]
@@ -632,7 +634,7 @@ def _process_on_message(self):
                         available_converters = self.__mapping_sub_topics[topic]
                         for converter in available_converters:
                             try:
-                                request_handled = self.put_data_to_convert(converter, message, content)
+                                request_handled = self.put_data_to_convert(converter, message, message.payload)
                             except Exception as e:
                                 self.__log.exception(e)
 
@@ -651,6 +653,8 @@ def _process_on_message(self):
                                   fullmatch(regex, message.topic)]
 
                 if topic_handlers:
+                    if content is None:
+                        content = TBUtility.decode(message)
                     for topic in topic_handlers:
                         handler = self.__connect_requests_sub_topics[topic]
 
@@ -677,6 +681,8 @@ def _process_on_message(self):
                 topic_handlers = [regex for regex in self.__disconnect_requests_sub_topics if
                                   fullmatch(regex, message.topic)]
                 if topic_handlers:
+                    if content is None:
+                        content = TBUtility.decode(message)
                     for topic in topic_handlers:
                         handler = self.__disconnect_requests_sub_topics[topic]
 
@@ -705,6 +711,8 @@ def _process_on_message(self):
                 topic_handlers = [regex for regex in self.__attribute_requests_sub_topics if
                                   fullmatch(regex, message.topic)]
                 if topic_handlers:
+                    if content is None:
+                        content = TBUtility.decode(message)
                     try:
                         for topic in topic_handlers:
                             handler = self.__attribute_requests_sub_topics[topic]

thingsboard_gateway/tb_utility/tb_utility.py

@@ -50,18 +50,25 @@ class TBUtility:
     @staticmethod
     def decode(message, return_raw=False):
         if return_raw:
-            return message.payload
-        try:
-            if isinstance(message.payload, bytes):
-                content = loads(message.payload.decode("utf-8", "ignore"))
-            else:
-                content = loads(message.payload)
-        except JSONDecodeError:
+            return getattr(message, 'payload', message)
+
+        payload = message if isinstance(message, bytes) else getattr(message, 'payload', None)
+        if payload is None:
+            return message
+
+        if isinstance(payload, bytes):
+            try:
+                return loads(payload.decode("utf-8", "ignore"))
+            except JSONDecodeError:
+                try:
+                    return payload.decode("utf-8")
+                except UnicodeDecodeError:
+                    return payload
+        else:
             try:
-                content = message.payload.decode("utf-8")
-            except UnicodeDecodeError:
-                content = message.payload
-        return content
+                return loads(payload)
+            except JSONDecodeError:
+                return payload
 
     @staticmethod
     def validate_converted_data(data: Union[dict, 'ConvertedData']):