Serial connector example

This tutorial walks through building a complete custom connector for the IoT Gateway — the SerialConnector — which reads data from a serial port and forwards it to ThingsBoard. The same connector ships in the gateway’s built-in extensions folder, so you can use it as a reference or starting point for your own implementation.

What we’re building:

  Serial device → bytes → SerialUplinkConverter → ConvertedData → ThingsBoard
  ThingsBoard   → RPC   → SerialDownlinkConverter → bytes → Serial device

Sample device payload:

48\r2430947595\n

48 — humidity value, terminated by \r
2430947595 — device serial number, from byte offset 4 to end of message

Step 1. Create the connector configuration

Create custom_serial.json in the same folder as your tb_gateway.json:

touch custom_serial.json

Add the following configuration:

{
  "name": "Custom serial connector",
  "logLevel": "DEBUG",
  "uplinkQueueSize": 100000,
  "devices": [
    {
      "name": "SerialDevice1",
      "type": "default",
      "port": "/dev/ttyUSB0",
      "baudrate": 9600,
      "converter": "SerialUplinkConverter",
      "downlink_converter": "SerialDownlinkConverter",
      "telemetry": [
        {
          "type": "float",
          "key": "humidity",
          "untilDelimiter": "\r"
        }
      ],
      "attributes": [
        {
          "key": "SerialNumber",
          "type": "string",
          "fromByte": 4,
          "toByte": -1
        }
      ],
      "attributeUpdates": [
        {
          "attributeOnPlatform": "attr1",
          "stringToDevice": "value = ${attr1}\n"
        }
      ],
      "serverSideRpc": [
        {
          "method": "setValue",
          "type": "int",
          "withResponse": true,
          "responseType": "string",
          "responseUntilDelimiter": "\r",
          "responseTimeoutSec": 5
        },
        {
          "method": "getValue",
          "type": "string",
          "withResponse": false
        }
      ]
    }
  ]
}

Top-level fields:

Field	Description
name	Connector name — must match the `"name"` entry in `tb_gateway.json`.
logLevel	Log verbosity: `TRACE`, `DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL`.
uplinkQueueSize	Maximum number of uplink data items to buffer before dropping.
devices	Array of device configurations.

Device fields:

Field	Description
name	Device name on the ThingsBoard platform.
type	Device profile name on the platform.
port	Serial port path.
baudrate	Serial port baud rate.
converter	Class name of the uplink converter.
downlink_converter	Class name of the downlink converter.
telemetry	Array of telemetry datapoint configurations.
attributes	Array of attribute datapoint configurations.
attributeUpdates	Array of attribute update configurations (platform → device).
serverSideRpc	Array of RPC method configurations (platform → device).

Step 2. Locate the extensions folder

Place the connector and converter files inside the extensions folder for your installation type:

Installation	Extensions folder path
Docker Compose (default volume)	`tb-gw-extensions`
Daemon	`/var/lib/thingsboard_gateway/extensions`
pip (system-wide)	`/usr/lib/python3/site-packages/thingsboard_gateway/extensions`
pip (user)	`/usr/local/lib/python3/dist-packages/thingsboard-gateway/extensions`

Create a subfolder named serial inside the extensions folder. All connector and converter files go there.

Step 3. Implement `SerialConnector`

Create extensions/serial/serial_connector.py with the following content. The connector manages SerialDevice worker threads and routes data to ThingsBoard.

from queue import Queue
from threading import Event, Thread, Lock
from typing import List, TYPE_CHECKING

import serial.tools
import serial.tools.list_ports
from thingsboard_gateway.tb_utility.tb_utility import TBUtility
from time import monotonic, sleep

try:
    import serial
except ImportError:
    print("pyserial library not found - installing...")
    TBUtility.install_package("pyserial")
    import serial

from thingsboard_gateway.connectors.connector import Connector
from thingsboard_gateway.gateway.entities.rpc_request import RPCType
from thingsboard_gateway.gateway.entities.rpc_response import RPCResponse
from thingsboard_gateway.tb_utility.tb_loader import TBModuleLoader
from thingsboard_gateway.tb_utility.tb_logger import init_logger

if TYPE_CHECKING:
    #  necessary for type checking to avoid circular import
    from thingsboard_gateway.gateway.tb_gateway_service import TBGatewayService


