Python 物聯網(IoT)與邊緣計算開發實戰(1)

Python 物聯網(IoT)與邊緣計算開發實戰
https://www.python.org/static/community_logos/python-logo-master-v3-TM.png

物聯網基礎與硬件交互
Raspberry Pi GPIO控制
python
import RPi.GPIO as GPIO
import time

# 設置GPIO模式
GPIO.setmode(GPIO.BCM)
GPIO.setwarnings(False)

# 定義引腳
LED_PIN = 17
BUTTON_PIN = 18

# 初始化引腳
GPIO.setup(LED_PIN, GPIO.OUT)
GPIO.setup(BUTTON_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)

def blink_led(times, speed=0.5):
? ? """LED閃爍效果"""
? ? for _ in range(times):
? ? ? ? GPIO.output(LED_PIN, GPIO.HIGH)
? ? ? ? time.sleep(speed)
? ? ? ? GPIO.output(LED_PIN, GPIO.LOW)
? ? ? ? time.sleep(speed)

try:
? ? print("按下按鈕控制LED (Ctrl+C退出)")
? ? while True:
? ? ? ? if GPIO.input(BUTTON_PIN) == GPIO.LOW:
? ? ? ? ? ? print("按鈕按下 - LED閃爍")
? ? ? ? ? ? blink_led(3, 0.3)
? ? ? ? ? ? time.sleep(0.5) ?# 防抖延遲
finally:
? ? GPIO.cleanup() ?# 清理GPIO設置
https://www.raspberrypi.com/documentation/computers/images/GPIO-Pinout-Diagram-2.png

傳感器數據采集
python
import Adafruit_DHT
import time

# 設置傳感器類型和引腳
DHT_SENSOR = Adafruit_DHT.DHT22
DHT_PIN = 4

def read_sensor():
? ? """讀取溫濕度傳感器數據"""
? ? humidity, temperature = Adafruit_DHT.read_retry(DHT_SENSOR, DHT_PIN)
? ? if humidity is not None and temperature is not None:
? ? ? ? return {
? ? ? ? ? ? 'temperature': round(temperature, 1),
? ? ? ? ? ? 'humidity': round(humidity, 1),
? ? ? ? ? ? 'timestamp': time.strftime('%Y-%m-%d %H:%M:%S')
? ? ? ? }
? ? else:
? ? ? ? print("傳感器讀取失敗!")
? ? ? ? return None

# 每5秒讀取一次數據
while True:
? ? sensor_data = read_sensor()
? ? if sensor_data:
? ? ? ? print(f"溫度: {sensor_data['temperature']}°C, 濕度: {sensor_data['humidity']}%")
? ? time.sleep(5)
物聯網通信協議
MQTT協議實現
python
import paho.mqtt.client as mqtt
import json
import time

# MQTT配置
MQTT_BROKER = "broker.example.com"
MQTT_PORT = 1883
MQTT_TOPIC_PUB = "sensor/data"
MQTT_TOPIC_SUB = "sensor/control"
CLIENT_ID = "python-iot-client"

# 傳感器模擬數據
def get_sensor_data():
? ? return {
? ? ? ? "device_id": CLIENT_ID,
? ? ? ? "temperature": 25.5 + (time.time() % 3),
? ? ? ? "humidity": 50 + (time.time() % 10),
? ? ? ? "timestamp": int(time.time())
? ? }

# 連接回調
def on_connect(client, userdata, flags, rc):
? ? print(f"連接MQTT服務器，返回碼: {rc}")
? ? client.subscribe(MQTT_TOPIC_SUB)

# 消息接收回調
def on_message(client, userdata, msg):
? ? payload = msg.payload.decode()
? ? print(f"收到消息 [{msg.topic}]: {payload}")
? ??
? ? try:
? ? ? ? command = json.loads(payload)
? ? ? ? if command.get('action') == 'reboot':
? ? ? ? ? ? print("執行重啟指令...")
? ? ? ? ? ? # 這里添加重啟邏輯

? ? except json.JSONDecodeError:
? ? ? ? print("無效的JSON消息")

# 創建MQTT客戶端
client = mqtt.Client(CLIENT_ID)
client.on_connect = on_connect
client.on_message = on_message

