
前阵子在朋友的葡萄棚里调树莓派温湿度采集脚本,他说之前网上抄的代码跑起来要么漏数据要么资源占用高,翻了下代码,发现他上来就套了个很重的工业级网关框架,完全大材小用。
直接手写串口轮询逻辑
这是上手门槛最低的写法,依赖极少,步骤也很直白:
- 先安装串口操作依赖,执行命令:
pip install pyserial - 确认硬件接线无误后,替换代码里的串口号、波特率,写循环读取数据即可,最小可运行代码如下:
import serial import time # Linux下串口号一般是/dev/ttyUSB0,Windows下多为COM3、COM4这类 ser = serial.Serial('/dev/ttyUSB0', 9600, timeout=2) while True: raw = ser.readline().decode('utf-8', errors='ignore').strip() if raw: print(f"采集到数值: {raw}") # 这里可以追加写文件、推送到云端的逻辑 time.sleep(10) # 采样间隔按实际需求调整
要注意的是,这种写法千万别在循环里塞耗时太长的同步操作,比如单次网络请求卡个一两秒,很容易把串口缓存挤爆丢数据,小项目用着最省心。
这种方案的好处是代码总共十几行,出问题一眼就能定位,适合单传感器、采样间隔在十秒以上的场景,比如家用温湿度监测、小型种植棚的环境记录;缺点是接入的传感器多了之后,轮询等待的时间会叠加,实时性会打折扣。
用异步IO做事件驱动采集
如果接入的传感器数量多,采样频率又高,比如需要秒级甚至更快的采样频率,就可以考虑异步的写法,不会被单步操作阻塞采集流程:
- 先安装异步串口依赖:
pip install pyserial-asyncio - 基于协议回调写读取逻辑,不用自己写空循环等数据:
import asyncio import serial_asyncio class SensorReader(asyncio.Protocol): def data_received(self, data): msg = data.decode('utf-8', errors='ignore').strip() print(f"异步收到传感器数据: {msg}") async def main(): loop = asyncio.get_running_loop() # 替换为自己的串口号和波特率 transport, _ = await serial_asyncio.create_serial_connection( loop, SensorReader, '/dev/ttyUSB0', baudrate=9600 ) await asyncio.Event().wait() # 保持持续运行 asyncio.run(main())
这种写法数据到达串口就会触发回调,不用空等轮询,接多个传感器也不容易卡,缺点是如果之前没接触过异步Python,很容易不小心在回调里写同步阻塞代码,直接把整个事件循环卡停,调试起来比同步写法麻烦点,适合点位多、实时性要求高的场景,比如小型机房的多参数监测。
对接硬件厂商的采集SDK
如果用的是市面上现成的商业采集网关,就没必要自己搓底层读写逻辑了,厂家一般都封装好了现成的Python SDK,操作步骤很简单:
- 按照硬件文档给出的源安装对应SDK,一般就是pip加厂家提供的包名
- 初始化的时候填上网关的局域网IP和设备密钥,直接调用接口读取数据即可,核心代码也就几行:
from commercial_iot_gw import Device gw = Device(local_ip="192.168.1.20", access_key="你的设备密钥") all_sensor_data = gw.read_all_channels() print(all_sensor_data)
这种写法最省时间,厂家一般都把Modbus协议解析、断网缓存这些通用逻辑做好了,不用自己踩硬件兼容的坑,缺点是灵活性差,想加个自定义的采集逻辑经常受SDK限制,而且绑定对应硬件,适合要快速交付的商用项目,自己玩DIY的话性价比不高。
反正选的时候不用盲目追最复杂的架构,够当前用,后续改起来不费劲,就比啥都强。
评论一下?