class SerialDevice(Thread):
    """
    Serial device class is used to represent a device that is connected to the serial port.
    It is used to read data from the device and send it to the platform.
    """
    def __init__(self, device_config, uplink_converter, stop_event: Event, logger, uplink_queue):
        super().__init__()
        self.__log = logger
        self.uplink_queue = uplink_queue
        self.daemon = True
        self.stopped = True
        self.__connector_stopped = stop_event
        self.config = device_config
        self.name = self.config.get('deviceName', self.config.get('name', 'SerialDevice'))
        self.type = self.config.get('deviceType', self.config.get('type', 'default'))
        self.uplink_converter = uplink_converter
        self.downlink_converter = None
        self.delimiter = self.config.get('delimiter', '\n')
        self.__rpc_in_progress = Event()
        self.__previous_connect = 0

        self.port = self.config.get('port', '/dev/ttyUSB0')
        self.baudrate = self.config.get('baudrate', 9600)
        self.timeout = self.config.get('timeout', 1)
        self.bytesize = self.config.get('bytesize', serial.EIGHTBITS)
        self.stopbits = self.config.get('stopbits', serial.STOPBITS_ONE)
        self.parity = self.config.get('parity', serial.PARITY_NONE)
        self.dsrdtr = self.config.get('dsrdtr', False)
        self.rtscts = self.config.get('rtscts', False)
        self.xonxoff = self.config.get('xonxoff', False)
        self.write_timeout = self.config.get('writeTimeout', None)
        self.inter_byte_timeout = self.config.get('interByteTimeout', None)
        self.exclusive = self.config.get('exclusive', None)
        self.__serial_lock = Lock()
        self.__serial = None

    def get_serial(self):
        """
        Method to get serial connection to the device.
        If connection is not established, it tries to connect to the device.
        """
        with self.__serial_lock:
            if self.__serial is None or not self.__serial.is_open:
                try:
                    self.__serial = serial.Serial(
                        port=self.port,
                        baudrate=self.baudrate,
                        timeout=self.timeout,
                        bytesize=self.bytesize,
                        stopbits=self.stopbits,
                        parity=self.parity,
                        dsrdtr=self.dsrdtr,
                        rtscts=self.rtscts,
                        xonxoff=self.xonxoff,
                        write_timeout=self.write_timeout,
                        inter_byte_timeout=self.inter_byte_timeout,
                        exclusive=self.exclusive
                    )
                    self.__log.info("Connected to device %s", self.name)
                except Exception as e:
                    self.__log.error("Failed to connect to device %s: %s", self.name, e)
                    self.__serial = None
        return self.__serial

    def run(self):
        """
        Main method to read data from the device and send it to the platform.
        """
        self.__log.info("Device %s started", self.name)
        self.stopped = False
        while not self.__connector_stopped.is_set() and not self.stopped:
            try:
                # Avoid excessive "Failed to connect to device" errors when __read_data_from_serial() calls get_serial()
                if self.get_serial() is None or not self.__serial.is_open:
                    sleep(3)
                else:
                    if not self.__rpc_in_progress.is_set():
                        data_from_device = self.__read_data_from_serial()
                        if data_from_device:
                            try:
                                converted_data = self.uplink_converter.convert(None, data_from_device)
                                self.uplink_queue.put(converted_data)
                            except Exception as e:
                                self.__log.error("Failed to convert data from device %s: %s", self.name, e)
            except Exception as e:
                self.__log.exception("Error in device %s: %s", self.name, e)
                self.stop()
        self.__log.info("Device %s stopped", self.name)

    def handle_rpc_request(self, rpc_method, params):
        """
        Method to process RPC requests from the platform.
        """
        result = {"success": True}
        processed = False
        for rpc_config in self.config.get("serverSideRpc", []):
            if rpc_method == rpc_config.get("method"):
                processed = True
                self.__rpc_in_progress.set()
                try:
                    if self.downlink_converter is not None:
                        converted_data = self.downlink_converter.convert(rpc_config, params)
                        if converted_data:
                            with_response = rpc_config.get("withResponse", False)
                            response_timeout = rpc_config.get("responseTimeoutSec", 5)
                            response = self.write(converted_data,
                                                  with_response=with_response,
                                                  response_timeout=response_timeout)
                            if with_response:
                                response_uplink_config = {}
                                if rpc_config.get("responseType"):
                                    response_uplink_config["type"] = rpc_config.get("responseType")
                                if rpc_config.get("responseFromByte"):
                                    response_uplink_config["fromByte"] = rpc_config.get("responseFromByte")
                                if rpc_config.get("responseToByte"):
                                    response_uplink_config["toByte"] = rpc_config.get("responseToByte")
                                if rpc_config.get("responseUntilDelimiter"):
                                    response_uplink_config["delimiter"] = rpc_config.get("responseUntilDelimiter")
                                if response_uplink_config and response:
                                    result = self.uplink_converter.convert(response_uplink_config, response)
                                else:
                                    result = {"error": "Cannot convert response with config: %r and response: %r" % (
                                        response_uplink_config, response), "success": False}
                        else:
                            result = {"error": "No data to send", "success": False}
                    else:
                        result = {"error": "Downlink converter not defined", "success": False}
                except Exception as e:
                    self.__log.error("Failed to process RPC with method: %r, params: %r, config: %r - Error: %s",
                                     rpc_method, params, rpc_config, e)
                    result = {"error": str(e), "success": False}
                finally:
                    self.__rpc_in_progress.clear()
        if not processed:
            result = {"error": "Method not found", "success": False}
        return result

    def write(self, data, with_response=False, response_timeout=5):
        """
        Method to write data to the device.
        If with_response is set to True, it waits for the response from the device.
        """
        try:
            serial_conn = self.get_serial()
            if serial_conn:
                with self.__serial_lock:
                    serial_conn.write(data)
                    self.__log.debug("Written to device %s: %s", self.name, data)
                if with_response:
                    return self.__read_data_from_serial(response_timeout)
        except Exception as e:
            self.__log.exception("Failed to write to device %s: %s", self.name, e)
        return None

    def __read_data_from_serial(self, timeout=1):
        """
        Method to read data from the device.
        It reads data until the delimiter is found.
        """
        data_from_device = b''
        try:
            serial_conn = self.get_serial()
            if serial_conn and serial_conn.is_open:
                # Ensure serial_conn is not None before accessing its timeout attribute
                previous_timeout = serial_conn.timeout
                try:
                    while not data_from_device.endswith(self.delimiter.encode('utf-8')):
                        serial_conn.timeout = timeout
                        chunk = serial_conn.read(1)
                        if chunk:
                            data_from_device += chunk
                        if self.__connector_stopped.is_set() or not chunk or self.stopped:
                            break
                finally:
                    serial_conn.timeout = previous_timeout
        except Exception as e:
            self.__log.exception("Failed to read from device %s: %s", self.name, e)
        return data_from_device

    def stop(self):
        self.stopped = True
        with self.__serial_lock:
            if self.__serial:
                self.__serial.close()
                self.__serial = None

    def is_connected_reconnect_if_needed(self):
        """
        Method to check if the device is connected.
        If the device is not connected, it tries to reconnect.
        """
        if self.__serial is None or not self.__serial.is_open:
            if monotonic() - self.__previous_connect > 1:
                self.__previous_connect = monotonic()
                self.__log.info("Reconnecting to device %s", self.name)
                self.get_serial()
                # Only return True if the serial connection is successful
                return self.__serial and self.__serial.is_open
        else:
            return True


