加入了学习《【DigiKey创意大赛】多通道微型气相色谱采集单元》，观看 多通道微型气相色谱采集单元

发表了主题帖： 【米尔-全志 T527 开发板-试用评测】第五篇 项目后记

在项目开发过程中，有几个有意思的小知识点分享给大家：   1，如何在米尔原生的安卓系统中编译python第三方库？ 本项目采用的是termux，可以在安卓系统中模拟linux环境。如果是纯python库，只需要使用pip安装即可。但是python很多高性能库使用的是rust编写的，比如pydantic, orjson之类的，而termux默认的是使用clang编译器，需要特殊设置。 本项目使用fastapi作为网络框架，它依赖了一堆rust写的库，在安装时需要需设置rust环境，运行: pkg install rust   pkg install binutils 编译库的时候要花点时间，耐心等待。   2. 软件框架fastapi+celery+redis 在使用 FastAPI编写后端的过程中，数据采集任务可以使用 Celery 通过消息任务队列的方式进行异步实现，从而提升应用的整体性能。采集和显示是两个异步任务。采集任务采集到的结果放到redis中，页面请求数据去redis中读取，这样显示和采集解耦，提升了响应速度。   3. 性能升级 尽管我们通过软件框架其实已经可以保证了产品的性能，但是如果想要更高的性能，可以考虑使用golang或者C语言这些静态语言来实现，不过这样开发效率可能会稍微慢些。个人认为在国产芯崛起，硬件越来越便宜的情况下，开发速度至关重要，pyhon的生态注定它的开发效率实在是太逆天了，另外有个开源的工业网关thingsboard gateway就是用python实现，在绝大多数的应用场景下绰绰有余。   最后祝米尔和全志的板卡大卖，后续能提供更多物美价廉的开发板。    

发表了主题帖： 【米尔-全志 T527 开发板-试用评测】第四篇 楼控网关展示平台