# 連接并啟動循環
client.connect(MQTT_BROKER, MQTT_PORT, 60)
client.loop_start()

try:
? ? while True:
? ? ? ? # 發布傳感器數據
? ? ? ? sensor_data = get_sensor_data()
? ? ? ? client.publish(MQTT_TOPIC_PUB, json.dumps(sensor_data))
? ? ? ? print(f"發布數據: {sensor_data}")
? ? ? ? time.sleep(10)
except KeyboardInterrupt:
? ? print("斷開連接...")
? ? client.loop_stop()
? ? client.disconnect()
https://mqtt.org/assets/img/mqtt-publish-subscribe.png

CoAP協議實現
python
from aiocoap import *
import asyncio
import time

async def coap_server():
? ? """CoAP服務器實現"""
? ? protocol = await Context.create_server_context(CoAPServer())

? ? # 持續運行
? ? print("CoAP服務器啟動...")
? ? await asyncio.get_running_loop().create_future()

class CoAPServer(Resource):
? ? """CoAP資源處理"""
? ??
? ? def __init__(self):
? ? ? ? super().__init__()
? ? ? ? self.sensor_data = {
? ? ? ? ? ? "temperature": 25.0,
? ? ? ? ? ? "humidity": 50.0
? ? ? ? }
? ??
? ? async def render_get(self, request):
? ? ? ? """處理GET請求"""
? ? ? ? self.update_sensor_data()
? ? ? ? payload = json.dumps(self.sensor_data).encode('utf-8')
? ? ? ? return Message(payload=payload)
? ??
? ? async def render_post(self, request):
? ? ? ? """處理POST請求"""
? ? ? ? try:
? ? ? ? ? ? payload = json.loads(request.payload.decode('utf-8'))
? ? ? ? ? ? if 'set_temp' in payload:
? ? ? ? ? ? ? ? self.sensor_data['temperature'] = payload['set_temp']
? ? ? ? ? ? if 'set_humidity' in payload:
? ? ? ? ? ? ? ? self.sensor_data['humidity'] = payload['set_humidity']
? ? ? ? ? ??
? ? ? ? ? ? return Message(code=CHANGED,?
? ? ? ? ? ? ? ? ? ? ? ? ?payload=b"Settings updated")
? ? ? ? except:
? ? ? ? ? ? return Message(code=BAD_REQUEST,?
? ? ? ? ? ? ? ? ? ? ? ? ?payload=b"Invalid request")

? ? def update_sensor_data(self):
? ? ? ? """更新傳感器數據(模擬)"""
? ? ? ? self.sensor_data = {
? ? ? ? ? ? "temperature": 25.0 + (time.time() % 3),
? ? ? ? ? ? "humidity": 50.0 + (time.time() % 10),
? ? ? ? ? ? "timestamp": int(time.time())
? ? ? ? }

async def coap_client():
? ? """CoAP客戶端實現"""
? ? protocol = await Context.create_client_context()
? ??
? ? # 獲取數據
? ? request = Message(code=GET, uri='coap://localhost/sensor')
? ? try:
? ? ? ? response = await protocol.request(request).response
? ? ? ? print(f"收到響應: {response.payload.decode()}")
? ? except Exception as e:
? ? ? ? print(f"請求失敗: {e}")
? ??
? ? # 設置數據
? ? payload = {"set_temp": 26.5, "set_humidity": 55.0}
? ? request = Message(code=POST,?
? ? ? ? ? ? ? ? ? ? ?payload=json.dumps(payload).encode(),
? ? ? ? ? ? ? ? ? ? ?uri='coap://localhost/sensor')
? ? try:
? ? ? ? response = await protocol.request(request).response
? ? ? ? print(f"設置響應: {response.payload.decode()}")
? ? except Exception as e:
? ? ? ? print(f"設置失敗: {e}")

# 運行示例
async def main():
? ? server_task = asyncio.create_task(coap_server())
? ? await asyncio.sleep(1) ?# 等待服務器啟動
? ? await coap_client()
? ? server_task.cancel()

asyncio.run(main())
邊緣計算框架
使用MicroPython
python
# ESP32 MicroPython示例
import machine
import network
import urequests
import ujson
from time import sleep

# 配置WiFi
WIFI_SSID = "your_wifi"
WIFI_PASS = "your_password"

