HarmonyOS开发｜ArkTS UI颜色API通用规则
一、为什么计算要下沉到边缘
把一台上千点位的设备直连云端，会立刻撞上三堵墙：工业现场的 Modbus、OPC UA 设备根本不支持 MQTT；按秒级上行的原始数据在 4G 流量上是一笔不小的开支；现场网络中断时，数据会毫无缓冲地直接丢失。
边缘网关的思路很直接——在离设备最近的地方完成协议转换、聚合与缓冲，云端只接收"已经处理过"的数据。这样一来，云端系统的复杂度也被顺带削掉了一层。
二、架构分层
一个务实的网关可以切成四层，每层职责单一、可独立替换：
| 层 | 职责 | 典型实现 |
| 南向采集 | 轮询 Modbus / 串口设备 | pymodbus |
| 数据处理 | 清洗、单位换算、时间窗聚合 | 纯 Python |
| 本地缓冲 | 断网时落盘排队 | SQLite |
| 北向上报 | 转为 MQTT 上报云端 | paho-mqtt |
核心设计原则是采集与上报解耦：采集线程只负责把最新值写进内存缓存，上报线程按自己的节奏读取。这样云端抖动或断连都不会阻塞采集，这是整套架构能长期稳定运行的前提。
三、协议转换：从 Modbus 到 MQTT
Modbus 是典型的"请求-响应 + 寄存器寻址"模型，读温度就是读某个保持寄存器再做一次缩放。
这一层最容易出错的往往不是协议本身，而是点位表管理——寄存器地址、数据类型、缩放系数应当放在配置里，而不是散落在代码中。
# modbus_to_mqtt.py —— 采集一个寄存器并上报
from pymodbus.client import ModbusTcpClient
import paho.mqtt.client as mqtt
import json, time
POINTS = [{"name": "temp", "addr": 0, "scale": 0.1}] # 点位表
mb = ModbusTcpClient("192.168.1.10", port=502)
mq = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="gw-1")
mq.connect("broker.local", 1883)
mb.connect()
while True:
for p in POINTS:
rr = mb.read_holding_registers(p["addr"], count=1)
if rr.isError():
continue # 单点失败不中断整轮采集
value = rr.registers[0] * p["scale"]
mq.publish(f"gw/dev1/{p['name']}", json.dumps({"v": value}), qos=1)
time.sleep(1)
这里有两处刻意为之：单点读取失败只跳过该点而非抛出中断整轮；上报使用 QoS 1，因为聚合后的数据每丢一条都意味着一个时间窗出现缺口。四、本地聚合：把 60 条压成 1 条
秒级采集、分钟级上报是最常见的配置。网关在本地维护一个滑动窗口，窗口结束时把均值、极值与样本数打包上报：
# 时间窗聚合：60 秒一个窗口
window, t0 = [], time.monotonic()
...
window.append(value)
if time.monotonic() - t0 >= 60:
payload = {"avg": sum(window) / len(window), "min": min(window),
"max": max(window), "n": len(window)}
mq.publish(topic, json.dumps(payload), qos=1)
window, t0 = [], time.monotonic() 收益是可量化的：上行消息数降到原来的 1/60，流量与云端写入压力同步下降。代价则是丢掉了原始分辨率——如果下游还要做异常波形分析，这一层就不能加，或者必须保留原始数据旁路上传。
五、断网缓冲：用 SQLite 做队列
网关最怕的是网络抖动。做一个"先落盘、再发送、收到确认才删除"的本地队列，就能把断网从"丢数"变成"延迟"：
# 简易落盘队列（表结构：id, topic, payload, ts）
def enqueue(topic, payload):
db.execute("INSERT INTO outbox(topic, payload, ts) VALUES(?, ?, ?)",
(topic, payload, time.time()))
db.commit()
def drain():
rows = db.execute("SELECT id, topic, payload FROM outbox ORDER BY id LIMIT 200")
for row in rows:
info = mq.publish(row[1], row[2], qos=1)
info.wait_for_publish(timeout=5) # 等确认，失败则留待下轮
if info.is_published():
db.execute("DELETE FROM outbox WHERE id = ?", (row[0],))
db.commit() 六、踩坑记录
- 别忘了给队列设上限。 断网三天、每秒一条数据，SQLite 文件会涨到几百 MB，网关的 eMMC 很容易被写满。必须加容量上限与丢弃策略，并明确丢弃时丢最旧还是最新。
- 现场时钟不可信。 网关 RTC 经常漂移，聚合窗口应基于单调时钟（time.monotonic），仅在最终上报时才附带墙钟时间。
- 聚合粒度要与业务对齐。 固定 60 秒切窗会在窗口边界丢掉尖峰，需要保尖峰的场景应改为"固定窗口 + 极值单独上报"。
七、部署与运维：让网关可被远程管理
网关一旦装到现场，维护成本就会成倍上升，因此"可运维"必须在设计阶段就纳入，而不是上线后再补。
首先是配置热更新。点位表、上报周期、Broker 地址这类参数应当通过 MQTT 下发，而不是改代码重刷固件。订阅 gw/<id>/config 主题，收到新配置后先校验、再落盘、最后平滑重建采集任务，整个过程不中断业务。
其次是心跳与健康上报。除业务数据外，网关还应定期上报自身的 CPU、内存、磁盘余量与队列积压长度，以及最后一次成功通信的时间。这几项指标能在设备"假在线"时第一时间暴露问题——网关能连上 Broker，底层串口却早已掉线，这是现场最常见的隐性故障。
最后是灰度与回滚。配置与固件的下发务必携带版本号与回滚入口，避免一次错误配置让所有现场设备同时失联；批量操作分批推进，永远给自己留一条退路。
八、总结
边缘网关的复杂度并不来自任何一项单点技术，而来自四层之间的解耦是否干净。判断标准可以很朴素：南向采集的代码里不应该出现 MQTT 字样，北向上报的代码里也不该知道 Modbus 的存在。做到这一点，替换协议或更换云平台都只是改一层的事。
