Added Async Modbus connector

setup.py

@@ -39,7 +39,7 @@
               'thingsboard_gateway.storage', 'thingsboard_gateway.storage.memory', 'thingsboard_gateway.gateway.shell',
               'thingsboard_gateway.storage.file', 'thingsboard_gateway.storage.sqlite', 'thingsboard_gateway.gateway.entities',
               'thingsboard_gateway.connectors', 'thingsboard_gateway.connectors.ble', 'thingsboard_gateway.connectors.socket',
-              'thingsboard_gateway.connectors.mqtt', 'thingsboard_gateway.connectors.xmpp',
+              'thingsboard_gateway.connectors.mqtt', 'thingsboard_gateway.connectors.xmpp', 'thingsboard_gateway.connectors.modbus_async',
               'thingsboard_gateway.connectors.opcua', 'thingsboard_gateway.connectors.request', 'thingsboard_gateway.connectors.ocpp',
               'thingsboard_gateway.connectors.modbus', 'thingsboard_gateway.connectors.can', 'thingsboard_gateway.connectors.bacnet',
               'thingsboard_gateway.connectors.bacnet.bacnet_utilities', 'thingsboard_gateway.connectors.odbc',

thingsboard_gateway/connectors/modbus_async/bytes_modbus_downlink_converter.py

@@ -0,0 +1,124 @@
+#     Copyright 2024. ThingsBoard
+#
+#     Licensed under the Apache License, Version 2.0 (the "License");
+#     you may not use this file except in compliance with the License.
+#     You may obtain a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#     Unless required by applicable law or agreed to in writing, software
+#     distributed under the License is distributed on an "AS IS" BASIS,
+#     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#     See the License for the specific language governing permissions and
+#     limitations under the License.
+
+from pymodbus.constants import Endian
+from pymodbus.payload import BinaryPayloadBuilder
+
+from thingsboard_gateway.connectors.modbus_async.modbus_converter import ModbusConverter
+from thingsboard_gateway.connectors.modbus_async.configs.bytes_downlink_converter_config import \
+    BytesDownlinkConverterConfig
+from thingsboard_gateway.gateway.statistics.decorators import CollectStatistics
+
+
+class BytesModbusDownlinkConverter(ModbusConverter):
+
+    def __init__(self, _, logger):
+        self._log = logger
+
+    @CollectStatistics(start_stat_type='allReceivedBytesFromTB',
+                       end_stat_type='allBytesSentToDevices')
+    def convert(self, config: BytesDownlinkConverterConfig, data):
+        byte_order = Endian.Big if config.byte_order.upper() == "BIG" else Endian.Little
+        word_order = Endian.Big if config.word_order.upper() == "BIG" else Endian.Little
+        repack = config.repack
+        builder = BinaryPayloadBuilder(byteorder=byte_order, wordorder=word_order, repack=repack)
+        builder_functions = {"string": builder.add_string,
+                             "bits": builder.add_bits,
+                             "8int": builder.add_8bit_int,
+                             "16int": builder.add_16bit_int,
+                             "32int": builder.add_32bit_int,
+                             "64int": builder.add_64bit_int,
+                             "8uint": builder.add_8bit_uint,
+                             "16uint": builder.add_16bit_uint,
+                             "32uint": builder.add_32bit_uint,
+                             "64uint": builder.add_64bit_uint,
+                             "16float": builder.add_16bit_float,
+                             "32float": builder.add_32bit_float,
+                             "64float": builder.add_64bit_float}
+
+        value = data["data"]["params"]
+
+        if config.lower_type == "error":
+            self._log.error('"type" and "tag" - not found in configuration.')
+
+        lower_type = config.lower_type
+        variable_size = config.objects_count * 16 if lower_type not in ["coils", "bits", "coil",
+                                                                        "bit"] else config.objects_count
+
+        if lower_type in ["integer", "dword", "dword/integer", "word", "int"]:
+            lower_type = str(variable_size) + "int"
+            assert builder_functions.get(lower_type) is not None
+            builder_functions[lower_type](int(value))
+        elif lower_type in ["uint", "unsigned", "unsigned integer", "unsigned int"]:
+            lower_type = str(variable_size) + "uint"
+            assert builder_functions.get(lower_type) is not None
+            builder_functions[lower_type](int(value))
+        elif lower_type in ["float", "double"]:
+            lower_type = str(variable_size) + "float"
+            assert builder_functions.get(lower_type) is not None
+            builder_functions[lower_type](float(value))
+        elif lower_type in ["coil", "bits", "coils", "bit"]:
+            assert builder_functions.get("bits") is not None
+            if variable_size > 1:
+                if isinstance(value, str):
+                    builder_functions["bits"](bytes(value, encoding='UTF-8'))
+                elif isinstance(value, list):
+                    builder_functions["bits"]([int(x) for x in value])
+                else:
+                    builder_functions["bits"]([int(x) for x in bin(value)[2:]])
+            else:
+                return int(value).to_bytes(1, byteorder='big')
+        elif lower_type in ["string"]:
+            assert builder_functions.get("string") is not None
+            builder_functions[lower_type](value)
+        elif lower_type in builder_functions and 'int' in lower_type:
+            builder_functions[lower_type](int(value))
+        elif lower_type in builder_functions and 'float' in lower_type:
+            builder_functions[lower_type](float(value))
+        elif lower_type in builder_functions:
+            builder_functions[lower_type](value)
+        else:
+            self._log.error("Unknown variable type")
+            return None
+
+        builder_converting_functions = {5: builder.to_coils,
+                                        15: builder.to_coils,
+                                        6: builder.to_registers,
+                                        16: builder.to_registers}
+
+        function_code = config.function_code
+
+        if function_code in builder_converting_functions:
+            builder = builder_converting_functions[function_code]()
+            self._log.debug("Created builder %r.", builder)
+            if "Exception" in str(builder):
+                self._log.exception(builder)
+                builder = str(builder)
+            # if function_code is 5 , is using first coils value
+            if function_code == 5:
+                if isinstance(builder, list):
+                    builder = builder[0]
+            else:
+                if variable_size <= 16:
+                    if isinstance(builder, list) and len(builder) not in (8, 16, 32, 64):
+                        builder = builder[0]
+                else:
+                    if isinstance(builder, list) and len(builder) not in (2, 4):
+                        self._log.warning("There is a problem with the value builder. "
+                                          "Only the first register is written.")
+                        builder = builder[0]
+            return builder
+        self._log.warning("Unsupported function code, for the device %s in the Modbus Downlink converter",
+                          config.device_name)
+        return None

