阿里云AMQT物联网订阅服务调试工具

  我们在阿里云物联网平台的开发中，常常需要接受设备发出的信息，用于调试和开发，如果一直翻看日志不仅耗费时间，所以我利用pyqt5设计了一个简易的调试界面，它可以将阿里云订阅的数据显示在文本框中。后面我会更新一个RRpc发送的小工具，无需到阿里云界面便可以快捷的调用调试测试设备。后面会放百度网盘链接。

  首先我们看看需要哪些信息，信息的获取需要Accesskey，AccessSecret，ConsumerGroup和iotInstanceId，这些我们可以在分别可以在阿里云Access管理和物联网平台服务端订阅中找到。

之后我们可以在服务端订阅里编辑订阅，把你想要的信息加入到推送消息类型，这样调试工具就会得到你所订阅的信息。

 在输入以上四个数据后，我们点击运行就可以成功连接上订阅端。

点击停止可以暂停接受消息，点击清空运行记录可以将右侧运行记录清除。

源代码如下：

from PyQt5 import QtCore, QtGui, QtWidgets
import ssl
from PyQt5.QtGui import QColor,QIcon
import hashlib
import hmac
import base64
import stomp
import schedule
import threading
import time
from PyQt5.QtCore import QThread, pyqtSignal



data=[1]
odl=1
userdata=[1,2,3,4]
# 创建线程1的类
class Worker(QThread):
    sig = pyqtSignal(str)

    def __init__(self, parent=None):
        super(Worker, self).__init__(parent)
        self.count = 0

    def run(self):

        # encoding=utf-8
        """amqp = input('请输入实例id： ')
        amqp = str(amqp)   '.amqp.iothub.aliyuncs.com'"""
        amqp = 'iot-06z00gv3rosifyx.amqp.iothub.aliyuncs.com'
        def connect_and_subscribe(conn):

            accessKey = str(userdata[0])
            accessSecret = str(userdata[1])
            consumerGroupId = str(userdata[2])
            # iotInstanceId：实例ID。
            iotInstanceId = str(userdata[3])
            clientId = "20205277104"
            # 签名方法：支持hmacmd5，hmacsha1和hmacsha256。
            signMethod = "hmacsha1"
            timestamp = current_time_millis()
            # userName组装方法，请参见AMQP客户端接入说明文档。
            # 若使用二进制传输，则userName需要添加encode=base64参数，服务端会将消息体base64编码后再推送。具体添加方法请参见下一章节“二进制消息体说明”。
            username = clientId   "|authMode=aksign"   ",signMethod="   signMethod \
                         ",timestamp="   timestamp   ",authId="   accessKey \
                         ",iotInstanceId="   iotInstanceId \
                         ",consumerGroupId="   consumerGroupId   "|"
            signContent = "authId="   accessKey   "&timestamp="   timestamp
            # 计算签名，password组装方法，请参见AMQP客户端接入说明文档。
            password = do_sign(accessSecret.encode("utf-8"), signContent.encode("utf-8"))

            conn.set_listener('', MyListener(conn))
            conn.connect(username, password, wait=True)
            # 清除历史连接检查任务，新建连接检查任务
            schedule.clear('conn-check')
            schedule.every(1).seconds.do(do_check, conn).tag('conn-check')

        class MyListener(stomp.ConnectionListener):
            def __init__(self, conn):
                self.conn = conn

            def on_error(self, frame):
                print('received an error "%s"' % frame.body)

            def on_message(self, frame):
                print('received a message "%s"' % frame.body)
                data[0]=frame.body
            def on_heartbeat_timeout(self):
                print('on_heartbeat_timeout')

            def on_connected(self, headers):
                data[0]="successfully connected"
                conn.subscribe(destination='/topic/#', id=1, ack='auto')
                data[0]="successfully subscribe"

            def on_disconnected(self):
                print('disconnected')
                connect_and_subscribe(self.conn)

        def current_time_millis():
            return str(int(round(time.time() * 1000)))

        def do_sign(secret, sign_content):
            m = hmac.new(secret, sign_content, digestmod=hashlib.sha1)
            return base64.b64encode(m.digest()).decode("utf-8")

        # 检查连接，如果未连接则重新建连
        def do_check(conn):
            print('check connection, is_connected: %s', conn.is_connected())
            if (not conn.is_connected()):
                try:
                    connect_and_subscribe(conn)
                except Exception as e:
                    data[0]='disconnected, '

        # 定时任务方法，检查连接状态
        def connection_check_timer():
            while 1:
                schedule.run_pending()
                time.sleep(10)

        #  接入域名，请参见AMQP客户端接入说明文档。这里直接填入域名，不需要带amqps://前缀
        conn = stomp.Connection([(amqp, 61614)], heartbeats=(0, 300))
        conn.set_ssl(for_hosts=[(amqp, 61614)], ssl_version=ssl.PROTOCOL_TLS)

        try:
            connect_and_subscribe(conn)
        except Exception as e:
            data[0]='connecting failed'
            raise e

        odl = 0
        # 异步线程运行定时任务，检查连接状态
        thread = threading.Thread(target=connection_check_timer)
        thread.start()
        while 1:

            if data[0] != odl:
                self.sig.emit(str(data[-1]))
            odl = data[0]

            time.sleep(0.5)