class SerialConnector(Thread, Connector):
    """
    Serial connector class is used to represent a serial connector.
    It is used to manage devices connected to the serial ports.
    """
    def __init__(self, gateway: 'TBGatewayService', config, connector_type):
        super().__init__()
        self._connector_type = connector_type  # required to have for get connector type method
        self.__config = config  # required to have for get config method
        self.__id = self.__config["id"]  # required to have for get id method
        self.__gateway = gateway  # required to have for send data to storage method or to use other gateway methods
        self.name = self.__config["name"]  # required to have for get name method
        self.__connected = False  # required to have for is connected method
        self.__uplink_queue = Queue(self.__config.get('uplinkQueueSize', 100000))
        self._log = init_logger(self.__gateway, self.name, level=self.__config.get('logLevel'),
                                enable_remote_logging=self.__config.get('enableRemoteLogging', False),
                                is_connector_logger=True)
        self._converter_log = init_logger(self.__gateway, self.name, level=self.__config.get('logLevel'),
                                          enable_remote_logging=self.__config.get('enableRemoteLogging', False),
                                          is_converter_logger=True)
        self._log.info("Starting %s connector", self.get_name())
        self.daemon = True
        self.stopped = Event()
        self.stopped.set()
        self.__devices: List[SerialDevice] = []
        self._log.info('Connector %s initialization success.', self.get_name())

    def __start_devices(self):
        failed_to_connect_devices = len(self.__devices)
        for device in self.__devices:
            try:
                device.start()
                failed_to_connect_devices -= 1
            except Exception as e:
                self._log.exception("Failed to start device %s, error: %s", device.name, e)
        self.__connected = failed_to_connect_devices == 0

    def open(self):
        """
        Service method to start the connector.
        """
        self.stopped.clear()
        self.start()

    def get_name(self):
        return self.name

    def get_type(self):
        return self._connector_type

    def is_connected(self):
        return self.__connected

    def is_stopped(self):
        return self.stopped.is_set()

    def get_config(self):
        return self.__config

    def get_id(self):
        return self.__id

    def __load_devices(self):
        """
        Method to create devices objects using configuration file and create converters for them.
        """
        devices_config = self.__config.get('devices')
        try:
            if devices_config is not None:
                for device_config in devices_config:
                    device = None
                    uplink_converter_class_name = device_config.get('converter', device_config.get('uplink_converter'))
                    if uplink_converter_class_name is not None:
                        converter_class = TBModuleLoader.import_module(self._connector_type,
                                                                       uplink_converter_class_name)
                        uplink_converter = converter_class(device_config, self._log)
                        device = SerialDevice(device_config, uplink_converter, self.stopped,
                                              self._log, self.__uplink_queue)
                    else:
                        self._log.error('Converter configuration for the connector %s -- \
                            not found, please check your configuration file.', self.get_name())
                    if device_config.get('downlink_converter') is not None:
                        downlink_converter_class = TBModuleLoader.import_module(self._connector_type,
                                                                                device_config.get('downlink_converter'))
                        if device is not None:
                            device.downlink_converter = downlink_converter_class(device_config, self._converter_log)
                    if device is not None:
                        self.__devices.append(device)
            else:
                self._log.error('Section "devices" in the configuration not found. \
                    Connector %s has been stopped.', self.get_name())
                self.close()
        except Exception as e:
            self._log.error('Failed to load devices, error: %s', e)

    def run(self):
        """
        Main method to manage devices connected to the serial ports and process data from them.
        """
        try:
            self.__load_devices()
            self.__start_devices()
            self._log.info("Devices in configuration file found: %s ",
                           '\n'.join(device.name for device in self.__devices))
            while not self.stopped.is_set():
                try:
                    connected_devices = len(self.__devices)
                    for device in self.__devices:
                        if not device.stopped and not device.is_connected_reconnect_if_needed():
                            connected_devices -= 1
                            self._log.error("Device %s is not connected", device.name)
                            device.stop()
                            device.join()
                            device = SerialDevice(device.config, device.uplink_converter, self.stopped,
                                                  self._log, self.__uplink_queue)
                            device.start()
                    self.__connected = connected_devices == len(self.__devices)
                    if not self.__uplink_queue.empty():
                        data = self.__uplink_queue.get()
                        self.__gateway.send_to_storage(self.name, self.__id, data)
                    else:
                        sleep(0.05)
                except Exception as e:
                    self._log.error("Failed to process data from device %s, error: %s", self.name, e)
        except Exception as e:
            self._log.error("Failed to process data from device %s, error: %s", self.name, e)

    def close(self):
        """
        Service method to stop the connector and all devices connected to it.
        """
        self.stopped.set()
        for device in self.__devices:
            self.__gateway.del_device(device.name)
            device.stop()
        self._log.stop()

    def on_attributes_update(self, content):
        """
        Callback method to process attribute updates from the platform.
        """
        self._log.debug("Received attribute update: %s", content)
        device_name = content.get("device")
        if device_name is not None:
            for device in self.__devices:
                if device_name == device.name:
                    request_config = device.config.get("attributeUpdates")
                    if request_config is not None:
                        attribute_config_found = False
                        for attribute_config in request_config:
                            attribute = attribute_config.get("attributeOnPlatform")
                            if attribute is not None and attribute in content["data"]:
                                attribute_config_found = True
                                try:
                                    value = content["data"][attribute]
                                    str_to_send = str(attribute_config["stringToDevice"]
                                                      .replace("${" + attribute + "}", str(value))
                                                      .replace("${deviceName}", device_name)
                                                      .replace("${deviceType}", device.type)
                                                      ).encode("UTF-8")
                                    device.write(str_to_send)
                                except Exception as e:
                                    self._log.error("Failed to send attribute update to device %s: %s",
                                                    device_name, e)
                        if not attribute_config_found:
                            self._log.error("Attribute update configuration for key %s for device %s not found",
                                            list(content['data'].keys())[0], device_name)
                    else:
                        self._log.error("Attribute update configuration for device %s not found", device_name)
        else:
            self._log.error("Device name is not provided in the attribute update request: %s", content)

    def server_side_rpc_handler(self, rpc_request) -> RPCResponse:
        """
        Callback method to process RPC requests from the platform.
        Returns RPCResponse, the gateway sends it to the platform as the RPC reply.
        """
        self._log.debug("Received RPC request: %s", rpc_request)
        response = RPCResponse(rpc_request.id, device=rpc_request.device_name)

        if rpc_request.rpc_type == RPCType.CONNECTOR:
            response.set_error_msg("RPC to connector is not supported by Serial connector")
            return response

        if rpc_request.rpc_type == RPCType.RESERVED:
            response.set_error_msg("Reserved RPC methods (get/set) are not supported by Serial connector")
            return response

        device = next((device for device in self.__devices if device.name == rpc_request.device_name), None)
        if device is None:
            response.set_error_msg(f"Device {rpc_request.device_name} not found")
            return response

        result = device.handle_rpc_request(rpc_request.method_name, rpc_request.params)
        if "error" in result:
            self._log.error("Failed to process RPC request for device %s, error: %s",
                            rpc_request.device_name, result["error"])
            response.set_error_msg(result["error"])
        else:
            response.set_message(result.get("result", result))

        return response

