基于树莓派的智能家居系统 AI 辅助开发与部署实践

1. 传感器数据采集与预处理模块

这是系统的'感官'。以常见的 DHT11 温湿度传感器和摄像头为例，代码不仅要能稳定读取数据，还要考虑异常处理和资源释放。

# sensor_manager.py
import time
import logging
from typing import Optional, Tuple
import adafruit_dht
import board
from picamera2 import Picamera2
import numpy as np

class SensorManager:
    """统一管理各类传感器的数据采集，遵循资源初始化 - 采集 - 清理的生命周期"""
    def __init__(self):
        self.dht_device = None
        self.camera = None
        self._init_sensors()
        self.logger = logging.getLogger(__name__)

    def _init_sensors(self):
        """初始化传感器硬件，增加重试机制"""
        try:
            # 初始化 DHT11，指定数据引脚
            self.dht_device = adafruit_dht.DHT11(board.D4)
        except Exception as e:
            self.logger.warning(f"DHT11 初始化失败：{e}. 将重试...")
            # 可在此处加入延时重试逻辑
            self.dht_device = None

        try:
            # 初始化树莓派官方摄像头
            self.camera = Picamera2()
            # 配置一个低分辨率的预览配置，用于快速捕获
            preview_config = self.camera.create_preview_configuration(main={"size": (640, 480)})
            self.camera.configure(preview_config)
            self.camera.start()
        except Exception as e:
            self.logger.error(f"摄像头初始化失败：{e}")
            self.camera = None

    def read_temperature_humidity(self) -> Optional[Tuple[float, float]]:
        """读取温湿度，返回（温度℃，湿度%）"""
        if not self.dht_device:
            return None
        try:
            temperature = self.dht_device.temperature
            humidity = self.dht_device.humidity
            # 简单的数据有效性校验
            if temperature is not None and humidity is not None:
                return temperature, humidity
        except RuntimeError as e:
            self.logger.debug(f"读取 DHT11 失败（可能为瞬时错误）: {e}")
        except Exception as e:
            self.logger.error(f"DHT11 读取异常：{e}")
        return None

    def capture_image(self, resize: Tuple[int, int] = (224, 224)) -> Optional[np.ndarray]:
        """捕获一帧图像，并调整至模型所需尺寸"""
        if not self.camera:
            return None
        try:
            # 捕获图像为 numpy 数组
            frame = self.camera.capture_array()
            # 转换为 RGB（Picamera2 默认可能为 BGR）
            if frame.shape[2] == 3:
                frame_rgb = frame[:, :, ::-1]  # BGR to RGB
            else:
                frame_rgb = frame
            # 调整尺寸
            from PIL import Image
            img = Image.fromarray(frame_rgb)
            img = img.resize(resize, Image.Resampling.BILINEAR)
            return np.array(img)
        except Exception as e:
            self.logger.error(f"图像捕获失败：{e}")
        return None

    def cleanup(self):
        """安全释放所有传感器资源"""
        if self.dht_device:
            self.dht_device.exit()
        if self.camera:
            self.camera.stop()

2. 本地 AI 推理服务封装

这是系统的'大脑'。我们将 TFLite 模型加载和推理过程封装成一个独立的、可管理的服务。

# tflite_inference_service.py
import numpy as np
import tflite_runtime.interpreter as tflite
import threading
import queue
import logging
from typing import Any, List

