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

from fastapi import FastAPI, HTTPException, Body, status
from paho.mqtt.client import Client as MQTTClient
from 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_connect
client.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_connect
client.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()