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

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

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

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

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

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

源代码如下：

from PyQt5 import QtCore, QtGui, QtWidgetsimport sslfrom PyQt5.QtGui import QColor,QIconimport hashlibimport hmacimport base64import stompimport scheduleimport threadingimport timefrom PyQt5.QtCore import QThread, pyqtSignaldata=[1]odl=1userdata=[1,2,3,4]# 创建线程1的类class Worker(QThread):sig = pyqtSignal(str)def __init__(self, parent=None):super(Worker, self).__init__(parent)self.count = 0def run(self):# encoding=utf-8"""amqp = input('请输入实例id： ')amqp = str(amqp) + '.amqp.'"""amqp = 'iot-06z00gv3rosifyx.amqp.'def connect_and_subscribe(conn):accessKey = str(userdata[0])accessSecret = str(userdata[1])consumerGroupId = str(userdata[2])# iotInstanceId：实例ID。iotInstanceId = str(userdata[3])clientId = "5277104"# 签名方法：支持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 = conndef 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.bodydef 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 eodl = 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.translateMainWindow.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 sysfrom 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_())

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

百度网盘链接：​​​​​​​百度网盘 请输入提取码

提取码：yzr1