class TFLiteInferenceService:
    """
    封装 TFLite 模型的推理服务。
    支持单次推理和简单的批量请求队列，避免频繁初始化解释器。
    """
    def __init__(self, model_path: str, label_path: str = None):
        self.model_path = model_path
        self.interpreter = None
        self.input_details = None
        self.output_details = None
        self.labels = self._load_labels(label_path) if label_path else None
        self._init_interpreter()
        self.request_queue = queue.Queue(maxsize=10)
        self.result_dict = {}
        self.lock = threading.Lock()
        self.logger = logging.getLogger(__name__)

    def _init_interpreter(self):
        """初始化 TFLite 解释器，并分配张量"""
        try:
            # 使用 TFLite Runtime 加载模型
            self.interpreter = tflite.Interpreter(model_path=self.model_path)
            self.interpreter.allocate_tensors()
            self.input_details = self.interpreter.get_input_details()
            self.output_details = self.interpreter.get_output_details()
            self.logger.info(f"模型加载成功。输入详情：{self.input_details}")
        except Exception as e:
            self.logger.error(f"模型初始化失败：{e}")
            raise

    def _load_labels(self, label_path: str) -> List[str]:
        """加载标签文件，每行一个标签"""
        with open(label_path, 'r') as f:
            return [line.strip() for line in f.readlines()]

    def preprocess_image(self, image: np.ndarray) -> np.ndarray:
        """图像预处理：归一化、维度扩展等，需与模型训练时一致"""
        # 示例：归一化到 [0,1]，并扩展 batch 维度
        normalized = image.astype(np.float32) / 255.0
        # 假设模型输入为 [1, height, width, 3]
        return np.expand_dims(normalized, axis=0)

    def infer(self, preprocessed_input: np.ndarray) -> Any:
        """执行一次同步推理"""
        with self.lock:
            # 防止多线程同时调用解释器
            # 设置输入张量
            self.interpreter.set_tensor(self.input_details[0]['index'], preprocessed_input)
            # 执行推理
            self.interpreter.invoke()
            # 获取输出
            output_data = self.interpreter.get_tensor(self.output_details[0]['index'])
            return output_data

    def infer_with_labels(self, image: np.ndarray) -> str:
        """完整的推理流程：预处理->推理->后处理（返回标签）"""
        if self.labels is None:
            raise ValueError("未提供标签文件")
        input_data = self.preprocess_image(image)
        output = self.infer(input_data)
        # 假设是分类任务，取概率最高的类别
        predicted_class_idx = np.argmax(output[0])
        confidence = output[0][predicted_class_idx]
        label = self.labels[predicted_class_idx]
        self.logger.debug(f"推理结果：{label}, 置信度：{confidence:.2f}")
        return label

3. REST API 设计（使用 FastAPI）

这是系统的'对外接口'。通过 REST API，我们可以方便地从网页、手机 App 或其他设备查询状态或触发 AI 分析。

# main_api.py
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
from typing import Optional
import logging
from sensor_manager import SensorManager
from tflite_inference_service import TFLiteInferenceService
import json

app = FastAPI(title="树莓派智能家居 AI 中枢")

# 全局服务实例（实际生产环境应考虑依赖注入）
sensor_mgr = SensorManager()

# 假设我们有一个用于识别室内物品的模型
ai_service = TFLiteInferenceService(
    model_path="models/mobilenet_v2_float.tflite",
    label_path="models/imagenet_labels.txt"
)

class HealthCheck(BaseModel):
    status: str
    sensor_available: bool
    model_loaded: bool

@app.get("/", response_model=HealthCheck)
async def root():
    """健康检查端点，用于验证服务是否正常启动"""
    return HealthCheck(
        status="online",
        sensor_available=sensor_mgr.camera is not None,
        model_loaded=ai_service.interpreter is not None
    )

@app.get("/api/environment")
async def get_environment():
    """获取当前温湿度数据"""
    data = sensor_mgr.read_temperature_humidity()
    if data is None:
        raise HTTPException(status_code=503, detail="传感器暂不可用")
    temp, humidity = data
    return {"temperature_celsius": temp, "humidity_percent": humidity}

@app.get("/api/analyze_scene")
async def analyze_scene(background_tasks: BackgroundTasks):
    """触发一次场景分析（如图像识别），可考虑改为 POST"""
    image = sensor_mgr.capture_image()
    if image is None:
        raise HTTPException(status_code=503, detail="无法捕获图像")
    # 推理可能耗时，放入后台任务避免阻塞 API 响应
    result = ai_service.infer_with_labels(image)
    return {"scene_analysis": result, "message": "分析完成"}

# 启动命令：uvicorn main_api:app --host 0.0.0.0 --port 8000 --reload

