通过物联网管理多台MQTT设备-基于米尔T527开发板

fastapi==0.108.0paho-mqtt==1.6.1

from fastapi import FastAPI, HTTPException, Body, status  from paho.mqtt.client import Client as MQTTClientfrom typing import List, Dict, Any  import asyncio  import json    app = FastAPI()  mqtt_client = None  device_data = {}  
subtopic="gateway/device/#"
  # MQTT回调函数  def on_message(client, userdata, msg):      payload = msg.payload.decode()      topic = msg.topic      device_id = topic.split('/')[-1]      device_data[device_id] = payload     print(f"Received message from {device_id}: {payload}")    # MQTT连接和订阅  def mqtt_connect_and_subscribe(broker_url, broker_port):      global mqtt_client      mqtt_client = MQTTClient()      mqtt_client.on_message = on_message      mqtt_client.connect(broker_url, broker_port, 60)      mqtt_client.subscribe(subtopic)      mqtt_client.loop_start()    # MQTT发布消息  async def mqtt_publish(topic: str, message: str):      if mqtt_client is not None and mqtt_client.is_connected():          mqtt_client.publish(topic, message)      else:          print("MQTT client is not connected!")    # 设备管理：添加设备  @app.post("/devices/", status_code=status.HTTP_201_CREATED)  async def add_device(device_id: str):      device_data[device_id] = None      return {"message": f"Device {device_id} added"}    # 设备管理：获取设备列表  @app.get("/devices/")  async def get_devices():      return list(device_data.keys())    # 设备管理：获取设备数据  @app.get("/devices/{device_id}/data")  async def get_device_data(device_id: str):      if device_id not in device_data:          raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Device {device_id} not found")      return device_data.get(device_id)    # 设备管理：发送数据到设备  @app.post("/devices/{device_id}/data")  async def send_data_to_device(device_id: str, data: Dict[str, Any] = Body(...)):      topic = f"devices/{device_id}"      message = json.dumps(data)      await mqtt_publish(topic, message)      return {"message": f"Data sent to {device_id}"}    # 设备控制：发送控制命令到设备  @app.post("/devices/{device_id}/control")  async def control_device(device_id: str, command: str):      topic = f"devices/device/{device_id}"      await mqtt_publish(topic, command)      return {"message": f"Control command sent to {device_id}"}    # FastAPI启动事件  @app.on_event("startup")  async def startup_event():      mqtt_connect_and_subscribe("127.0.0.1", 1883)    # FastAPI关闭事件  @app.on_event("shutdown")  async def shutdown_event():      if mqtt_client is not None:          mqtt_client.loop_stop()          mqtt_client.disconnect()    # 运行FastAPI应用  if __name__ == "__main__":      import uvicorn      uvicorn.run(app, host="127.0.0.1", port=8000)   

import paho.mqtt.client as mqtt
# 连接成功回调def on_connect(client, userdata, flags, rc):    print('Connected with result code '+str(rc))    client.subscribe('devices/1')
# 消息接收回调def on_message(client, userdata, msg):    print(msg.topic+" "+str(msg.payload))    client.publish('gateway/device/1',payload=f'echo {msg.payload}',qos=0)
client = mqtt.Client()
# 指定回调函数client.on_connect = on_connectclient.on_message = on_message
# 建立连接client.connect('127.0.0.1', 1883)# 发布消息client.publish('gateway/device/1',payload='Hello, I am device',qos=0)
client.loop_forever()

import paho.mqtt.client as mqtt# 连接成功回调def on_connect(client, userdata, flags, rc):    print('Connected with result code '+str(rc))    client.subscribe('devices/2')# 消息接收回调def on_message(client, userdata, msg):    print(msg.topic+" "+str(msg.payload))    client.publish('gateway/device/2',payload=f'echo {msg.payload}',qos=0)client = mqtt.Client()# 指定回调函数client.on_connect = on_connectclient.on_message = on_message# 建立连接client.connect('127.0.0.1', 1883)# 发布消息client.publish('gateway/device/2',payload='Hello, I am device',qos=0)client.loop_forever()