# 物联网边缘计算案例
# 1. 边缘节点部署
# 硬件：树莓派4
# 操作系统：Raspbian
# 数据库：MariaDB
# 2. 数据采集
# 连接传感器
# 编写数据采集脚本
#!/usr/bin/env python3
import mysql.connector
import time
import random
# 连接数据库
conn = mysql.connector.connect(
host=”fgedu.localhost”,
user=”root”,
password=”password”,
database=”iot_db”
)
cursor = conn.cursor()
# 创建数据表
cursor.execute(“CREATE TABLE IF NOT EXISTS sensor_data (id INT AUTO_INCREMENT PRIMARY KEY, sensor_id VARCHAR(50), temperature FLOAT, humidity FLOAT, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP)”)
# 模拟数据采集
while True:
sensor_id = “sensor_” + str(random.randint(1, 10))
temperature = round(random.uniform(20, 30), 2)
humidity = round(random.uniform(40, 60), 2)
# 插入数据
sql = “INSERT INTO sensor_data (sensor_id, temperature, humidity) VALUES (%s, %s, %s)”
val = (sensor_id, temperature, humidity)
cursor.execute(sql, val)
conn.commit()
print(f”Inserted data: sensor_id={sensor_id}, temperature={temperature}, humidity={humidity}”)
time.sleep(5)
# 3. 数据处理
# 编写数据处理脚本
#!/usr/bin/env python3
import mysql.connector
# 连接数据库
conn = mysql.connector.connect(
host=”fgedu.localhost”,
user=”root”,
password=”password”,
database=”iot_db”
)
cursor = conn.cursor()
# 计算平均值
cursor.execute(“SELECT AVG(temperature), AVG(humidity) FROM sensor_data WHERE timestamp > DATE_SUB(NOW(), INTERVAL 1 HOUR)”)
result = cursor.fetchone()
print(f”Average temperature: {result[0]}, Average humidity: {result[1]}”)
# 4. 数据同步
# 编写数据同步脚本
#!/bin/bash
mysqldump -u root -p iot_db > iot_db.sql
scp iot_db.sql user@cloud-server:/path/to/backup/
mysql -u root -p -h cloud-server iot_db < iot_db.sql

执行结果：

# 物联网边缘计算结果
# 数据采集：成功
# 数据处理：成功
# 数据同步：成功
# 系统运行：稳定

# 智能交通边缘计算案例
# 1. 边缘节点部署
# 硬件：工业计算机
# 操作系统：Ubuntu Server
# 数据库：MariaDB
# 2. 数据采集
# 连接摄像头和传感器
# 编写数据采集脚本
#!/bin/bash
while true;
do
# 模拟交通流量数据
vehicle_count=$(shuf -i 10-100 -n 1)
average_speed=$(shuf -i 20-80 -n 1)
# 插入数据
mysql -u root -p -e “INSERT INTO traffic_db.traffic_data (vehicle_count, average_speed) VALUES ($vehicle_count, $average_speed);
”
sleep 10;
done
# 3. 数据处理
# 编写数据处理脚本
#!/bin/bash
# 计算每小时交通流量
mysql -u root -p -e “SELECT HOUR(timestamp) as hour, SUM(vehicle_count) as total_vehicles FROM traffic_db.traffic_data WHERE DATE(timestamp) = CURDATE() GROUP BY HOUR(timestamp);
”
# 4. 数据同步
# 编写数据同步脚本
#!/bin/bash
# 每小时同步一次数据
while true;
do
mysqldump -u root -p traffic_db > traffic_db.sql
scp traffic_db.sql user@cloud-server:/path/to/backup/
mysql -u root -p -h cloud-server traffic_db < traffic_db.sql
sleep 3600;
done

执行结果：

# 智能交通边缘计算结果
# 数据采集：成功
# 数据处理：成功
# 数据同步：成功
# 交通分析：有效

# 工业互联网边缘计算案例
# 1. 边缘节点部署
# 硬件：工业网关
# 操作系统：Linux
# 数据库：MariaDB
# 2. 数据采集
# 连接工业设备
# 编写数据采集脚本
#!/usr/bin/env python3
import mysql.connector
import time
import random
# 连接数据库
conn = mysql.connector.connect(
host=”fgedu.localhost”,
user=”root”,
password=”password”,
database=”industry_db”
)
cursor = conn.cursor()
# 创建数据表
cursor.execute(“CREATE TABLE IF NOT EXISTS device_data (id INT AUTO_INCREMENT PRIMARY KEY, device_id VARCHAR(50), temperature FLOAT, pressure FLOAT, vibration FLOAT, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP)”)
# 模拟设备数据
while True:
device_id = “device_” + str(random.randint(1, 5))
temperature = round(random.uniform(50, 80), 2)
pressure = round(random.uniform(1.0, 5.0), 2)
vibration = round(random.uniform(0.1, 1.0), 2)
# 插入数据
sql = “INSERT INTO device_data (device_id, temperature, pressure, vibration) VALUES (%s, %s, %s, %s)”
val = (device_id, temperature, pressure, vibration)
cursor.execute(sql, val)
conn.commit()
print(f”Inserted data: device_id={device_id}, temperature={temperature}, pressure={pressure}, vibration={vibration}”)
time.sleep(2)
# 3. 数据处理
# 编写数据处理脚本
#!/usr/bin/env python3
import mysql.connector
# 连接数据库
conn = mysql.connector.connect(
host=”fgedu.localhost”,
user=”root”,
password=”password”,
database=”industry_db”
)
cursor = conn.cursor()
# 检测异常
cursor.execute(“SELECT device_id, temperature, pressure, vibration FROM device_data WHERE timestamp > DATE_SUB(NOW(), INTERVAL 5 MINUTE) AND (temperature > 75 OR pressure > 4.5 OR vibration > 0.8)”)
results = cursor.fetchall()
if results:
print(“Anomalies detected:”)
for row in results:
print(f”Device: {row[0]}, Temperature: {row[1]}, Pressure: {row[2]}, Vibration: {row[3]}”)
# 4. 数据同步
# 编写数据同步脚本
#!/bin/bash
# 每5分钟同步一次数据
while true;
do
mysqldump -u root -p industry_db > industry_db.sql
scp industry_db.sql user@cloud-server:/path/to/backup/
mysql -u root -p -h cloud-server industry_db < industry_db.sql
sleep 300;
done

执行结果：

# 工业互联网边缘计算结果
# 数据采集：成功
# 数据处理：成功
# 异常检测：有效
# 数据同步：成功