4、边缘计算与本地逻辑：边缘节点概念、使用Python在树莓派上进行数据预处理、本地告警规则引擎设计

好，咱们进入第四章。这一章我特别想跟你聊聊边缘计算。说实话，在电梯物联网这个领域，边缘节点不是个可选项，而是个必选项。为什么？你想想看，一台电梯每天产生的数据量，如果全部往云端送，带宽和延迟都是大问题。我在一个老旧小区改造项目里就吃过这个亏——当时把所有传感器数据都往服务器推，结果网络一波动，电梯困人了报警都发不出去。从那以后，我对边缘计算的态度就变成了：能本地处理的，绝不往上传。

4.1 边缘节点到底是什么？

边缘节点，说白了就是靠近数据源头的一台小电脑。在电梯场景里，它通常是一块树莓派，或者性能更强的 Jetson Nano。它的任务很简单：采集、处理、决策。

我个人习惯把边缘节点分成三层：

• 采集层：接传感器，读数据。比如电梯的加速度、门状态、楼层、速度。
• 处理层：做滤波、去噪、特征提取。这一步很关键，能过滤掉大量无效数据。
• 决策层：跑规则引擎，判断要不要报警，要不要本地存储。

嗯，这里要注意：边缘节点不是万能的。它资源有限，所以你不能把云端那套复杂的深度学习模型直接塞进去。我建议的做法是——边缘做轻量级规则，云端做深度分析。

4.2 树莓派上的数据预处理

好，咱们上代码。我用树莓派 4B 做过一个原型，Python 3.9，传感器是 MPU6050（六轴加速度+陀螺仪）。数据预处理这一步，我踩过不少坑。

先看一个最简单的数据采集和滤波示例：

import smbus
import time
import numpy as np

# 初始化 I2C 总线
bus = smbus.SMBus(1)
MPU6050_ADDR = 0x68

# 唤醒传感器
bus.write_byte_data(MPU6050_ADDR, 0x6B, 0)

def read_accel():
    # 读取原始加速度数据
    accel_x = bus.read_byte_data(MPU6050_ADDR, 0x3B) << 8 | bus.read_byte_data(MPU6050_ADDR, 0x3C)
    accel_y = bus.read_byte_data(MPU6050_ADDR, 0x3D) << 8 | bus.read_byte_data(MPU6050_ADDR, 0x3E)
    accel_z = bus.read_byte_data(MPU6050_ADDR, 0x3F) << 8 | bus.read_byte_data(MPU6050_ADDR, 0x40)
    # 转换为 g 值
    return [x / 16384.0 for x in [accel_x, accel_y, accel_z]]

# 滑动平均滤波
def moving_average(data, window_size=5):
    return np.convolve(data, np.ones(window_size)/window_size, mode='valid')

# 主循环
buffer = []
while True:
    raw = read_accel()
    buffer.append(raw[0])  # 只取 X 轴做演示
    if len(buffer) > 10:
        buffer.pop(0)
        filtered = moving_average(buffer, 3)
        print(f"原始值: {raw[0]:.3f}g, 滤波后: {filtered[-1]:.3f}g")
    time.sleep(0.1)

这段代码里，我用了滑动平均滤波。为什么不用卡尔曼？因为树莓派的算力有限，卡尔曼滤波虽然精度高，但计算量大。我在一个项目里试过，跑卡尔曼时 CPU 占用率直接飙到 60%，而滑动平均只有 5%。对于电梯的振动监测，滑动平均完全够用。

4.3 本地告警规则引擎设计

规则引擎，是边缘节点的灵魂。它决定了什么时候该报警，什么时候该忽略。

我设计过一个简单的规则引擎，基于 Python 的字典和 lambda 函数。结构很清晰：

class RuleEngine:
    def __init__(self):
        self.rules = []
    
    def add_rule(self, name, condition, action):
        """
        condition: 一个函数，接收数据字典，返回 True/False
        action: 一个函数，接收数据字典，执行报警动作
        """
        self.rules.append({
            'name': name,
            'condition': condition,
            'action': action
        })
    
    def evaluate(self, data):
        for rule in self.rules:
            if rule['condition'](data):
                rule['action'](data)
                return True  # 一旦触发，停止后续规则
        return False

# 定义规则
def overspeed_condition(data):
    return data['speed'] > 2.5  # 速度超过 2.5 m/s

def overspeed_action(data):
    print(f"[紧急] 超速报警！当前速度: {data['speed']:.2f} m/s")
    # 这里可以接 GPIO 控制蜂鸣器

def door_open_moving_condition(data):
    return data['door_open'] and data['speed'] > 0.1

def door_open_moving_action(data):
    print(f"[危险] 门开着还在运行！")

# 使用
engine = RuleEngine()
engine.add_rule('超速', overspeed_condition, overspeed_action)
engine.add_rule('门开运行', door_open_moving_condition, door_open_moving_action)

# 模拟数据
sample_data = {'speed': 2.8, 'door_open': False}
engine.evaluate(sample_data)

这个引擎虽然简单，但很实用。我在一个 12 层楼的电梯项目里部署过，跑了半年没出过问题。规则可以热更新——你只需要在云端修改规则配置，边缘节点定期拉取就行。

4.4 边缘节点的本地存储策略

数据预处理完了，规则也跑了，那数据存不存？我的建议是：存，但只存精华。

我设计过一个环形缓冲区：

import sqlite3
from collections import deque

class LocalStorage:
    def __init__(self, db_path='/home/pi/elevator.db', max_records=10000):
        self.conn = sqlite3.connect(db_path)
        self.cursor = self.conn.cursor()
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT,
                event_type TEXT,
                data TEXT
            )
        ''')
        self.buffer = deque(maxlen=max_records)
    
    def add_event(self, event_type, data):
        import json, datetime
        record = {
            'timestamp': datetime.datetime.now().isoformat(),
            'event_type': event_type,
            'data': json.dumps(data)
        }
        self.buffer.append(record)
        # 每 100 条批量写入一次
        if len(self.buffer) % 100 == 0:
            self._flush()
    
    def _flush(self):
        for rec in self.buffer:
            self.cursor.execute(
                'INSERT INTO events (timestamp, event_type, data) VALUES (?, ?, ?)',
                (rec['timestamp'], rec['event_type'], rec['data'])
            )
        self.conn.commit()
        self.buffer.clear()

这里有个细节：我用了环形缓冲区，而不是每条数据都写数据库。为什么？因为树莓派的 SD 卡写入寿命有限。频繁写入会加速 SD 卡损坏。我在一个项目里就遇到过，SD 卡用了 3 个月就挂了，后来换成工业级 eMMC 模块才解决。

4.5 边缘与云端的协同

最后，聊聊边缘和云端怎么配合。我的做法是：

数据类型	处理位置	说明
实时告警	边缘	毫秒级响应，本地触发蜂鸣器/灯光
运行日志	边缘存储，定时上传	每 5 分钟批量上传一次，节省带宽
振动波形	边缘提取特征，云端做 FFT	边缘只传 RMS、峰值等统计量
模型更新	云端训练，边缘推理	每周更新一次规则参数

你想想看，如果所有数据都往云端传，一个 1000 台电梯的项目，每天光网络流量就得几十个 G。而用了边缘预处理之后，流量能降到原来的 1/10 以下。这就是边缘计算的价值。