树莓派Python嵌入式开发：从环境搭建到物联网应用实战

1. 从“玩具”到“利器”：重新认识树莓派上的Python

很多人第一次接触树莓派，可能都是从点亮一个LED灯或者跑一个简单的“Hello World”开始的。在嵌入式开发的传统认知里，C/C++是当之无愧的王者，它们直接操作寄存器、管理内存，追求极致的性能和效率。而Python，常常被看作是“脚本语言”，是数据分析、Web开发的后台工具，似乎和底层硬件隔着十万八千里。这种刻板印象，让很多嵌入式开发者对“用Python做嵌入式软件”这件事嗤之以鼻，认为它太慢、太“重”、不够“硬核”。

但我想说的是，时代变了，我们看待工具的方式也需要变。今天，当我们谈论在树莓派上用Python进行嵌入式软件开发时，我们谈论的绝不仅仅是“玩具项目”或“教学演示”。我们谈论的是一种全新的开发范式：快速原型、高效迭代和系统整合。树莓派本身就是一个运行着完整Linux操作系统的微型计算机，它模糊了传统MCU（微控制器）和MPU（微处理器）的界限。在这个平台上，Python凭借其丰富的生态库和极高的开发效率，正从一个“可选工具”演变为解决特定复杂问题的“首选利器”。

举个例子，一个传统的基于STM32的温湿度监测系统，你可能需要花几天时间调试I2C驱动、编写数据采集逻辑、再通过串口或者简单的网络模块上传数据。而在树莓派上，用Python，你可以在一个下午内完成：用smbus2或Adafruit_DHT库读取传感器，用requests库通过HTTP将数据发送到云端服务器，同时还能用Flask快速搭建一个本地Web页面来实时显示数据，甚至用matplotlib生成历史曲线图。Python让你能快速整合网络、数据可视化和业务逻辑，这是传统嵌入式开发流程难以比拟的效率。

所以，这篇内容不是要教你用Python去写一个取代RTOS的内核，也不是去挑战毫秒级中断响应的极限。而是要和你分享，如何将Python在树莓派这个独特的嵌入式平台上的能力发挥到极致，解决那些需要连接、计算和快速呈现的复合型问题。我们将从环境搭建的“坑”说起，深入到与硬件交互的核心，再探讨如何让代码变得可靠、高效，最终构建出真正可用的嵌入式应用。

2. 基石：构建稳定且高效的Python开发环境

在树莓派上玩Python，第一步就是搭建环境。这听起来简单，但很多人在这里就踩了坑，导致后续开发各种不顺。树莓派官方系统Raspbian（现称Raspberry Pi OS）默认就安装了Python，但这恰恰是第一个陷阱。

2.1 系统Python与虚拟环境：为什么不要动/usr/bin/python3

树莓派OS为了系统组件的正常运行，自身就依赖特定版本的Python。你通过python3 --version看到的，很可能是一个较旧的版本（比如Python 3.9）。如果你一时兴起，用sudo apt upgrade python3或者从源码编译一个更新的版本直接覆盖安装，极有可能导致系统桌面环境崩溃、软件包管理器apt出错等灾难性后果。

注意：永远不要尝试升级或替换系统自带的Python解释器。你的所有项目开发，都应该在独立的虚拟环境中进行。

解决方案是使用venv。这是Python官方推荐的虚拟环境管理工具，它能为你每个项目创建一个独立的Python运行环境，包括独立的解释器、pip和库目录，与系统环境完全隔离。

# 1. 确保安装了python3-venv工具（通常已安装） # 2. 为你的项目创建一个目录并进入 mkdir my_embedded_project && cd my_embedded_project # 3. 创建虚拟环境，环境目录名为‘venv’ python3 -m venv venv # 4. 激活虚拟环境 source venv/bin/activate

激活后，你的命令行提示符前会出现(venv)字样。此时，你执行的python和pip命令，都指向你项目目录下的独立环境。在这里，你可以安全地安装任何版本的包（比如pip install numpy），而不会影响系统。

2.2 编辑器/IDE的选择：VSCode远程开发实战

在树莓派那小屏幕上直接写代码体验很差。更专业的做法是在你的主力电脑（Windows/Mac）上写代码，然后同步或直接在树莓派上运行。这里强烈推荐使用VSCode的远程开发功能。