class Ui_MainWindow(object):
    def __init__(self):
        super(Ui_MainWindow,self).__init__()
        self.thread = Worker()
        self.thread.sig.connect(self.updateLabel)



    def setupUi(self, MainWindow):
        MainWindow.setWindowIcon(QIcon("./birdicon.ico"))
        MainWindow.setObjectName("MainWindow")
        MainWindow.resize(800, 600)
        self.centralwidget = QtWidgets.QWidget(MainWindow)
        self.centralwidget.setObjectName("centralwidget")
        self.pushButton = QtWidgets.QPushButton(self.centralwidget)
        self.pushButton.setGeometry(QtCore.QRect(80, 410, 140, 51))
        self.pushButton.setObjectName("pushButton")
        self.pushButton_2 = QtWidgets.QPushButton(self.centralwidget)
        self.pushButton_2.setGeometry(QtCore.QRect(80, 480, 140, 51))
        self.pushButton_2.setObjectName("pushButton_3")
        self.pushButton_3 = QtWidgets.QPushButton(self.centralwidget)
        self.pushButton_3.setGeometry(QtCore.QRect(230, 410, 140, 51))
        self.pushButton_3.setObjectName("pushButton_3")
        self.label = QtWidgets.QLabel(self.centralwidget)
        self.label.setGeometry(QtCore.QRect(10, 40, 150, 31))
        self.label.setObjectName("label")
        self.label_2 = QtWidgets.QLabel(self.centralwidget)
        self.label_2.setGeometry(QtCore.QRect(10, 140, 150, 31))
        self.label_2.setObjectName("label_2")
        self.label_3 = QtWidgets.QLabel(self.centralwidget)
        self.label_3.setGeometry(QtCore.QRect(10, 250, 150, 31))
        self.label_3.setObjectName("label_3")
        self.label_4 = QtWidgets.QLabel(self.centralwidget)
        self.label_4.setGeometry(QtCore.QRect(10, 340, 150, 31))
        self.label_4.setObjectName("label_4")
        self.lineEdit = QtWidgets.QLineEdit(self.centralwidget)
        self.lineEdit.setGeometry(QtCore.QRect(160, 40, 260, 41))
        self.lineEdit.setObjectName("lineEdit")
        self.lineEdit_2 = QtWidgets.QLineEdit(self.centralwidget)
        self.lineEdit_2.setGeometry(QtCore.QRect(160, 140, 260, 41))
        self.lineEdit_2.setObjectName("lineEdit_2")
        self.lineEdit_3 = QtWidgets.QLineEdit(self.centralwidget)
        self.lineEdit_3.setGeometry(QtCore.QRect(160, 240, 260, 41))
        self.lineEdit_3.setObjectName("lineEdit_3")
        self.lineEdit_4 = QtWidgets.QLineEdit(self.centralwidget)
        self.lineEdit_4.setGeometry(QtCore.QRect(160, 340, 260, 41))
        self.lineEdit_4.setObjectName("lineEdit_4")
        self.textBrowser = QtWidgets.QTextBrowser(self.centralwidget)
        self.textBrowser.setGeometry(QtCore.QRect(440, 40, 311, 431))
        self.textBrowser.setObjectName("textBrowser")
        MainWindow.setCentralWidget(self.centralwidget)
        self.menubar = QtWidgets.QMenuBar(MainWindow)
        self.menubar.setGeometry(QtCore.QRect(0, 0, 800, 26))
        self.menubar.setObjectName("menubar")
        MainWindow.setMenuBar(self.menubar)
        self.statusbar = QtWidgets.QStatusBar(MainWindow)
        self.statusbar.setObjectName("statusbar")
        MainWindow.setStatusBar(self.statusbar)

        self.retranslateUi(MainWindow)
        QtCore.QMetaObject.connectSlotsByName(MainWindow)

    def retranslateUi(self, MainWindow):
        _translate = QtCore.QCoreApplication.translate
        MainWindow.setWindowTitle(_translate("MainWindow", "阿里云订阅服务端 元孜然 version:1.0.1"))
        self.pushButton.setText(_translate("MainWindow", "运行"))
        self.pushButton.clicked.connect(self.test)
        self.pushButton_2.setText(_translate("MainWindow", "停止"))
        self.pushButton_2.clicked.connect(self.quit)
        self.pushButton_3.setText(_translate("MainWindow", "清空运行记录"))
        self.pushButton_3.clicked.connect(self.claer)
        self.label.setText(_translate("MainWindow", "用户名Accesskey"))
        self.label_2.setText(_translate("MainWindow", "密匙AccessSecret"))
        self.label_3.setText(_translate("MainWindow", "用户组ConsumerGroupId"))
        self.label_4.setText(_translate("MainWindow", "实例名iotInstanceId"))



    def test(self):
        userdata[0] = str(self.lineEdit.text())
        userdata[1] = str(self.lineEdit_2.text())
        userdata[2] = str(self.lineEdit_3.text())
        userdata[3] = str(self.lineEdit_4.text())
        self.thread.start()

    def updateLabel(self, text):
        self.textBrowser.append('received a message ' text)
        self.cursor = self.textBrowser.textCursor()
        self.textBrowser.moveCursor(self.cursor.End)
        time.sleep(0.2)  # 光标移到最后，这样就会自动显示出来


    def quit(self):
        self.thread.terminate()
    def claer(self):
        self.textBrowser.clear()





if __name__ == '__main__':
    import sys
    from PyQt5.QtWidgets import QApplication, QMainWindow
    """import pics_ui_rc""" # 导入添加的资源（根据实际情况填写文件名）

    app = QApplication(sys.argv)
    MainWindow = QMainWindow()
    ui = Ui_MainWindow()
    ui.setupUi(MainWindow)
    MainWindow.show()
    sys.exit(app.exec_())

更新一下新版。这次输入错误不会退出。

下载链接：

https://download.csdn.net/download/qq_35682718/87828791

这篇好文章是转载于：学新通技术网