def connect_wifi():
? ? sta_if = network.WLAN(network.STA_IF)
? ? if not sta_if.isconnected():
? ? ? ? print("連接WiFi...")
? ? ? ? sta_if.active(True)
? ? ? ? sta_if.connect(WIFI_SSID, WIFI_PASS)
? ? ? ? while not sta_if.isconnected():
? ? ? ? ? ? pass
? ? print("網絡配置:", sta_if.ifconfig())

# 讀取傳感器(模擬)
def read_sensor():
? ? return {
? ? ? ? "temp": 25 + machine.rng() % 5,
? ? ? ? "humidity": 50 + machine.rng() % 10
? ? }

# 邊緣計算處理
def process_data(data):
? ? # 簡單異常檢測
? ? if data['temp'] > 30 or data['humidity'] > 80:
? ? ? ? data['alert'] = True
? ? else:
? ? ? ? data['alert'] = False
? ? return data

# 主循環
connect_wifi()
while True:
? ? sensor_data = read_sensor()
? ? processed_data = process_data(sensor_data)
? ??
? ? if processed_data['alert']:
? ? ? ? print("警報狀態! 發送數據...")
? ? ? ? response = urequests.post(
? ? ? ? ? ? "api.example.com/alerts",
? ? ? ? ? ? json=processed_data,
? ? ? ? ? ? headers={'Content-Type': 'application/json'}
? ? ? ? )
? ? ? ? print("響應:", response.text)
? ? ? ? response.close()
? ??
? ? sleep(60) ?# 每分鐘檢查一次
使用EdgeX Foundry
python
import requests
import json
import time

# EdgeX配置
EDGEX_URL = "localhost:48080/api/v1"
DEVICE_NAME = "temperature-sensor"

def register_device():
? ? """注冊設備到EdgeX"""
? ? device = {
? ? ? ? "name": DEVICE_NAME,
? ? ? ? "description": "Python IoT溫度傳感器",
? ? ? ? "adminState": "UNLOCKED",
? ? ? ? "operatingState": "ENABLED",
? ? ? ? "protocols": {
? ? ? ? ? ? "other": {
? ? ? ? ? ? ? ? "Address": "virtual01",
? ? ? ? ? ? ? ? "Protocol": "300"
? ? ? ? ? ? }
? ? ? ? }
? ? }
? ??
? ? response = requests.post(
? ? ? ? f"{EDGEX_URL}/device",
? ? ? ? json=device,
? ? ? ? headers={"Content-Type": "application/json"}
? ? )
? ? return response.json()

def send_reading(value):
? ? """發送傳感器讀數"""
? ? reading = {
? ? ? ? "device": DEVICE_NAME,
? ? ? ? "readings": [
? ? ? ? ? ? {
? ? ? ? ? ? ? ? "name": "Temperature",
? ? ? ? ? ? ? ? "value": str(value)
? ? ? ? ? ? }
? ? ? ? ]
? ? }
? ??
? ? response = requests.post(
? ? ? ? f"{EDGEX_URL}/reading",
? ? ? ? json=reading,
? ? ? ? headers={"Content-Type": "application/json"}
? ? )
? ? return response.status_code == 200

# 模擬設備運行
print("注冊設備...")
register_device()

print("開始發送傳感器數據...")
while True:
? ? temp = 20 + (time.time() % 10) ?# 模擬溫度波動
? ? if send_reading(temp):
? ? ? ? print(f"發送溫度數據: {temp}°C")
? ? else:
? ? ? ? print("發送數據失敗")
? ? time.sleep(5)
https://docs.edgexfoundry.org/1.3/_images/EdgeX_arch.png

物聯網數據處理
實時數據流處理
python
import pyarrow.flight as flight
import pandas as pd
import numpy as np

class FlightServer(flight.FlightServerBase):
? ? """Arrow Flight服務器實現"""
? ??
? ? def __init__(self, location):
? ? ? ? super().__init__(location)
? ? ? ? self.data = pd.DataFrame({
? ? ? ? ? ? 'timestamp': pd.date_range('2023-01-01', periods=100, freq='s'),
? ? ? ? ? ? 'value': np.random.randn(100)
? ? ? ? })
? ??
? ? def do_get(self, context, ticket):
? ? ? ? """處理數據獲取請求"""
? ? ? ? df = self.data[self.data['timestamp'] > pd.Timestamp.now() - pd.Timedelta('1min')]
? ? ? ? table = pa.Table.from_pandas(df)
? ? ? ? return flight.RecordBatchStream(table)