首先，确保你的树莓派和电脑在同一个局域网，并开启了SSH服务（树莓派OS默认开启）。在电脑的VSCode中安装官方扩展“Remote - SSH”。然后，你可以通过SSH连接到树莓派的IP地址，VSCode的界面会无缝切换到远程模式，所有操作（打开终端、编辑文件、运行调试）都像是在本地操作树莓派一样。

这样做有几个巨大优势：

1. 性能：代码补全、语法检查等消耗资源的任务在你的高性能电脑上完成。
2. 便捷：直接使用你熟悉的电脑键盘、鼠标和编辑器配置。
3. 隔离：开发环境干净，文件天然存储在树莓派上，避免同步麻烦。

配置好远程连接后，在VSCode中打开树莓派上的项目文件夹，并选择你创建的虚拟环境（venv）中的Python解释器（通常在~/.vscode-server相关目录下能找到，或通过命令which python在激活的虚拟环境中查看路径）。这样，你的代码分析、调试和运行就完全基于项目独立的虚拟环境了。

2.3 依赖管理：从requirements.txt到pyproject.toml

项目依赖管理是保证可复现性的关键。传统的做法是生成一个requirements.txt文件：

# 在虚拟环境中，导出当前安装的所有包及其精确版本 pip freeze > requirements.txt # 在新环境中恢复安装 pip install -r requirements.txt

但更现代、功能更强大的方式是使用pyproject.toml文件（结合setuptools或poetry）。它不仅能管理依赖，还能定义项目元数据、构建配置等。对于嵌入式项目，明确依赖至关重要，尤其是那些需要编译原生扩展的库（如numpy,opencv-python），在树莓派上编译可能耗时很长甚至失败。使用固定版本可以确保一致性。

一个简单的pyproject.toml示例：

