Python實例題：基于邊緣計算的智能物聯網系統

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())