diagram.json

# ThingsBoard ID: jangoh1974@naver.com
# ThingsBoard PW: su8155103283

from umqtt.simple import MQTTClient
from machine import Pin
import time, network, ujson
from tb_device_mqtt import TBDeviceMqttClient

WIFI_SSID = 'Wokwi-GUEST'
WIFI_PASS = ''

print("Connecting to WiFi network '{}'".
format(WIFI_SSID))
wifi = network.WLAN(network.STA_IF)
wifi.active(True)
wifi.connect(WIFI_SSID, WIFI_PASS)
while not wifi.isconnected():
  time.sleep(1)
  print('WiFi connect retry ...')
print('WiFi IP:', wifi.ifconfig()[0])

MQTT_CLIENT_ID = "7f69dbf0-9196-11ee-8a25-070bfaaff76d"
MQTT_BROKER = "mqtt.thingsboard.cloud"
MQTT_USER = "XuDl5xasFW81edjxaarp"
MQTT_PASSWORD = ""
MQTT_PORT = "1883"
MQTT_TOPIC = "v1/devices/me/rpc/request/+"

print("Connecting to MQTT server... ", end="")
client = MQTTClient(MQTT_CLIENT_ID, MQTT_BROKER,
MQTT_PORT, user=MQTT_USER, password=MQTT_PASSWORD)
client.connect()
print("Connected!")

led_r = Pin(26, Pin.OUT)
led_r.value(0)
led_g = Pin(25, Pin.OUT)
led_g.value(0)
led_b = Pin(33, Pin.OUT)
led_b.value(0)

flag = 0

def led():
  global flag 
  if flag == 0: 
    led_r.value(0)
    led_g.value(1)
    led_b.value(0)
  elif flag == 1:
    led_r.value(0)
    led_g.value(1)
    led_b.value(1)
  else:
    led_r.value(0)
    led_g.value(0)
    led_b.value(0)

def sub_cb(topic, msg):
    print((topic, msg))
    jsLoads = ujson.loads(msg)
    print(type(jsLoads), jsLoads)
    if jsLoads["method"] == "watering":
        print("Start Watering Now, Current Temp is: ", jsLoads["params"]['temp'])

client.set_callback(sub_cb)

client.subscribe(MQTT_TOPIC)

while True:
    client.check_msg()


wateringStatus = 0

def main():
  global wateringStatus
  print(wateringStatus)
  if jsLoads("method") == "wateringON":
    flag = 0
    wateringStatus = 1
    print(wateringStatus)
  elif jsLoads("method") == "wateringOFF":
    wateringStatus = 0
    flag = 1
    print(wateringStatus)

if __name__ == '__main__':
  main()


main.py

#      Copyright 2023. 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.
#

import time
from math import ceil

from sdk_utils import verify_checksum
from umqtt import MQTTClient, MQTTException
import ujson
import ubinascii
import machine


FW_TITLE_ATTR = "fw_title"
FW_VERSION_ATTR = "fw_version"
FW_CHECKSUM_ATTR = "fw_checksum"
FW_CHECKSUM_ALG_ATTR = "fw_checksum_algorithm"
FW_SIZE_ATTR = "fw_size"
FW_STATE_ATTR = "fw_state"

REQUIRED_SHARED_KEYS = "{0},{1},{2},{3},{4}".format(FW_CHECKSUM_ATTR, FW_CHECKSUM_ALG_ATTR, FW_SIZE_ATTR, FW_TITLE_ATTR,
                                                    FW_VERSION_ATTR)

RPC_REQUEST_TOPIC = 'v1/devices/me/rpc/request/'
RPC_RESPONSE_TOPIC = 'v1/devices/me/rpc/response/'
ATTRIBUTES_TOPIC = 'v1/devices/me/attributes'
ATTRIBUTE_REQUEST_TOPIC = 'v1/devices/me/attributes/request/'
ATTRIBUTE_TOPIC_RESPONSE = 'v1/devices/me/attributes/response/'