[build-system] requires = ["setuptools>=61.0", "wheel"] build-backend = "setuptools.build_meta" [project] name = "my_embedded_app" version = "0.1.0" dependencies = [ "RPi.GPIO>=0.7.0", "smbus2>=0.4.0", "pyserial>=3.5", "requests>=2.28.0", # 指定版本以避免不兼容 "numpy==1.24.0", ]

使用pip install -e .即可安装当前项目及其依赖。这种方式比裸奔的requirements.txt更结构化，是迈向专业项目管理的标志。

3. 与硬件对话：GPIO、总线与传感器驱动

环境搭好了，接下来就是让Python“摸得着”硬件。树莓派提供了40针的GPIO接口，这是连接外部世界的主要通道。

3.1 GPIO控制：不止于RPi.GPIO

最广为人知的库是RPi.GPIO。它简单直接，对于点亮LED、读取按钮状态等基础操作绰绰有余。

import RPi.GPIO as GPIO import time LED_PIN = 17 BUTTON_PIN = 27 GPIO.setmode(GPIO.BCM) # 使用BCM编号 GPIO.setup(LED_PIN, GPIO.OUT) GPIO.setup(BUTTON_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP) # 启用内部上拉电阻 try: while True: if GPIO.input(BUTTON_PIN) == GPIO.LOW: # 按钮按下 GPIO.output(LED_PIN, GPIO.HIGH) else: GPIO.output(LED_PIN, GPIO.LOW) time.sleep(0.05) # 简单防抖，但并非最佳实践 except KeyboardInterrupt: pass finally: GPIO.cleanup() # 清理！非常重要！

但RPi.GPIO有个问题：它是纯Python实现，通过读写/dev/mem来操作GPIO，在需要高频、精确时序（如PWM、软件模拟协议）时，性能会受限于Python的解释执行和操作系统的调度，产生不可忽略的抖动。

对于更高要求的应用，可以考虑：

• gpiozero：树莓派官方推荐的更高层抽象库。它面向对象，代码更清晰，并且内置了防抖、后台线程管理等实用功能。对于大多数应用，它的性能足够好，且API更友好。
• pigpio：这是一个C语言编写的守护进程，提供了网络接口。你的Python程序通过socket与守护进程通信，由守护进程来执行精确的硬件操作。它能提供硬件级别的PWM和时序控制，精度远高于RPi.GPIO，甚至可以实现软件模拟的精准UART、I2C。对于需要稳定PWM控制舵机、生成精确脉冲序列的应用，pigpio是首选。

3.2 I2C/SPI通信：与传感器和模块交互

树莓派内置了硬件I2C和SPI控制器。在Python中与之交互，你需要先通过raspi-config或在/boot/config.txt中启用这些接口。

对于I2C，smbus2库是smbus的现代替代品，提供了更Pythonic的接口。它底层使用Linux的I2C设备文件（如/dev/i2c-1）。

from smbus2 import SMBus, i2c_msg import time # 假设BME280气压温度湿度传感器的I2C地址是0x76 DEVICE_ADDR = 0x76 BME280_REG_ID = 0xD0 with SMBus(1) as bus: # 使用I2C总线1 # 读取芯片ID chip_id = bus.read_byte_data(DEVICE_ADDR, BME280_REG_ID) print(f"Chip ID: {chip_id:#04x}") # 应输出0x60 # 更复杂的操作：写入配置寄存器，然后读取多字节数据 # 设置测量模式 bus.write_byte_data(DEVICE_ADDR, 0xF4, 0x27) time.sleep(0.01) # 等待转换完成 # 读取温度和气压的原始数据（6个字节） read_msg = i2c_msg.read(DEVICE_ADDR, 6) bus.i2c_rdwr(read_msg) data = list(read_msg) # ... 后续进行数据转换

关键点：I2C通信是同步的，bus.read_byte_data()这类调用会阻塞直到完成。在图形界面或网络服务的主线程中长时间阻塞会导致界面卡顿。此时，应将硬件访问操作放入单独的线程中。

对于SPI，可以使用spidev库。其模式类似，但需要注意SPI的模式（CPOL, CPHA）、速度和位序。

3.3 串口通信：连接老派设备与调试

树莓派的硬件串口（/dev/ttyAMA0或/dev/serial0）默认可能被蓝牙占用，需要配置才能用于通用串口通信。使用pyserial库可以方便地进行串口读写。

import serial import time # 配置串口 ser = serial.Serial( port='/dev/ttyAMA0', # 或 '/dev/serial0' baudrate=9600, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout=1 # 读超时1秒 ) if ser.is_open: try: # 发送数据 command = b'READ_DATA\r\n' ser.write(command) # 读取响应 response = ser.readline() # 读取直到换行符或超时 print(f"Received: {response.decode('ascii').strip()}") # 或者读取指定字节 # response = ser.read(10) except Exception as e: print(f"Serial error: {e}") finally: ser.close()

串口通信的坑在于编码和缓冲。很多设备返回的是二进制数据，而非文本。使用readline()的前提是数据有明确的终止符（如\n）。对于二进制协议，你需要根据协议规范，精确计算需要读取的字节数，使用ser.read(size)。同时，注意pyserial的内部缓冲区，有时需要ser.flushInput()和ser.flushOutput()来清空旧数据。

4. 超越脚本：构建健壮的嵌入式应用架构

当你的项目从简单的“脚本”成长为需要长时间稳定运行、处理多种任务的“应用”时，代码架构就变得至关重要。直接一个while True循环里塞满所有逻辑，很快就会变得难以维护和调试。

4.1 并发与异步：不让I/O等待卡住整个世界

嵌入式应用经常需要同时做多件事：监听按钮、采集传感器数据、上传网络、更新本地显示。Python有几种并发处理方式：

1. 多线程 (threading)：适用于I/O密集型任务。例如，用一个线程专门阻塞读取串口数据，另一个线程处理用户界面。但由于GIL（全局解释器锁）的存在，多线程无法充分利用多核CPU进行并行计算。对于硬件访问，必须注意线程安全，避免多个线程同时操作同一个GPIO或总线设备。

import threading import queue import time # 创建一个线程安全的队列，用于线程间传递传感器数据 data_queue = queue.Queue() def sensor_reading_thread(): """模拟传感器读取线程""" while True: # 模拟从硬件读取数据 simulated_data = {"temp": 25.0, "humi": 50.0} data_queue.put(simulated_data) time.sleep(2) # 每2秒读一次 def data_processing_thread(): """数据处理线程""" while True: try: data = data_queue.get(timeout=1) # 处理数据，例如上传、存储、显示 print(f"Processing: {data}") # 标记任务完成 data_queue.task_done() except queue.Empty: continue # 启动线程 t1 = threading.Thread(target=sensor_reading_thread, daemon=True) t2 = threading.Thread(target=data_processing_thread, daemon=True) t1.start() t2.start() # 主线程可以去做其他事，比如运行一个Web服务器 try: while True: time.sleep(0.1) except KeyboardInterrupt: print("Exiting...")

2. 异步IO (asyncio)：这是处理大量网络连接或高并发I/O的现代方案。它使用单线程，通过协程和事件循环来实现并发，开销远小于线程。如果你的应用核心是处理HTTP请求、MQTT消息等网络I/O，asyncio是绝佳选择。但需要注意的是，大多数硬件访问库（如RPi.GPIO, smbus2, pyserial）是同步阻塞的，直接在异步函数中调用它们会阻塞整个事件循环。解决方案是使用loop.run_in_executor()将这些阻塞调用放到线程池中执行。

import asyncio import aiohttp from concurrent.futures import ThreadPoolExecutor # 创建一个线程池用于执行阻塞的硬件操作 executor = ThreadPoolExecutor(max_workers=2) async def read_sensor_async(): loop = asyncio.get_event_loop() # 将阻塞的sensor_read()函数放到线程池中运行 data = await loop.run_in_executor(executor, sensor_read) # sensor_read是同步函数 return data async def upload_data_async(data): async with aiohttp.ClientSession() as session: async with session.post('http://api.example.com/data', json=data) as resp: return await resp.text() async def main(): while True: # 并发地读取传感器和上传数据（如果上传慢，不会阻塞下一次读取） sensor_task = asyncio.create_task(read_sensor_async()) # 可以在这里做其他不依赖sensor_task的事情... data = await sensor_task upload_task = asyncio.create_task(upload_data_async(data)) # 处理上传结果或做其他事... result = await upload_task print(result) await asyncio.sleep(5) # 异步等待5秒 # 运行异步主函数 asyncio.run(main())

4.2 状态管理与事件驱动

复杂的应用会有多种状态（如“待机”、“测量”、“报警”、“上传”）。使用一个简单的状态机可以让逻辑更清晰。

from enum import Enum, auto import time class AppState(Enum): IDLE = auto() MEASURING = auto() UPLOADING = auto() ERROR = auto() class EmbeddedApp: def __init__(self): self.state = AppState.IDLE self.measurement_data = None self.error_message = "" def run_cycle(self): """主循环每次迭代调用""" if self.state == AppState.IDLE: if self._check_start_condition(): # 例如按钮按下 self.state = AppState.MEASURING elif self.state == AppState.MEASURING: self.measurement_data = self._read_sensors() if self.measurement_data: self.state = AppState.UPLOADING else: self.state = AppState.ERROR self.error_message = "Sensor read failed" elif self.state == AppState.UPLOADING: if self._upload_data(): print("Upload success") self.state = AppState.IDLE self.measurement_data = None else: self.state = AppState.ERROR self.error_message = "Upload failed" elif self.state == AppState.ERROR: print(f"Error: {self.error_message}") # 错误恢复逻辑，例如重试或复位 time.sleep(5) self.state = AppState.IDLE self.error_message = "" def _check_start_condition(self): # 检查硬件或逻辑条件 return True # 简化示例 def _read_sensors(self): # 读取硬件 return {"value": 42} def _upload_data(self): # 网络上传 return True # 使用 app = EmbeddedApp() while True: app.run_cycle() time.sleep(0.1) # 主循环频率

这种显式的状态管理，比在复杂的if-else嵌套中追踪逻辑要清晰和健壮得多，也更容易调试和扩展。

4.3 日志与错误处理：让系统可观测

一个需要7x24小时运行的嵌入式应用，完善的日志系统是诊断问题的生命线。不要再用print()了，使用logging模块。

import logging import logging.handlers def setup_logging(): logger = logging.getLogger('EmbeddedApp') logger.setLevel(logging.DEBUG) # 设置最低记录级别 # 控制台处理器 ch = logging.StreamHandler() ch.setLevel(logging.INFO) console_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') ch.setFormatter(console_formatter) logger.addHandler(ch) # 文件处理器，自动按日期或大小滚动 fh = logging.handlers.RotatingFileHandler( '/var/log/my_embedded_app/app.log', maxBytes=1024*1024, # 1MB backupCount=5 ) fh.setLevel(logging.DEBUG) file_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - [%(module)s:%(lineno)d] - %(message)s') fh.setFormatter(file_formatter) logger.addHandler(fh) return logger logger = setup_logging() try: value = read_from_i2c_sensor() if value is None: logger.error("Failed to read from I2C sensor at address 0x%02x", DEVICE_ADDR) else: logger.debug("Sensor raw value: %d", value) logger.info("Temperature: %.2f C", calculate_temperature(value)) except IOError as e: logger.exception("I2C communication error occurred") # 会自动记录异常堆栈

将日志分级（DEBUG, INFO, WARNING, ERROR, CRITICAL）并输出到文件，你可以在出问题时回头查看历史记录，而不是对着一个已经消失的控制台输出发呆。RotatingFileHandler能防止日志文件无限膨胀占满树莓派有限的存储空间。

5. 性能优化与资源管理：在限制下跳舞

树莓派性能虽强于传统MCU，但资源仍是有限的。Python应用若编写不当，容易成为“内存吞噬兽”和“CPU发烧友”。

5.1 内存管理：警惕循环引用与全局缓存

Python有自动垃圾回收（GC），但并非万能。在长时间运行的应用中，内存泄漏依然可能发生。最常见的原因是循环引用，尤其是在你自定义的类中。

class SensorNode: def __init__(self, name): self.name = name self.next_node = None # 可能指向另一个SensorNode # 创建循环引用 node_a = SensorNode('A') node_b = SensorNode('B') node_a.next_node = node_b node_b.next_node = node_a # 循环引用！ # 即使删除外部引用，node_a和node_b由于互相引用，引用计数不为零，GC需要周期回收才能处理 del node_a del node_b

对于可能创建大量临时对象的循环（如持续解析网络数据包），要注意让对象及时离开作用域。对于需要缓存的数据，使用weakref模块创建弱引用，或者定期清理缓存字典。

另一个内存杀手是无限制增长的列表或字典，例如用来存储所有历史数据。对于嵌入式应用，你需要设定上限，或者使用collections.deque指定最大长度。

from collections import deque maxlen = 1000 recent_readings = deque(maxlen=maxlen) while True: data = read_sensor() recent_readings.append(data) # 当超过1000个元素时，会自动丢弃最老的 # 这样内存占用始终恒定

5.2 CPU优化：减少不必要的计算与阻塞

• 避免在紧凑循环中频繁创建对象：例如字符串拼接，使用"".join(list_of_strings)比循环中+=效率高得多。
• 使用本地变量：在循环内频繁访问全局变量或对象的属性会有额外开销。将其赋值给一个本地变量可以提速。
• 善用time.sleep()：在轮询循环中，一个很小的sleep（如time.sleep(0.001)）可以大幅降低CPU占用率，从接近100%降到几乎为0，而响应速度依然能满足很多应用需求。这比忙等待（while True: pass）要友好得多。
• 对于密集计算，考虑使用C扩展或NumPy：如果算法涉及大量数值运算（如图像处理、信号滤波），纯Python会非常慢。使用numpy数组运算（底层是C和Fortran）可以获得数百倍的性能提升。对于最核心的算法，可以将其用Cython或C/C++实现为Python扩展模块。

5.3 电源与可靠性考量

树莓派作为Linux设备，不像单片机那样容易实现超低功耗。但对于电池供电或需要应对意外断电的场景，仍有注意事项：

• 软件看门狗：使用watchdog库或Linux的硬件看门狗（需要内核模块支持）。你的主程序需要定期“喂狗”，如果程序卡死，看门狗会触发系统重启。
• 优雅关机：监听SIGTERM或SIGINT信号，在程序退出前，安全地关闭硬件（如将GPIO设为输入、关闭文件、断开网络连接）。
• 文件系统同步：对于重要的数据文件（如配置、日志），写入后考虑调用os.fsync()或使用with open(...) as f:确保数据刷入磁盘，防止断电丢失。
• 只读文件系统：对于部署后不再更改的应用，可以将根文件系统挂载为只读，极大提高SD卡寿命和系统抗断电损坏的能力。这需要一些额外的系统配置。

6. 实战：构建一个物联网环境监测节点

让我们综合以上所有知识，设计一个简单的但具备生产级代码雏形的项目：一个室内环境监测节点，它读取BME280传感器的温湿度气压数据，通过MQTT协议发布到云端，同时在本地提供一个简单的Web界面显示实时数据。

6.1 项目结构与依赖

my_environment_monitor/ ├── pyproject.toml # 项目依赖和配置 ├── src/ │ └── env_monitor/ │ ├── __init__.py │ ├── config.py # 配置文件读取 │ ├── sensor.py # 传感器硬件抽象层 │ ├── mqtt_client.py # MQTT通信模块 │ ├── web_server.py # 简易Web服务器 │ └── main.py # 主程序入口 ├── data/ # 数据存储（如果需要） ├── logs/ # 日志目录 └── tests/ # 单元测试

pyproject.toml依赖：

[project] dependencies = [ "paho-mqtt>=1.6.0", "flask>=2.0.0", "smbus2>=0.4.0", "bme280>=0.2.0", # 一个BME280的Python驱动库 "gpiozero>=1.6.0", # 用于可能的按钮/LED状态指示 ]

6.2 核心模块拆解

sensor.py- 硬件抽象

import logging from smbus2 import SMBus import bme280 from threading import Lock logger = logging.getLogger(__name__) class BME280Sensor: def __init__(self, bus_number=1, address=0x76): self.bus_number = bus_number self.address = address self._lock = Lock() # 保证多线程下对I2C总线的安全访问 self._calibration_params = None self._initialize() def _initialize(self): """初始化传感器，读取校准参数""" try: with SMBus(self.bus_number) as bus: self._calibration_params = bme280.load_calibration_params(bus, self.address) logger.info(f"BME280 sensor initialized at I2C bus {self.bus_number}, address 0x{self.address:02x}") except Exception as e: logger.error(f"Failed to initialize BME280 sensor: {e}") self._calibration_params = None def read(self): """读取一次传感器数据。返回字典或None。""" if self._calibration_params is None: logger.warning("Sensor not properly initialized.") return None with self._lock: # 确保同一时间只有一个线程在访问I2C try: with SMBus(self.bus_number) as bus: raw_data = bme280.sample(bus, self.address, self._calibration_params) return { 'temperature_c': raw_data.temperature, 'humidity': raw_data.humidity, 'pressure_hpa': raw_data.pressure, 'timestamp': raw_data.timestamp.isoformat() if raw_data.timestamp else None } except Exception as e: logger.error(f"Error reading from sensor: {e}") return None

这个类封装了硬件细节，提供了线程安全的read()方法，并进行了基本的错误处理和日志记录。

mqtt_client.py- 网络通信

import paho.mqtt.client as mqtt import json import logging import threading import time logger = logging.getLogger(__name__) class MQTTPublisher: def __init__(self, broker, port, topic_prefix, client_id=""): self.broker = broker self.port = port self.topic_prefix = topic_prefix.rstrip('/') self.client = mqtt.Client(client_id if client_id else mqtt.base62(uuid.uuid4().int, padding=22)) self.client.on_connect = self._on_connect self.client.on_disconnect = self._on_disconnect # 连接状态和重连机制 self._connected = False self._reconnect_delay = 5 self._reconnect_thread = None def _on_connect(self, client, userdata, flags, rc): if rc == 0: self._connected = True logger.info(f"Connected to MQTT broker at {self.broker}:{self.port}") else: logger.error(f"Failed to connect, return code {rc}") def _on_disconnect(self, client, userdata, rc): self._connected = False logger.warning(f"Disconnected from MQTT broker. Code: {rc}") # 触发后台重连 self._start_reconnect() def _start_reconnect(self): if self._reconnect_thread and self._reconnect_thread.is_alive(): return self._reconnect_thread = threading.Thread(target=self._reconnect_loop, daemon=True) self._reconnect_thread.start() def _reconnect_loop(self): while not self._connected: logger.info(f"Attempting to reconnect in {self._reconnect_delay} seconds...") time.sleep(self._reconnect_delay) try: self.client.reconnect() except Exception as e: logger.error(f"Reconnect attempt failed: {e}") def connect(self, username=None, password=None): if username and password: self.client.username_pw_set(username, password) try: self.client.connect(self.broker, self.port, keepalive=60) self.client.loop_start() # 启动网络循环线程 except Exception as e: logger.error(f"Initial connection failed: {e}") self._start_reconnect() def publish_data(self, sensor_data): if not self._connected or sensor_data is None: return False topic = f"{self.topic_prefix}/environment" payload = json.dumps(sensor_data) result = self.client.publish(topic, payload, qos=1) # QoS 1: 至少送达一次 if result.rc == mqtt.MQTT_ERR_SUCCESS: logger.debug(f"Published to {topic}: {payload}") return True else: logger.error(f"Failed to publish to {topic}") return False def disconnect(self): self.client.loop_stop() self.client.disconnect()

这个类处理了MQTT的连接、断开、自动重连和消息发布，将网络通信的复杂性隔离起来。

main.py- 应用主循环

import logging import signal import sys import time from threading import Event from src.env_monitor.config import load_config from src.env_monitor.sensor import BME280Sensor from src.env_monitor.mqtt_client import MQTTPublisher from src.env_monitor.web_server import start_web_server_in_thread logger = logging.getLogger(__name__) shutdown_event = Event() def signal_handler(sig, frame): logger.info("Shutdown signal received.") shutdown_event.set() def main(): # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') # 加载配置 config = load_config('config.yaml') # 初始化组件 sensor = BME280Sensor(bus_number=config['sensor']['i2c_bus'], address=config['sensor']['i2c_addr']) mqtt_pub = MQTTPublisher( broker=config['mqtt']['broker'], port=config['mqtt']['port'], topic_prefix=config['mqtt']['topic_prefix'] ) mqtt_pub.connect(config['mqtt'].get('username'), config['mqtt'].get('password')) # 启动Web服务器（非阻塞） web_thread = start_web_server_in_thread(config['web']['host'], config['web']['port']) # 注册信号处理 signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) logger.info("Environment monitor started.") # 主循环 read_interval = config['general']['read_interval_seconds'] last_read_time = 0 while not shutdown_event.is_set(): current_time = time.time() if current_time - last_read_time >= read_interval: data = sensor.read() if data: # 更新Web服务器共享的数据（可通过全局变量或队列） update_web_data(data) # 发布到MQTT mqtt_pub.publish_data(data) logger.info(f"Read and published: Temp={data['temperature_c']:.1f}C, Humi={data['humidity']:.1f}%") last_read_time = current_time # 非忙等待，降低CPU占用 shutdown_event.wait(timeout=0.1) # 清理资源 logger.info("Shutting down...") mqtt_pub.disconnect() # 等待Web服务器线程结束 web_thread.join(timeout=5) logger.info("Shutdown complete.") if __name__ == '__main__': main()

这个主程序整合了各个模块，实现了定时读取、数据发布，并优雅地响应关机信号。

6.3 部署与自启动

开发完成后，你需要让应用在树莓派启动时自动运行。最可靠的方式是创建一个systemd服务。

在/etc/systemd/system/下创建文件env-monitor.service：

[Unit] Description=Environment Monitor Service After=network.target [Service] Type=simple User=pi WorkingDirectory=/home/pi/my_environment_monitor Environment="PATH=/home/pi/my_environment_monitor/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" ExecStart=/home/pi/my_environment_monitor/venv/bin/python -m src.env_monitor.main Restart=on-failure RestartSec=10 StandardOutput=journal StandardError=journal [Install] WantedBy=multi-user.target

然后启用并启动服务：

sudo systemctl daemon-reload sudo systemctl enable env-monitor.service sudo systemctl start env-monitor.service # 查看状态和日志 sudo systemctl status env-monitor.service sudo journalctl -u env-monitor.service -f

使用systemd管理，你的应用就具备了开机自启、崩溃重启、集中日志等生产级特性。

走到这一步，你已经不再只是写一个脚本，而是在构建一个结构清晰、模块化、可维护、可部署的嵌入式软件系统。Python在树莓派上的价值，正是在于它能让你以如此高的效率，跨越硬件控制、数据处理、网络通信和用户交互的鸿沟，将想法快速、稳健地转化为现实。