class DataProcessor:
? ? """實時數據處理"""
? ??
? ? def __init__(self, server_url):
? ? ? ? self.client = flight.FlightClient(server_url)
? ??
? ? def process_stream(self):
? ? ? ? """處理實時數據流"""
? ? ? ? while True:
? ? ? ? ? ? try:
? ? ? ? ? ? ? ? # 獲取最近1分鐘數據
? ? ? ? ? ? ? ? descriptor = flight.FlightDescriptor.for_command(b"latest_data")
? ? ? ? ? ? ? ? flight_info = self.client.get_flight_info(descriptor)
? ? ? ? ? ? ? ??
? ? ? ? ? ? ? ? for endpoint in flight_info.endpoints:
? ? ? ? ? ? ? ? ? ? for location in endpoint.locations:
? ? ? ? ? ? ? ? ? ? ? ? reader = self.client.do_get(location.ticket)
? ? ? ? ? ? ? ? ? ? ? ? batch = reader.read_all()
? ? ? ? ? ? ? ? ? ? ? ? df = batch.to_pandas()
? ? ? ? ? ? ? ? ? ? ? ??
? ? ? ? ? ? ? ? ? ? ? ? # 實時分析
? ? ? ? ? ? ? ? ? ? ? ? if not df.empty:
? ? ? ? ? ? ? ? ? ? ? ? ? ? avg = df['value'].mean()
? ? ? ? ? ? ? ? ? ? ? ? ? ? max_val = df['value'].max()
? ? ? ? ? ? ? ? ? ? ? ? ? ? print(f"平均值: {avg:.2f}, 最大值: {max_val:.2f}")
? ? ? ? ? ??
? ? ? ? ? ? except Exception as e:
? ? ? ? ? ? ? ? print(f"處理錯誤: {e}")
? ? ? ? ? ??
? ? ? ? ? ? time.sleep(10)

# 啟動服務器
server = FlightServer("grpc://0.0.0.0:8815")
server_thread = threading.Thread(target=server.serve)
server_thread.start()

# 啟動客戶端處理
processor = DataProcessor("grpc://localhost:8815")
processor.process_stream()
時序數據庫集成
python
import influxdb_client
from influxdb_client.client.write_api import SYNCHRONOUS

class InfluxDBManager:
? ? """InfluxDB時序數據庫管理"""
? ??
? ? def __init__(self, url, token, org, bucket):
? ? ? ? self.client = influxdb_client.InfluxDBClient(
? ? ? ? ? ? url=url,
? ? ? ? ? ? token=token,
? ? ? ? ? ? org=org
? ? ? ? )
? ? ? ? self.bucket = bucket
? ? ? ? self.write_api = self.client.write_api(write_options=SYNCHRONOUS)
? ? ? ? self.query_api = self.client.query_api()
? ??
? ? def write_data(self, measurement, tags, fields):
? ? ? ? """寫入數據點"""
? ? ? ? point = influxdb_client.Point(measurement)
? ? ? ??
? ? ? ? # 添加標簽
? ? ? ? for tag_key, tag_value in tags.items():
? ? ? ? ? ? point.tag(tag_key, tag_value)
? ? ? ??
? ? ? ? # 添加字段
? ? ? ? for field_key, field_value in fields.items():
? ? ? ? ? ? point.field(field_key, field_value)
? ? ? ??
? ? ? ? # 寫入數據庫
? ? ? ? self.write_api.write(bucket=self.bucket, record=point)
? ??
? ? def query_data(self, query):
? ? ? ? """查詢數據"""
? ? ? ? result = self.query_api.query(query)
? ? ? ? records = []
? ? ? ??
? ? ? ? for table in result:
? ? ? ? ? ? for record in table.records:
? ? ? ? ? ? ? ? records.append({
? ? ? ? ? ? ? ? ? ? 'time': record.get_time(),
? ? ? ? ? ? ? ? ? ? 'measurement': record.get_measurement(),
? ? ? ? ? ? ? ? ? ? **record.values
? ? ? ? ? ? ? ? })
? ? ? ??
? ? ? ? return records

