منو
منو

کیت توسعه‌دهنده کلاینت پایتون

این کیت توسعه‌دهنده از موارد زیر پشتیبانی می‌کند:

  • اتصال رمزگذاری نشده و رمزگذاری شده (TLS نسخه ۱.۲)
  • کیفیت سرویس (QoS) سطح ۰ و ۱
  • اتصال مجدد خودکار
  • تمامی APIهای MQTT برای دستگاه‌ها
  • تمامی APIهای MQTT برای درگاه‌ها

این کیت توسعه‌دهنده مبتنی بر کتابخانه Paho MQTT است.

نصب

برای نصب از طریق pip:

				
					pip3 install tb-mqtt-client

شروع به کار

راه‌اندازی اولیه کلاینت و ارسال داده‌های تله‌متری.

				
					from tb_device_mqtt import TBDeviceMqttClient, TBPublishInfo


telemetry = {"temperature": 41.9, "enabled": False, "currentFirmwareVersion": "v1.2.2"}
client = TBDeviceMqttClient("127.0.0.1", username="A1_TEST_TOKEN")
# Connect to ThingsBoard
client.connect()
# Sending telemetry without checking the delivery status
client.send_telemetry(telemetry) 
# Sending telemetry and checking the delivery status (QoS = 1 by default)
result = client.send_telemetry(telemetry)
# get is a blocking call that awaits delivery status  
success = result.get() == TBPublishInfo.TB_ERR_SUCCESS
# Disconnect from ThingsBoard
client.disconnect()

اتصال با استفاده از TLS

اتصال TLS به localhost. برای اطلاعات بیشتر درباره پیکربندی کلاینت و ThingsConnect با استفاده از MQTT روی SSL، به مستندات MQTT روی SSL مراجعه کنید. برای اتصال به ThingsConnect با MQTT از طریق SSL، ابتدا باید یک گواهی‌نامه تولید کنید و کدی مشابه نمونه زیر داشته باشید:

				
					from socket import gethostname
from tb_device_mqtt import TBDeviceMqttClient


client = TBDeviceMqttClient(gethostname())
client.connect(tls=True,
               ca_certs="mqttserver.pub.pem",
               cert_file="mqttclient.nopass.pem")
client.disconnect()

استفاده از APIهای دستگاه

TBDeviceMqttClient دسترسی به APIهای MQTT دستگاه‌های پلتفرم ThingsConnect را فراهم می‌کند. این قابلیت به شما اجازه می‌دهد تا داده‌های تله‌متری و به‌روزرسانی‌های ویژگی‌ها را ارسال کرده، تغییرات ویژگی‌ها را مشترک شوید، و همچنین دستورات RPC را ارسال و دریافت کنید.

اشتراک در ویژگی‌ها

اگر نیاز دارید که به‌روزرسانی‌های ویژگی‌های مشترک را دریافت کنید، می‌توانید از کدی مشابه نمونه زیر استفاده کنید:

				
					from time import sleep
from tb_device_mqtt import TBDeviceMqttClient


def callback(result):
    print(result)

client = TBDeviceMqttClient("127.0.0.1", username="A1_TEST_TOKEN")
client.connect()
client.subscribe_to_attribute("uploadFrequency", callback)
client.subscribe_to_all_attributes(callback)
while True:
    sleep(1)

ارسال بسته تله‌متری

برای ارسال داده به ThingsConnect، می‌توانید از کدی مشابه نمونه زیر استفاده کنید:

				
					from time import time
from tb_device_mqtt import TBDeviceMqttClient, TBPublishInfo


telemetry_with_ts = {"ts": int(round(time() * 1000)), "values": {"temperature": 42.1, "humidity": 70}}
client = TBDeviceMqttClient("127.0.0.1", username="A1_TEST_TOKEN")
# we set maximum amount of messages sent to send them at the same time. it may stress memory but increases performance
client.max_inflight_messages_set(100)
client.connect()
results = []
result = True
for i in range(0, 100):
    results.append(client.send_telemetry(telemetry_with_ts))
for tmp_result in results:
    result &= tmp_result.get() == TBPublishInfo.TB_ERR_SUCCESS
print("Result", str(result))
client.disconnect()

درخواست ویژگی‌ها از سرور

برای درخواست مقادیر ویژگی‌های مشترک از ThingsConnect، می‌توانید از مثال زیر استفاده کنید:

				
					from time import sleep
from tb_device_mqtt import TBDeviceMqttClient


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

