本篇测评由电子工程世界的优秀测评者“JerryZhen”提供。

本文将介绍基于米尔电子MYD-LT527开发板的网关方案测试。

一、系统概述

基于米尔-全志 T527设计一个简易的物联网网关，该网关能够管理多台MQTT设备，通过MQTT协议对设备进行读写操作，同时提供HTTP接口，允许用户通过HTTP协议与网关进行交互，并对设备进行读写操作。

二、系统架构

• 网关服务：基于FastAPI框架构建的Web服务，提供HTTP接口。
• MQTT客户端：负责与MQTT设备通信，管理设备连接、消息发布和订阅。
• 设备管理：维护一个设备列表，记录设备的基本信息和状态。
  
• 数据存储：使用内存或数据库存储设备数据，确保数据持久化。
  
  

三、组件设计

• MQTT组件：
  
  • 负责与MQTT broker建立连接。
  • 订阅设备主题，接收设备发送的消息。
  • 发布消息到设备，实现远程控制。
    
• 设备管理组件：
  
  • 维护一个设备列表，记录设备的唯一标识符（如设备ID）、MQTT主题、连接状态等信息。
  • 提供设备增删改查的方法。
    
• HTTP组件：
  
  • 基于FastAPI定义HTTP接口。
  • 接收用户请求，调用MQTT组件和设备管理组件进行相应操作。
  • 返回操作结果给用户。
    

四、接口设计

• 设备列表：
  
  • GET /devices：返回所有设备的列表。
  • POST /devices：添加新设备到网关。
  • DELETE /devices/{device_id}：从网关中删除指定设备。
    
• 设备详情：
  
  • GET /devices/{device_id}：返回指定设备的详细信息。
    
• 设备数据：
  
  • GET /devices/{device_id}/data：获取指定设备的最新数据。
  • POST /devices/{device_id}/data：发送数据到指定设备。
    
• 设备控制：
  
  • POST /devices/{device_id}/control：发送控制命令到指定设备。
    

五、数据结构设计

• 设备信息：
  
  • 设备ID (device_id)：唯一标识设备的字符串。
  • MQTT主题 (mqtt_topic)：设备在MQTT broker上的主题。
  • 连接状态 (connection_status)：表示设备是否在线的布尔值。
  • 其他设备属性（如名称、描述等）。
    
• 设备数据：
  
  • 设备ID (device_id)：关联设备信息的设备ID。
  • 时间戳 (timestamp)：数据发送或接收的时间。
  • 数据内容 (data)：设备发送或接收的具体数据，可以是JSON格式或其他格式。
    

六、安全性考虑

• 使用HTTPS协议提供安全的HTTP通信。
• 实现用户认证和授权机制，确保只有授权用户可以访问和操作设备。
• 对于敏感操作（如删除设备），要求用户进行二次确认或提供额外的安全措施。
  

七、部署与扩展

• 使用Docker容器化部署网关服务，便于管理和扩展。
• 根据需要，可以水平扩展网关实例以处理更多的设备连接和请求。
  

八、实现步骤

• 安装所需的Python库：fastapi, uvicorn, paho-mqtt等。
• 创建FastAPI应用并定义路由。
• 实现MQTT组件，包括与MQTT broker的连接、订阅、发布等功能。
• 实现设备管理组件，维护设备列表并提供增删改查的方法。
• 实现HTTP组件，调用MQTT组件和设备管理组件处理用户请求。
• 编写测试代码，验证网关的各项功能是否正常工作。
• 部署网关服务并监控其运行状态。
  

该设计方案仅仅是概述，具体实现细节可能需要根据实际需求和项目环境进行调整和优化。在实际开发中，还需要考虑异常处理、日志记录、性能优化等方面的问题。基于上述设计方案，以下是一个简化版的参考代码，展示了如何使用FastAPI和paho-mqtt库来创建一个物联网网关。需要注意，示例中不包含完整的错误处理、用户认证和授权机制，这些在实际生产环境中都是必不可少的。依赖的主要库版本：

fastapi==0.108.0

paho-mqtt==1.6.1

网关模拟代码gateway.py：

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)

设备1模拟代码 dev1.py：

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()

设备2模拟代码 dev2.py

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()