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

发布时间:2024-5-11 18:08    发布者:swiftman
关键词: 全志 , T527 , 开发板 , 嵌入式 , 物联网
本篇测评由电子工程世界的优秀测评者“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:
  1. from fastapi import FastAPI, HTTPException, Body, status
  2. from paho.mqtt.client import Client as MQTTClient
  3. from typing import List, Dict, Any
  4. import asyncio
  5. import json

  6. app = FastAPI()
  7. mqtt_client = None
  8. device_data = {}  

  9. subtopic="gateway/device/#"

  10. # MQTT回调函数
  11. def on_message(client, userdata, msg):
  12.      payload = msg.payload.decode()
  13.      topic = msg.topic
  14.      device_id = topic.split('/')[-1]
  15.      device_data[device_id] = payload
  16.      print(f"Received message from {device_id}: {payload}")  
  17.      
  18. # MQTT连接和订阅
  19. def mqtt_connect_and_subscribe(broker_url, broker_port):
  20.       global mqtt_client
  21.       mqtt_client = MQTTClient()
  22.       mqtt_client.on_message = on_message
  23.       mqtt_client.connect(broker_url, broker_port, 60)
  24.       mqtt_client.subscribe(subtopic)
  25.       mqtt_client.loop_start()
  26.       
  27. # MQTT发布消息
  28. async def mqtt_publish(topic: str, message: str):
  29.       if mqtt_client is not None and mqtt_client.is_connected():
  30.          mqtt_client.publish(topic, message)
  31.       else:
  32.          print("MQTT client is not connected!")
  33.          
  34. # 设备管理:添加设备
  35. @app.post("/devices/", status_code=status.HTTP_201_CREATED)
  36. async def add_device(device_id: str):
  37.       device_data[device_id] = None
  38.       return {"message": f"Device {device_id} added"}
  39.       
  40. # 设备管理:获取设备列表
  41. @app.get("/devices/")
  42. async def get_devices():
  43.      return list(device_data.keys())
  44.      
  45. # 设备管理:获取设备数据
  46. @app.get("/devices/{device_id}/data")
  47. async def get_device_data(device_id: str):
  48.       if device_id not in device_data:
  49.           raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Device {device_id} not found")
  50.       return device_data.get(device_id)
  51.       
  52. # 设备管理:发送数据到设备
  53. @app.post("/devices/{device_id}/data")
  54. async def send_data_to_device(device_id: str, data: Dict[str, Any] = Body(...)):
  55.       topic = f"devices/{device_id}"
  56.       message = json.dumps(data)
  57.       await mqtt_publish(topic, message)
  58.       return {"message": f"Data sent to {device_id}"}
  59.       
  60. # 设备控制:发送控制命令到设备
  61. @app.post("/devices/{device_id}/control")
  62. async def control_device(device_id: str, command: str):
  63.       topic = f"devices/device/{device_id}"
  64.       await mqtt_publish(topic, command)
  65.       return {"message": f"Control command sent to {device_id}"}
  66.       
  67. # FastAPI启动事件
  68. @app.on_event("startup")
  69. async def startup_event():
  70.       mqtt_connect_and_subscribe("127.0.0.1", 1883)
  71.       
  72. # FastAPI关闭事件
  73. @app.on_event("shutdown")
  74. async def shutdown_event():
  75.       if mqtt_client is not None:
  76.          mqtt_client.loop_stop()
  77.          mqtt_client.disconnect()
  78.       
  79. # 运行FastAPI应用
  80. if __name__ == "__main__":
  81.       import uvicorn
  82.       uvicorn.run(app, host="127.0.0.1", port=8000)
复制代码

设备1模拟代码 dev1.py:
  1. import paho.mqtt.client as mqtt

  2. # 连接成功回调
  3. def on_connect(client, userdata, flags, rc):
  4.     print('Connected with result code '+str(rc))
  5.     client.subscribe('devices/1')

  6. # 消息接收回调
  7. def on_message(client, userdata, msg):
  8.     print(msg.topic+" "+str(msg.payload))
  9.     client.publish('gateway/device/1',payload=f'echo {msg.payload}',qos=0)
  10.    
  11. client = mqtt.Client()

  12. # 指定回调函数
  13. client.on_connect = on_connect
  14. client.on_message = on_message

  15. # 建立连接
  16. client.connect('127.0.0.1', 1883)
  17. # 发布消息
  18. client.publish('gateway/device/1',payload='Hello, I am device',qos=0)

  19. client.loop_forever()
复制代码

设备2模拟代码 dev2.py
  1. import paho.mqtt.client as mqtt

  2. # 连接成功回调
  3. def on_connect(client, userdata, flags, rc):
  4.     print('Connected with result code '+str(rc))
  5.     client.subscribe('devices/2')

  6. # 消息接收回调
  7. def on_message(client, userdata, msg):
  8.     print(msg.topic+" "+str(msg.payload))
  9.     client.publish('gateway/device/2',payload=f'echo {msg.payload}',qos=0)

  10. client = mqtt.Client()

  11. # 指定回调函数
  12. client.on_connect = on_connect
  13. client.on_message = on_message

  14. # 建立连接
  15. client.connect('127.0.0.1', 1883)
  16. # 发布消息
  17. client.publish('gateway/device/2',payload='Hello, I am device',qos=0)

  18. client.loop_forever()
复制代码

运行网关代码,打开网页得到api接口: 2.jpg 通过api分别添加设备1和设备2, 3.jpg 在另外两个控制台中分别运行模拟设备1和模拟设备2的代码通过网页API向设备1发送数据 4.jpg 通过网页API获得设备回复的数据,设备代码中只是简单的把网关发过来的数据进行回传 5.jpg 我们在网关的后台可以看到完整的数据流 6.jpg 至此一个简易的网关已经实现了,接下来将会尝试实现楼宇里的最常见的bacnet设备进行通讯管理。
本文地址:https://www.eechina.com/thread-856853-1-1.html     【打印本页】

本站部分文章为转载或网友发布,目的在于传递和分享信息,并不代表本网赞同其观点和对其真实性负责;文章版权归原作者及原出处所有,如涉及作品内容、版权和其它问题,我们将根据著作权人的要求,第一时间更正或删除。
您需要登录后才可以发表评论 登录 | 立即注册

厂商推荐

  • Microchip视频专区
  • PIC64GX 64位四核MPU
  • 全新32位dsPIC33A DSC的主要特性和功能介绍
  • 新产品!PolarFire® SoC Discovery工具包——探索RISC-V®和FPGA技术的低成本方案
  • 10BASE-T1S以太网的应用开发培训教程
  • 贸泽电子(Mouser)专区

相关视频

关于我们  -  服务条款  -  使用指南  -  站点地图  -  友情链接  -  联系我们
电子工程网 © 版权所有   京ICP备16069177号 | 京公网安备11010502021702
快速回复 返回顶部 返回列表