#边缘计算：树莓派、手机端与边缘AI部署详解

"""
边缘计算核心概念：

1. 数据本地化：
   - 数据在边缘设备上处理
   - 减少数据传输到云端的需求
   - 提高数据隐私和安全性

2. 低延迟：
   - 减少网络传输时间
   - 实现实时响应
   - 适用于实时应用

3. 带宽优化：
   - 减少网络流量
   - 降低通信成本
   - 提高网络效率
"""

def edge_computing_benefits():
    """
    边缘计算优势
    """
    benefits = {
        "低延迟": "毫秒级响应，适合实时应用",
        "隐私保护": "数据本地处理，减少泄露风险",
        "带宽节省": "减少数据传输，节省网络资源",
        "可靠性": "网络中断时仍可正常工作",
        "成本效益": "减少云端计算资源使用"
    }
    
    print("边缘计算优势:")
    for benefit, desc in benefits.items():
        print(f"• {benefit}: {desc}")

edge_computing_benefits()

def edge_ai_architecture():
    """
    边缘AI架构
    """
    architecture = {
        "云端": "模型训练、复杂计算、数据存储",
        "边缘网关": "数据聚合、初步处理、缓存",
        "边缘设备": "实时推理、传感器数据处理",
        "终端设备": "数据采集、简单处理、用户交互"
    }
    
    print("边缘AI架构:")
    for layer, desc in architecture.items():
        print(f"• {layer}: {desc}")

edge_ai_architecture()

def raspberry_pi_setup():
    """
    树莓派部署示例
    """
    print("树莓派AI推理示例:")
    print("""
import torch
import cv2
import numpy as np
from PIL import Image
import time

class RaspberryPiInference:
    def __init__(self, model_path):
        self.device = torch.device('cpu')  # 树莓派使用CPU
        self.model = torch.load(model_path, map_location=self.device)
        self.model.eval()
        
        # 预处理变换
        self.transform = transforms.Compose([
            transforms.Resize((224, 224)),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406], 
                               std=[0.229, 0.224, 0.225])
        ])
    
    def predict(self, image_path):
        # 加载图像
        image = Image.open(image_path).convert('RGB')
        
        # 预处理
        input_tensor = self.transform(image).unsqueeze(0).to(self.device)
        
        # 推理
        start_time = time.time()
        with torch.no_grad():
            output = self.model(input_tensor)
        
        inference_time = time.time() - start_time
        
        # 后处理
        probabilities = torch.nn.functional.softmax(output[0], dim=0)
        predicted_class = torch.argmax(probabilities).item()
        confidence = probabilities[predicted_class].item()
        
        return {
            'class': predicted_class,
            'confidence': confidence,
            'inference_time': inference_time
        }

# 使用示例
inference = RaspberryPiInference('model.pth')
result = inference.predict('test_image.jpg')
print(f"预测类别: {result['class']}, 置信度: {result['confidence']:.2f}")
""")

def raspberry_pi_optimization():
    """
    树莓派性能优化策略
    """
    optimizations = [
        "使用轻量化模型 (MobileNet, EfficientNet)",
        "启用NEON指令集加速",
        "调整CPU频率和调度策略",
        "使用内存映射文件",
        "优化数据加载和预处理",
        "使用多线程并行处理"
    ]
    
    print("树莓派性能优化策略:")
    for i, opt in enumerate(optimizations, 1):
        print(f"{i}. {opt}")

raspberry_pi_optimization()

def raspberry_pi_monitoring():
    """
    树莓派资源监控
    """
    print("树莓派系统监控:")
    print("""
import psutil
import os

def get_system_info():
    info = {
        'cpu_percent': psutil.cpu_percent(interval=1),
        'memory_percent': psutil.virtual_memory().percent,
        'disk_usage': psutil.disk_usage('/').percent,
        'temperature': get_cpu_temperature(),
        'uptime': psutil.boot_time()
    }
    return info

def get_cpu_temperature():
    try:
        with open('/sys/class/thermal/thermal_zone0/temp', 'r') as f:
            temp = float(f.read()) / 1000.0
        return temp
    except:
        return None

# 监控系统资源
system_info = get_system_info()
print(f"CPU使用率: {system_info['cpu_percent']}%")
print(f"内存使用率: {system_info['memory_percent']}%")
print(f"CPU温度: {system_info['temperature']}°C")
""")