client = TBDeviceMqttClient("127.0.0.1", username="A1_TEST_TOKEN")
client.connect()
client.request_attributes(["configuration","targetFirmwareVersion"], callback=on_attributes_change)


while True:
    sleep(1)

پاسخ به فراخوانی RPC سرور

اگر می‌خواهید به یک درخواست RPC پاسخ دهید، می‌توانید از منطق موجود در کد زیر استفاده کنید. مثال زیر به نمونه محلی ThingsConnect متصل شده و منتظر درخواست RPC می‌ماند.
زمانی که درخواست RPC دریافت شد، کلاینت پاسخی حاوی داده‌های دستگاه با نام "Test Device A1" را به ThingsConnect ارسال می‌کند.

				
					from psutil import cpu_percent, virtual_memory
from time import sleep
from tb_device_mqtt import TBDeviceMqttClient


# dependently of request method we send different data back
def on_server_side_rpc_request(client, request_id, request_body):
    print(request_id, request_body)
    if request_body["method"] == "getCPULoad":
        client.send_rpc_reply(request_id, {"CPU percent": cpu_percent()})
    elif request_body["method"] == "getMemoryUsage":
        client.send_rpc_reply(request_id, {"Memory": virtual_memory().percent})

client = TBDeviceMqttClient("127.0.0.1", username="A1_TEST_TOKEN")
client.set_server_side_rpc_request_handler(on_server_side_rpc_request)
client.connect()


while True:
    sleep(1)

استفاده از APIهای درگاه

TBGatewayMqttClient که گسترشی از TBDeviceMqttClient است، به تمام APIهای آن به عنوان یک دستگاه معمولی دسترسی دارد. علاوه بر این، درگاه قادر است چندین دستگاه متصل به خود را نیز نمایندگی کند. برای مثال، می‌تواند تله‌متری یا ویژگی‌ها را از جانب دستگاه‌های دیگر و محدود ارسال کند. برای اطلاعات بیشتر درباره درگاه، به اینجا مراجعه کنید.

ارسال تله‌متری و ویژگی‌ها

برای ارسال داده به ThingsConnect از طرف دستگاهی با نام "Test Device A1"، می‌توانید از کدی مشابه نمونه زیر استفاده کنید:

				
					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()

درخواست ویژگی‌ها از سرور

برای درخواست مقادیر ویژگی‌های مشترک از ThingsConnect برای دستگاهی با نام "Test Device A1"، می‌توانید از مثال زیر استفاده کنید:

				
					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)

پاسخ به فراخوانی RPC سرور

اگر می‌خواهید به یک درخواست RPC پاسخ دهید، می‌توانید از منطق موجود در کد زیر استفاده کنید.
مثال زیر به نمونه محلی ThingsConnect متصل شده و منتظر دریافت درخواست RPC می‌ماند.
هنگامی که درخواست RPC دریافت شد، کلاینت پاسخی حاوی داده‌های دستگاهی با نام "Test Device A1" را به ThingsConnect ارسال می‌کند.

				
					from time import sleep
from psutil import cpu_percent, virtual_memory
from tb_gateway_mqtt import TBGatewayMqttClient


def rpc_request_response(client, request_id, request_body):
    # request body contains id, method and other parameters
    print(request_body)
    method = request_body["data"]["method"]
    device = request_body["device"]
    req_id = request_body["data"]["id"]
    # dependently of request method we send different data back
    if method == 'getCPULoad':
        gateway.gw_send_rpc_reply(device, req_id, {"CPU load": cpu_percent()})
    elif method == 'getMemoryLoad':
        gateway.gw_send_rpc_reply(device, req_id, {"Memory": virtual_memory().percent})
    else:
        print('Unknown method: ' + method)

gateway = TBGatewayMqttClient("127.0.0.1", username="TEST_GATEWAY_TOKEN")
gateway.connect()
# now rpc_request_response will process rpc requests from servers
gateway.gw_set_server_side_rpc_request_handler(rpc_request_response)
# without device connection it is impossible to get any messages
gateway.gw_connect_device("Test Device A1")

while True:
    sleep(1)

سایر مثال‌ها

مثال‌های بیشتری برای هر دو دستگاه و درگاه در پوشه‌های مربوطه موجود است.

شرایط استفاده از خدمات

aparat-icon-svg Telegram

عناوین هر بخش

شرایط استفاده از خدمات

aparat-icon-svg Telegram
بستن