class TBDeviceMqttClient:
    def __init__(self, host, port=1883, access_token=None, quality_of_service=None, client_id=None, chunk_size=0):
        self._host = host
        self._port = port
        self.quality_of_service = quality_of_service if quality_of_service is not None else 1
        self.current_firmware_info = {
            "current_" + FW_TITLE_ATTR: "Initial",
            "current_" + FW_VERSION_ATTR: "v0"
        }
        self._attr_request_dict = {}
        self.__device_client_rpc_dict = {}
        self.__device_sub_dict = {}
        self.__device_on_server_side_rpc_response = None
        self.__attr_request_number = 0
        self.__device_client_rpc_number = 0
        self.__device_max_sub_id = 0
        self.__firmware_request_id = 0
        self.__request_id = 0
        self.__chunk_size = chunk_size
        self.connected = False

        if not access_token:
            print("token is not set, connection without tls wont be established")
        self._access_token = access_token

        if not client_id:
            client_id = ubinascii.hexlify(machine.unique_id())
        self._client_id = client_id
        self._client = MQTTClient(self._client_id, self._host, self._port, self._access_token, 'pswd', keepalive=120)

    def connect(self):
        try:
            response = self._client.connect()
            self._client.set_callback(self._callback)
            self._client.subscribe(ATTRIBUTES_TOPIC, qos=self.quality_of_service)
            self._client.subscribe(ATTRIBUTES_TOPIC + "/response/+", qos=self.quality_of_service)
            self._client.subscribe(RPC_REQUEST_TOPIC + '+', qos=self.quality_of_service)
            self._client.subscribe(RPC_RESPONSE_TOPIC + '+', qos=self.quality_of_service)

            self.connected = True

            return response
        except MQTTException as e:
            self.connected = False
            print(e)

    def disconnect(self):
        self._client.disconnect()
        self.connected = False

    def _callback(self, topic, msg):
        topic = topic.decode('utf-8')
        print('callback', topic, msg)

        update_response_pattern = "v2/fw/response/" + str(self.__firmware_request_id) + "/chunk/"

        if topic.startswith(update_response_pattern):
            firmware_data = msg

            self.firmware_data = self.firmware_data + firmware_data
            self.__current_chunk = self.__current_chunk + 1

            print('Getting chunk with number: %s. Chunk size is : %r byte(s).' % (
                self.__current_chunk, self.__chunk_size))

            if len(self.firmware_data) == self.__target_firmware_length:
                self.__process_firmware()
            else:
                self.__get_firmware()
        else:
            self._on_decode_message(topic, msg)

    def _on_decode_message(self, topic, msg):
        if topic.startswith(RPC_REQUEST_TOPIC):
            self._handle_rpc_request(topic, msg)
        elif topic.startswith(RPC_RESPONSE_TOPIC):
            self._handle_rpc_response(topic, msg)
        elif topic == ATTRIBUTES_TOPIC:
            self._handle_attributes(msg)
        elif topic.startswith(ATTRIBUTE_TOPIC_RESPONSE):
            self._handle_attribute_response(topic, msg)

        if topic.startswith(ATTRIBUTES_TOPIC):
            self.firmware_info = ujson.loads(msg)

            if '/response/' in topic:
                self._handle_firmware_info()

            if (self.firmware_info.get(FW_VERSION_ATTR) is not None and self.firmware_info.get(
                    FW_VERSION_ATTR) != self.current_firmware_info.get("current_" + FW_VERSION_ATTR)) or \
                    (self.firmware_info.get(FW_TITLE_ATTR) is not None and self.firmware_info.get(
                        FW_TITLE_ATTR) != self.current_firmware_info.get("current_" + FW_TITLE_ATTR)):
                self._handle_firmware_update()

    def _handle_firmware_info(self):
        self.firmware_info = self.firmware_info.get("shared", {}) if isinstance(self.firmware_info,
                                                                                dict) else {}

    def _handle_firmware_update(self):
        print('Firmware is not the same')
        self.firmware_data = b''
        self.__current_chunk = 0

        self.current_firmware_info[FW_STATE_ATTR] = "DOWNLOADING"
        self.send_telemetry(self.current_firmware_info)
        time.sleep(1)

        self.__firmware_request_id = self.__firmware_request_id + 1
        self.__target_firmware_length = self.firmware_info[FW_SIZE_ATTR]
        self.__chunk_count = 0 if not self.__chunk_size else ceil(
            self.firmware_info[FW_SIZE_ATTR] / self.__chunk_size)
        self.__get_firmware()

    def __get_firmware(self):
        payload = '' if not self.__chunk_size or self.__chunk_size > self.firmware_info.get(FW_SIZE_ATTR, 0) else str(
            self.__chunk_size).encode()
        self._client.publish("v2/fw/request/{0}/chunk/{1}".format(self.__firmware_request_id, self.__current_chunk),
                             payload, qos=1)

    def __process_firmware(self):
        self.current_firmware_info[FW_STATE_ATTR] = "DOWNLOADED"
        self.send_telemetry(self.current_firmware_info)
        time.sleep(1)

        verification_result = verify_checksum(self.firmware_data, self.firmware_info.get(FW_CHECKSUM_ALG_ATTR),
                                              self.firmware_info.get(FW_CHECKSUM_ATTR))

        if verification_result:
            print('Checksum verified!')
            self.current_firmware_info[FW_STATE_ATTR] = "VERIFIED"
            self.send_telemetry(self.current_firmware_info)
            time.sleep(1)

            self.__on_firmware_received(self.firmware_info.get(FW_VERSION_ATTR))

            self.current_firmware_info = {
                "current_" + FW_TITLE_ATTR: self.firmware_info.get(FW_TITLE_ATTR),
                "current_" + FW_VERSION_ATTR: self.firmware_info.get(FW_VERSION_ATTR),
                FW_STATE_ATTR: "UPDATED"
            }
            self.send_telemetry(self.current_firmware_info)
        else:
            print('Checksum verification failed!')
            self.current_firmware_info[FW_STATE_ATTR] = "FAILED"
            self.send_telemetry(self.current_firmware_info)
            self.__request_firmware_info()
            return
        self.firmware_received = True

    def __on_firmware_received(self, version_to):
        with open(self.firmware_info.get(FW_TITLE_ATTR), "wb") as firmware_file:
            firmware_file.write(self.firmware_data)
        print('Firmware is updated!\n Current firmware version is: {0}'.format(version_to))
        machine.reset()

    def __request_firmware_info(self):
        self.__request_id = self.__request_id + 1
        self._client.publish("v1/devices/me/attributes/request/{0}".format(self.__request_id),
                             ujson.dumps({"sharedKeys": REQUIRED_SHARED_KEYS}))

    def _handle_rpc_request(self, topic, msg):
        request_id = topic[len(RPC_REQUEST_TOPIC):len(topic)]
        if self.__device_on_server_side_rpc_response:
            self.__device_on_server_side_rpc_response(request_id, ujson.loads(msg))

    def _handle_rpc_response(self, topic, msg):
        request_id = int(topic[len(RPC_RESPONSE_TOPIC):len(topic)])
        callback = self.__device_client_rpc_dict.pop(request_id)
        callback(request_id, ujson.loads(msg), None)

    def set_server_side_rpc_request_handler(self, handler):
        self.__device_on_server_side_rpc_response = handler

    def _handle_attributes(self, msg):
        msg = ujson.loads(msg)
        dict_results = []
        # callbacks for everything
        if self.__device_sub_dict.get("*"):
            for subscription_id in self.__device_sub_dict["*"]:
                dict_results.append(self.__device_sub_dict["*"][subscription_id])
        # specific callback
        keys = msg.keys()
        keys_list = []
        for key in keys:
            keys_list.append(key)
        # iterate through message
        for key in keys_list:
            # find key in our dict
            if self.__device_sub_dict.get(key):
                for subscription in self.__device_sub_dict[key]:
                    dict_results.append(self.__device_sub_dict[key][subscription])
        for res in dict_results:
            res(msg, None)

    def _handle_attribute_response(self, topic, msg):
        req_id = int(topic[len(ATTRIBUTES_TOPIC + "/response/"):])
        callback = self._attr_request_dict.pop(req_id)
        if isinstance(callback, tuple):
            callback[0](ujson.loads(msg), None, callback[1])
        else:
            callback(ujson.loads(msg), None)

    def send_telemetry(self, data):
        telemetry_topic = 'v1/devices/me/telemetry'
        self._client.publish(telemetry_topic, ujson.dumps(data))

    def send_attributes(self, data):
        self._client.publish(ATTRIBUTES_TOPIC, ujson.dumps(data))

    def request_attributes(self, client_keys=None, shared_keys=None, callback=None):
        msg = {}
        if client_keys:
            tmp = ""
            for key in client_keys:
                tmp += key + ","
            tmp = tmp[:len(tmp) - 1]
            msg.update({"clientKeys": tmp})
        if shared_keys:
            tmp = ""
            for key in shared_keys:
                tmp += key + ","
            tmp = tmp[:len(tmp) - 1]
            msg.update({"sharedKeys": tmp})
        self._add_attr_request_callback(callback)
        self._client.publish(ATTRIBUTE_REQUEST_TOPIC + str(self.__attr_request_number),
                             ujson.dumps(msg),
                             qos=self.quality_of_service)
        self._client.wait_msg()

    def _add_attr_request_callback(self, callback):
        self.__attr_request_number += 1
        self._attr_request_dict.update({self.__attr_request_number: callback})
        attr_request_number = self.__attr_request_number
        return attr_request_number

    def send_rpc_call(self, method, params, callback):
        self.__device_client_rpc_number += 1
        self.__device_client_rpc_dict.update({self.__device_client_rpc_number: callback})
        rpc_request_id = self.__device_client_rpc_number
        payload = {"method": method, "params": params}
        self._client.publish(RPC_REQUEST_TOPIC + str(rpc_request_id),
                             ujson.dumps(payload),
                             qos=self.quality_of_service)
        self._client.wait_msg()

    def unsubscribe_from_attribute(self, subscription_id):
        for attribute in self.__device_sub_dict:
            if self.__device_sub_dict[attribute].get(subscription_id):
                del self.__device_sub_dict[attribute][subscription_id]
                print("Unsubscribed from {0}, subscription id {1}".format(attribute, subscription_id))
        if subscription_id == '*':
            self.__device_sub_dict = {}
        self.__device_sub_dict = dict((k, v) for k, v in self.__device_sub_dict.items() if v)

    def clean_device_sub_dict(self):
        self.__device_sub_dict = {}

    def subscribe_to_all_attributes(self, callback):
        return self.subscribe_to_attribute("*", callback)

    def subscribe_to_attribute(self, key, callback):
        self.__device_max_sub_id += 1
        if key not in self.__device_sub_dict:
            self.__device_sub_dict.update({key: {self.__device_max_sub_id: callback}})
        else:
            self.__device_sub_dict[key].update({self.__device_max_sub_id: callback})
        print("Subscribed to {0} with id {1}".format(key, self.__device_max_sub_id))
        return self.__device_max_sub_id

    def wait_for_msg(self):
        self._client.wait_msg()


