当前位置：首页 > news >正文

Python实例题：基于边缘计算的智能物联网系统

news 来源：原创 2025/6/22 5:35:16

Python实例题

题目

基于边缘计算的智能物联网系统

问题描述

开发一个基于边缘计算的智能物联网系统，包含以下功能：

边缘设备管理：连接和管理大量物联网设备
边缘计算节点：在边缘处理数据，减少云传输
实时数据分析：实时处理和分析传感器数据
智能决策：基于 AI 模型的自动化决策
云边协同：边缘节点与云端的协同工作

解题思路

设计边缘设备与边缘节点的通信协议
开发边缘计算框架处理本地数据
训练和部署轻量级 AI 模型到边缘节点
实现云边数据同步和协同机制
构建可视化界面监控系统状态

关键代码框架

# 边缘节点主程序
import asyncio
import json
import time
import logging
from typing import Dict, List, Any
import numpy as np
from sklearn.ensemble import IsolationForest
import paho.mqtt.client as mqtt
from sqlalchemy import create_engine, Column, Integer, Float, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import datetime# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)# 配置
CONFIG = {'mqtt_broker': 'localhost','mqtt_port': 1883,'mqtt_username': 'edge_node','mqtt_password': 'edge_password','data_topic': 'iot/data/#','control_topic': 'iot/control/','edge_id': 'edge-node-01','db_url': 'sqlite:///edge_data.db','anomaly_detection_window': 100,'sync_interval': 60  # 数据同步到云端的间隔(秒)
}# 数据库模型
Base = declarative_base()class SensorData(Base):__tablename__ = 'sensor_data'id = Column(Integer, primary_key=True)device_id = Column(String(50))sensor_type = Column(String(50))value = Column(Float)timestamp = Column(DateTime)processed = Column(Boolean, default=False)is_anomaly = Column(Boolean, default=False)class AnomalyAlert(Base):__tablename__ = 'anomaly_alerts'id = Column(Integer, primary_key=True)device_id = Column(String(50))sensor_type = Column(String(50))value = Column(Float)timestamp = Column(DateTime)confidence = Column(Float)resolved = Column(Boolean, default=False)# 初始化数据库
engine = create_engine(CONFIG['db_url'])
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)# 异常检测模型
class AnomalyDetector:def __init__(self, window_size=100):self.window_size = window_sizeself.models = {}  # 每个传感器一个模型def train(self, device_id, sensor_type, data):"""训练异常检测模型"""if len(data) < 10:  # 至少需要10个数据点return# 重塑数据以适应模型X = np.array(data).reshape(-1, 1)# 创建或更新模型model_key = f"{device_id}_{sensor_type}"if model_key not in self.models:self.models[model_key] = IsolationForest(contamination=0.1, random_state=42)# 训练模型self.models[model_key].fit(X)def detect(self, device_id, sensor_type, value):"""检测异常"""model_key = f"{device_id}_{sensor_type}"if model_key not in self.models:return False, 0.0# 预测X = np.array([value]).reshape(1, -1)prediction = self.models[model_key].predict(X)decision_function = self.models[model_key].decision_function(X)# 转换为异常分数 (越负越异常)anomaly_score = -decision_function[0]is_anomaly = prediction[0] == -1return is_anomaly, anomaly_score# MQTT客户端
class EdgeMQTTClient:def __init__(self, config, data_processor, cloud_sync):self.config = configself.data_processor = data_processorself.cloud_sync = cloud_syncself.client = mqtt.Client(client_id=config['edge_id'])# 设置回调函数self.client.on_connect = self.on_connectself.client.on_message = self.on_messageself.client.on_disconnect = self.on_disconnect# 设置认证self.client.username_pw_set(config['mqtt_username'], config['mqtt_password'])def connect(self):"""连接到MQTT代理"""try:self.client.connect(self.config['mqtt_broker'], self.config['mqtt_port'], 60)self.client.loop_start()logger.info(f"已连接到MQTT代理: {self.config['mqtt_broker']}:{self.config['mqtt_port']}")except Exception as e:logger.error(f"连接MQTT代理失败: {e}")raisedef disconnect(self):"""断开MQTT连接"""self.client.loop_stop()self.client.disconnect()logger.info("已断开MQTT连接")def on_connect(self, client, userdata, flags, rc):"""连接成功回调"""if rc == 0:logger.info("MQTT连接成功")# 订阅数据主题client.subscribe(self.config['data_topic'])logger.info(f"已订阅主题: {self.config['data_topic']}")else:logger.error(f"MQTT连接失败，错误码: {rc}")def on_message(self, client, userdata, msg):"""消息接收回调"""try:topic = msg.topicpayload = msg.payload.decode('utf-8')data = json.loads(payload)logger.debug(f"收到消息 - 主题: {topic}, 数据: {data}")# 处理数据self.data_processor.process_data(data)except json.JSONDecodeError as e:logger.error(f"JSON解析错误: {e}, 数据: {msg.payload}")except Exception as e:logger.error(f"处理消息时出错: {e}")def on_disconnect(self, client, userdata, rc):"""断开连接回调"""logger.warning(f"MQTT连接断开，错误码: {rc}")# 尝试重新连接try:time.sleep(5)self.client.reconnect()except Exception as e:logger.error(f"重新连接失败: {e}")def publish_control_message(self, device_id, command, payload=None):"""发布控制消息"""topic = f"{self.config['control_topic']}{device_id}"message = {'command': command,'payload': payload,'timestamp': datetime.datetime.now().isoformat()}try:self.client.publish(topic, json.dumps(message))logger.info(f"发布控制消息 - 主题: {topic}, 消息: {message}")except Exception as e:logger.error(f"发布控制消息失败: {e}")# 数据处理器
class DataProcessor:def __init__(self, config, db_session, anomaly_detector):self.config = configself.db_session = db_sessionself.anomaly_detector = anomaly_detectorself.sensor_data_buffers = {}  # 存储传感器数据缓冲区def process_data(self, data):"""处理传感器数据"""try:# 提取数据device_id = data.get('device_id')sensor_type = data.get('sensor_type')value = data.get('value')timestamp = data.get('timestamp')if not device_id or not sensor_type or value is None:logger.warning("无效的传感器数据，缺少必要字段")return# 转换时间戳if timestamp:try:timestamp = datetime.datetime.fromisoformat(timestamp)except ValueError:timestamp = datetime.datetime.now()else:timestamp = datetime.datetime.now()# 检测异常buffer_key = f"{device_id}_{sensor_type}"if buffer_key not in self.sensor_data_buffers:self.sensor_data_buffers[buffer_key] = []self.sensor_data_buffers[buffer_key].append(value)# 保持缓冲区大小if len(self.sensor_data_buffers[buffer_key]) > self.config['anomaly_detection_window']:self.sensor_data_buffers[buffer_key] = self.sensor_data_buffers[buffer_key][-self.config['anomaly_detection_window']:]# 训练异常检测模型self.anomaly_detector.train(device_id, sensor_type, self.sensor_data_buffers[buffer_key])# 检测异常is_anomaly, confidence = self.anomaly_detector.detect(device_id, sensor_type, value)# 保存数据到数据库sensor_data = SensorData(device_id=device_id,sensor_type=sensor_type,value=value,timestamp=timestamp,is_anomaly=is_anomaly)self.db_session.add(sensor_data)# 如果是异常，记录警报if is_anomaly:alert = AnomalyAlert(device_id=device_id,sensor_type=sensor_type,value=value,timestamp=timestamp,confidence=confidence)self.db_session.add(alert)logger.warning(f"检测到异常 - 设备: {device_id}, 传感器: {sensor_type}, 值: {value}, 置信度: {confidence}")# 提交事务self.db_session.commit()logger.info(f"处理数据 - 设备: {device_id}, 传感器: {sensor_type}, 值: {value}, 是否异常: {is_anomaly}")except Exception as e:self.db_session.rollback()logger.error(f"处理数据时出错: {e}")# 云同步服务
class CloudSyncService:def __init__(self, config, db_session, mqtt_client):self.config = configself.db_session = db_sessionself.mqtt_client = mqtt_clientself.cloud_topic = "cloud/data"async def start_periodic_sync(self):"""启动定期数据同步"""while True:try:logger.info("开始云同步...")self.sync_data_to_cloud()logger.info("云同步完成")# 等待下一次同步await asyncio.sleep(self.config['sync_interval'])except Exception as e:logger.error(f"云同步失败: {e}")await asyncio.sleep(self.config['sync_interval'])def sync_data_to_cloud(self):"""将数据同步到云端"""try:# 获取未同步的数据unsynced_data = self.db_session.query(SensorData).filter(SensorData.processed == False).limit(100).all()if not unsynced_data:logger.info("没有需要同步的数据")return# 准备数据发送data_to_send = []for data in unsynced_data:data_to_send.append({'id': data.id,'device_id': data.device_id,'sensor_type': data.sensor_type,'value': data.value,'timestamp': data.timestamp.isoformat(),'is_anomaly': data.is_anomaly})# 发送数据到云端payload = {'edge_id': self.config['edge_id'],'data': data_to_send,'timestamp': datetime.datetime.now().isoformat()}self.mqtt_client.publish(self.cloud_topic, json.dumps(payload))# 标记数据为已同步for data in unsynced_data:data.processed = Trueself.db_session.commit()logger.info(f"已同步 {len(unsynced_data)} 条数据到云端")except Exception as e:self.db_session.rollback()logger.error(f"同步数据到云端时出错: {e}")# 主应用
async def main():# 初始化数据库会话session = Session()# 初始化异常检测器anomaly_detector = AnomalyDetector(window_size=CONFIG['anomaly_detection_window'])# 初始化数据处理器data_processor = DataProcessor(CONFIG, session, anomaly_detector)# 初始化云同步服务cloud_sync = CloudSyncService(CONFIG, session, None)  # MQTT客户端稍后设置# 初始化MQTT客户端mqtt_client = EdgeMQTTClient(CONFIG, data_processor, cloud_sync)mqtt_client.connect()# 设置MQTT客户端到云同步服务cloud_sync.mqtt_client = mqtt_client# 启动云同步任务sync_task = asyncio.create_task(cloud_sync.start_periodic_sync())# 保持应用运行try:await sync_taskexcept KeyboardInterrupt:logger.info("应用被用户中断")finally:# 清理资源mqtt_client.disconnect()session.close()logger.info("应用已停止")if __name__ == "__main__":asyncio.run(main())