Step 4. Implement `SerialUplinkConverter`

Create extensions/serial/uplink_serial_converter.py. The uplink converter parses raw bytes from the device and produces a ConvertedData object.

from typing import Any, Tuple
from simplejson import loads

from thingsboard_gateway.connectors.converter import Converter
from thingsboard_gateway.gateway.constants import REPORT_STRATEGY_PARAMETER, TELEMETRY_PARAMETER, TIMESERIES_PARAMETER
from thingsboard_gateway.gateway.entities.converted_data import ConvertedData
from thingsboard_gateway.gateway.entities.datapoint_key import DatapointKey
from thingsboard_gateway.gateway.entities.report_strategy_config import ReportStrategyConfig
from thingsboard_gateway.gateway.entities.telemetry_entry import TelemetryEntry
from thingsboard_gateway.tb_utility.tb_utility import TBUtility


class SerialUplinkConverter(Converter):
    """
    Uplink converter is used to convert incoming data to the format that platform expects.
    We create one uplink converter per configured device,
    because the config passed to __init__ is device-specific,
    so this converter is used to convert incoming data from only one device.
    If your connector can handle multiple devices, you can create one converter for all devices.
    """

    def __init__(self, config, logger):
        self._log = logger
        self.__config = config
        self.__device_report_strategy = None
        self.__device_name = self.__config.get('deviceName', self.__config.get('name', 'SerialDevice'))
        self.__device_type = self.__config.get('deviceType', self.__config.get('type', 'default'))
        try:
            self.__device_report_strategy = ReportStrategyConfig(self.__config.get(REPORT_STRATEGY_PARAMETER))
        except ValueError as e:
            self._log.trace("Report strategy config is not specified for device %s: %s", self.__device_name, e)

    def convert(self, config, data: bytes):
        """Converts incoming data to the format that platform expects. Config is specified only for RPC responses."""
        self._log.debug("Data to convert: %s", data)
        if config is not None:
            converted_data = {"result": self.__convert_value_to_type(data, config)}
            return converted_data
        else:
            converted_data = ConvertedData(self.__device_name, self.__device_type)
            for datapoint_config in self.__config.get(TIMESERIES_PARAMETER, self.__config.get(TELEMETRY_PARAMETER, [])):
                try:
                    telemetry_entry = self.__convert_telemetry_datapoint(data, datapoint_config)
                    if telemetry_entry:
                        converted_data.add_to_telemetry(telemetry_entry)
                except Exception as e:
                    self._log.error("Error converting telemetry datapoint: %s", e)
            for datapoint_config in self.__config.get('attributes', []):
                try:
                    attribute_data = self.__convert_attributes_datapoint(data, datapoint_config)
                    if attribute_data:
                        converted_data.add_to_attributes(*attribute_data)
                except Exception as e:
                    self._log.error("Error converting attribute datapoint: %s", e)
            self._log.debug("Converted data: %s", converted_data)
        return converted_data

    def __convert_telemetry_datapoint(self, data, dp_config) -> TelemetryEntry:
        key = dp_config.get('key')
        datapoint_key = self.__convert_datapoint_key(key, dp_config, self.__device_report_strategy, self._log)
        value = self.__convert_value_to_type(data, dp_config)
        if not datapoint_key or not value:
            self._log.trace("Datapoint %s - not found in incoming data: %s", key, data.hex())
            return None
        return TelemetryEntry({datapoint_key: value})

    def __convert_attributes_datapoint(self, data, dp_config) -> Tuple[DatapointKey, Any]:
        key = dp_config.get('key')
        datapoint_key = self.__convert_datapoint_key(key, dp_config, self.__device_report_strategy, self._log)
        value = self.__convert_value_to_type(data, dp_config)
        if not datapoint_key or not value:
            self._log.trace("Datapoint %s - not found in incoming data: %s", key, data.hex())
            return None
        return (datapoint_key, value)

    @staticmethod
    def __convert_value_to_type(data, dp_config):
        type_ = dp_config.get('type')
        data_for_conversion = data
        if dp_config.get("untilDelimiter") or dp_config.get("fromDelimiter"):
            fromDelimiter = dp_config.get("fromDelimiter")
            untilDelimiter = dp_config.get("untilDelimiter")
            fromDelimiterPosition = data_for_conversion.find(
                fromDelimiter.encode('UTF-8')) if fromDelimiter else 0
            untilDelimiterPosition = data_for_conversion.find(
                untilDelimiter.encode('UTF-8')) if untilDelimiter else -1
            if fromDelimiterPosition != -1 \
                    and untilDelimiterPosition != -1 \
                    and fromDelimiterPosition < untilDelimiterPosition:
                data_for_conversion = data_for_conversion[fromDelimiterPosition:untilDelimiterPosition]
            elif fromDelimiterPosition != -1 and fromDelimiterPosition < len(data_for_conversion):
                data_for_conversion = data_for_conversion[fromDelimiterPosition:]
            elif untilDelimiterPosition != -1 and untilDelimiterPosition < len(data_for_conversion):
                data_for_conversion = data_for_conversion[:untilDelimiterPosition]
        elif dp_config.get("fromByte") or dp_config.get("toByte"):
            if dp_config.get("fromByte") and dp_config.get("toByte") \
                    and dp_config["fromByte"] < dp_config["toByte"] \
                    and len(data_for_conversion) > dp_config["toByte"]:
                data_for_conversion = data_for_conversion[dp_config["toByte"]:dp_config["fromByte"]]
            else:
                if dp_config.get("fromByte") and len(data_for_conversion) > dp_config.get("fromByte", 0):
                    data_for_conversion = data_for_conversion[dp_config["fromByte"]:]
                if dp_config.get("toByte") and \
                        (len(data_for_conversion) > dp_config.get("toByte", 0) or dp_config["toByte"] == -1):
                    data_for_conversion = data_for_conversion[:dp_config["toByte"]]

        if type_ == 'string':
            value = data_for_conversion.decode('UTF-8').strip()
        elif type_ == 'json':
            value = loads(data_for_conversion.decode('UTF-8'))
        elif type_ == 'int':
            value = int(data_for_conversion)
        elif type_ == 'float' or type_ == 'double':
            value = float(data_for_conversion)
        elif type_ == 'bool':
            try:
                value = bool(int(data_for_conversion))
            except ValueError:
                return data_for_conversion.decode('UTF-8').strip().lower() == 'true'
        else:
            value = data_for_conversion.hex()
        return value

    @staticmethod
    def __convert_datapoint_key(key, dp_config, device_report_strategy, logger):
        return TBUtility.convert_key_to_datapoint_key(key, device_report_strategy, dp_config, logger)