tb_device_mqtt.py

import usocket as socket
import ustruct as struct
from ubinascii import hexlify


class MQTTException(Exception):
    pass


class MQTTClient:
    def __init__(
        self,
        client_id,
        server,
        port=0,
        user=None,
        password=None,
        keepalive=0,
        ssl=False,
        ssl_params={},
    ):
        if port == 0:
            port = 8883 if ssl else 1883
        self.client_id = client_id
        self.sock = None
        self.server = server
        self.port = port
        self.ssl = ssl
        self.ssl_params = ssl_params
        self.pid = 0
        self.cb = None
        self.user = user
        self.pswd = password
        self.keepalive = keepalive
        self.lw_topic = None
        self.lw_msg = None
        self.lw_qos = 0
        self.lw_retain = False

    def _send_str(self, s):
        self.sock.write(struct.pack("!H", len(s)))
        self.sock.write(s)

    def _recv_len(self):
        n = 0
        sh = 0
        while 1:
            b = self.sock.read(1)[0]
            n |= (b & 0x7F) << sh
            if not b & 0x80:
                return n
            sh += 7

    def set_callback(self, f):
        self.cb = f

    def set_last_will(self, topic, msg, retain=False, qos=0):
        assert 0 <= qos <= 2
        assert topic
        self.lw_topic = topic
        self.lw_msg = msg
        self.lw_qos = qos
        self.lw_retain = retain

    def connect(self, clean_session=True):
        self.sock = socket.socket()
        addr = socket.getaddrinfo(self.server, self.port)[0][-1]
        self.sock.connect(addr)
        if self.ssl:
            import ussl

            self.sock = ussl.wrap_socket(self.sock, **self.ssl_params)
        premsg = bytearray(b"\x10\0\0\0\0\0")
        msg = bytearray(b"\x04MQTT\x04\x02\0\0")

        sz = 10 + 2 + len(self.client_id)
        msg[6] = clean_session << 1
        if self.user is not None:
            sz += 2 + len(self.user) + 2 + len(self.pswd) if self.pswd else 0
            msg[6] |= 0xC0
        if self.keepalive:
            assert self.keepalive < 65536
            msg[7] |= self.keepalive >> 8
            msg[8] |= self.keepalive & 0x00FF
        if self.lw_topic:
            sz += 2 + len(self.lw_topic) + 2 + len(self.lw_msg)
            msg[6] |= 0x4 | (self.lw_qos & 0x1) << 3 | (self.lw_qos & 0x2) << 3
            msg[6] |= self.lw_retain << 5

        i = 1
        while sz > 0x7F:
            premsg[i] = (sz & 0x7F) | 0x80
            sz >>= 7
            i += 1
        premsg[i] = sz

        self.sock.write(premsg, i + 2)
        self.sock.write(msg)
        # print(hex(len(msg)), hexlify(msg, ":"))
        self._send_str(self.client_id)
        if self.lw_topic:
            self._send_str(self.lw_topic)
            self._send_str(self.lw_msg)
        if self.user is not None:
            self._send_str(self.user)
        if self.pswd is not None:
            self._send_str(self.pswd)
        resp = self.sock.read(4)
        assert resp[0] == 0x20 and resp[1] == 0x02
        if resp[3] != 0:
            raise MQTTException(resp[3])
        return resp[2] & 1

    def disconnect(self):
        self.sock.write(b"\xe0\0")
        self.sock.close()

    def ping(self):
        self.sock.write(b"\xc0\0")

    def publish(self, topic, msg, retain=False, qos=0):
        pkt = bytearray(b"\x30\0\0\0")
        pkt[0] |= qos << 1 | retain
        sz = 2 + len(topic) + len(msg)
        if qos > 0:
            sz += 2
        assert sz < 2097152
        i = 1
        while sz > 0x7F:
            pkt[i] = (sz & 0x7F) | 0x80
            sz >>= 7
            i += 1
        pkt[i] = sz
        # print(hex(len(pkt)), hexlify(pkt, ":"))
        self.sock.write(pkt, i + 1)
        self._send_str(topic)
        if qos > 0:
            self.pid += 1
            pid = self.pid
            struct.pack_into("!H", pkt, 0, pid)
            self.sock.write(pkt, 2)
        self.sock.write(msg)
        if qos == 1:
            while 1:
                op = self.wait_msg()
                if op == 0x40:
                    sz = self.sock.read(1)
                    assert sz == b"\x02"
                    rcv_pid = self.sock.read(2)
                    rcv_pid = rcv_pid[0] << 8 | rcv_pid[1]
                    if pid == rcv_pid:
                        return
        elif qos == 2:
            assert 0

    def subscribe(self, topic, qos=0):
        assert self.cb is not None, "Subscribe callback is not set"
        pkt = bytearray(b"\x82\0\0\0")
        self.pid += 1
        struct.pack_into("!BH", pkt, 1, 2 + 2 + len(topic) + 1, self.pid)
        # print(hex(len(pkt)), hexlify(pkt, ":"))
        self.sock.write(pkt)
        self._send_str(topic)
        self.sock.write(qos.to_bytes(1, "little"))
        while 1:
            op = self.wait_msg()
            if op == 0x90:
                resp = self.sock.read(4)
                # print(resp)
                assert resp[1] == pkt[2] and resp[2] == pkt[3]
                if resp[3] == 0x80:
                    raise MQTTException(resp[3])
                return

    # Wait for a single incoming MQTT message and process it.
    # Subscribed messages are delivered to a callback previously
    # set by .set_callback() method. Other (internal) MQTT
    # messages processed internally.
    def wait_msg(self):
        res = self.sock.read(1)
        self.sock.setblocking(True)
        if res is None:
            return None
        if res == b"":
            raise OSError(-1)
        if res == b"\xd0":  # PINGRESP
            sz = self.sock.read(1)[0]
            assert sz == 0
            return None
        op = res[0]
        if op & 0xF0 != 0x30:
            return op
        sz = self._recv_len()
        topic_len = self.sock.read(2)
        topic_len = (topic_len[0] << 8) | topic_len[1]
        topic = self.sock.read(topic_len)
        sz -= topic_len + 2
        if op & 6:
            pid = self.sock.read(2)
            pid = pid[0] << 8 | pid[1]
            sz -= 2
        msg = self.sock.read(sz)
        self.cb(topic, msg)
        if op & 6 == 2:
            pkt = bytearray(b"\x40\x02\0\0")
            struct.pack_into("!H", pkt, 2, pid)
            self.sock.write(pkt)
        elif op & 6 == 4:
            assert 0
        return op

    # Checks whether a pending message from server is available.
    # If not, returns immediately with None. Otherwise, does
    # the same processing as wait_msg.
    def check_msg(self):
        self.sock.setblocking(False)
        return self.wait_msg()


umqtt.py

#      Copyright 2023. 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 random import getrandbits
from hashlib import sha256


def verify_checksum(firmware_data, checksum_alg, checksum):
    if firmware_data is None:
        print('Firmware was not received!')
        return False
    if checksum is None:
        print('Checksum was\'t provided!')
        return False
    checksum_of_received_firmware = None
    print('Checksum algorithm is: %s' % checksum_alg)
    if checksum_alg.lower() == "sha256":
        checksum_of_received_firmware = "".join(["%.2x" % i for i in sha256(firmware_data).digest()])
    else:
        print('Client error. Unsupported checksum algorithm.')

    print(checksum_of_received_firmware)

    return checksum_of_received_firmware == checksum