# 使用示例
influx_mgr = InfluxDBManager(
? ? url="localhost:8086",
? ? token="your-token",
? ? org="your-org",
? ? bucket="iot-data"
)

# 寫入傳感器數據
influx_mgr.write_data(
? ? measurement="temperature",
? ? tags={"location": "room1", "device": "sensor1"},
? ? fields={"value": 25.3}
)

# 查詢最近1小時數據
query = """
from(bucket: "iot-data")
? |> range(start: -1h)
? |> filter(fn: (r) => r._measurement == "temperature")
"""
data = influx_mgr.query_data(query)
print("查詢結果:", data)
物聯網安全
設備認證與加密
python
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.backends import default_backend
import os

class IoTDeviceSecurity:
? ? """物聯網設備安全類"""
? ??
? ? def __init__(self):
? ? ? ? # 生成ECDSA密鑰對
? ? ? ? self.private_key = ec.generate_private_key(
? ? ? ? ? ? ec.SECP256R1(), default_backend()
? ? ? ? )
? ? ? ? self.public_key = self.private_key.public_key()
? ? ? ??
? ? ? ? # 預共享密鑰(實際應用中應從安全存儲獲取)
? ? ? ? self.shared_secret = os.urandom(32)
? ??
? ? def get_public_key_pem(self):
? ? ? ? """獲取PEM格式的公鑰"""
? ? ? ? return self.public_key.public_bytes(
? ? ? ? ? ? encoding=serialization.Encoding.PEM,
? ? ? ? ? ? format=serialization.PublicFormat.SubjectPublicKeyInfo
? ? ? ? )
? ??
? ? def sign_data(self, data):
? ? ? ? """簽名數據"""
? ? ? ? if isinstance(data, str):
? ? ? ? ? ? data = data.encode('utf-8')
? ? ? ??
? ? ? ? signature = self.private_key.sign(
? ? ? ? ? ? data,
? ? ? ? ? ? ec.ECDSA(hashes.SHA256())
? ? ? ? )
? ? ? ? return signature
? ??
? ? def verify_signature(self, public_key_pem, data, signature):
? ? ? ? """驗證簽名"""
? ? ? ? public_key = serialization.load_pem_public_key(
? ? ? ? ? ? public_key_pem,
? ? ? ? ? ? backend=default_backend()
? ? ? ? )
? ? ? ??
? ? ? ? if isinstance(data, str):
? ? ? ? ? ? data = data.encode('utf-8')
? ? ? ??
? ? ? ? try:
? ? ? ? ? ? public_key.verify(
? ? ? ? ? ? ? ? signature,
? ? ? ? ? ? ? ? data,
? ? ? ? ? ? ? ? ec.ECDSA(hashes.SHA256())
? ? ? ? ? ? )
? ? ? ? ? ? return True
? ? ? ? except:
? ? ? ? ? ? return False
? ??
? ? def derive_session_key(self, peer_public_key_pem):
? ? ? ? """派生會話密鑰"""
? ? ? ? peer_public_key = serialization.load_pem_public_key(
? ? ? ? ? ? peer_public_key_pem,
? ? ? ? ? ? backend=default_backend()
? ? ? ? )
? ? ? ??
? ? ? ? shared_key = self.private_key.exchange(
? ? ? ? ? ? ec.ECDH(), peer_public_key
? ? ? ? )
? ? ? ??
? ? ? ? # 使用HKDF派生密鑰
? ? ? ? derived_key = HKDF(
? ? ? ? ? ? algorithm=hashes.SHA256(),
? ? ? ? ? ? length=32,
? ? ? ? ? ? salt=None,
? ? ? ? ? ? info=b'session-key',
? ? ? ? ? ? backend=default_backend()
? ? ? ? ).derive(shared_key)
? ? ? ??
? ? ? ? return derived_key

# 使用示例
device1 = IoTDeviceSecurity()
device2 = IoTDeviceSecurity()

# 交換公鑰
device1_pubkey = device1.get_public_key_pem()
device2_pubkey = device2.get_public_key_pem()