After processing 48\r2430947595\n, the converter produces:

Device name: "SerialDevice1"
Device type: "default"
Telemetry:   [{"humidity": 48.0}]
Attributes:  {"SerialNumber": "2430947595"}

Step 5. Implement `SerialDownlinkConverter`

Create extensions/serial/downlink_serial_converter.py. The downlink converter turns ThingsBoard RPC payloads into raw bytes to send to the device.

from math import ceil
from struct import pack, unpack

from thingsboard_gateway.connectors.converter import Converter


class SerialDownlinkConverter(Converter):
    """
    Downlink converter is used to convert data that can be sent to device.
    We create one downlink converter per configured device,
    because the config passed to __init__ is device-specific,
    so this converter is used to convert data that can be sent to only one device.
    If your connector can handle multiple devices, you can create one converter for all devices.
    """

    def __init__(self, config, logger):
        self._log = logger
        self.__config = config

    def convert(self, config, data) -> bytes:
        """Method to convert data that can be sent to the serial port."""
        self._log.debug("Data to convert: %s", data)
        byteorder = self.__config.get('byteorder', 'big').lower()
        if data is None:
            return None
        type_ = config.get("type")
        if type_ == "int":
            length = ceil(data.bit_length() / 8)
            return data.to_bytes(length, byteorder=byteorder)
        elif type_ == "float" or type_ == "double":
            fmt_single_precision = ('>' if byteorder == 'big' else '<') + 'f'
            single_precision_bytes = pack(fmt_single_precision, data)
            if unpack(fmt_single_precision, single_precision_bytes)[0] == data:
                return single_precision_bytes
            fmt_double_precision = ('>' if byteorder == 'big' else '<') + 'd'
            return pack(fmt_double_precision, data)
        return data.encode("UTF-8")