本帖最后由 JerryZhen 于 2024-6-5 14:17 编辑 最近一直在忙着设计楼控网关的展示箱，用于展会宣传，现在不同以往，好酒也怕巷子深，卖点宣传很重要。从设计到硬件到软件等一系列工作，其实很繁琐的，经过一个月的奋战，现在已经基本完工，先上张图和视频，展示下效果。     [localvideo]f8042271f3c205391deb54aecec5bd13[/localvideo]   项目硬件部分设计思路如下，T527作为数据采集网关，显示部分通过HDMI接显示屏，数据采集部分通过bacnet协议读取楼控DDC设备中的逻辑点位信息，进而将网络中的设备状态在屏幕上进行可视化。项目的原理图如下。可以看到，我们是通过DDC控制两个电机，两个led灯，以及两个温湿度传感器。主控T527会每秒去采集一次DDC中的点位信息，然后将采集到的数据可视化到屏幕上。   项目的软件部分：软件的核心就是数据采集，具体代码如下： import BAC0 from app.services.cache import redis_client points = { "led1": { "read":'192.168.20.51:47808 binaryOutput 501 presentValue', "default":"inactive", }, "led2":{ "read": '192.168.20.51:47808 binaryOutput 502 presentValue', "default":"inactive", }, "motor1": { "read": '192.168.20.108:47808 binaryOutput 501 presentValue', "default":"inactive", }, "motor2": { "read": '192.168.20.108:47808 binaryOutput 502 presentValue', "default":"inactive", } , "temperature": { "read": '192.168.20.108:47808 analogInput 301 presentValue', "default": "0.0" }, "humidity": { "read": '192.168.20.108:47808 analogInput 302 presentValue', "default": "0.0" }, # "switch1": '192.168.20.51:47808 binaryInput 201 presentValue', # "switch2": '192.168.20.51:47808 binaryInput 202 presentValue', # "switch3": '192.168.20.108:47808 binaryInput 201 presentValue', # "switch4": '192.168.20.108:47808 binaryInput 202 presentValue', } class BacnetClient: def __init__(self, host:str): self.conn = BAC0.connect(host) def close(self): self.conn.disconnect() def read_points(self): for point, pointinfo in points.items(): try: point_value = self.conn.read(pointinfo["read"]) if point in ["temperature", "humidity"]: point_value = '%.1f'%point_value redis_client.set(point, point_value) except Exception as e: print("read error", e) bacnet_client = BacnetClient("192.168.20.65/24") if __name__ == "__main__": ret = bacnet_client.read_points() print(ret) bacnet_client.close()   总结：米尔-全志 T527，配备八核A55高性能处理器，RISC-V协处理器，支持2Tops NPU，拥有丰富的通讯接口，功能强大。非常适合工控行业，本次采用米尔的这款板卡用于原型设计，由于米尔配套的资料齐全，大大缩短了项目周期，为米尔和全志芯点赞。  

发表了主题帖： 【米尔-全志 T527 开发板-试用评测】-第三篇 网关方案

本帖最后由 JerryZhen 于 2024-4-27 20:17 编辑 一、系统概述 基于米尔-全志 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()   运行网关代码，打开网页得到api接口：  通过api分别添加设备1和设备2，     在另外两个控制台中分别运行模拟设备1和模拟设备2的代码 通过网页API向设备1发送数据   通过网页API获得设备回复的数据，设备代码中只是简单的把网关发过来的数据进行回传   我们在网关的后台可以看到完整的数据流   至此一个简易的网关已经实现了，接下来将会尝试实现楼宇里的最常见的bacnet设备进行通讯管理。  

发表了主题帖： 【米尔-全志 T527 开发板-试用评测】-第二篇 网关协议管理功能的实现

在第一篇中在米尔T527开发板自带的安卓系统上搭建了linux开发环境，并测试了mqtt协议。本篇我们进一步实现一个简易的协议管理器，这对于边缘网关来说很重要。因为在边缘网关中通常需要处理各种常见的物联网通讯协议，一个优秀的协议管理器对项目成败至关重要。因为用来做产品原型测试，所以我们继续用python作为开发语言，原因无他，主要是因为生态好，库多，上手快。为了简单起见，接下来我们主要展示如何同时管理socket 和 MQTT，并根据不同的协议进行数据的收发。   因为要管理多种协议，我们自然想到使用工厂模式实现 ProtocolManager 类，从而可以使代码更加灵活和可扩展。工厂模式允许我们根据传入的参数动态地创建和返回协议实例，而不需要在 ProtocolManager 中硬编码所有协议的具体实现。 class Protocol: def send(self, data): raise NotImplementedError("Protocol must implement send method") def receive(self): raise NotImplementedError("Protocol must implement receive method") class TCPProtocol(Protocol): def __init__(self, host, port): self.host = host self.port = port # 初始化 TCP 连接等 def send(self, data): # 实现 TCP 发送逻辑 pass def receive(self): # 实现 TCP 接收逻辑 pass class MQTTProtocol(Protocol): def __init__(self, broker_address): self.broker_address = broker_address # 初始化 MQTT 连接等 def send(self, data): # 实现 MQTT 发送逻辑 pass def receive(self): # 实现 MQTT 接收逻辑 pass class ProtocolFactory: @staticmethod def create_protocol(protocol_type, *args, **kwargs): if protocol_type == 'tcp': return TCPProtocol(*args, **kwargs) elif protocol_type == 'mqtt': return MQTTProtocol(*args, **kwargs) else: raise ValueError(f"Unknown protocol type: {protocol_type}") class ProtocolManager: def __init__(self): self.protocols = {} def connect(self, protocol_type, *args, **kwargs): protocol_instance = ProtocolFactory.create_protocol(protocol_type, *args, **kwargs) self.protocols[protocol_type] = protocol_instance def send_data(self, protocol_type, data): if protocol_type in self.protocols: protocol_instance = self.protocols[protocol_type] protocol_instance.send(data) else: print(f"No connection established for protocol: {protocol_type}") def receive_data(self, protocol_type): if protocol_type in self.protocols: protocol_instance = self.protocols[protocol_type] return protocol_instance.receive() else: print(f"No connection established for protocol: {protocol_type}") return None # 使用示例 manager = ProtocolManager() # 连接 TCP 和 MQTT manager.connect('tcp', 'localhost', 12345) manager.connect('mqtt', 'localhost') # 发送数据 manager.send_data('tcp', b'Hello, TCP!') manager.send_data('mqtt', 'Hello, MQTT!') # 接收数据（这里只是示例，实际使用时可能需要异步接收或轮询） tcp_data = manager.receive_data('tcp') mqtt_data = manager.receive_data('mqtt') print(f"Received TCP data: {tcp_data}") print(f"Received MQTT data: {mqtt_data}") 在上面的示例中，我们定义了 Protocol 基类，并创建了两个继承自 Protocol 的具体协议类 TCPProtocol 和 MQTTProtocol。ProtocolFactory 类是一个静态工厂，负责根据传入的协议类型创建相应的协议实例。ProtocolManager 类使用 connect 方法通过工厂创建协议实例，并将其存储在 protocols 字典中。send_data 和 receive_data 方法现在根据协议类型从字典中获取协议实例，并调用其 send 和 receive 方法。这种方式使得添加新的协议变得非常简单，只需要定义一个新的协议类，并在 ProtocolFactory 中添加相应的创建逻辑即可，而不需要修改 ProtocolManager 类的代码。   但是在实际应用中，接收数据通常涉及到监听数据流的异步事件或回调，并且通常使用队列或缓存来存储接收到的数据，直到它们被处理。python中可以使用asyncio库和queue模块来实现异步队列， 我们将上述接收数据部分改写为异步缓存处理，修改后的代码如下： import asyncio import queue class Protocol: async def receive(self): raise NotImplementedError async def send_data(self, data): raise NotImplementedError class AsyncProtocol(Protocol): def __init__(self): self.receive_queue = asyncio.Queue() async def receive(self): while True: data = await self._receive_data() await self.receive_queue.put(data) async def _receive_data(self): # 这里模拟从网络或其他IO源接收数据 await asyncio.sleep(1) return f"Received data from {self.__class__.__name__}" async def get_received_data(self): return await self.receive_queue.get() async def send_data(self, data): # 这里应该实现具体的发送逻辑 # 暂时打印数据作为模拟 print(f"Sending data to {self.__class__.__name__}: {data}") class TCPProtocol(AsyncProtocol): pass class MQTTProtocol(AsyncProtocol): pass class ProtocolFactory: @staticmethod async def create_protocol(protocol_type): if protocol_type == 'tcp': return TCPProtocol() elif protocol_type == 'mqtt': return MQTTProtocol() else: raise ValueError(f"Unknown protocol type: {protocol_type}") class ProtocolManager: def __init__(self): self.protocols = {} async def connect(self, protocol_type): protocol_instance = await ProtocolFactory.create_protocol(protocol_type) self.protocols[protocol_type] = protocol_instance asyncio.create_task(protocol_instance.receive()) async def send_data(self, protocol_type, data): if protocol_type in self.protocols: await self.protocols[protocol_type].send_data(data) else: print(f"No connection established for protocol: {protocol_type}") async def receive_data(self, protocol_type): if protocol_type in self.protocols: protocol_instance = self.protocols[protocol_type] return await protocol_instance.get_received_data() else: print(f"No connection established for protocol: {protocol_type}") return None # 使用示例 async def main(): manager = ProtocolManager() await manager.connect('tcp') await manager.connect('mqtt') # 发送数据 await manager.send_data('tcp', 'Hello TCP!') await manager.send_data('mqtt', 'Hello MQTT!') # 接收数据 tcp_data = await manager.receive_data('tcp') mqtt_data = await manager.receive_data('mqtt') print(f"TCP received: {tcp_data}") print(f"MQTT received: {mqtt_data}") # 运行事件循环 asyncio.run(main()) 修改后的代码中，send_data方法现在只是简单地打印出要发送的数据和协议类型，以模拟发送逻辑。在实际应用中，需要在这些方法中加入真正的网络发送和接收逻辑。 运行代码，结果如下：   至此，一个简易的协议管理框架已经大家完毕，下一篇会基于该框架进行更多的物联网协议测试，以实现网关基础的收发功能。  

发表了主题帖： 【米尔-全志 T527 开发板-试用评测】-第一篇 小试牛刀之MQTT

本帖最后由 JerryZhen 于 2024-3-17 08:57 编辑 米尔开发套件 MYD-LT527由核心板 MYC-LT527 和底板MYB-LT527 组成，核心板与底板采用 LGA贴片焊接方式。随同开发套件 MYIR    提供了软件资源以及文档资料。由于全志科技 T527 系列高性能处理器是一款基于八核 Cortex-A55 + HiFi4 DSP+RISC-V 多核异构工业级处理器，因此性能肯定没得说。笔者很幸运能够在第一时间拿到板子。开发板的硬件参数以及外观可以参考官网，这里不作赘述。放上两张实物照片让大家饱饱眼福，做工很扎实，外观漂亮。   板子出厂系统是安卓 13，采用 AOSP 版本项目进行构建，包含完整的硬件驱动，常用的系统工具，调试工具等。支持使用 Java 进行应用开发。至于linux系统，官方表示稍后会提供 ubuntu22.04 以及buildroot系统。   安卓系统现在在车机之类的行业中应用很广泛。但是由于笔者主要从事工控，因此还是更习惯用linux开发。虽然现在官方暂未提供linux系统，但是问题不大，我们可以通过termux在安卓上实现类似linux虚拟机。Termux是一个专为安卓设备设计的开源Linux环境和命令行工具。通过Termux，用户可以在安卓设备上运行各种Linux命令和软件包，用来进行编程开发、网络安全测试、数据分析等各种任务。 Termux具有以下特点和功能： 1. 完全开源：Termux的所有代码都是开源的，用户可以自由地修改和定制。 2. 支持包管理：Termux内置了包管理工具，用户可以通过它来安装、更新和删除各种软件包。 3. 支持终端模拟器：Termux内置了功能强大的终端模拟器，用户可以在其中执行各种命令。 4. 支持插件：Termux还支持通过插件扩展功能，用户可以根据需要安装各种插件。 5. 社区活跃：Termux有一个激活的用户社区，用户可以在社区中获取帮助、分享经验和交流。 总的来说，Termux是一个强大的工具，可以让用户在安卓设备上轻松地进行各种Linux操作和开发任务。   下面笔者简要介绍下，如何在MYD-LT527的安卓系统上搭建Termux开发环境，并实现一个简单的MQTT通信demo。 首先安装两个apk(Termux, Termux:Boot)，然后打开Termux.   pkg换源:  termux-change-repo 升级系统包：  pkg update -y 获取开发板存储权限： termux-setup-storage 安装openssh： pkg install -y openssh 启动ssh服务：sshd 设置密码：passwd 查看当前用户名：whoami 查看当前IP地址： ifconfig   现在可以在电脑上通过SSH远程登录开发板并进行各种操作，大大提高了开发板的可玩性。 现在我们通过python在开发板上实现mqtt收发数据为例进行讲解。 在系统中安装pyhon：pkg install python 创建项目文件夹：mkdir mqtt 进入项目目录：cd mqtt 创建python环境: python -m venv .venv 激活环境：source .venv/bin/activate 安装mqtt库： pip install paho-mqtt     然后分别实现发送客户端pub.py: import time import paho.mqtt.client as mqtt broker = 'broker.emqx.io' port = 1883 topic = "/Python/mqtt" mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2) mqttc.connect(broker, port) mqttc.loop_start() msg_count = 0 while True: time.sleep(1) msg = f"messages: {msg_count}" msg_info = mqttc.publish(topic, msg) print(f"Send `{msg}` to topic `{topic}`") msg_count += 1 mqttc.disconnect() mqttc.loop_stop() 再实现接收客户端sub.py: import paho.mqtt.client as mqtt broker = 'broker.emqx.io' port = 1883 topic = "/Python/mqtt" def on_message(client, userdata, message): print(f"Received `{message.payload.decode()}` from `{message.topic}` topic") def on_connect(client, userdata, flags, reason_code, properties): if reason_code.is_failure: print(f"Failed to connect: {reason_code}. loop_forever() will retry connection") else: client.subscribe(topic) mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2) mqttc.on_connect = on_connect mqttc.on_message = on_message mqttc.connect(broker, port) mqttc.loop_forever()   代码中broker选择公网的emqx官方服务器，配置如下: broker = 'broker.emqx.io' port = 1883 topic = "/Python/mqtt"   然后在两个窗口中分别运行发送客户端和接收客户端, 并观察结果。   至此第一篇分享结束，我们已经成功的在米尔T527开发板实现了一个简单的mqtt通讯应用。后面章节我们会基于T527做工业边缘网关，实现更强大的功能。  

回复了主题帖： 测评入围名单：米尔基于全志T527开发板

个人信息无误，确认可以完成评测计划。