本篇测评由电子工程世界的优秀测评者“JerryZhen”提供。 本文将介绍基于米尔电子MYD-LT527开发板的网关方案测试。
一、系统概述 基于米尔-全志 T527设计一个简易的物联网网关,该网关能够管理多台MQTT设备,通过MQTT协议对设备进行读写操作,同时提供HTTP接口,允许用户通过HTTP协议与网关进行交互,并对设备进行读写操作。 二、系统架构 三、组件设计 MQTT组件:
- 负责与MQTT broker建立连接。
- 订阅设备主题,接收设备发送的消息。
发布消息到设备,实现远程控制。
设备管理组件:
- 维护一个设备列表,记录设备的唯一标识符(如设备ID)、MQTT主题、连接状态等信息。
提供设备增删改查的方法。
HTTP组件:
- 基于FastAPI定义HTTP接口。
- 接收用户请求,调用MQTT组件和设备管理组件进行相应操作。
返回操作结果给用户。
四、接口设计 五、数据结构设计 设备信息:
- 设备ID (device_id):唯一标识设备的字符串。
- MQTT主题 (mqtt_topic):设备在MQTT broker上的主题。
- 连接状态 (connection_status):表示设备是否在线的布尔值。
其他设备属性(如名称、描述等)。
设备数据:
六、安全性考虑 七、部署与扩展 八、实现步骤 - 安装所需的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()
复制代码
运行网关代码,打开网页得到api接口:
通过api分别添加设备1和设备2,
在另外两个控制台中分别运行模拟设备1和模拟设备2的代码通过网页API向设备1发送数据
通过网页API获得设备回复的数据,设备代码中只是简单的把网关发过来的数据进行回传
我们在网关的后台可以看到完整的数据流
至此一个简易的网关已经实现了,接下来将会尝试实现楼宇里的最常见的bacnet设备进行通讯管理。
|
网友评论