mqtt-control/esp8266/esp_mqtt.py

# Hardware Platform: FireBeetle-ESP32
# Result: input MQTTlibrary and remote controls LED by mqtt communication.
import os
from machine import Pin
from machine import Timer
import ustruct as struct
from umqtt.simple import MQTTClient

led = Pin(2, Pin.OUT, value=0)

SERVER = 'mqtts.heclouds.com'  # 服务器地址
PRO_ID = '522417'  # 产品ID
DEV_NAME = 'esp8266'  # 设备名称
# accesskey = "dgEP7iOL2/1kVNc1ykDvIhXSVqhb0tLnKinjU2RYpDs="  # 设备秘钥
password = "version=2018-10-31&res=products%2F522417&et=1685551369&method=sha1&sign=9l9gkFQx7A%2B7B3fu8uU%2F089azps%3D"

class MqttOneNet():
    def __init__(self, device_name=DEV_NAME) -> None:
        self.broker_addr = 'mqtts.heclouds.com'
        self.broker_port = 1883
        self.product_id = "522417"
        # 由 token 算法得出，有效期为 2023年5月31日01:41:21
        self.password = "version=2018-10-31&res=products%2F522417&et=1685551369&method=sha1&sign=9l9gkFQx7A%2B7B3fu8uU%2F089azps%3D"
        self.device_name = device_name
        # 暂未用到
        self.accesskey = ""
        self.device_id = ""
        self.client = self.client_init()
        self.data_point_id = 0
        self.upload_init()

    def upload_init(self):
        self.data_point_id = 0
        self.tim = Timer(-1)
        # 每 2000ms 定时回调一次 self.upload_data()
        self.tim.init(period=2000, mode=Timer.PERIODIC,
                      callback=self.publish_timer_callback)
        print("init finish, start upload_data")

    def client_init(self):
        client = MQTTClient(client_id=self.device_name, server=self.broker_addr, port=self.broker_port, user=self.product_id,
                            password=self.password, keepalive=360)  # create a mqtt client
        client.connect()  # connect mqtt
        print("mqtt connect")
        client.set_callback(self.client_msg_callback)
        # 订阅所有消息
        topic = f"$sys/{self.product_id}/{self.device_name}/#"
        client.subscribe(topic)
        print("subscribed to %s topic" % (topic))
        return client

    # timer 是定时器回调函数需要的参数，实际上是定时器本身(self)，如果没有这个参数，定时器会一直打印一个错误
    # http://docs.micropython.org/en/latest/library/machine.Timer.html#machine-timer
    def publish_timer_callback(self, timer):
        topic_publish = '$sys/%s/%s/dp/post/json' % (
            self.product_id, self.device_name)
        payload = self.get_upload_data()
        
        print("------------- publish ------------- \n")
        print("topic:%s\n" % topic_publish)
        print("payload:%s\n" % payload)
        self.client.publish(topic=topic_publish, msg=payload, qos=1)

    def client_msg_callback(self, topic, msg):
        print("\n------------- receive ------------- \n")
        print("topic:%s\n" % topic)
        print("payload:%s\n" % msg)

    # 获取板子数据，用于上传到服务器
    def get_upload_data(self):
        self.data_point_id += 1
        message = {
            "id": int(self.data_point_id),
            "dp": {
                "sensor": [{      # 距离传感器采集的数据
                    "v": {
                        "temperature": self.get_temperature(),
                        "humidity": self.get_humidity()
                    }
                }],
                "system": [{        
                    "v": self.get_system()
                }]
            }
        }
        message = bytes("{}".format(message), 'utf8')
        return message

    def get_temperature(self):
        return struct.unpack('@H', os.urandom(2))[0]
    
    def get_humidity(self):
        return struct.unpack('@H', os.urandom(2))[0]

    def get_system(self):
        info = {}
        info["power"]=os.urandom(1)[0] % 42
        gps_location = struct.unpack('@BBBB', os.urandom(4))
        info["GPS"] = "%d°%d'N',%d°%d'E" % (gps_location[0], gps_location[1], gps_location[2], gps_location[3])
        info["altitude"] = struct.unpack('@H', os.urandom(2))[0]
        return info
    

            
# device = MqttOneNet()
if __name__ == '__main__':
    try:
        MqttOneNet()
    except Exception as e:
        print(e)