做物联网开发这几年,我最大的感受是:Python在这个领域的地位很微妙——很多人觉得它慢、不适合底层嵌入式,可真正到了接设备、写网关协议、做云端数据处理的时候,又离不开它。尤其当你从零搭建一套IoT系统,从边缘侧的数据采集,到边缘网关的预处理与转发,再到云端接收、清洗、存储和分析展示,走完这条链路,你会发现Python几乎是唯一一门能从端到云“一套语言打通”的工具。这篇文章不是一个概念科普,而是结合我实际做过的仓库环境监测、设备振动监控这类项目,把Python在IoT里的核心应用拆开讲清楚:边缘计算环节怎么用Python处理数据、云端怎么接MQTT消息、数据怎么落地、聚合和告警怎么做。如果你正准备用Python入门物联网,或者已经在做设备接入但被数据链路搞得焦头烂额,这篇文章应该能给你一条可以直接照着走的路径。
1. 项目整体设计:看清从传感器到云端这条链路的全貌
1.1 一条完整的物联网数据链路到底分几层
很多新手拿到一块ESP32开发板,想着把传感器数据发到云端,第一反应是“直接POST到服务器”,实际项目里这种方式几乎不可行。真实物联网系统的数据链路,至少分成四层:感知层、边缘层、传输层和云端处理层。
感知层就是传感器本身,比如温湿度传感器、震动传感器、电流互感器,它们的输出形式各不相同,有的是I2C数字信号,有的是4-20mA模拟量,有的走Modbus RTU协议。边缘层负责跟传感器“打交道”,把物理世界的信号变成结构化数据,同时做第一道预处理,比如滤波、单位换算、异常值剔除。传输层解决的是“数据怎么到云端”的问题,物联网场景里90%以上会用MQTT协议,因为它的发布订阅模型非常适合大量低功耗设备频繁上报小数据包。云端处理层则是数据真正产生价值的环节,消息接入之后要落库、清洗、聚合、存时序数据库、触发告警,最后通过API或可视化面板展示。
这一层架构里,边缘层和云端处理层就是Python最活跃的阵地。你用MicroPython嵌入式代码在ESP32上读传感器,用完整版Python在树莓派或工控机上做网关协议转换,再用Python写云端订阅服务和数据处理脚本,整个数据管道都是同一门语言,团队协作、代码维护都会轻松很多。
1.2 为什么选Python:边缘与云端两侧的居中优势
说Python不适合做IoT的人,大多是站在“嵌入式实时控制”的角度看问题。坦白讲,如果产品需要微秒级响应的电机控制、需要跑在几KB内存的MCU上,Python确实不合适,那是C和汇编的主场。但IoT项目的工程量并不只在那一小段底层控制里,更多的工作量在设备接入、协议解析、数据上传、云端存储分析这些环节。
举个例子,你在仓库里装了一套温湿度监测系统,一个网关要接二三十个MODBUS传感器节点。用C写这个网关,Modbus CRC校验、帧解析、异常重试、断线重连、JSON组包,一套下来少说几千行代码,而且改一个协议就要编译一次。用Python的话,pymodbus库就能处理Modbus协议,paho-mqtt用来转发MQTT消息,再配合一个轻量的规则引擎,维护成本低得多。而且边缘网关用的是Linux系统,Python部署起来就像在服务器上一样自然。
云端就更不用说了。数据到了云端,要做清洗、聚合、异常检测、数据可视化,这些刚好是Python生态最强的地方。Pandas做数据处理,InfluxDB客户端写时序数据,FastAPI暴露查询接口,一套组合拳下来,从边缘采集到云端展示的所有环节都能覆盖。这也是为什么Python在IoT领域的位置越来越靠前——它未必是最快的那把刀,但它是你从厨房走到餐桌时一直能攥在手里的那把刀。
1.3 方案选型避坑:先别急着写代码
接手一个新的IoT项目,我建议你至少先花一个小时,把下面几个问题梳理清楚,再做技术选型。不然代码写到一半,返工的成本会让你非常难受。
第一是设备的离线策略。边缘设备断网是常态,不是例外。数据是直接丢弃,还是先缓存在本地、等网络恢复了再补传?如果补传,补传的数据跟实时数据怎么区分?我通常的做法是,边缘网关用SQLite做本地环形缓冲,数据带时间戳,云端根据“数据时间”和“到达时间”两个字段区分历史补传与实时上报,这样补传数据不会污染实时统计。
第二是数据量级的预判。一套系统接入10个设备和接入1万个设备,架构完全不同。10个设备一台云服务器加一个SQLite可能就够了,但上万设备就必须上MQTT集群、时序数据库分片、边缘聚合降低上报频率。很多项目就是因为前期没想清楚接入规模,后期全部重推。
第三是时间同步问题。边缘设备、网关、云端服务器的时钟很可能不一致。数据到了云端做时序分析,时间戳到底以谁为准?我的习惯是设备端只上报传感器读数,时间戳统一在网关层打,网关用NTP校时,这样能保证同一条数据链路里时间口径一致。
2. 边缘计算落地:Python在设备端的三种玩法
2.1 玩法一:单片机上的MicroPython——用Python直接点亮传感器
很多人的入门路径是从MicroPython开始,我也一样。烧录了MicroPython固件的ESP32,可以直接在REPL里写代码控制GPIO、读取传感器,开发效率比C语言高好几个量级。对于温湿度采集、继电器控制这类低速IO场景,MicroPython的性能完全够用。
下面是一个典型的ESP32 + DHT22温湿度采集代码:
from machine import Pin import dht import time import json import network from umqtt.simple import MQTTClient # 初始化DHT22传感器,数据引脚接GPIO4 sensor = dht.DHT22(Pin(4)) # 连接WiFi wlan = network.WLAN(network.STA_IF) wlan.active(True) wlan.connect("your_ssid", "your_password") while not wlan.isconnected(): time.sleep(0.5) # MQTT连接云端 client = MQTTClient("warehouse_node_01", "192.168.1.100", 1883) client.connect() while True: sensor.measure() payload = json.dumps({ "device_id": "wh_node_01", "temperature": sensor.temperature(), "humidity": sensor.humidity(), "ts": time.time() }) client.publish("warehouse/env/node01", payload) time.sleep(30)这段代码核心就干了三件事:读传感器、转JSON、发布MQTT消息。这里需要特别说的一个坑是,dht.DHT22读数据是有要求的,两次measure()之间间隔最好大于2秒,否则容易读到无效数据。DHT22本身响应慢,你在循环里如果加了其他耗时操作,比如网络重连逻辑,很容易把采样周期拉长,所以建议把“读取传感器”和“上报数据”拆成两个协程或者两个线程,读数据保持固定节奏,上报可以积攒几条再批量发。
另外,MicroPython的内存非常紧张,ESP32只有大约320KB的RAM。JSON组包、字符串拼接这类操作要特别小心,频繁创建大字符串会导致内存碎片化。我的习惯是能直接构造bytes就不生成中间字符串,能定长分配就不动态生长,不然跑几天之后设备会莫名重启,那多半就是内存炸了。
2.2 玩法二:边缘网关——最值得花时间的一环
如果说传感器端是“能跑就行”,网关端就是整个系统里最值得花时间打磨的一环。网关的位置正好卡在设备与云端的中间,它往下要兼容各种乱七八糟的传感器协议,往上要跟云端稳定通信,同时还承担着本地数据预处理和规则判断的任务。它的稳定性直接决定了整条数据链路的质量。
我现在的标准做法是:边缘网关用树莓派或工控机,装精简版Linux,跑一个Python常驻服务。服务内部用多线程加队列的方式组织:一个线程负责轮询Modbus节点,一个线程做数据预处理,一个线程负责推送MQTT,线程之间用queue.Queue通信,避免互相阻塞。代码如下:
import threading import queue import json import time from pymodbus.client import ModbusSerialClient import paho.mqtt.client as mqtt data_queue = queue.Queue(maxsize=1000) def read_modbus_loop(): client = ModbusSerialClient( port='/dev/ttyUSB0', baudrate=9600, timeout=2 ) client.connect() while True: try: # 读取1号设备的保持寄存器,起始地址0,读10个寄存器 resp = client.read_holding_registers(0, 10, slave=1) if not resp.isError(): values = resp.registers payload = { "device_id": "sensor_01", "values": values, "ts": int(time.time() * 1000) } data_queue.put(payload) except Exception as e: print(f"Modbus read error: {e}") time.sleep(2) def process_and_publish_loop(): mqtt_client = mqtt.Client() mqtt_client.connect("your-cloud-broker", 1883, 60) mqtt_client.loop_start() while True: item = data_queue.get() # 这里可以加本地规则判断,比如温度超过阈值直接推告警 if item["values"][0] > 800: mqtt_client.publish("alerts/temperature", json.dumps(item)) mqtt_client.publish("warehouse/env/gateway01", json.dumps(item)) threading.Thread(target=read_modbus_loop, daemon=True).start() threading.Thread(target=process_and_publish_loop, daemon=True).start() time.sleep(60 * 60 * 24)这里有几个细节:queue.Queue(maxsize=1000)是背压机制,如果云端网络抖动、MQTT发送不及时,队列积压到1000条就会阻塞读取线程,避免内存无限制增长;MQTT客户端的loop_start()放在单独线程里,保证了网络收发的响应性;每个payload都带毫秒级时间戳ts,这是后续云端做时序分析的关键字段,一定要在网关层统一打点,不要在设备端打。
网关层还承担着“本地规则引擎”的工作,这可能是被不少人忽略的。有些告警必须在边缘就地触发,比如温度超过80度就要声光报警,如果等数据上传云端再下发指令,一上一下可能已经过去好几秒,现场早就出了状况。在网关里用Python写几行简单的阈值判断,毫秒级就能触发本地IO。等到云端告警到达时,边缘侧的应急响应已经做完了。
2.3 玩法三:协议适配与边缘缓存
网关面对的现实是:传感器协议五花八门,有Modbus RTU的,有Modbus TCP的,有BACnet的,还有些私有协议,甚至有的设备只提供串口调试输出,要自己去解析ASCII报文。Python的优势在于第三方库丰富,pymodbus、bacpypes这些库能省掉大量底层协议代码。如果遇到没有现成库的私有协议,用Python写一个串口解析器也不难,pyserial配合struct模块就能把裸报文解析成结构化数据。
但协议适配还不能只看“能不能读到数”,更要考虑“读不到数的时候怎么办”。工业现场总会有设备掉线、线路故障、寄存器返回异常的情况。网关必须有异常处理机制:连续读取失败多少次判定设备离线,离线后是否重试、重试间隔多少,这些都要有清晰的策略。
我再强调一下边缘缓存的重要性。我之前做过一个项目,云端部署在阿里云上,而现场是西部一个偏远厂区,网络质量很差,动不动就丢包断线。一开始没有做缓存,网络一断数据就全丢,客户根本没法接受。后来在网关里加了一层SQLite缓存,数据先写本地,等MQTT恢复连接后按时间顺序补传,这才解决了问题。SQLite在嵌入式环境里跑得非常稳,单文件存储,不需要额外服务,天然适合做边缘缓存。
3. 云端数据处理:从MQTT消息到可查询的数据资产
3.1 消息接入层:MQTT Broker选型与主题设计
设备数据经过网关转发,最终汇聚到MQTT Broker。Broker的选型直接影响系统的承载能力和稳定性,目前主流选择是EMQX和Mosquitto。我的经验是:设备量在百台以内,Mosquitto完全够用,轻量、内存占用小,部署在2核4G的云主机上毫无压力;设备量上千甚至上万,或者需要规则引擎、数据桥接这些高级功能,直接用EMQX,它的集群能力和内置的Web管理界面会省很多事,而且EMQX有开源版,功能对大多数项目已经够用了。
主题设计是很多人前期不重视、后期改起来很痛苦的事。我建议从一开始就按“项目/地点/设备类型/设备ID/数据类型”这样的层级来设计主题,例如:
warehouse/site01/env/node01/data warehouse/site01/env/node01/status warehouse/site01/alarm/temperature主题的设计原则是:把可变的设备ID放在尾部,便于Broker做通配符订阅;把数据类型区分开,数据流和状态流不要混在一起;告警主题单独建层级,这样告警订阅方不需要订阅全量数据。主题结构清晰之后,云端做数据分流、权限控制、数据归档都会方便很多。
3.2 数据落地:时序数据库的选择与建模
MQTT消息到了云端,最终要落库才能查询分析。物联网设备上报的数据有几个明显特征:持续写入、时间有序、极少更新、按时间维度聚合查询,这些特征跟传统关系型数据库完全不对付,所以你要用专门为这种场景设计的时序数据库。目前生态比较成熟的时序库有InfluxDB、TDengine、TimescaleDB,其中InfluxDB用户群体最大、文档和示例最多,TDengine则在国产化场景里越来越流行,单机性能更强。
我个人用的最多的是InfluxDB 2.x,它的数据模型是“Measurement + Tag + Field”。Measurement是表名,Tag是索引字段,Field是实际数值。比如温湿度数据建模:
measurement: env_sensor tags: device_id, site_id fields: temperature(float), humidity(float) timestamp: 自动带上这里要特别注意Tag和Field的选择,Tag是字符串且需要建索引,适合放设备ID、站点ID这类低基数数据;Field是数值,放实际传感器读数。Tag设计不好会导致索引膨胀,查询速度下降,比如把温度值本身设成Tag就是灾难级设计。建表之前想清楚哪些字段需要“过滤”和“分组”,哪些字段只是“查出来展示”,前者放Tag,后者放Field。
3.3 数据处理管道:清洗、聚合与告警
数据落库之后,还远没到“能用”的程度。原始数据是很脏的:传感器偶发跳变、网络延迟导致乱序到达、边缘补传导致某段时间数据密度异常。云端必须做一套数据处理管道。我的管道分三段:清洗、聚合、告警。
清洗阶段做的是剔异常值。常见的方法是用滚动窗口的中位数或均值做基线,如果某个数据点偏离基线超过N倍标准差,就认为它是异常点,打上标签但不直接删除。为什么不直接删?因为异常值可能是真有故障,直接误删了故障证据,后续排查问题时会非常被动。我的做法是给每一条数据加一个quality字段,正常是good,可疑是suspect,分析时可以通过字段过滤。
聚合阶段把原始数据压缩成不同粒度的统计值。比如原始数据每10秒上报一次,一小时就是360条,一年下来数据量非常庞大,长期保存成本很高。云端可以用定时任务按5分钟、1小时、1天粒度做聚合,生成新的数据表,原始数据保留3个月后自动清理。Pandas+APScheduler就能很轻松地实现这套逻辑。
告警模块是业务价值的直接出口。我通常用规则引擎加阈值配置的方式,温度大于80度持续1分钟触发高频告警、湿度低于20%触发低频提醒等等。规则要支持动态配置,不要在代码里写死阈值。一个简单的做法是告警规则存MySQL或配置文件,定时任务每30秒扫描一次最新数据与规则匹配,触发后推送Webhook到钉钉或企业微信。用Python写这个模块非常顺手,schedule库可以跑定时任务,requests库直接调Webhook接口,几十行代码就能搞定。
4. 实操案例:从零搭建一个小型仓库环境监测系统
4.1 硬件与软件环境准备
光讲架构没有具体项目,总感觉隔着一层。我这边用一个我自己做过的仓库环境监测系统作为完整案例,把从边缘到云端的代码串起来讲一遍。这个项目规模不大,但链路是完整的,很适合作为参考模板。
硬件方面,我用了一个ESP32开发板,接了一个DHT22温湿度传感器,作为边缘采集端点。网关用树莓派4B,通过串口连接一个Modbus温湿度传感器来模拟多样设备接入。云端服务器是一台2核4G的云主机,装了Ubuntu、Docker、EMQX Broker和InfluxDB。软件开发环境方面,边缘端用MicroPython,网关端用Python 3.9 + paho-mqtt + pymodbus,云端用Python 3.10 + FastAPI + influxdb-client + pandas。整套环境搭建下来,大概半天时间可以完成。
4.2 边缘端代码实现:ESP32采集与网关转发
ESP32端代码跟前面展示的类似,这里不再重复,核心就是读取DHT22,通过MQTT上报到网关端或者直接到云端Broker。在实际案例里我让ESP32直接上报到云端,网关独立轮询Modbus设备,这样两条数据链路可以对比验证,也更好排查问题。
网关端是这套系统的关键节点。我给它分配了三个任务:第一,每5秒钟轮询Modbus传感器,读取温湿度寄存器值;第二,把读取到的数据先写本地SQLite做环形缓存,以应对云端网络抖动;第三,通过MQTT把数据转发到云端Broker。这里我额外加了数据本地规则判断:如果温度超过60度,网关直接驱动一个蜂鸣器IO口报警,不依赖云端决策。
网关端代码里有一个需要重点解释的地方:SQLite缓存和MQTT发送是解耦的,数据先写缓存、再尝试发送,发送成功后从缓存中删除,发送失败则保留并在下次循环中重试。这样即使云端Broker宕机半小时,本地数据也不会丢,恢复后自动补传。
4.3 云端服务实现:MQTT订阅、数据入库与告警触发
云端这里我写了一个常驻的Python服务,它的职责是:订阅所有网关上报的数据主题,收到消息后解析JSON、做基础清洗、写入InfluxDB,同时检查是否需要触发告警。
import json import paho.mqtt.client as mqtt from influxdb_client import InfluxDBClient, Point from influxdb_client.client.write_api import SYNCHRONOUS # 连接InfluxDB influx = InfluxDBClient(url="http://localhost:8086", token="your-token", org="iot-org") write_api = influx.write_api(write_options=SYNCHRONOUS) # 告警阈值配置,可以放到配置文件里动态调整 ALARM_RULES = { "temperature_max": 80, "humidity_min": 20, } def on_message(client, userdata, msg): try: payload = json.loads(msg.payload.decode()) device_id = payload.get("device_id") temperature = payload.get("temperature") humidity = payload.get("humidity") ts = payload.get("ts") # 数据清洗:超出合理范围的值标记为异常 quality = "good" if temperature is None or temperature < -40 or temperature > 100: quality = "suspect" # 写入InfluxDB point = Point("env_sensor") \ .tag("device_id", device_id) \ .field("temperature", temperature) \ .field("humidity", humidity) \ .time(ts, write_precision="ms") write_api.write(bucket="iot-bucket", record=point) # 告警检查 if temperature is not None and temperature > ALARM_RULES["temperature_max"]: mqtt_client.publish("alerts/temperature_high", json.dumps(payload)) except Exception as e: print(f"消息处理失败: {e}") mqtt_client = mqtt.Client() mqtt_client.on_message = on_message mqtt_client.connect("localhost", 1883, 60) mqtt_client.subscribe("warehouse/+/env/#") mqtt_client.loop_forever()这段代码里有个细节我要强调,就是时间戳的处理。InfluxDB的Point.time()要求传入毫秒级时间戳write_precision="ms",如果边缘设备上报的是秒级时间戳,这里不转换会导致所有数据的时间全部错乱,查询出来的曲线完全没有参考价值。所以云端在写入之前要统一做时间戳的单位校验。
还有一点,告警检查放在云端虽然实时性不如边缘侧,但对“温湿度持续偏高”这类缓慢变化场景已经足够。真正需要毫秒级响应的紧急操作,永远应该在边缘侧做,云端告警定位的是“趋势性风险”和“需要跨设备综合判断”的场景。
4.4 数据验证:查看一条完整的采集与聚合链路
服务部署完成后,我用真实数据跑了一整天,然后通过InfluxDB的Flux查询验证数据是否正确入库,再做一个简单的聚合验证。
from influxdb_client import InfluxDBClient client = InfluxDBClient(url="http://localhost:8086", token="your-token", org="iot-org") query_api = client.query_api() # 查询最近一小时原始数据 raw_query = ''' from(bucket:"iot-bucket") |> range(start: -1h) |> filter(fn:(r) => r._measurement == "env_sensor") |> filter(fn:(r) => r._field == "temperature") ''' raw_result = query_api.query(raw_query) for table in raw_result: for record in table.records: print(f"时间: {record.get_time()}, 设备: {record.values.get('device_id')}, 温度: {record.get_value()}") # 按每小时聚合取平均值 agg_query = ''' from(bucket:"iot-bucket") |> range(start: -24h) |> filter(fn:(r) => r._measurement == "env_sensor") |> filter(fn:(r) => r._field == "temperature") |> aggregateWindow(every: 1h, fn: mean) '''这只是最简单的数据验证,实际项目中,你会发现数据链路中还有很多需要盯着的东西,比如网关离线报警、MQTT消息积压监控、数据库磁盘空间预警。这些监控合在一起,才是一套完整的cloud-side运维体系。
5. 常见问题与排查技巧实录
物联网系统的排查难度比纯软件系统高很多,因为它的链路太长了,任何一个环节出问题,表现都可能是“云端没数据”。这里我整理了一份高频问题和排查思路,都是实际踩过的坑。
5.1 云端收不到设备数据,问题出在哪
云端收不到数据,通常按这个顺序排查:先看设备端有没有读取到传感器数据,再看MQTT Broker是否收到消息,最后看云端订阅服务是否正常消费。我见过太多人一上来就改云端代码,折腾半天发现是设备压根没上电。
排查时可以用最朴素的招数:在MQTT Broker上跑一个订阅所有主题的调试客户端,看消息到底有没有进来。如果消息到了Broker但云端服务没反应,大概率是订阅主题写错了,或者QoS设置了消息不匹配。实时打日志比什么都重要,每个环节的关键节点都要留日志,特别是网关端要记录“读到什么”和“发出什么”两步,云端要记录“收到什么”和“写入什么”两步,出现问题时对着日志一看就知道卡在哪。
5.2 数据时有时无,断断续续
这个现象多半是网络质量波动导致的。设备上报用的TCP长连接,一旦网络抖动,连接断开后没有自动重连,数据就丢了。解决方法是:设备端和网关端都必须实现自动重连机制。paho-mqtt的connect方法要放在循环里,断线后等待几秒重新连接,而且重连后要重新订阅主题。
另外,网络波动时如果数据没有本地缓存,就会直接丢数据。这也是我一直强调边缘缓存的原因,数据可靠性不能单靠网络保障,边缘缓存是最廉价也最有效的兜底方案。
5.3 数据有,但时间对不上,分析结果全乱
时序数据最核心的属性就是时间,时间错了,后面的分析全白做。时间对不上的原因主要有三个:设备端没有NTP校时,时钟漂移严重;网关层打的时间戳和设备端上报的时间戳混用;单位不统一,有的地方用秒,有的地方用毫秒。
我的做法是设备端只采集数据不记时间,网关层统一打毫秒级时间戳,云端以此为准。这样整条链路只有一个时间来源,即使设备端时钟不准,也不影响最终数据质量。还有一个细节是各个服务器之间要配置NTP自动校时,否则云端多台服务器处理同一批数据时也会出现时间乱序。
5.4 高频数据写入数据库,存储成本扛不住
上万个设备、每10秒上报一次数据,一天就是864万条记录,一年的数据量非常可怕。解决思路有两个方向:一是边缘端做聚合,网关先把1分钟内数据求平均,再上报一条均值,把上报频率从每秒降到每分钟;二是云端做降采样,原始数据保留3个月,聚合数据保留2年,历史查询走聚合表。
用Python做聚合任务时,我一般配合schedule库或者直接在FastAPI里用BackgroundTasks,每天凌晨跑一次前一天的聚合统计。聚合任务一定要做幂等设计,避免重复执行产生双倍数据,我习惯用“日期 + 聚合粒度”作为唯一键,写库前先检查是否已经存在。
6. 几个容易被忽略的细节与总结建议
写到这里,主线内容差不多讲完了。有几个细节我放在最后单独说一下,因为它们不处理也不会立刻出问题,但一旦出了问题都是比较难查的坑。
第一个是MQTT的QoS级别选择。很多人图省事全程用QoS 0,只要网络稍微抖动一下,消息就丢了。我的建议是控制指令类消息用QoS 1,数据上报类消息根据重要程度选择QoS 0或QoS 1。QoS 2慎用,性能开销太大,而且大多数场景根本不需要。
第二个是Broker的连接鉴权。IoT设备“裸奔”连接Broker是我见过最危险的事之一。设备端写死了Broker地址和端口,没有任何用户名密码保护,一旦Broker暴露在公网,任何人都能订阅你的数据主题。轻则数据泄露,重则有人伪造设备上报假数据,把整个数据体系污染掉。EMQX和Mosquitto都支持用户名密码和TLS证书,设备量大的话建议直接上证书认证。
第三个是数据模型的“活版本”问题。物联网设备分布在现场,固件升级很麻烦,你不可能让现场几百台设备同时更新到新版本的数据格式。所以云端解析数据时一定要做字段级兼容,新老字段同时支持,设备端新老版本交替时才能平滑过渡。Python处理这个很方便,dict.get()可以给字段设置默认值,新版本来了就读新字段,老版本来了就读老字段,做不到的字段给个默认值,日志记录一下。
第四个是关于Python版本的选择。现在新项目我一般直接上3.10+,语法更简洁,类型标注也更完善。但你要注意边缘设备上跑的MicroPython跟PC上的CPython是两回事,MicroPython只实现了Python 3语法的一个子集,像enum、functools.lru_cache这种标准库模块在MicroPython里可能没有,写边缘代码的时候要主动避开这些依赖,尽量只用基础语法,同时跑在云端的脚本则不用有这些顾虑。
最后一个建议是关于持续集成与日志采集。不要觉得IoT系统搭好能用就行了,设备现场环境复杂,固件升级、配置变更、云端代码更新,任何一个环节都可能引入问题。我现在的习惯是所有边缘设备和网关都保留远程SSH通道,通过Ansible批量推送Python脚本,配置用YAML管理,代码入库走GitLab CI自动同步到云端服务器。告警日志统一接入ELK或Loki,这样设备端、网关端、云端的问题都能在一个面板里看到。虽然这套体系初期搭建要花些时间,但是项目上线三个月之后,你会感谢当初做了这些基础设施。
物联网这个领域里,Python的位置取决于你从哪个角度看它。如果你只看单片机上的裸机开发,Python确实不是首选;但如果你把眼光放在“从边缘到云端”的完整数据链路上,Python提供了一个平滑的、贯穿始终的解决方案。我做的项目不算特别大,但每个阶段都踩过不少坑,希望这篇文章里的思路和代码,能让你在搭建自己的Python IoT系统时少走一些弯路。