当前位置：首页 > news >正文

test_mqtt/python

news 来源：原创 2024/9/21 12:27:41

#subscript端的功能
import csv
import os
import random
import timefrom paho.mqtt import client as mqtt_client
broker = '175.24.64.90'
port = 1883
# topic = "/v1_1/up/device_upload/EMS1"
topic = "/api/ems_cloud/v1_1/operate_time_command/{HZ01}"
# generate client ID with pub prefix rando
client_id = f'python-mqtt-{random.randint(1000, 2000)}'file = open("output.csv", 'w').close()  # 清空文件之前的内容def write_csv(mPayload, mTopic):# 创建文件夹和表头# os.makedirs(os.path.join('..', 'data'), exist_ok=True)  # 创建数据文件夹# data_file = os.path.join('..', 'data', 'output.csv')# with open(data_file, 'w', encoding='utf-8', newline='') as f:#    writer = csv.writer(f)#   writer.writerow(['payload', 'topic'])# 追加数据with open("output.csv", 'a', encoding='utf-8', newline='') as f:f.seek(0)writer = csv.writer(f)writer.writerow([mPayload, mTopic])

# publish端的功能
import random
import time
import datetime
from paho.mqtt import client as mqtt_clientbroker = '175.24.64.90'
port = 1883
topic = "/v1_1/down/remote_control/EMS1"
# generate client ID with pub prefix randomly
client_id = f'python-mqtt-{random.randint(0, 1000)}'
pub_data = {"cmdid": 1024,                          # 命令id，便于边缘侧做命令反馈"ts": datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),                            # 秒级时间戳"identify": "simulator",                      # 受控设备identify"data": [{"identity": "cmd_bool",                # 受控点identity"value": 1,                      # 受控点写入值},{"identity": "cmd_ushort","value": 10,},{"identity": "cmd_float",  # 受控点identity"value": -11.98,  # 受控点写入值}]
}# msg = json.dumps(pub_data)
msg = str(pub_data)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_connectclient.username_pw_set("emsuser","emsuser")client.connect(broker, port, 60)return clientdef publish(client):msg_count = 0while True:time.sleep(1)result = client.publish(topic, msg)# result: [0, 1]status = result[0]if status == 0:print(f"Send `{msg}` to topic `{topic}`")else:print(f"Failed to send message to topic {topic}")msg_count += 1def run():client = connect_mqtt()client.loop_start()publish(client)if __name__ == '__main__':run()