Step 6. Register the connector in `tb_gateway.json`

Add the following entry to the "connectors" array in tb_gateway.json:

{
  "name": "Serial Connector",
  "type": "serial",
  "configuration": "custom_serial.json",
  "class": "SerialConnector"
}

Field	Description
name	Connector name — must match the `"name"` in `custom_serial.json`.
type	Extensions subfolder name (`serial`).
configuration	Path to the connector config file, relative to the gateway config folder.
class	Connector class name inside the connector file.

sudo systemctl restart thingsboard-gateway

sudo python3 -c 'from thingsboard_gateway.gateway.tb_gateway_service import TBGatewayService; TBGatewayService("./tb_gateway.json")'

Default log locations:

Installation	Log folder
Docker Compose	`tb-gw-logs` volume
Daemon	`/var/log/thingsboard-gateway/`
Python module (pip)	`./logs/`

Step 8. Verify the result

Connect the serial device, then open Devices in the ThingsBoard UI. You should see a device named SerialDevice1. Open it and go to the Latest telemetry tab — the humidity key should appear with the value parsed from the serial stream.

Connector method reference

The SerialConnector class implements all required methods from the Connector interface. Key methods used in this example:

Method	Role in this connector
`__init__`	Initialises the uplink queue, logger, and device list from config.
`open`	Starts the connector thread.
`run`	Main loop: loads devices, starts threads, drains the uplink queue, handles reconnects.
`close`	Stops all device threads and the logger.
`on_attributes_update`	Formats the attribute value as a UTF-8 string and writes it to the device’s serial port.
`server_side_rpc_handler`	Delegates RPC handling to `SerialDevice.handle_rpc_request` and returns an `RPCResponse`, which the gateway sends to ThingsBoard.

Converter method reference

The SerialUplinkConverter and SerialDownlinkConverter classes implement the Converter interface.

Class	`convert(config, data)`
`SerialUplinkConverter`	`config=None` for telemetry; config dict for RPC responses. `data` is the raw bytes from the port. Returns `ConvertedData`.
`SerialDownlinkConverter`	`config` is the RPC config section. `data` is the value from ThingsBoard. Returns `bytes`.

Was this helpful?