thingsboard_gateway/connectors/modbus_async/bytes_modbus_uplink_converter.py

@@ -0,0 +1,195 @@
+#     Copyright 2024. ThingsBoard
+#
+#     Licensed under the Apache License, Version 2.0 (the "License");
+#     you may not use this file except in compliance with the License.
+#     You may obtain a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#     Unless required by applicable law or agreed to in writing, software
+#     distributed under the License is distributed on an "AS IS" BASIS,
+#     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#     See the License for the specific language governing permissions and
+#     limitations under the License.
+from typing import List, Union
+
+from pymodbus.constants import Endian
+from pymodbus.exceptions import ModbusIOException
+from pymodbus.payload import BinaryPayloadDecoder
+from pymodbus.pdu import ExceptionResponse
+
+from thingsboard_gateway.connectors.modbus_async.configs.bytes_uplink_converter_config import BytesUplinkConverterConfig
+from thingsboard_gateway.connectors.modbus_async.modbus_converter import ModbusConverter
+from thingsboard_gateway.gateway.entities.converted_data import ConvertedData
+from thingsboard_gateway.gateway.statistics.decorators import CollectStatistics
+from thingsboard_gateway.gateway.statistics.statistics_service import StatisticsService
+
+
+class BytesModbusUplinkConverter(ModbusConverter):
+    def __init__(self, config: BytesUplinkConverterConfig, logger):
+        self._log = logger
+        self.__config = config
+
+    @CollectStatistics(start_stat_type='receivedBytesFromDevices',
+                       end_stat_type='convertedBytesFromDevice')
+    def convert(self, _, data: List[dict]) -> Union[ConvertedData, None]:
+        result = ConvertedData(self.__config.device_name, self.__config.device_type)
+        converted_data_append_methods = {
+            'attributes': result.add_to_attributes,
+            'telemetry': result.add_to_telemetry
+        }
+        for device_data in data:
+            StatisticsService.count_connector_message(self._log.name, 'convertersMsgProcessed')
+
+            for config_section in converted_data_append_methods:
+                for config in getattr(self.__config, config_section):
+                    encoded_data = device_data[config_section].get(config['tag'])
+
+                    if encoded_data:
+                        endian_order = Endian.Big if self.__config.byte_order.upper() == "BIG" else Endian.Little
+                        word_endian_order = Endian.Big if self.__config.word_order.upper() == "BIG" else Endian.Little
+
+                        decoded_data = self.decode_data(encoded_data, config, endian_order, word_endian_order)
+
+                        if decoded_data is not None:
+                            converted_data_append_methods[config_section]({config['tag']: decoded_data})
+
+            self._log.trace("Decoded data: %s", result)
+            StatisticsService.count_connector_message(self._log.name, 'convertersAttrProduced',
+                                                      count=result.attributes_datapoints_count)
+            StatisticsService.count_connector_message(self._log.name, 'convertersTsProduced',
+                                                      count=result.telemetry_datapoints_count)
+
+            return result
+
+    def decode_data(self, encoded_data, config, endian_order, word_endian_order):
+        decoded_data = None
+
+        if not isinstance(encoded_data, ModbusIOException) and not isinstance(encoded_data, ExceptionResponse):
+            if config['functionCode'] in (1, 2):
+                try:
+                    decoder = self.from_coils(encoded_data.bits, endian_order=endian_order,
+                                              word_endian_order=word_endian_order)
+                except TypeError:
+                    decoder = self.from_coils(encoded_data.bits, word_endian_order=word_endian_order)
+
+                decoded_data = self.decode_from_registers(decoder, config)
+            elif config['functionCode'] in (3, 4):
+                decoder = BinaryPayloadDecoder.fromRegisters(encoded_data.registers, byteorder=endian_order,
+                                                             wordorder=word_endian_order)
+                decoded_data = self.decode_from_registers(decoder, config)
+
+                if config.get('divider'):
+                    decoded_data = float(decoded_data) / float(config['divider'])
+                elif config.get('multiplier'):
+                    decoded_data = decoded_data * config['multiplier']
+        else:
+            self._log.exception("Error while decoding data: %s, with config: %s", encoded_data, config)
+            decoded_data = None
+
+        return decoded_data
+
+    @staticmethod
+    def from_coils(coils, endian_order=Endian.Little, word_endian_order=Endian.Big):
+        _is_wordorder = '_wordorder' in BinaryPayloadDecoder.fromCoils.__code__.co_varnames
+        if _is_wordorder:
+            try:
+                decoder = BinaryPayloadDecoder.fromCoils(coils, byteorder=endian_order,
+                                                         wordorder=word_endian_order)
+            except TypeError:
+                decoder = BinaryPayloadDecoder.fromCoils(coils, wordorder=word_endian_order)
+        else:
+            try:
+                decoder = BinaryPayloadDecoder.fromCoils(coils, byteorder=endian_order,
+                                                         wordorder=word_endian_order)
+            except TypeError:
+                decoder = BinaryPayloadDecoder.fromCoils(coils, wordorder=word_endian_order)
+
+        return decoder
+
+    def decode_from_registers(self, decoder, configuration):
+        objects_count = configuration.get("objectsCount",
+                                          configuration.get("registersCount", configuration.get("registerCount", 1)))
+        lower_type = configuration["type"].lower()
+
+        decoder_functions = {
+            'string': decoder.decode_string,
+            'bytes': decoder.decode_string,
+            'bit': decoder.decode_bits,
+            'bits': decoder.decode_bits,
+            '8int': decoder.decode_8bit_int,
+            '8uint': decoder.decode_8bit_uint,
+            '16int': decoder.decode_16bit_int,
+            '16uint': decoder.decode_16bit_uint,
+            '16float': decoder.decode_16bit_float,
+            '32int': decoder.decode_32bit_int,
+            '32uint': decoder.decode_32bit_uint,
+            '32float': decoder.decode_32bit_float,
+            '64int': decoder.decode_64bit_int,
+            '64uint': decoder.decode_64bit_uint,
+            '64float': decoder.decode_64bit_float,
+            }
+
+        decoded = None
+
+        if lower_type in ['bit', 'bits']:
+            decoded = decoder_functions[lower_type]()
+            decoded_lastbyte = decoder_functions[lower_type]()
+            decoded += decoded_lastbyte
+            decoded = decoded[len(decoded)-objects_count:]
+
+        elif lower_type == "string":
+            decoded = decoder_functions[lower_type](objects_count * 2)
+
+        elif lower_type == "bytes":
+            decoded = decoder_functions[lower_type](size=objects_count * 2)
+
+        elif decoder_functions.get(lower_type) is not None:
+            decoded = decoder_functions[lower_type]()
+
+        elif lower_type in ['int', 'long', 'integer']:
+            type_ = str(objects_count * 16) + "int"
+            assert decoder_functions.get(type_) is not None
+            decoded = decoder_functions[type_]()
+
+        elif lower_type in ["double", "float"]:
+            type_ = str(objects_count * 16) + "float"
+            assert decoder_functions.get(type_) is not None
+            decoded = decoder_functions[type_]()
+
+        elif lower_type == 'uint':
+            type_ = str(objects_count * 16) + "uint"
+            assert decoder_functions.get(type_) is not None
+            decoded = decoder_functions[type_]()
+
+        else:
+            self._log.error("Unknown type: %s", lower_type)
+
+        if isinstance(decoded, int):
+            result_data = decoded
+        elif isinstance(decoded, bytes) and lower_type == "string":
+            try:
+                result_data = decoded.decode('UTF-8')
+            except UnicodeDecodeError as e:
+                self._log.error("Error decoding string from bytes, will be saved as hex: %s", decoded, exc_info=e)
+                result_data = decoded.hex()
+        elif isinstance(decoded, bytes) and lower_type == "bytes":
+            result_data = decoded.hex()
+        elif isinstance(decoded, list):
+            if configuration.get('bit') is not None:
+                result_data = int(decoded[configuration['bit'] if
+                configuration['bit'] < len(decoded) else len(decoded) - 1])
+            else:
+                bitAsBoolean = configuration.get('bitTargetType', 'bool') == 'bool'
+                if objects_count == 1:
+                    result_data = bool(decoded[-1]) if bitAsBoolean else int(decoded[-1])
+                else:
+                    result_data = [bool(bit) if bitAsBoolean else int(bit) for bit in decoded]
+        elif isinstance(decoded, float):
+            result_data = decoded
+        elif decoded is not None:
+            result_data = int(decoded, 16)
+        else:
+            result_data = decoded
+
+        return result_data

thingsboard_gateway/connectors/modbus_async/configs/bytes_downlink_converter_config.py

@@ -0,0 +1,10 @@
+class BytesDownlinkConverterConfig:
+    def __init__(self, device_name, byte_order, word_order, repack, objects_count, function_code, lower_type, address):
+        self.device_name = device_name
+        self.byte_order = byte_order
+        self.word_order = word_order
+        self.repack = repack
+        self.objects_count = objects_count
+        self.function_code = function_code
+        self.lower_type = lower_type.lower()
+        self.address = address

thingsboard_gateway/connectors/modbus_async/configs/bytes_uplink_converter_config.py

@@ -0,0 +1,8 @@
+class BytesUplinkConverterConfig:
+    def __init__(self, **kwargs):
+        self.device_name = kwargs['deviceName']
+        self.device_type = kwargs.get('deviceType', 'default')
+        self.byte_order = kwargs.get('byteOrder', 'LITTLE')
+        self.word_order = kwargs.get('wordOrder', 'LITTLE')
+        self.telemetry = kwargs.get('timeseries', [])
+        self.attributes = kwargs.get('attributes', [])