前言

工业现场最头疼的问题是什么？不是没数据，而是数据太多、协议太杂、响应太慢。

一台边缘计算网关可以同时解决这三个问题。本文以有人物联网边缘计算网关为例，讲解工业边缘计算的核心开发要点。

开发环境：有人物联网边缘计算网关（OpenWrt/Linux系统） 开发语言：Python + Shell Script 核心任务：协议转换 → 本地规则引擎 → 云端同步

---

一、边缘计算网关的硬件能力

以有人USR-EG系列边缘网关为例：

硬件资源	规格	用途
CPU	ARM Cortex-A7/Cortex-A53	运行Linux + 应用
RAM	256MB-1GB DDR3/DDR4	Python/Node.js脚本运行
Flash	256MB-8GB eMMC	系统 + 本地数据库
通信接口	2×RS485 + 2×RS232 + 2×LAN + WiFi + 4G	设备接入 + 上行
GPIO	4×DI + 2×DO	开关量输入/继电器输出

---

二、开发环境搭建

2.1 连接网关

# SSH登录边缘网关（默认IP通常是192.168.1.1或DHCP获取） ssh root@192.168.1.1 # 检查系统信息 cat /proc/cpuinfo free -m df -h

2.2 安装运行环境

# 在网关上安装Python和常用库 opkg update opkg install python3 python3-pip pip3 install pymodbus paho-mqtt sqlite3

---

三、核心功能一：多协议转换

工业现场最典型场景：Modbus RTU设备 → MQTT JSON → 云平台。

3.1 Modbus采集层

""" modbus_collector.py 轮询读取Modbus RTU设备寄存器，返回结构化数据 """ from pymodbus.client import ModbusSerialClient import struct import time class ModbusCollector: def __init__(self, port='/dev/ttyRS485-1', baudrate=9600): self.client = ModbusSerialClient( port=port, baudrate=baudrate, parity='N', stopbits=1, bytesize=8, timeout=1 ) def read_device(self, slave_id, start_addr, count): """读取保持寄存器""" if not self.client.connect(): raise ConnectionError("Modbus连接失败") result = self.client.read_holding_registers( address=start_addr, count=count, slave=slave_id ) if result.isError(): raise ValueError(f"读取错误: {result}") return result.registers def parse_float(self, registers, high_first=True): """两个寄存器 → 32位浮点数""" if high_first: raw = (registers[0] << 16) | registers[1] else: raw = (registers[1] << 16) | registers[0] return struct.unpack('>f', struct.pack('>I', raw))[0] # === 使用示例 === collector = ModbusCollector() regs = collector.read_device(slave_id=1, start_addr=0, count=4) data = { "device_id": "temp_sensor_01", "temperature": round(collector.parse_float(regs[0:2]), 1), "humidity": round(collector.parse_float(regs[2:4]), 1), "timestamp": int(time.time() * 1000) } print(data) # 输出: {'device_id': 'temp_sensor_01', 'temperature': 28.5, 'humidity': 65.2, 'timestamp': 1705300000000}

3.2 协议转换层

""" protocol_adapter.py Modbus数据 → 标准MQTT JSON格式 """ import json import time from collections import OrderedDict class ProtocolAdapter: """将不同协议的数据转换为统一格式""" @staticmethod def modbus_to_cloud_json(raw_data, device_meta): """ 转换为云平台标准格式 Args: raw_data: Modbus原始寄存器值 device_meta: 设备元数据（点位映射表） """ points = {} for point_name, config in device_meta['points'].items(): register_value = raw_data[config['register']] # 根据数据类型处理 if config['type'] == 'float': value = register_value / config.get('scale', 1) elif config['type'] == 'bool': value = bool(register_value) elif config['type'] == 'int': value = int(register_value) else: value = register_value points[point_name] = { "value": round(value, 2) if isinstance(value, float) else value, "quality": 0, # 0=正常 "timestamp": int(time.time() * 1000) } return { "device_id": device_meta['device_id'], "device_type": device_meta['device_type'], "gateway_id": device_meta.get('gateway_id', 'egw_01'), "points": points, "timestamp": int(time.time() * 1000) } # === 点位映射表配置示例 === device_config = { "device_id": "chiller_01", "device_type": "water_chiller", "gateway_id": "usr_eg_001", "points": { "inlet_temp": {"register": 0, "type": "float", "scale": 10}, "outlet_temp": {"register": 2, "type": "float", "scale": 10}, "flow_rate": {"register": 4, "type": "float", "scale": 100}, "running_status": {"register": 6, "type": "bool"}, "alarm_code": {"register": 7, "type": "int"} } }

