Python Gateway SDK

The ThingsBoard Python Gateway SDK (tb-mqtt-client) lets Python applications connect to ThingsBoard as a gateway — managing multiple constrained devices and transmitting data on their behalf.

TBGatewayMqttClient extends TBDeviceMqttClient, so it has access to all Device SDK APIs plus gateway-specific methods for device lifecycle, telemetry, attributes, and RPC.

Installation

pip3 install tb-mqtt-client

Send telemetry and attributes

Connect a device through the gateway, send telemetry and attributes, then disconnect:

from time import time
from tb_gateway_mqtt import TBGatewayMqttClient

gateway = TBGatewayMqttClient("127.0.0.1", username="TEST_GATEWAY_TOKEN")
gateway.connect()
gateway.gw_connect_device("Test Device A1")

gateway.gw_send_telemetry("Test Device A1", {"ts": int(round(time() * 1000)), "values": {"temperature": 42.2}})
gateway.gw_send_attributes("Test Device A1", {"firmwareVersion": "2.3.1"})

gateway.gw_disconnect_device("Test Device A1")
gateway.disconnect()

Request attributes from server

Retrieve shared attributes for a connected device:

from time import sleep
from tb_gateway_mqtt import TBGatewayMqttClient

def callback(result, exception):
  if exception is not None:
      print("Exception: " + str(exception))
  else:
      print(result)

gateway = TBGatewayMqttClient("127.0.0.1", username="TEST_GATEWAY_TOKEN")
gateway.connect()
gateway.gw_request_shared_attributes("Test Device A1", ["temperature"], callback)

while True:
  sleep(1)

Respond to server RPC calls

Handle RPC requests for devices connected through the gateway:

import time
from tb_gateway_mqtt import TBGatewayMqttClient

try:
  import psutil
except ImportError:
  print("Please install psutil using 'pip install psutil' command")
  exit(1)

def rpc_request_response(request_id, request_body):
  print(request_body)
  method = request_body["data"]["method"]
  device = request_body["device"]
  req_id = request_body["data"]["id"]
  if method == 'getCPULoad':
      gateway.gw_send_rpc_reply(device, req_id, {"CPU load": psutil.cpu_percent()})
  elif method == 'getMemoryLoad':
      gateway.gw_send_rpc_reply(device, req_id, {"Memory": psutil.virtual_memory().percent})
  else:
      print('Unknown method: ' + method)

gateway = TBGatewayMqttClient("127.0.0.1", username="TEST_GATEWAY_TOKEN")
gateway.connect()
# Register RPC handler before connecting devices
gateway.gw_set_server_side_rpc_request_handler(rpc_request_response)
gateway.gw_connect_device("Test Device A1")

while True:
  time.sleep(1)

More examples

Additional examples are available in the thingsboard-python-client-sdk GitHub repository.

Was this helpful?