import random
import time
from paho.mqtt import client as mqtt_client
import json
from datetime import datetime
broker = 'broker.emqx.io'
port = 1883
topic = "/python/mqtt/li"
client_id = f'python-mqtt-{random.randint(0, 1000)}' # 隨機生成客戶端id
def connect_mqtt():
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT Broker!")
else:
print("Failed to connect, return code %d\n", rc)
client = mqtt_client.Client(client_id)
client.on_connect = on_connect
client.connect(broker, port)
return client
def publish(client):
while True:
time.sleep(0.01)
msg = json.dumps({"MAC": "0123456789",
"samplerate": 12,
"sampletime": str(datetime.utcnow().strftime('%Y/%m/%d-%H:%M:%S.%f')[:-3]),
"battery": 0.5,
"acc": [
[random.randint(200, 350), -random.randint(200, 350), -random.randint(200, 350), random.randint(200, 350), random.randint(200, 350), random.randint(200, 350)],
[random.randint(200, 350), -random.randint(200, 350), -random.randint(200, 350), random.randint(200, 350), random.randint(200, 350), random.randint(200, 350)],
[random.randint(200, 350), -random.randint(200, 350), -random.randint(200, 350), random.randint(200, 350), random.randint(200, 350), random.randint(200, 350)],
[random.randint(200, 350), -random.randint(200, 350), -random.randint(200, 350), random.randint(200, 350), random.randint(200, 350), random.randint(200, 350)],
[random.randint(200, 350), -random.randint(200, 350), -random.randint(200, 350), random.randint(200, 350), random.randint(200, 350), random.randint(200, 350)],
[random.randint(200, 350), -random.randint(200, 350), -random.randint(200, 350), random.randint(200, 350), random.randint(200, 350), random.randint(200, 350)],
[random.randint(200, 350), -random.randint(200, 350), -random.randint(200, 350), random.randint(200, 350), random.randint(200, 350), random.randint(200, 350)],
[random.randint(200, 350), -random.randint(200, 350), -random.randint(200, 350), random.randint(200, 350), random.randint(200, 350), random.randint(200, 350)],
[random.randint(200, 350), -random.randint(200, 350), -random.randint(200, 350), random.randint(200, 350), random.randint(200, 350), random.randint(200, 350)],
[random.randint(200, 350), -random.randint(200, 350), -random.randint(200, 350), random.randint(200, 350), random.randint(200, 350), random.randint(200, 350)],
[random.randint(200, 350), -random.randint(200, 350), -random.randint(200, 350), random.randint(200, 350), random.randint(200, 350), random.randint(200, 350)],
[random.randint(200, 350), -random.randint(200, 350), -random.randint(200, 350), random.randint(200, 350), random.randint(200, 350), random.randint(200, 350)],
]})
result = client.publish(topic, msg)
status = result[0]
if status == 0:
print(f"Send `{msg}` to topic `{topic}`")
else:
print(f"Failed to send message to topic {topic}")
def run():
client = connect_mqtt()
client.loop_start()
publish(client)
if __name__ == '__main__':
run()
import json
import random
from paho.mqtt import client as mqtt_client
import time
import datetime
from math import ceil, floor
import matplotlib.pyplot as plt
import _thread
# 公共變量
broker = 'broker.emqx.io'
topic = "/python/mqtt/li"
port = 1883
client_id = f'python-mqtt-li-{random.randint(0, 100)}'
show_num = 300
x_num = [-1] # 計數(shù)
acc1 = []
acc2 = []
acc3 = []
acc4 = []
acc5 = []
acc6 = []
stime = []
"""mqtt subscribe topic"""
def str_microsecond_datetime2int_13timestamp(str_microsecond_datetime):
"""將字符串型【毫秒級】格式化時間 轉(zhuǎn)為 【13位】整型時間戳"""
datetime_obj = datetime.datetime.strptime(str_microsecond_datetime, "%Y/%m/%d-%H:%M:%S.%f")
obj_stamp = int(time.mktime(datetime_obj.timetuple()) * 1000.0 + datetime_obj.microsecond / 1000.0) / 1000.0
return obj_stamp
def int2datetime(int_float_timestamp):
"""
有小數(shù)點:分離小數(shù)點,整數(shù)轉(zhuǎn)為格式化時間,小數(shù)點直接跟在后面
無小數(shù)點:從第10位進(jìn)行分離,
所以本函數(shù)只適用于時間戳整數(shù)位數(shù)大于9且小于11.
"""
if '.' in str(int_float_timestamp):
int_float = str(int_float_timestamp).split('.')
date = time.localtime(int(int_float[0]))
tempDate = time.strftime("%Y/%m/%d-%H:%M:%S", date)
secondafter = '.' + str(int_float[1])
return str(tempDate) + secondafter
def parse_mqttmsg(msg):
"""解析mqt頭數(shù)據(jù) MAC samplerate sampletime battery acc"""
content = json.loads(msg.payload.decode())
span = 1000 / content['samplerate'] * 10
time_span = [ceil(span) / 10 / 1000, floor(span) / 10 / 1000]
sampletime = content['sampletime']
sampletime_int = str_microsecond_datetime2int_13timestamp(sampletime)
acc = content['acc']
for i in range(len(acc)):
x_num.append(x_num[-1] + 1)
acc1.append(acc[i][0])
acc2.append(acc[i][1])
acc3.append(acc[i][2])
acc4.append(acc[i][3])
acc5.append(acc[i][4])
acc6.append(acc[i][5])
if i != 0:
sampletime_int += time_span[i % 2]
stime.append(int2datetime(round(sampletime_int * 1000, 0) / 1000))
else:
stime.append(sampletime)
print(x_num[-1], stime[-1], acc1[-1], acc2[-1], acc3[-1], acc4[-1], acc5[-1], acc6[-1])
def connect_mqtt():
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT Broker!")
else:
print("Failed to connect, return code %d\n", rc)
pass
client = mqtt_client.Client(client_id)
client.on_connect = on_connect
client.connect(broker, port)
return client
def subscribe(client: mqtt_client):
def on_message(client, userdata, msg):
# print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")
parse_mqttmsg(msg)
client.subscribe(topic)
client.on_message = on_message
def run():
client = connect_mqtt()
subscribe(client)
client.loop_forever()
""" draw figures """
def draw_figure():
plt.ion() # 開啟一個畫圖的窗口進(jìn)入交互模式,用于實時更新數(shù)據(jù)
plt.rcParams['figure.figsize'] = (10, 10) # 圖像顯示大小
plt.rcParams['font.sans-serif'] = ['SimHei'] # 防止中文標(biāo)簽亂碼,還有通過導(dǎo)入字體文件的方法
plt.rcParams['axes.unicode_minus'] = False
plt.rcParams['lines.linewidth'] = 0.5 # 設(shè)置曲線線條寬度
count = 0
while True:
plt.clf() # 清除刷新前的圖表,防止數(shù)據(jù)量過大消耗內(nèi)存
plt.suptitle("總標(biāo)題", fontsize=30) # 添加總標(biāo)題,并設(shè)置文字大小
plt.tight_layout()
# 圖表1
agraphic = plt.subplot(2, 1, 1)
agraphic.set_title('子圖表標(biāo)題1') # 添加子標(biāo)題
agraphic.set_xlabel('x軸', fontsize=10) # 添加軸標(biāo)簽
agraphic.set_ylabel('y軸', fontsize=20)
plt.plot(x_num[1:][-show_num:], acc1[-show_num:], 'g-')
try:
xtricks = list(range(len(acc1) - show_num, len(acc1), 10)) # **1**
xlabels = [stime[i] for i in xtricks] # **2**
plt.xticks(xtricks, xlabels, rotation=15)
except:
pass
# 圖表2
bgraghic = plt.subplot(2, 1, 2)
bgraghic.set_title('子圖表標(biāo)題2')
bgraghic.set_xlabel('x軸', fontsize=10) # 添加軸標(biāo)簽
bgraghic.set_ylabel('y軸', fontsize=20)
bgraghic.plot(x_num[1:][-show_num:], acc2[-show_num:], 'r^')
plt.pause(0.001) # 設(shè)置暫停時間,太快圖表無法正常顯示
count = count + 1
if __name__ == '__main__':
# 多線程
_thread.start_new_thread(run, ())
draw_figure()
到此這篇關(guān)于Python中使用matplotlib繪制mqtt數(shù)據(jù)實時圖像的文章就介紹到這了,更多相關(guān)matplotlib繪制mqtt數(shù)據(jù)實時圖像內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!