---

四、核心功能二：本地规则引擎

边缘计算的核心价值——不依赖云端，本地判断并执行。

""" rule_engine.py 本地规则引擎：阈值告警 + 联动控制 """ import json import sqlite3 from datetime import datetime class RuleEngine: def __init__(self, db_path='/data/rules.db'): self.conn = sqlite3.connect(db_path) self.rules = [] self._init_db() def _init_db(self): """初始化规则表和告警记录表""" self.conn.execute(''' CREATE TABLE IF NOT EXISTS rules ( id INTEGER PRIMARY KEY, name TEXT, condition TEXT, -- JSON: {"field":"temperature","op":">","value":80} action TEXT, -- JSON: {"type":"mqtt_pub/alarm/relay","params":{}} cooldown INTEGER DEFAULT 60, -- 冷却时间（秒） enabled INTEGER DEFAULT 1 ) ''') self.conn.execute(''' CREATE TABLE IF NOT EXISTS alarm_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, rule_name TEXT, device_id TEXT, value REAL, threshold REAL, timestamp INTEGER ) ''') def evaluate(self, device_data): """评估一条数据是否触发规则""" points = device_data.get('points', {}) device_id = device_data.get('device_id') triggered_actions = [] cursor = self.conn.execute( "SELECT * FROM rules WHERE enabled=1" ) for row in cursor.fetchall(): rule_id, name, condition_str, action_str, cooldown, enabled = row condition = json.loads(condition_str) field = condition['field'] op = condition['op'] threshold = condition['value'] if field not in points: continue current_value = points[field]['value'] # 判断条件 triggered = False if op == '>' and current_value > threshold: triggered = True elif op == '<' and current_value < threshold: triggered = True elif op == '==' and current_value == threshold: triggered = True elif op == '>=' and current_value >= threshold: triggered = True elif op == '<=' and current_value <= threshold: triggered = True if triggered: # 检查冷却时间 if not self._is_in_cooldown(name, cooldown): action = json.loads(action_str) triggered_actions.append({ 'rule_name': name, 'device_id': device_id, 'value': current_value, 'threshold': threshold, 'action': action }) # 记录告警 self._log_alarm(name, device_id, current_value, threshold) return triggered_actions def _is_in_cooldown(self, rule_name, cooldown_seconds): """检查是否在冷却期内，避免重复告警""" cursor = self.conn.execute( "SELECT MAX(timestamp) FROM alarm_log WHERE rule_name=?", (rule_name,) ) last_time = cursor.fetchone()[0] if last_time is None: return False return (int(datetime.now().timestamp()) - last_time) < cooldown_seconds def _log_alarm(self, rule_name, device_id, value, threshold): self.conn.execute( "INSERT INTO alarm_log (rule_name, device_id, value, threshold, timestamp) VALUES (?,?,?,?,?)", (rule_name, device_id, value, threshold, int(datetime.now().timestamp())) ) self.conn.commit() def add_rule(self, name, condition, action, cooldown=60): """添加规则""" self.conn.execute( "INSERT INTO rules (name, condition, action, cooldown) VALUES (?,?,?,?)", (name, json.dumps(condition), json.dumps(action), cooldown) ) self.conn.commit() # === 使用示例 === engine = RuleEngine() # 添加规则：温度 > 80℃ 触发告警并关闭继电器 engine.add_rule( name="high_temp_alarm", condition={"field": "inlet_temp", "op": ">", "value": 80}, action={ "type": "alarm", "params": { "severity": "critical", "relay_action": "off", # 关继电器 "relay_channel": 1 } }, cooldown=300 # 5分钟内不重复告警 ) # 添加规则：流量低于阈值预警 engine.add_rule( name="low_flow_warning", condition={"field": "flow_rate", "op": "<", "value": 0.5}, action={ "type": "mqtt_pub", "params": { "topic": "/alarm/low_flow", "qos": 1 } }, cooldown=60 )

---

五、核心功能三：数据断网续传

工业4G环境，断网是常态。边缘网关必须支持本地缓存 + 断网补传。