# 派生會話密鑰
session_key1 = device1.derive_session_key(device2_pubkey)
session_key2 = device2.derive_session_key(device1_pubkey)

print("會話密鑰匹配:", session_key1 == session_key2)
安全固件更新
python
import hashlib
import hmac
import requests
import tempfile
import subprocess

class SecureFirmwareUpdater:
? ? """安全固件更新"""
? ??
? ? def __init__(self, device_id, secret_key, update_server):
? ? ? ? self.device_id = device_id
? ? ? ? self.secret_key = secret_key.encode('utf-8')
? ? ? ? self.update_server = update_server
? ??
? ? def check_update(self):
? ? ? ? """檢查更新"""
? ? ? ? # 創建認證簽名
? ? ? ? nonce = os.urandom(16).hex()
? ? ? ? message = f"{self.device_id}:{nonce}".encode('utf-8')
? ? ? ? signature = hmac.new(
? ? ? ? ? ? self.secret_key,?
? ? ? ? ? ? message,?
? ? ? ? ? ? hashlib.sha256
? ? ? ? ).hexdigest()
? ? ? ??
? ? ? ? # 發送認證請求
? ? ? ? response = requests.get(
? ? ? ? ? ? f"{self.update_server}/check-update",
? ? ? ? ? ? headers={
? ? ? ? ? ? ? ? "Device-ID": self.device_id,
? ? ? ? ? ? ? ? "Nonce": nonce,
? ? ? ? ? ? ? ? "Signature": signature
? ? ? ? ? ? }
? ? ? ? )
? ? ? ??
? ? ? ? if response.status_code == 200:
? ? ? ? ? ? return response.json()
? ? ? ? else:
? ? ? ? ? ? print(f"檢查更新失敗: {response.text}")
? ? ? ? ? ? return None
? ??
? ? def download_firmware(self, version, checksum):
? ? ? ? """下載固件"""
? ? ? ? # 創建臨時文件
? ? ? ? temp_file = tempfile.NamedTemporaryFile(delete=False)
? ? ? ??
? ? ? ? try:
? ? ? ? ? ? # 流式下載固件
? ? ? ? ? ? response = requests.get(
? ? ? ? ? ? ? ? f"{self.update_server}/firmware/{version}",
? ? ? ? ? ? ? ? stream=True
? ? ? ? ? ? )
? ? ? ? ? ??
? ? ? ? ? ? # 計算下載文件的哈希
? ? ? ? ? ? sha256 = hashlib.sha256()
? ? ? ? ? ??
? ? ? ? ? ? for chunk in response.iter_content(chunk_size=8192):
? ? ? ? ? ? ? ? temp_file.write(chunk)
? ? ? ? ? ? ? ? sha256.update(chunk)
? ? ? ? ? ??
? ? ? ? ? ? temp_file.close()
? ? ? ? ? ??
? ? ? ? ? ? # 驗證校驗和
? ? ? ? ? ? if sha256.hexdigest() != checksum:
? ? ? ? ? ? ? ? os.unlink(temp_file.name)
? ? ? ? ? ? ? ? raise ValueError("固件校驗和不匹配")
? ? ? ? ? ??
? ? ? ? ? ? return temp_file.name
? ? ? ? except:
? ? ? ? ? ? os.unlink(temp_file.name)
? ? ? ? ? ? raise
? ??
? ? def apply_update(self, firmware_path):
? ? ? ? """應用更新"""
? ? ? ? # 驗證固件簽名(示例)
? ? ? ? if not self.verify_firmware(firmware_path):
? ? ? ? ? ? raise ValueError("固件簽名驗證失敗")
? ? ? ??
? ? ? ? # 執行更新腳本(實際實現依平臺而定)
? ? ? ? result = subprocess.run(
? ? ? ? ? ? ["/bin/sh", firmware_path],
? ? ? ? ? ? capture_output=True,
? ? ? ? ? ? text=True
? ? ? ? )
? ? ? ??
? ? ? ? if result.returncode != 0:
? ? ? ? ? ? print(f"更新失敗: {result.stderr}")
? ? ? ? ? ? return False
? ? ? ??
? ? ? ? print("固件更新成功!")
? ? ? ? return True
? ??
? ? def verify_firmware(self, firmware_path):
? ? ? ? """驗證固件簽名"""
? ? ? ? # 這里應實現實際的簽名驗證邏輯
? ? ? ? # 示例中僅檢查文件大小
? ? ? ? return os.path.getsize(firmware_path) > 0