import tensorflow as tf
import numpy as np

def tflite_basics():
    """
    TensorFlow Lite基础概念
    """
    print("TensorFlow Lite基础:")
    print("• 轻量级: 适合移动和边缘设备")
    print("• 高效: 优化的推理引擎")
    print("• 多平台: 支持Android、iOS、嵌入式设备")
    print("• 量化支持: INT8、INT16、FLOAT16量化")

def convert_to_tflite(model_path):
    """
    转换模型为TensorFlow Lite格式
    """
    print("TensorFlow Lite转换示例:")
    print("""
# 方法1: 从SavedModel转换
converter = tf.lite.TFLiteConverter.from_saved_model('saved_model_dir')
tflite_model = converter.convert()

# 方法2: 从Keras模型转换
model = tf.keras.models.load_model('model.h5')
converter = tf.lite.TFLiteConverter.from_keras_model(model)
tflite_model = converter.convert()

# 方法3: 动态量化
converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
tflite_model = converter.convert()

# 方法4: 完整量化 (需要校准数据)
def representative_dataset():
    for _ in range(100):
        data = np.random.rand(1, 224, 224, 3).astype(np.float32)
        yield [data]

converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
converter.representative_dataset = representative_dataset
converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]
converter.inference_input_type = tf.uint8
converter.inference_output_type = tf.uint8
tflite_model = converter.convert()

# 保存模型
with open('model.tflite', 'wb') as f:
    f.write(tflite_model)
""")

convert_to_tflite("model_path")

def tflite_inference_example():
    """
    TensorFlow Lite推理示例
    """
    print("TensorFlow Lite推理示例:")
    print("""
import tensorflow as tf
import numpy as np
from PIL import Image

class TFLiteInference:
    def __init__(self, model_path):
        # 加载TFLite模型
        self.interpreter = tf.lite.Interpreter(model_path=model_path)
        self.interpreter.allocate_tensors()
        
        # 获取输入输出信息
        self.input_details = self.interpreter.get_input_details()
        self.output_details = self.interpreter.get_output_details()
        
        # 获取输入形状
        self.input_shape = self.input_details[0]['shape']
    
    def preprocess_image(self, image_path):
        # 加载并预处理图像
        image = Image.open(image_path).resize((224, 224))
        image_array = np.array(image).astype(np.float32)
        
        # 归一化
        image_array = (image_array / 255.0 - 0.5) / 0.5
        
        # 添加批次维度
        image_array = np.expand_dims(image_array, axis=0)
        
        return image_array
    
    def predict(self, image_path):
        # 预处理图像
        input_data = self.preprocess_image(image_path)
        
        # 设置输入
        self.interpreter.set_tensor(self.input_details[0]['index'], input_data)
        
        # 执行推理
        import time
        start_time = time.time()
        self.interpreter.invoke()
        inference_time = time.time() - start_time
        
        # 获取输出
        output = self.interpreter.get_tensor(self.output_details[0]['index'])
        
        # 后处理
        predictions = tf.nn.softmax(output[0]).numpy()
        predicted_class = np.argmax(predictions)
        confidence = predictions[predicted_class]
        
        return {
            'predictions': predictions,
            'predicted_class': predicted_class,
            'confidence': confidence,
            'inference_time': inference_time
        }

# 使用示例
tflite_model = TFLiteInference('model.tflite')
result = tflite_model.predict('test_image.jpg')
print(f"预测结果: 类别{result['predicted_class']}, 置信度{result['confidence']:.3f}")
""")

tflite_inference_example()

def model_optimization_techniques():
    """
    模型优化技术
    """
    techniques = {
        "模型量化": "将浮点模型转换为低精度整数模型",
        "模型剪枝": "移除不重要的权重连接",
        "知识蒸馏": "用大模型训练小模型",
        "网络架构搜索": "自动设计轻量化架构",
        "层融合": "合并相邻层减少计算",
        "稀疏化": "利用模型稀疏性减少计算"
    }
    
    print("模型优化技术:")
    for tech, desc in techniques.items():
        print(f"• {tech}: {desc}")