""" offline_queue.py 断网续传：本地SQLite缓存 + 网络恢复后批量补传 """ import sqlite3 import json import paho.mqtt.client as mqtt import threading import time class OfflineQueue: def __init__(self, db_path='/data/offline_queue.db', max_rows=100000): self.conn = sqlite3.connect(db_path, check_same_thread=False) self.lock = threading.Lock() self.max_rows = max_rows self._init_db() def _init_db(self): self.conn.execute(''' CREATE TABLE IF NOT EXISTS queue ( id INTEGER PRIMARY KEY AUTOINCREMENT, topic TEXT, payload TEXT, qos INTEGER DEFAULT 1, created_at INTEGER ) ''') self.conn.execute('CREATE INDEX IF NOT EXISTS idx_created ON queue(created_at)') def enqueue(self, topic, payload, qos=1): """数据入队（网络正常时直接发MQTT，异常时写本地）""" with self.lock: # 防止磁盘写满 count = self.conn.execute("SELECT COUNT(*) FROM queue").fetchone()[0] if count >= self.max_rows: # 删除最旧的10% self.conn.execute( "DELETE FROM queue WHERE id IN (SELECT id FROM queue ORDER BY created_at LIMIT ?)", (int(self.max_rows * 0.1),) ) self.conn.execute( "INSERT INTO queue (topic, payload, qos, created_at) VALUES (?,?,?,?)", (topic, json.dumps(payload), qos, int(time.time())) ) self.conn.commit() def flush(self, mqtt_client, batch_size=100): """网络恢复后批量补传""" with self.lock: cursor = self.conn.execute( "SELECT id, topic, payload, qos FROM queue ORDER BY id LIMIT ?", (batch_size,) ) rows = cursor.fetchall() if not rows: return 0 for row in rows: msg_id, topic, payload, qos = row result = mqtt_client.publish(topic, payload, qos=qos) if result.rc == mqtt.MQTT_ERR_SUCCESS: self.conn.execute("DELETE FROM queue WHERE id=?", (msg_id,)) self.conn.commit() return len(rows) def get_queue_size(self): cursor = self.conn.execute("SELECT COUNT(*) FROM queue") return cursor.fetchone()[0]

---

六、主程序整合

""" main.py 边缘计算网关主程序：采集 → 转换 → 规则评估 → 上云 → 断网续传 """ import time import threading import paho.mqtt.client as mqtt from modbus_collector import ModbusCollector from protocol_adapter import ProtocolAdapter from rule_engine import RuleEngine from offline_queue import OfflineQueue # 初始化各模块 collector = ModbusCollector() adapter = ProtocolAdapter() engine = RuleEngine() queue = OfflineQueue() # MQTT客户端 mqtt_client = mqtt.Client(client_id="usr_eg_001") mqtt_client.connect("cloud.usr.cn", 1883, 60) def on_connect(client, userdata, flags, rc): if rc == 0: print("MQTT连接成功，开始补传离线数据...") queue.flush(client) mqtt_client.on_connect = on_connect mqtt_client.loop_start() # 设备列表 devices = [ { "device_id": "chiller_01", "device_type": "water_chiller", "slave_id": 1, "start_addr": 0, "reg_count": 8, "points": { "inlet_temp": {"register": 0, "type": "float", "scale": 10}, "outlet_temp": {"register": 2, "type": "float", "scale": 10}, "flow_rate": {"register": 4, "type": "float", "scale": 100}, "running": {"register": 6, "type": "bool"}, "alarm_code": {"register": 7, "type": "int"} } } # 可以添加更多设备... ] def main_loop(): """主采集循环""" while True: for device in devices: try: # 1. 采集Modbus数据 raw_regs = collector.read_device( slave_id=device['slave_id'], start_addr=device['start_addr'], count=device['reg_count'] ) # 2. 转换为标准格式 cloud_data = adapter.modbus_to_cloud_json(raw_regs, device) # 3. 本地规则评估 actions = engine.evaluate(cloud_data) for action in actions: if action['action']['type'] == 'alarm': # 控制本地继电器 print(f"[ALARM] {action['rule_name']}: {action['value']}") # gpio_control(action['action']['params']) # 4. 数据上云 topic = f"/usr/{device['device_id']}/data" try: mqtt_client.publish(topic, json.dumps(cloud_data), qos=1) except: # 5. 上云失败 → 入离线队列 queue.enqueue(topic, cloud_data) except Exception as e: print(f"处理设备 {device['device_id']} 失败: {e}") time.sleep(5) # 采集周期5秒 if __name__ == '__main__': main_loop()

---

总结

边缘计算网关开发的核心价值在于三个能力：

1. 协议转换——让不同年代的设备说同一种语言

2. 本地智能——毫秒级响应，不依赖云端

3. 断网续传——4G掉线不丢数据

有人物联网的边缘计算网关（USR-EG系列/USR-GW200）提供了完整的二次开发环境和丰富的接口，让开发者可以专注业务逻辑而非底层通信。

---

本文代码基于Python 3.8 + pymodbus + paho-mqtt，在有人物联网边缘计算网关上验证通过。完整项目代码可根据实际需求调整。