4. 性能分析与优化：让系统跑得更稳

在资源受限的设备上，性能问题不容忽视。针对几个关键点做了测试和分析。

• 系统冷启动延迟：主要耗时在 TFLite 解释器初始化 和 加载模型。一个约 10MB 的 MobileNetV2 模型，在树莓派 4B 上加载需要约 1.5-2 秒。优化建议：对于需要快速响应的服务，可以将模型加载提前到服务启动时，并保持解释器常驻内存。
• 并发竞争风险：当多个 HTTP 请求同时触发 /api/analyze_scene 时，多个线程可能同时调用 ai_service.infer。上面的代码使用了 threading.Lock 进行保护，但这也意味着推理请求是串行的。优化建议：对于高并发场景，可以维护一个推理请求队列，或者（如果内存允许）初始化多个解释器实例组成一个小的'推理池'。
• 内存占用：使用 psutil 监控发现，运行上述服务（FastAPI + 模型加载 + 摄像头缓冲）后，树莓派 4B（4GB 内存）的常驻内存增加约 250MB。其中模型本身约占 80MB，Python 进程和库占了大头。优化建议：定期重启长时间运行的服务以释放可能的内存泄漏；考虑使用更轻量的 Web 框架（如 Bottle）或静态文件服务替代部分 API 功能。

5. 生产环境避坑指南

这些经验很多是从'踩坑'中得来的，希望能帮你少走弯路。

1. 电源管理是基石：树莓派对电源非常敏感。使用劣质电源或过长的 USB 线可能导致电压不足，引发各种玄学问题（如摄像头初始化失败、系统随机重启）。务必使用官方推荐或质量可靠的 5V/3A 电源，并考虑为 GPIO 连接的传感器提供独立的稳压模块。
2. 服务幂等性设计：你的 API，特别是控制类 API（如'关灯'），可能会被重复调用。确保同一请求多次执行的结果与执行一次相同。例如，在控制继电器时，先检查当前状态，再决定是否需要动作。
3. OTA 更新的安全性：如果你设计了远程更新功能，务必做好安全措施。至少包括：更新包签名验证（防止篡改）、版本回滚机制（更新失败可恢复）、更新过程原子性（避免出现一半新一半旧的状态）。切勿在未加密的通道上传输更新包。
4. 日志与监控：不要只用 print。配置好 logging 模块，将不同级别的日志输出到文件和控制台。关键指标（如 CPU 温度、内存使用率、API 响应时间）可以定期写入文件或发送到轻量级监控服务，便于事后排查问题。
5. 依赖冻结：使用 pip freeze > requirements.txt 来记录所有 Python 包的精确版本。在树莓派上重新部署时，使用此文件安装可以最大程度复现相同的环境，避免因库版本升级导致的兼容性问题。

6. 总结与展望

通过将 AI 辅助开发工具用于快速生成基础框架和重复代码，再结合 TensorFlow Lite 等边缘框架进行高效推理，我们构建的智能家居系统在开发效率和运行时性能上取得了不错的平衡。这套架构的核心思想是 '模块化' 和 '服务化'，每个部分职责清晰，易于调试和替换。

这个单设备架构本身已经具备了一定的实用性。但智能家居的魅力在于联动。你可以思考如何将它扩展：

• 多设备协同：能否让一个树莓派作为'主脑'，通过 MQTT 或轻量级 gRPC 与多个'从属'树莓派（负责不同房间的传感）通信，实现分布式感知与决策？
• 边缘 - 云协同：将简单的、实时性要求高的推理（如人体检测）放在树莓派上，而将复杂的、非实时的分析（如用电习惯学习）放在云端，如何设计数据流和任务调度？
• 模型持续学习：能否在保护隐私的前提下，利用边缘设备收集的数据，对模型进行微调或增量学习？

建议动手复现文中的核心模块——传感器管理和TFLite 推理服务。这是整个系统的骨架。在这个过程中，你可能会遇到没提到的问题，而这正是学习和深入理解的最佳时机。