# 使用示例
updater = SecureFirmwareUpdater(
? ? device_id="device-123",
? ? secret_key="your-secret-key",
? ? update_server="firmware.example.com"
)

update_info = updater.check_update()
if update_info and update_info['available']:
? ? print(f"發現新版本: {update_info['version']}")
? ? try:
? ? ? ? firmware_path = updater.download_firmware(
? ? ? ? ? ? update_info['version'],
? ? ? ? ? ? update_info['checksum']
? ? ? ? )
? ? ? ? if updater.apply_update(firmware_path):
? ? ? ? ? ? print("設備需要重啟以完成更新")
? ? except Exception as e:
? ? ? ? print(f"更新失敗: {str(e)}")
else:
? ? print("設備固件已是最新")
物聯網可視化
實時儀表盤
python
import dash
from dash import dcc, html
from dash.dependencies import Input, Output
import plotly.graph_objs as go
import pandas as pd
import random
from datetime import datetime

# 創建Dash應用
app = dash.Dash(__name__)

# 初始數據
initial_data = pd.DataFrame({
? ? 'timestamp': [datetime.now()],
? ? 'temperature': [25.0],
? ? 'humidity': [50.0]
})

# 應用布局
app.layout = html.Div([
? ? html.H1("物聯網設備監控儀表盤"),
? ??
? ? dcc.Interval(
? ? ? ? id='interval-component',
? ? ? ? interval=5*1000, ?# 5秒
? ? ? ? n_intervals=0
? ? ),
? ??
? ? html.Div([
? ? ? ? html.Div([
? ? ? ? ? ? dcc.Graph(id='temp-gauge'),
? ? ? ? ? ? html.H3("當前溫度", style={'text-align': 'center'})
? ? ? ? ], className="six columns"),
? ? ? ??
? ? ? ? html.Div([
? ? ? ? ? ? dcc.Graph(id='humidity-gauge'),
? ? ? ? ? ? html.H3("當前濕度", style={'text-align': 'center'})
? ? ? ? ], className="six columns")
? ? ], className="row"),
? ??
? ? dcc.Graph(id='temp-trend'),
? ??
? ? html.Div(id='alerts-container')
], className="container")