model_optimization_techniques()

def hardware_acceleration():
    """
    硬件加速技术
    """
    accelerators = {
        "GPU": "并行计算，适合深度学习推理",
        "NPU": "神经网络处理单元，专用AI芯片",
        "TPU": "张量处理单元，Google专用芯片",
        "VPU": "视觉处理单元，图像处理优化",
        "FPGA": "现场可编程门阵列，可定制计算"
    }
    
    print("硬件加速器:")
    for accel, desc in accelerators.items():
        print(f"• {accel}: {desc}")

hardware_acceleration()

def deployment_architectures():
    """
    边缘部署架构
    """
    architectures = {
        "纯边缘部署": "所有推理在边缘设备完成",
        "边缘-云协同": "边缘预处理，云端精处理",
        "联邦学习": "分布式训练，中心聚合",
        "边缘缓存": "缓存模型和结果减少重复计算"
    }
    
    print("部署架构:")
    for arch, desc in architectures.items():
        print(f"• {arch}: {desc}")

deployment_architectures()

def performance_monitoring():
    """
    性能监控指标
    """
    metrics = {
        "推理延迟": "单次推理所需时间",
        "吞吐量": "单位时间内处理的请求数",
        "内存使用": "模型和数据占用的内存",
        "功耗": "设备运行时的能耗",
        "CPU/GPU利用率": "处理器使用情况",
        "准确率": "模型预测准确性"
    }
    
    print("性能监控指标:")
    for metric, desc in metrics.items():
        print(f"• {metric}: {desc}")

performance_monitoring()

def edge_device_monitoring():
    """
    边缘设备监控示例
    """
    print("边缘设备监控:")
    print("""
import psutil
import time
import json

class EdgeDeviceMonitor:
    def __init__(self):
        self.metrics = []
    
    def collect_metrics(self):
        metrics = {
            'timestamp': time.time(),
            'cpu_percent': psutil.cpu_percent(interval=1),
            'memory_percent': psutil.virtual_memory().percent,
            'disk_usage': psutil.disk_usage('/').percent,
            'temperature': self.get_temperature(),
            'inference_count': self.get_inference_count(),
            'avg_latency': self.get_avg_latency()
        }
        self.metrics.append(metrics)
        return metrics
    
    def get_temperature(self):
        try:
            with open('/sys/class/thermal/thermal_zone0/temp', 'r') as f:
                return float(f.read()) / 1000.0
        except:
            return None
    
    def save_report(self, filename='edge_monitor_report.json'):
        with open(filename, 'w') as f:
            json.dump(self.metrics, f, indent=2)

# 使用示例
monitor = EdgeDeviceMonitor()
current_metrics = monitor.collect_metrics()
print(f"CPU使用率: {current_metrics['cpu_percent']}%")
print(f"内存使用率: {current_metrics['memory_percent']}%")
""")

def edge_ai_trends():
    """
    边缘AI发展趋势
    """
    trends = [
        "边缘AI芯片专用化: 更多专用AI芯片推出",
        "5G与边缘计算融合: 低延迟网络支持",
        "联邦学习普及: 隐私保护的分布式学习",
        "AutoML在边缘: 自动化模型优化",
        "边缘安全增强: 硬件级安全防护",
        "边缘云协同: 混合计算架构"
    ]
    
    print("边缘AI发展趋势:")
    for trend in trends:
        print(f"• {trend}")

edge_ai_trends()

def edge_ai_applications():
    """
    边缘AI应用场景
    """
    applications = {
        "智能家居": "语音助手、安防监控、环境感知",
        "自动驾驶": "实时感知、路径规划、决策控制",
        "工业物联网": "设备监控、预测维护、质量检测",
        "医疗健康": "可穿戴设备、远程诊断、健康监测",
        "智慧城市": "交通管理、环境监测、公共安全",
        "零售电商": "智能货架、顾客分析、个性化推荐"
    }
    
    print("边缘AI应用场景:")
    for app, desc in applications.items():
        print(f"• {app}: {desc}")

edge_ai_applications()