# 回調函數 - 更新數據
@app.callback(
? ? Output('temp-gauge', 'figure'),
? ? Output('humidity-gauge', 'figure'),
? ? Output('temp-trend', 'figure'),
? ? Output('alerts-container', 'children'),
? ? Input('interval-component', 'n_intervals')
)
def update_metrics(n):
? ? # 模擬新數據
? ? new_data = {
? ? ? ? 'timestamp': datetime.now(),
? ? ? ? 'temperature': 25 + random.uniform(-2, 2),
? ? ? ? 'humidity': 50 + random.uniform(-5, 5)
? ? }
? ??
? ? # 更新數據集
? ? global initial_data
? ? initial_data = initial_data.append(new_data, ignore_index=True)
? ??
? ? # 溫度儀表
? ? temp_gauge = go.Figure(go.Indicator(
? ? ? ? mode="gauge+number",
? ? ? ? value=new_data['temperature'],
? ? ? ? domain={'x': [0, 1], 'y': [0, 1]},
? ? ? ? gauge={
? ? ? ? ? ? 'axis': {'range': [None, 40]},
? ? ? ? ? ? 'steps': [
? ? ? ? ? ? ? ? {'range': [0, 20], 'color': "lightgray"},
? ? ? ? ? ? ? ? {'range': [20, 30], 'color': "gray"},
? ? ? ? ? ? ? ? {'range': [30, 40], 'color': "red"}
? ? ? ? ? ? ],
? ? ? ? ? ? 'threshold': {
? ? ? ? ? ? ? ? 'line': {'color': "black", 'width': 4},
? ? ? ? ? ? ? ? 'thickness': 0.75,
? ? ? ? ? ? ? ? 'value': 30
? ? ? ? ? ? }
? ? ? ? }
? ? ))
? ? temp_gauge.update_layout(height=300, margin=dict(t=0, b=0))
? ??
? ? # 濕度儀表
? ? humidity_gauge = go.Figure(go.Indicator(
? ? ? ? mode="gauge+number",
? ? ? ? value=new_data['humidity'],
? ? ? ? domain={'x': [0, 1], 'y': [0, 1]},
? ? ? ? gauge={
? ? ? ? ? ? 'axis': {'range': [0, 100]},
? ? ? ? ? ? 'steps': [
? ? ? ? ? ? ? ? {'range': [0, 30], 'color': "red"},
? ? ? ? ? ? ? ? {'range': [30, 70], 'color': "lightgray"},
? ? ? ? ? ? ? ? {'range': [70, 100], 'color': "blue"}
? ? ? ? ? ? ],
? ? ? ? ? ? 'threshold': {
? ? ? ? ? ? ? ? 'line': {'color': "black", 'width': 4},
? ? ? ? ? ? ? ? 'thickness': 0.75,
? ? ? ? ? ? ? ? 'value': 70
? ? ? ? ? ? }
? ? ? ? }
? ? ))
? ? humidity_gauge.update_layout(height=300, margin=dict(t=0, b=0))
? ??
? ? # 溫度趨勢圖
? ? temp_trend = go.Figure()
? ? temp_trend.add_trace(go.Scatter(
? ? ? ? x=initial_data['timestamp'],
? ? ? ? y=initial_data['temperature'],
? ? ? ? name='溫度',
? ? ? ? line=dict(color='red', width=2)
? ? ))
? ? temp_trend.add_trace(go.Scatter(
? ? ? ? x=initial_data['timestamp'],
? ? ? ? y=initial_data['humidity'],
? ? ? ? name='濕度',
? ? ? ? yaxis='y2',
? ? ? ? line=dict(color='blue', width=2)
? ? ))
? ? temp_trend.update_layout(
? ? ? ? yaxis=dict(title='溫度 (°C)'),
? ? ? ? yaxis2=dict(
? ? ? ? ? ? title='濕度 (%)',
? ? ? ? ? ? overlaying='y',
? ? ? ? ? ? side='right'
? ? ? ? ),
? ? ? ? hovermode="x unified"
? ? )
? ??
? ? # 警報信息
? ? alerts = []
? ? if new_data['temperature'] > 28:
? ? ? ? alerts.append(html.Div(
? ? ? ? ? ? f"高溫警報! 當前溫度: {new_data['temperature']:.1f}°C",
? ? ? ? ? ? style={
? ? ? ? ? ? ? ? 'color': 'white',
? ? ? ? ? ? ? ? 'background': 'red',
? ? ? ? ? ? ? ? 'padding': '10px',
? ? ? ? ? ? ? ? 'margin': '10px 0',
? ? ? ? ? ? ? ? 'border-radius': '5px'
? ? ? ? ? ? }
? ? ? ? ))
? ? if new_data['humidity'] > 70:
? ? ? ? alerts.append(html.Div(
? ? ? ? ? ? f"高濕度警報! 當前濕度: {new_data['humidity']:.1f}%",
? ? ? ? ? ? style={
? ? ? ? ? ? ? ? 'color': 'white',
? ? ? ? ? ? ? ? 'background': 'blue',
? ? ? ? ? ? ? ? 'padding': '10px',
? ? ? ? ? ? ? ? 'margin': '10px 0',
? ? ? ? ? ? ? ? 'border-radius': '5px'
? ? ? ? ? ? }
? ? ? ? ))
? ??
? ? return temp_gauge, humidity_gauge, temp_trend, alerts

# 運行應用
if __name__ == '__main__':
? ? app.run_server(debug=True, host='0.0.0.0')
https://plotly.com/python/static/images/dash-dashboards/iot-dashboard.png

結語與學習路徑
https://www.python.org/static/community_logos/python-logo-master-v3-TM.png

通過這十一篇系列教程，你已經掌握了：

物聯網硬件交互與傳感器集成

物聯網通信協議(MQTT/CoAP)

邊緣計算框架與應用

物聯網數據處理與分析

物聯網安全實踐

實時可視化儀表盤開發

進階學習方向：

專業領域深入：

工業物聯網(IIoT)平臺開發

智能家居系統集成