用 Python 玩转树莓派与物联网：从零基础到专业的 100 个示例

树莓派是一台小型电脑，可以读取传感器、控制继电器，并通过网络发送数据，因此常被用作物联网（IoT）项目的基础。本文用 100 个示例，带你从第一个 Python 程序一直做到完整的传感器数据流水线。你将学习 Python 基础、处理传感器数据、用 GPIO 控制引脚、通过 I2C、SPI 和 1-Wire 读取传感器、用 HTTP 和 MQTT 发送数据、搭建小型网页仪表盘、把读数保存到 SQLite，并让程序作为服务在开机时自动运行。诚实说明：能在普通电脑上运行的示例已实际运行，并在代码之后显示真实输出。其余示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

需要准备的东西

运行 Raspberry Pi OS 的树莓派 3、4 或 5、一张 microSD 卡、电源以及网络连接。Raspberry Pi OS 已包含 Python 3.10 或更高版本。硬件示例需要 LED、电阻、按键、杜邦线以及若干传感器。大多数可运行的示例可以在任何装有 Python 3.10 或更高版本的电脑上运行，这也是可运行示例的测试方式。

如何运行示例

把示例保存为 main.py，然后用 python3 main.py 运行。标注为可运行的示例会显示在电脑上实际运行的输出。bash 示例是树莓派上的配置文件，这里没有运行。

知识点 1: 在树莓派上开始使用 Python

Python 易读，并且几乎每种传感器都有对应的库，因此是树莓派项目的常用语言。前几个示例涵盖输出、变量、类型和运算。

运行第一个程序

保存一个扩展名为 .py 的文件，然后在树莓派上用 python3 运行。print 把文字输出到终端，f 字符串用 {name} 把变量放进文字中。Python 从上到下读取文件，所以代码行的顺序很重要。

类型及其含义

数字可以是 int（整数）或 float（小数），文字是 str，True 和 False 是 bool，None 表示没有值。用 type() 可以检查一个值的类型。传感器读数通常是 float，而引脚编号通常是 int。

传感器计算所需的算术

/ 总是得到 float，// 整除并舍去小数部分，% 得到余数。round() 限制显示的小数位数。用这些运算符可以把原始读数换算成有意义的数值，并对数值分组。

示意图

示例 1: 来自 Pi 的问候

在 Raspberry Pi OS 上编写第一个 Python 程序。

print("Hello from my Raspberry Pi!")

输出:

Hello from my Raspberry Pi!

示例 2: 变量与 f 字符串

保存值，并在文字中显示。

sensor_name = "DHT22"
temperature = 27.5
print(f"{sensor_name} reads {temperature} °C")

输出:

DHT22 reads 27.5 °C

示例 3: 数据类型

Python 的基本类型及其名称。

for value in (42, 3.14, "pi", True, None):
    print(repr(value), type(value).__name__)

输出:

42 int
3.14 float
'pi' str
True bool
None NoneType

示例 4: 算术、整除与取整

为传感器计算使用的除法、取余与四舍五入。

readings = 17
per_group = 5
print(readings / per_group)
print(readings // per_group, readings % per_group)
print(round(3.14159, 2))

输出:

3.4
3 2
3.14

示例 5: 文本处理

清理设备名称并检查前缀。

status = "  ONLINE  "
clean = status.strip().lower()
print(clean, clean.startswith("on"), len(clean))

输出:

online True 6

示例 6: 单位换算

用函数把摄氏度转换为华氏度。

def celsius_to_fahrenheit(c):
    return c * 9 / 5 + 32

for c in (0, 25, 37.5):
    print(f"{c} C = {celsius_to_fahrenheit(c):.1f} F")

输出:

0 C = 32.0 F
25 C = 77.0 F
37.5 C = 99.5 F

知识点 2: 判断与循环

树莓派项目会对输入作出反应。条件决定会发生什么，循环则重复执行工作，例如每隔几秒采样一次传感器。

if、elif 与 else

条件从上到下依次检查，第一个为真的分支会执行。应把最具体的判断放在前面。像 20 <= t <= 25 这样的链式比较可以在一行里检查整个范围。

for、while、break 与 continue

for 在一个范围或序列上重复执行。while 在条件改变之前一直重复。break 立即跳出循环，continue 跳过本轮。当你需要忽略一个错误读数并继续运行时，这些语句很有用。

真值判断与 match 语句

空字符串或空列表等空值被视为假。Python 3.10 引入的 match 语句可以把命令与多个模式匹配，是处理手机应用发来的文字命令的简洁方式。

示例 7: if、elif 与 else

根据温度选择动作。

temperature = 31
if temperature >= 35:
    print("Too hot: stop the fan test")
elif temperature >= 28:
    print("Warm: run the fan")
else:
    print("Comfortable")

输出:

Warm: run the fan

示例 8: 带 range 的 for 循环

按固定次数重复执行。

for minute in range(0, 15, 5):
    print(f"minute {minute}")

输出:

minute 0
minute 5
minute 10

示例 9: while 循环

重复执行直到条件改变。

countdown = 3
while countdown > 0:
    print("sampling in", countdown)
    countdown -= 1
print("start")

输出:

sampling in 3
sampling in 2
sampling in 1
start

示例 10: break 与 continue

跳过错误读数，并在故障时停止。

for reading in [21, None, 23, 99, 24]:
    if reading is None:
        continue
    if reading > 50:
        print("sensor fault, stop")
        break
    print("ok", reading)

输出:

ok 21
ok 23
sensor fault, stop

示例 11: 真值判断

哪些值被视为真，哪些为假。

for value in ("", "Pi", 0, 5, [], [1]):
    print(repr(value), "is true" if value else "is false")

输出:

'' is false
'Pi' is true
0 is false
5 is true
[] is false
[1] is true

示例 12: 链式比较

检查一个值是否在范围之内。

humidity = 55
print(40 <= humidity <= 60)
print(humidity > 60 or humidity < 40)

输出:

True
False

示例 13: 条件表达式

一行写出的 if-else，并返回一个值。

reading = 12
state = "ON" if reading < 15 else "OFF"
print(state)

输出:

ON

示例 14: match 语句

把文本命令映射到动作。

def describe(command):
    match command:
        case "on" | "1":
            return "turn relay on"
        case "off" | "0":
            return "turn relay off"
        case _:
            return "unknown command"

for c in ("on", "0", "blink"):
    print(c, "->", describe(c))

输出:

on -> turn relay on
0 -> turn relay off
blink -> unknown command

知识点 3: 函数与标准库

函数为你重复执行的任务起一个名字。标准库已经包含数学、统计和时间相关的函数，因此你很少需要自己编写。

定义函数

def 用于创建函数，return 把值返回。当调用者省略参数时，默认参数提供合理的值；关键字参数让调用更易读，例如 sample(4, unit='volts')。

返回与传递多个值

函数可以把多个值作为元组返回，由调用者解包。*args 接收任意数量的位置参数，**kwargs 接收任意数量的关键字参数，很适合日志辅助函数。

lambda、map、filter 与递归

lambda 是一个很小的匿名函数，与 map 和 filter 配合可以变换列表。递归函数会调用自身，必须有终止条件，阶乘是典型例子。

示例 15: 带默认参数的函数

为函数设置合理的默认值。

def sample(pin, samples=3, unit="raw"):
    return f"read pin {pin} x{samples} ({unit})"

print(sample(4))
print(sample(4, 10, unit="volts"))

输出:

read pin 4 x3 (raw)
read pin 4 x10 (volts)

示例 16: 返回多个值

同时返回最小值和最大值。

def min_max(values):
    return min(values), max(values)

low, high = min_max([19.5, 22.0, 18.2])
print(f"low {low}, high {high}")

输出:

low 18.2, high 22.0

示例 17: *args 与 **kwargs

接收任意数量的位置参数和关键字参数。

def log_event(*parts, **fields):
    text = " ".join(str(p) for p in parts)
    extra = ", ".join(f"{k}={v}" for k, v in fields.items())
    print(text, "|", extra)

log_event("door", "opened", sensor="reed-1", room="kitchen")

输出:

door opened | sensor=reed-1, room=kitchen

示例 18: lambda、map 与 filter

用一行代码转换原始 ADC 数值。

raw = [512, 1023, 0, 768]
volts = list(map(lambda x: round(x * 3.3 / 1023, 2), raw))
nonzero = list(filter(lambda v: v > 0, volts))
print(volts)
print(nonzero)

输出:

[1.65, 3.3, 0.0, 2.48]
[1.65, 3.3, 2.48]

示例 19: 标准库：math 与 statistics

使用内置的数学和统计模块。

import math
import statistics

samples = [2.0, 4.0, 4.5, 3.5]
print(round(math.sqrt(16), 2), math.pi > 3.14)
print(statistics.mean(samples), statistics.median(samples))

输出:

4.0 True
3.5 3.75

示例 20: 递归

调用自身的函数，并带有终止条件。

def factorial(n):
    if n <= 1:
        return 1
    return n * factorial(n - 1)

print(factorial(5), factorial(10))

输出:

120 3628800

知识点 4: 设备与读数的数据结构

树莓派项目需要保存读数列表、设备设置的字典，以及不重复的名称集合。选对数据结构可以让代码更短、更快。

列表与推导式

列表是可以修改的有序序列。列表推导式可以用一行代码由旧列表生成新列表，例如把一组摄氏温度转换为华氏温度。

用字典保存带名称的值

字典把键映射到值，适合保存某台设备的设置，或一次带有多个字段的读数。对可能缺失的键使用 get()，用 sorted() 按顺序输出键。

元组、集合与切片

元组是固定的一组值。集合保存不重复的元素，并支持并集和交集。像 values[-2:] 这样的切片取出列表的一部分，[::-1] 则把列表反转。

示例 21: 列表

添加、插入和删除元素。

readings = [20.1, 20.4]
readings.append(20.9)
readings.insert(0, 19.8)
readings.remove(20.4)
print(readings, len(readings), readings[-1])

输出:

[19.8, 20.1, 20.9] 3 20.9

示例 22: 列表推导式

由旧列表生成新列表。

celsius = [0, 10, 20, 30]
fahrenheit = [c * 9 / 5 + 32 for c in celsius if c != 10]
print(fahrenheit)

输出:

[32.0, 68.0, 86.0]

示例 23: 元组与解包

不可更改的值的分组。

location = ("kitchen", 2, 3.5)
room, floor, height = location
print(room, floor, height)
try:
    location[0] = "garage"
except TypeError as e:
    print("tuples cannot change:", type(e).__name__)

输出:

kitchen 2 3.5
tuples cannot change: TypeError

示例 24: 字典

为设备保存带名称的值。

device = {"name": "node-1", "pin": 17, "online": True}
device["pin"] = 22
device["room"] = "garage"
print(device.get("name"), device.get("temp", "n/a"), len(device))

输出:

node-1 n/a 4

示例 25: 有序输出字典

按顺序打印每种事件的计数。

counts = {"motion": 4, "door": 1, "temp": 9}
for name in sorted(counts):
    print(f"{name:<8}{counts[name]:>3}")

输出:

door      1
motion    4
temp      9

示例 26: 集合

唯一的值以及集合运算。

seen = {"kitchen", "garage"}
seen.add("kitchen")
new = {"garage", "attic"}
print(len(seen), sorted(seen | new), sorted(seen & new))

输出:

2 ['attic', 'garage', 'kitchen'] ['garage']

示例 27: 切片

截取列表的一部分，并反转。

samples = [1, 2, 3, 4, 5, 6, 7]
print(samples[:3], samples[-2:], samples[::2])
print(samples[::-1])

输出:

[1, 2, 3] [6, 7] [1, 3, 5, 7]
[7, 6, 5, 4, 3, 2, 1]

示例 28: 字典组成的列表

筛选在线的设备列表。

devices = [
    {"id": "a1", "online": True},
    {"id": "b2", "online": False},
    {"id": "c3", "online": True},
]
online = [d["id"] for d in devices if d["online"]]
print("online:", online)

输出:

online: ['a1', 'c3']

知识点 5: 文本、文件与数据格式

传感器数据需要保存和共享。文本文件、CSV 和 JSON 是大多数物联网项目使用的格式，标准库可以读写这三种格式。

读写文件

用 with 代码块打开文件，它会自动关闭；处理文本时使用 encoding='utf-8'。pathlib 中的 Path 对象可以在任何操作系统上工作，并且便于拼接文件名。

CSV 与 JSON

csv 模块可以写出带表头的行，电子表格和数据工具都能识别。json 模块把字典转换为文本，用于网络服务和 MQTT 消息，也能再转换回来。

表格与日期格式化

像 {name:<10} 这样的宽度格式可以让终端中的列对齐。datetime 用 strftime 格式化时间戳，并用 timedelta 进行时间加减，因此可以按时间对读数分组。

示例 29: 格式化表格输出

用格式宽度对齐列。

rows = [("Temp", 27.456), ("Humidity", 55.0), ("Pressure", 1013.2)]
for name, value in rows:
    print(f"{name:<10}{value:>10.1f}")

输出:

Temp            27.5
Humidity        55.0
Pressure      1013.2

示例 30: 读写文本文件

把传感器数值保存到文件并读回。

from pathlib import Path

log = Path("sensor_log.txt")
with log.open("w", encoding="utf-8") as f:
    f.write("21.5\n22.0\n22.4\n")
with log.open("r", encoding="utf-8") as f:
    values = [float(line) for line in f]
log.unlink()
print("read", len(values), "values, last", values[-1])

输出:

read 3 values, last 22.4

示例 31: CSV 文件

写入并读取测量数据表。

import csv
from pathlib import Path

path = Path("readings.csv")
with path.open("w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f)
    writer.writerow(["time", "temp_c"])
    writer.writerows([["08:00", 21.5], ["08:05", 21.9]])
with path.open(newline="", encoding="utf-8") as f:
    for row in csv.DictReader(f):
        print(row["time"], float(row["temp_c"]) + 1)
path.unlink()

输出:

08:00 22.5
08:05 22.9

示例 32: JSON

在数据和 JSON 文本之间转换。

import json

payload = {"device": "node-1", "readings": [21.5, 21.9], "ok": True}
text = json.dumps(payload, indent=2)
print(text)
print(json.loads(text)["readings"][1])

输出:

{
  "device": "node-1",
  "readings": [
    21.5,
    21.9
  ],
  "ok": true
}
21.9

示例 33: 路径

安全地处理文件名和文件夹。

from pathlib import PurePosixPath

path = PurePosixPath("/home/pi/projects/iot/config.json")
print(path.name, path.suffix, path.parent)
print(path.parent / "logs")

输出:

config.json .json /home/pi/projects/iot
/home/pi/projects/iot/logs

示例 34: 日期和时间

格式化日期时间，并进行加减。

from datetime import datetime, timedelta

start = datetime(2026, 1, 15, 8, 0)
later = start + timedelta(minutes=90)
print(start.strftime("%Y-%m-%d %H:%M"), "->", later.strftime("%H:%M"))

输出:

2026-01-15 08:00 -> 09:30

知识点 6: 错误处理、日志与计时

设备可能在无人看管的情况下运行数月。它必须能承受错误读数、记录发生的事情，并测量自身的运行时间。

用 try 与 except 处理错误

把可能失败的代码放进 try，捕获特定的异常，然后继续运行。无论是否发生错误，finally 都会执行，因此它是关闭输出或关闭文件的合适位置。

用 logging 代替 print

logging 模块以 INFO、WARNING 等级别记录消息，并可以选择显示哪个级别。当树莓派没有屏幕时，日志需要写入文件，这一点很重要。

防御性检查与计时

assert 在开发过程中尽早拒绝不可能的值。time.perf_counter 能精确测量经过的时间，帮助你在小型电脑上找到运行缓慢的代码。

示例 35: try 与 except

处理错误读数，程序不中断。

def parse_temperature(text):
    try:
        return float(text)
    except ValueError:
        print(f"bad reading: {text!r}")
        return None

print(parse_temperature("23.4"))
print(parse_temperature("23,4"))

输出:

23.4
bad reading: '23,4'
None

示例 36: finally 与清理

即使出错也始终执行清理。

from pathlib import Path

target = Path("demo_state.txt")
try:
    with target.open("w", encoding="utf-8") as f:
        f.write("on")
    print("saved")
finally:
    target.unlink(missing_ok=True)
    print("cleaned up")

输出:

saved
cleaned up

示例 37: 自定义异常

对不可能的传感器数值给出清晰的错误。

class SensorError(Exception):
    """Raised when a sensor gives an impossible value."""

def check(value):
    if value < -40 or value > 85:
        raise SensorError(f"value {value} is outside the sensor range")
    return value

try:
    check(120)
except SensorError as e:
    print("sensor error:", e)

输出:

sensor error: value 120 is outside the sensor range

示例 38: 日志记录

用带级别的日志代替 print。

import logging

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger("node-1")
log.info("started")
log.warning("battery at 18%")
log.debug("this line is hidden at INFO level")

输出:

示例 39: 用 assert 做防御性检查

尽早拒绝无效输入。

def set_pwm(duty):
    assert 0 <= duty <= 100, "duty must be 0 to 100"
    return duty

print(set_pwm(50))
try:
    set_pwm(150)
except AssertionError as e:
    print("rejected:", e)

输出:

50
rejected: duty must be 0 to 100

示例 40: 测量经过的时间

用精确的时钟为代码计时。

import time

start = time.perf_counter()
total = sum(i * i for i in range(200_000))
elapsed = time.perf_counter() - start
print("sum of squares:", total)
print("took measurable time:", elapsed >= 0)

输出:

sum of squares: 2666646666700000
took measurable time: True

知识点 7: 处理传感器数据

原始传感器数值通常有噪声，而且单位不对。这些示例把原始数字变成有用的读数：经过校准、平滑处理、检查异常值，并与阈值比较。

校准与单位换算

ADC 返回的是原始计数值。两个已知的参考点可以给出一条直线校准，从而把计数值换算成电压。每个传感器都不同，因此应把校准值与设备一起保存。

平滑与滞回

对最近几个读数做移动平均，可以去掉突发的尖峰。滞回使用两个阈值，一个用于开启，一个用于关闭，这样当数值停在某个界限附近时，继电器不会反复跳动。

统计与异常值

平均值、中位数和标准差描述一段时间内的读数。中位数不受极端值影响，因此在求平均之前，用它去除错误的尖峰是很好的做法。

示例 41: 校准原始 ADC 数值

用两个参考点把原始数值换算为电压。

RAW_MIN, RAW_MAX = 102, 921      # readings at 0 V and at full scale
VOLT_MAX = 3.3

def to_volts(raw):
    return (raw - RAW_MIN) * VOLT_MAX / (RAW_MAX - RAW_MIN)

for raw in (102, 512, 921):
    print(raw, "->", round(to_volts(raw), 3), "V")

输出:

102 -> 0.0 V
512 -> 1.652 V
921 -> 3.3 V

示例 42: 移动平均滤波

在一个窗口内平滑有噪声的读数。

from collections import deque

window = deque(maxlen=3)
def smooth(value):
    window.append(value)
    return sum(window) / len(window)

for reading in [20.0, 25.0, 21.0, 40.0, 22.0]:
    print(f"raw {reading:5.1f}  smoothed {smooth(reading):6.2f}")

输出:

raw  20.0  smoothed  20.00
raw  25.0  smoothed  22.50
raw  21.0  smoothed  22.00
raw  40.0  smoothed  28.67
raw  22.0  smoothed  27.67

示例 43: 风扇继电器的滞回

防止继电器频繁地开关。

def relay_state(temp, currently_on, on_at=24.0, off_at=22.0):
    if not currently_on and temp >= on_at:
        return True
    if currently_on and temp <= off_at:
        return False
    return currently_on

state = False
for t in [21.0, 23.0, 24.5, 23.5, 22.5, 21.9]:
    state = relay_state(t, state)
    print(t, "fan", "ON" if state else "OFF")

输出:

21.0 fan OFF
23.0 fan OFF
24.5 fan ON
23.5 fan ON
22.5 fan ON
21.9 fan OFF

示例 44: 气压单位与取整

把 hPa 转换为 inHg，并正确取整。

def hpa_to_inhg(hpa):
    return hpa * 0.02953

pressure = 1013.25
print(f"{pressure} hPa = {hpa_to_inhg(pressure):.2f} inHg")
print(round(2.5), round(3.5))

输出:

1013.25 hPa = 29.92 inHg
2 4

示例 45: 窗口内的统计

最近读数的平均值、离散程度和范围。

import statistics

window = [21.4, 21.6, 21.5, 21.9, 21.7]
print("mean", round(statistics.mean(window), 2))
print("stdev", round(statistics.pstdev(window), 3))
print("range", round(max(window) - min(window), 2))

输出:

mean 21.62
stdev 0.172
range 0.5

示例 46: 用中位数去除异常值

忽略明显偏离典型读数的数值。

import statistics

def remove_outliers(values, tolerance=3.0):
    med = statistics.median(values)
    return [v for v in values if abs(v - med) <= tolerance]

raw = [21.5, 21.7, 85.0, 21.6, -40.0, 21.8]
print(remove_outliers(raw))

输出:

[21.5, 21.7, 21.6, 21.8]

示例 47: 变化速率

数值每分钟上升了多少。

def rate_per_minute(prev_value, prev_minute, value, minute):
    if minute == prev_minute:
        return 0.0
    return (value - prev_value) / (minute - prev_minute)

print(round(rate_per_minute(20.0, 0, 21.2, 4), 2), "C per minute")

输出:

0.3 C per minute

示例 48: 把读数做成数据类

用一个小类描述一次测量。

from dataclasses import dataclass, asdict

@dataclass
class Reading:
    device: str
    temperature_c: float
    humidity_pct: float

r = Reading("node-1", 21.5, 48.0)
print(r)
print(asdict(r))

输出:

Reading(device='node-1', temperature_c=21.5, humidity_pct=48.0)
{'device': 'node-1', 'temperature_c': 21.5, 'humidity_pct': 48.0}

知识点 8: GPIO：控制 LED、按键、电机与继电器

40 针的排针让树莓派可以切换输出、读取输入。这些示例包括 LED、按键、PWM 调光、舵机、距离传感器和继电器。接通电源之前，务必对照引脚图检查接线。

输出、输入与上拉电阻

把引脚设为输出，可以通过电阻驱动 LED；设为输入，可以读取按键。上拉电阻使输入保持高电平，直到按键把它接到地，这样引脚就不会悬空。

PWM 与舵机

脉宽调制会快速地开关引脚，导通时间所占的比例决定亮度或舵机的位置。gpiozero 库隐藏了大部分的计时工作。

继电器、传感器与安全关闭

继电器用于切换功率较大的设备，例如水泵，因此树莓派只需驱动它小小的控制线圈。应始终在 finally 代码块中关闭输出，并调用 cleanup，这样即使程序崩溃，电机也不会一直运转。

示例 49: 让 LED 闪烁（GPIO）

从 Raspberry Pi 的引脚开关 LED。

import time
import RPi.GPIO as GPIO

LED_PIN = 17
GPIO.setmode(GPIO.BCM)
GPIO.setup(LED_PIN, GPIO.OUT)

try:
    for _ in range(5):
        GPIO.output(LED_PIN, GPIO.HIGH)
        time.sleep(0.5)
        GPIO.output(LED_PIN, GPIO.LOW)
        time.sleep(0.5)
finally:
    GPIO.cleanup()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 50: 用上拉电阻读取按钮

用去抖延时安全地检测按键。

import time
import RPi.GPIO as GPIO

BUTTON_PIN = 27
GPIO.setmode(GPIO.BCM)
GPIO.setup(BUTTON_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)

try:
    while True:
        if GPIO.input(BUTTON_PIN) == GPIO.LOW:   # pressed connects the pin to GND
            print("button pressed")
            time.sleep(0.2)                      # simple debounce
        time.sleep(0.01)
finally:
    GPIO.cleanup()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 51: 用 PWM 调光

用脉宽调制改变 LED 的亮度。

import time
import RPi.GPIO as GPIO

GPIO.setmode(GPIO.BCM)
GPIO.setup(18, GPIO.OUT)
pwm = GPIO.PWM(18, 1000)          # 1 kHz
pwm.start(0)
try:
    for duty in range(0, 101, 10):
        pwm.ChangeDutyCycle(duty)
        time.sleep(0.1)
finally:
    pwm.stop()
    GPIO.cleanup()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 52: 用 gpiozero 控制 LED

更简单、更安全的 GPIO 库。

from gpiozero import LED
from time import sleep

led = LED(17)
for _ in range(3):
    led.on()
    sleep(0.5)
    led.off()
    sleep(0.5)

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 53: 按钮回调

不需要轮询循环即可响应按下。

from gpiozero import Button
from signal import pause

button = Button(27)
button.when_pressed = lambda: print("pressed")
button.when_released = lambda: print("released")
pause()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 54: 舵机控制

把舵机转到设定的位置。

from gpiozero import Servo
from time import sleep

servo = Servo(12)
for position in (-1, 0, 1, 0):   # -1 = left, 0 = centre, 1 = right
    servo.value = position
    sleep(1)

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 55: 超声波测距传感器

用 HC-SR04 传感器测量距离。

import time
import RPi.GPIO as GPIO

TRIG, ECHO = 23, 24
GPIO.setmode(GPIO.BCM)
GPIO.setup(TRIG, GPIO.OUT)
GPIO.setup(ECHO, GPIO.IN)

def distance_cm():
    GPIO.output(TRIG, False)
    time.sleep(0.05)
    GPIO.output(TRIG, True)
    time.sleep(0.00001)
    GPIO.output(TRIG, False)
    while GPIO.input(ECHO) == 0:
        start = time.time()
    while GPIO.input(ECHO) == 1:
        end = time.time()
    return (end - start) * 34300 / 2   # speed of sound, there and back

print(round(distance_cm(), 1), "cm")
GPIO.cleanup()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 56: 默认安全的继电器

安全地切换水泵，并保证结束时关闭。

from gpiozero import OutputDevice

relay = OutputDevice(22, active_high=True, initial_value=False)
relay.on()     # power the pump
relay.off()    # always end with the relay off

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 57: 蜂鸣器音调

在蜂鸣器上播放简单的音调。

from gpiozero import TonalBuzzer
from time import sleep

buzzer = TonalBuzzer(26)
for note in (0.0, 0.5, 1.0):
    buzzer.play(note)
    sleep(0.3)
buzzer.stop()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 58: 安全关闭 GPIO

程序退出前让输出处于安全状态。

import RPi.GPIO as GPIO

GPIO.setmode(GPIO.BCM)
GPIO.setup(17, GPIO.OUT)
try:
    GPIO.output(17, GPIO.HIGH)
    raise RuntimeError("simulated fault")
except RuntimeError as e:
    print("error:", e)
finally:
    GPIO.output(17, GPIO.LOW)   # make the output safe before cleanup
    GPIO.cleanup()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

知识点 9: 通过 I2C、SPI、1-Wire 与摄像头连接的传感器

许多传感器通过总线与树莓派通信。I2C 使用两根线和一个地址，SPI 使用四根线和片选，1-Wire 只使用一根数据线。摄像头模块有自己的库，是一种特殊情况。

I2C：扫描与读取寄存器

每个 I2C 设备都有一个地址。先扫描总线，找出连接了什么。读取芯片 ID 寄存器来确认传感器，然后写入控制寄存器，再读取数据字节。

SPI 与模拟输入

树莓派没有模拟输入，因此需要像 MCP3008 这样的 ADC 通过 SPI 转换电压。发送通道号，然后从返回的字节中组合出 10 位的数值。

1-Wire、DHT 传感器与摄像头

1-Wire 上的 DS18B20 会以文件的形式出现在 /sys/bus/w1 中；需要先在 raspi-config 中启用该接口。DHT 传感器有时会读取失败，因此应当重试。摄像头模块使用 picamera2 拍摄静态图片和帧。

示例 59: 扫描 I2C 总线

查找连接了哪些传感器。

from smbus2 import SMBus

with SMBus(1) as bus:
    found = []
    for address in range(0x03, 0x78):
        try:
            bus.read_byte(address)
            found.append(hex(address))
        except OSError:
            pass
print("devices:", found)

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 60: 读取 BMP280 气压传感器

检查芯片 ID，并通过 I2C 读取原始气压字节。

from smbus2 import SMBus

ADDRESS = 0x76
with SMBus(1) as bus:
    chip_id = bus.read_byte_data(ADDRESS, 0xD0)
    print("chip id", hex(chip_id), "(expect 0x58)")
    bus.write_byte_data(ADDRESS, 0xF4, 0b00100111)   # normal mode, oversampling x1
    raw = bus.read_i2c_block_data(ADDRESS, 0xF7, 6)
print("raw bytes", raw)

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 61: DHT22 温度与湿度

读取 DHT22，未就绪时重试。

import time
import board
import adafruit_dht

dht = adafruit_dht.DHT22(board.D4)
for _ in range(5):
    try:
        print(f"{dht.temperature:.1f} C  {dht.humidity:.0f} %")
    except RuntimeError as e:
        print("sensor not ready:", e)   # DHT sensors fail sometimes; just retry
    time.sleep(2)

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 62: 用 MCP3008 进行模拟输入

通过 SPI 读取模拟电压。

import spidev

spi = spidev.SpiDev()
spi.open(0, 0)                  # bus 0, device 0
spi.max_speed_hz = 1_350_000

def read_channel(channel):
    cmd = [1, (8 + channel) << 4, 0]
    reply = spi.xfer2(cmd)
    return ((reply[1] & 3) << 8) | reply[2]   # 10-bit value, 0 to 1023

print("channel 0 raw:", read_channel(0))
spi.close()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 63: 通过 1-Wire 读取 DS18B20 温度

从内核读取防水型温度传感器。

from pathlib import Path
import glob

device_dirs = glob.glob("/sys/bus/w1/devices/28-*")
if not device_dirs:
    raise SystemExit("no DS18B20 found: enable 1-Wire in raspi-config")
lines = (Path(device_dirs[0]) / "w1_slave").read_text().splitlines()
if lines[0].endswith("YES"):
    milli = int(lines[1].split("t=")[1])
    print(f"{milli / 1000:.3f} C")

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 64: 不稳定传感器的重试

放弃之前再读一次。

import time

def read_with_retries(read_fn, attempts=3, delay=1.0):
    for attempt in range(1, attempts + 1):
        try:
            return read_fn()
        except OSError as e:
            print(f"attempt {attempt} failed: {e}")
            time.sleep(delay)
    raise RuntimeError("sensor unavailable after retries")

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 65: 把校准值保存到 JSON

将校准值保存在文件中，并能重新加载。

import json
from pathlib import Path

CONFIG = Path("calibration.json")
DEFAULTS = {"offset_c": 0.0, "scale": 1.0}

def load():
    if CONFIG.exists():
        return {**DEFAULTS, **json.loads(CONFIG.read_text(encoding="utf-8"))}
    return dict(DEFAULTS)

def save(cfg):
    CONFIG.write_text(json.dumps(cfg, indent=2), encoding="utf-8")

cfg = load()
cfg["offset_c"] = -0.4
save(cfg)
print("saved:", load())
CONFIG.unlink()

输出:

saved: {'offset_c': -0.4, 'scale': 1.0}

示例 66: 拍摄静态照片

用树莓派摄像头模块拍照。

from picamera2 import Picamera2
from time import sleep

cam = Picamera2()
cam.configure(cam.create_still_configuration())
cam.start()
sleep(2)                      # let the auto-exposure settle
cam.capture_file("photo.jpg")
cam.stop()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 67: 根据摄像头画面检测运动

比较相邻画面来发现移动。

import numpy as np
from picamera2 import Picamera2

cam = Picamera2()
cam.configure(cam.create_preview_configuration(main={"size": (320, 240), "format": "YUV420"}))
cam.start()
previous = None
for _ in range(20):
    frame = cam.capture_array()[:240].astype(np.int16)
    if previous is not None and np.mean(np.abs(frame - previous)) > 10:
        print("motion")
    previous = frame
cam.stop()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

知识点 10: 使用套接字与 HTTP 进行网络通信

物联网设备必须与其他电脑通信。套接字是底层基础，HTTP 是网络服务与 API 的工作方式，健壮的代码要处理超时、重试和离线的情况。

TCP 与 UDP 套接字

TCP 套接字是可靠的连接，UDP 则在没有连接的情况下发送单条消息。绑定到端口 0 时，操作系统会选择一个空闲端口，这样示例之间就不会冲突。

带超时的 HTTP 请求

requests 用一行代码就能完成 HTTP 调用。务必设置超时，否则一个无响应的网络会让程序卡住。raise_for_status 会把错误响应转换为可以处理的异常。

重试、退避与离线队列

对失败的请求使用指数退避重试，每次等待的时间更长，以免压垮繁忙的服务器。网络中断时，把读数保存在队列中，等网络恢复后再发送。

示例 68: TCP 回显服务器与客户端

向服务器发送字节并读取回复。

import socket
import threading

def serve(server):
    conn, _ = server.accept()
    with conn:
        data = conn.recv(1024)
        conn.sendall(b"echo: " + data)

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(("127.0.0.1", 0))            # port 0: let the OS choose a free port
server.listen(1)
port = server.getsockname()[1]
thread = threading.Thread(target=serve, args=(server,))
thread.start()

with socket.create_connection(("127.0.0.1", port)) as client:
    client.sendall(b"temp=21.5")
    print(client.recv(1024).decode())
thread.join()
server.close()

输出:

echo: temp=21.5

示例 69: 环回地址上的 UDP 消息

无需建立连接即可发送数据报。

import socket

receiver = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
receiver.bind(("127.0.0.1", 0))
addr = receiver.getsockname()

sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sender.sendto(b"humidity=48", addr)
data, _ = receiver.recvfrom(1024)
print("received:", data.decode())
sender.close()
receiver.close()

输出:

received: humidity=48

示例 70: 向 Web 服务 POST 读数

以 JSON 通过 HTTPS 发送传感器读数。

import requests

reading = {"device": "node-1", "temperature_c": 21.5}
response = requests.post(
    "https://api.example.com/readings",
    json=reading,
    timeout=5,
)
response.raise_for_status()
print("server said:", response.status_code)

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 71: 带超时的 API JSON 获取

获取数据并处理失败，不会卡住。

import requests

try:
    r = requests.get("https://api.example.com/forecast", params={"city": "Bangkok"}, timeout=5)
    r.raise_for_status()
    data = r.json()
    print(data.get("temp_c"))
except requests.RequestException as e:
    print("request failed:", e)

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 72: 指数退避重试

每次失败后等待更长时间。

import random
import time

def with_backoff(fn, tries=4, base=0.1):
    for n in range(tries):
        try:
            return fn()
        except ConnectionError:
            wait = base * (2 ** n)
            print(f"try {n + 1} failed, waiting {wait:.1f}s")
            time.sleep(wait)
    raise ConnectionError("gave up")

random.seed(3)
calls = {"n": 0}
def flaky():
    calls["n"] += 1
    if calls["n"] < 3:
        raise ConnectionError("network down")
    return "uploaded"

print(with_backoff(flaky))

输出:

try 1 failed, waiting 0.1s
try 2 failed, waiting 0.2s
uploaded

示例 73: 用 urllib 发送 JSON

只用标准库发送请求。

import json
import urllib.request

body = json.dumps({"device": "node-1", "humidity_pct": 48}).encode()
req = urllib.request.Request(
    "https://api.example.com/readings",
    data=body,
    headers={"Content-Type": "application/json"},
    method="POST",
)
with urllib.request.urlopen(req, timeout=5) as resp:
    print("status", resp.status)

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 74: 状态网页服务器

在树莓派上提供一个小型状态页面。

from http.server import BaseHTTPRequestHandler, HTTPServer
import json

class Status(BaseHTTPRequestHandler):
    def do_GET(self):
        body = json.dumps({"status": "ok", "uptime_s": 42}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(body)

HTTPServer(("0.0.0.0", 8000), Status).serve_forever()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 75: 发送前检查网络

网络断开时先把数据排队。

import socket

def internet_ok(host="1.1.1.1", port=53, timeout=3):
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False

print("online" if internet_ok() else "offline, queue the data")

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

知识点 11: MQTT：发布与订阅

MQTT 是物联网的标准消息协议。设备把读数发布到主题，代理服务器负责投递，任意数量的订阅者都能收到。它足够轻量，适合小型设备，也能容忍不稳定的网络。

主题、发布与订阅

主题是类似 home/kitchen/temperature 的路径。发布通过代理服务器把消息发送到某个主题。订阅可以使用通配符：+ 匹配一层，# 匹配其下的所有内容。

服务质量与保留消息

QoS 0 是发后即忘，QoS 1 保证至少送达一次，QoS 2 保证恰好送达一次。保留消息由代理服务器存储，因此新的订阅者会立即收到最后一个值。

遗嘱、TLS 与发现

当设备意外断开时，代理服务器会发布遗嘱消息，这可以用来发现离线的设备。对于互联网上的代理服务器，应使用 TLS，并从环境变量中读取密码。Home Assistant 可以通过发送到特定主题的发现消息找到传感器。

示例 76: MQTT：发布遥测数据

把读数发送到主题上的代理。

import json
import time
import paho.mqtt.client as mqtt

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="node-1")
client.connect("broker.local", 1883, keepalive=60)
client.loop_start()

payload = {"temp_c": 21.5, "ts": int(time.time())}
client.publish("home/kitchen/temperature", json.dumps(payload), qos=1)
time.sleep(1)
client.loop_stop()
client.disconnect()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 77: MQTT：订阅消息

用通配符接收多台设备的数据。

import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe("home/+/temperature")

def on_message(client, userdata, msg):
    print(msg.topic, "->", msg.payload.decode())

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.local", 1883)
client.loop_forever()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 78: MQTT：QoS 与保留消息

选择传输保证，并保留最后一个值。

import paho.mqtt.client as mqtt

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.connect("broker.local", 1883)
client.loop_start()

# QoS 0 = at most once, 1 = at least once, 2 = exactly once
client.publish("home/door/state", "closed", qos=1, retain=True)   # last value is kept
client.loop_stop()
client.disconnect()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 79: MQTT：遗嘱消息检测离线

设备掉线时通知所有人。

import paho.mqtt.client as mqtt

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="node-1")
client.will_set("home/node-1/status", payload="offline", qos=1, retain=True)
client.publish("home/node-1/status", "online", qos=1, retain=True)   # set on connect
client.connect("broker.local", 1883)
client.loop_forever()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 80: 基于 TLS 的 MQTT

加密与代理的连接。

import ssl
import paho.mqtt.client as mqtt

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.tls_set(ca_certs="/etc/ssl/certs/ca-certificates.crt",
               tls_version=ssl.PROTOCOL_TLS_CLIENT)
client.username_pw_set("node-1", "use-a-secret-from-env-not-code")
client.connect("broker.example.com", 8883)
client.loop_forever()

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 81: Home Assistant 发现消息

发布传感器，使 Home Assistant 自动找到它。

import json

config = {
    "name": "Kitchen temperature",
    "unique_id": "node-1_kitchen_temp",
    "state_topic": "home/kitchen/temperature",
    "unit_of_measurement": "°C",
    "device_class": "temperature",
    "value_template": "{{ value_json.temp_c }}",
}
topic = "homeassistant/sensor/node-1_kitchen_temp/config"
print(topic)
print(json.dumps(config, ensure_ascii=False, indent=2))

输出:

homeassistant/sensor/node-1_kitchen_temp/config
{
  "name": "Kitchen temperature",
  "unique_id": "node-1_kitchen_temp",
  "state_topic": "home/kitchen/temperature",
  "unit_of_measurement": "°C",
  "device_class": "temperature",
  "value_template": "{{ value_json.temp_c }}"
}

示例 82: MQTT 主题通配符

检查订阅匹配哪些主题。

def topic_matches(pattern, topic):
    p_parts = pattern.split("/")
    t_parts = topic.split("/")
    for i, part in enumerate(p_parts):
        if part == "#":
            return True
        if i >= len(t_parts):
            return False
        if part != "+" and part != t_parts[i]:
            return False
    return len(p_parts) == len(t_parts)

tests = [
    ("home/+/temperature", "home/kitchen/temperature", True),
    ("home/#", "home/kitchen/humidity", True),
    ("home/+/temperature", "home/kitchen/humidity", False),
]
for pattern, topic, expected in tests:
    matched = topic_matches(pattern, topic)
    print(pattern, topic, "match" if matched else "no match", "(as expected)" if matched == expected else "(UNEXPECTED)")

输出:

home/+/temperature home/kitchen/temperature match (as expected)
home/# home/kitchen/humidity match (as expected)
home/+/temperature home/kitchen/humidity no match (as expected)

知识点 12: 用 Flask 构建网页仪表盘与 API

Flask 是一个小型 Python 网络框架。只需几行代码，它就能提供状态页面、返回 JSON，或接收其他设备发来的读数，让你可以在手机或浏览器中查看树莓派。

路由与响应

像 @app.get('/') 这样的路由装饰器把 URL 与函数关联起来。函数可以返回文本、页面，或用 jsonify 返回 JSON。使用 0.0.0.0 地址可以接受来自局域网的连接。

带验证的数据接收

POST 路由从请求中读取 JSON。在保存任何内容之前，先检查类型和范围；数据有误时返回 400 状态码，让发送方知道该修改什么。

后台线程与仪表盘

后台线程在网络服务器响应请求的同时读取传感器。共享的数据应保持简单且安全。一个小的 HTML 模板可以把最新读数呈现为网络中任何人都能查看的页面。

示例 83: Flask 的 hello 接口

在树莓派上提供第一个网页路由。

from flask import Flask

app = Flask(__name__)

@app.get("/")
def hello():
    return "Raspberry Pi is alive"

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 84: Flask 的 JSON 接口

以 JSON 返回最新读数。

from flask import Flask, jsonify

app = Flask(__name__)
latest = {"temp_c": 21.5, "humidity_pct": 48}

@app.get("/api/latest")
def get_latest():
    return jsonify(latest)

app.run(host="0.0.0.0", port=5000)

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 85: 带校验的 Flask POST

接收读数，并用错误码拒绝无效数值。

from flask import Flask, request, jsonify

app = Flask(__name__)
readings = []

@app.post("/api/readings")
def add_reading():
    data = request.get_json(silent=True) or {}
    temp = data.get("temp_c")
    if not isinstance(temp, (int, float)) or not -40 <= temp <= 85:
        return jsonify(error="temp_c must be a number from -40 to 85"), 400
    readings.append(temp)
    return jsonify(saved=len(readings)), 201

app.run(port=5000)

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 86: 一个小型 HTML 仪表盘

渲染显示最新温度的页面。

from flask import Flask, render_template_string

app = Flask(__name__)
PAGE = """
<h1>Kitchen</h1>
<p>Temperature: {{ temp }} °C</p>
"""

@app.get("/")
def home():
    return render_template_string(PAGE, temp=21.5)

app.run(port=5000)

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 87: Flask 加后台传感器线程

Web 服务器运行时持续读取传感器。

import threading
import time
from flask import Flask, jsonify

app = Flask(__name__)
state = {"temp_c": None}

def sensor_loop():
    while True:
        state["temp_c"] = 21.5          # replace with the real sensor read
        time.sleep(5)

threading.Thread(target=sensor_loop, daemon=True).start()

@app.get("/api/temp")
def temp():
    return jsonify(state)

app.run(port=5000)

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

知识点 13: 保存与汇总读数

一台要记录一整年数据的设备，需要简单且不会占满 SD 卡的存储。SQLite 是随 Python 附带的单文件数据库，CSV 文件则适合简单的历史记录。

SQLite 表与插入

SQLite 把表保存在一个文件中。INSERT 语句中使用 ? 占位符，可以防止 SQL 注入，并自动处理引号。写入之后调用 commit，数据才会被保存。

聚合与时间分桶

GROUP BY 配合 AVG 和 COUNT 可以按设备汇总读数。把时间戳按分钟分组，例如取 ISO 时间的前五个字符，就能得到简单的每分钟平均值。

数据保留与仅追加日志

按计划删除旧数据，让存储保持较小。仅追加的 CSV 文件每次读数添加一行，表头只写一次，简单又可靠。

示例 88: SQLite：建表与插入

把读数保存到数据库文件中。

import sqlite3

conn = sqlite3.connect(":memory:")   # use "readings.db" to keep data on the Pi
conn.execute("""CREATE TABLE readings (
    ts TEXT NOT NULL, device TEXT NOT NULL, temp_c REAL NOT NULL)""")
rows = [("2026-01-15T08:00", "node-1", 21.5), ("2026-01-15T08:05", "node-1", 21.9)]
conn.executemany("INSERT INTO readings VALUES (?, ?, ?)", rows)
conn.commit()
print(conn.execute("SELECT COUNT(*) FROM readings").fetchone()[0], "rows stored")
conn.close()

输出:

2 rows stored

示例 89: SQLite：聚合查询

按设备计算平均值和数量。

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE readings (device TEXT, temp_c REAL)")
conn.executemany("INSERT INTO readings VALUES (?, ?)",
                 [("node-1", 21.5), ("node-1", 22.5), ("node-2", 19.0)])
for device, avg, n in conn.execute(
        "SELECT device, AVG(temp_c), COUNT(*) FROM readings GROUP BY device ORDER BY device"):
    print(f"{device}: average {avg:.2f} from {n} readings")
conn.close()

输出:

node-1: average 22.00 from 2 readings
node-2: average 19.00 from 1 readings

示例 90: CSV 历史记录器

把读数追加到带表头的历史文件中。

import csv
from pathlib import Path

LOG = Path("sensor_history.csv")
FIELDS = ["ts", "temp_c"]

def log_reading(ts, temp_c):
    new_file = not LOG.exists()
    with LOG.open("a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        if new_file:
            writer.writeheader()
        writer.writerow({"ts": ts, "temp_c": temp_c})

log_reading("08:00", 21.5)
log_reading("08:05", 21.9)
print(LOG.read_text(encoding="utf-8"))
LOG.unlink()

输出:

ts,temp_c
08:00,21.5
08:05,21.9

示例 91: 按分钟分组读数

把读数合并为每分钟的平均值。

from collections import defaultdict
from statistics import mean

readings = [("08:00:10", 21.4), ("08:00:40", 21.6), ("08:01:05", 21.9)]
buckets = defaultdict(list)
for ts, value in readings:
    buckets[ts[:5]].append(value)     # keep HH:MM
for minute in sorted(buckets):
    print(minute, round(mean(buckets[minute]), 2))

输出:

08:00 21.5
08:01 21.9

示例 92: 数据保留

删除旧数据，让树莓派的存储保持较小。

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE readings (day INTEGER, temp_c REAL)")
conn.executemany("INSERT INTO readings VALUES (?, ?)", [(d, 20.0 + d / 10) for d in range(1, 11)])
keep_from = 6
deleted = conn.execute("DELETE FROM readings WHERE day < ?", (keep_from,)).rowcount
conn.commit()
remaining = conn.execute("SELECT COUNT(*) FROM readings").fetchone()[0]
print("deleted", deleted, "kept", remaining)
conn.close()

输出:

deleted 5 kept 5

知识点 14: 作为服务运行与配置

项目只有在断电后能自行启动并持续运行时，才真正有用。systemd 在开机时运行服务，cron 按计划运行任务，环境变量则把设置放在代码之外。

systemd 服务

单元文件描述了如何运行程序、工作目录、运行用户，以及失败时的处理方式（Restart=on-failure）。启用服务后，它会在开机时启动，并可以用 journalctl 查看日志。

用 cron 定时任务

crontab 中的一行可以在固定时间运行命令，例如每五分钟运行一次。把输出重定向到日志文件，你离开时发生的事情也能查到。

从环境变量读取配置

设备 ID、上报间隔和代理服务器地址等设置，可以从环境变量中读取，并提供安全的默认值。你无需修改代码，只需在服务文件中更改它们。

示例 93: 开机启动的 systemd 服务

树莓派开机时自动运行记录程序。

# /etc/systemd/system/sensor-logger.service
[Unit]
Description=Sensor logger
After=network-online.target
Wants=network-online.target

[Service]
ExecStart=/usr/bin/python3 /home/pi/iot/logger.py
WorkingDirectory=/home/pi/iot
Restart=on-failure
RestartSec=5
User=pi

[Install]
WantedBy=multi-user.target

# enable and start it:
# sudo systemctl daemon-reload
# sudo systemctl enable --now sensor-logger.service
# journalctl -u sensor-logger -f

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 94: 用 cron 定时任务

每五分钟运行一次脚本。

# edit the schedule with: crontab -e
# every 5 minutes, record the time and the readings file size
*/5 * * * * /usr/bin/python3 /home/pi/iot/snapshot.py >> /home/pi/iot/cron.log 2>&1
# every day at 02:30, remove logs older than 30 days
30 2 * * * find /home/pi/iot/logs -name "*.csv" -mtime +30 -delete

此示例需要树莓派、网络服务或系统服务，因此只做了语法检查，不显示输出。

示例 95: 看门狗的心跳文件

写入时间戳，让看门狗知道程序仍在运行。

import time
from pathlib import Path

HEARTBEAT = Path("heartbeat.txt")

def beat(count):
    HEARTBEAT.write_text(f"{count} {time.time():.0f}", encoding="utf-8")

for i in range(3):
    beat(i)
    print("beat", i, "written")
    time.sleep(0.05)
print("last:", HEARTBEAT.read_text(encoding="utf-8").split()[0])
HEARTBEAT.unlink()

输出:

beat 0 written
beat 1 written
beat 2 written
last: 2

示例 96: 从环境变量读取配置

以安全的默认值读取设置。

import os

def get_config():
    return {
        "device_id": os.environ.get("DEVICE_ID", "node-1"),
        "interval_s": int(os.environ.get("INTERVAL_S", "60")),
        "broker": os.environ.get("MQTT_HOST", "localhost"),
    }

print(get_config())

输出:

{'device_id': 'node-1', 'interval_s': 60, 'broker': 'localhost'}

知识点 15: 日志、安全与完整的流水线

最后的主题把所有内容整合起来：滚动日志防止 SD 卡被占满，秘密信息不写入代码，对接收到的数据进行验证，以及从传感器到报警的完整流水线。

滚动日志与心跳

滚动文件处理器限制每个日志文件的大小，并保留几个旧文件。带时间戳的心跳文件可以让看门狗检查程序是否仍在运行。

秘密信息与验证

把密码和令牌保存在环境变量中，绝不打印出来。在数据进入存储或报警之前，检查每个传入值的类型和范围，因为传感器和网络可能发送任何内容。

组装流水线

流水线读取传感器，用窗口对数值进行平滑，与阈值比较，设置报警标志，并保存每一步的结果。每个环节都很小且可以测试，这正是专业物联网代码的组织方式。

示例 97: 滚动日志文件

防止日志占满 SD 卡。

import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path

path = Path("iot.log")
handler = RotatingFileHandler(path, maxBytes=200, backupCount=2, encoding="utf-8")
log = logging.getLogger("iot")
log.setLevel(logging.INFO)
log.addHandler(handler)
for i in range(20):
    log.info("reading %d: temp=21.%d", i, i % 10)
handler.close()
print("files:", sorted(p.name for p in Path(".").glob("iot.log*")))
for p in Path(".").glob("iot.log*"):
    p.unlink()

输出:

files: ['iot.log', 'iot.log.1', 'iot.log.2']

示例 98: 从环境变量读取机密

从环境加载密码，且绝不打印出来。

import os

def require_secret(name):
    value = os.environ.get(name)
    if not value:
        raise SystemExit(f"set {name} in the environment (do not write it in the code)")
    return value

def mask(secret):
    return secret[:2] + "*" * (len(secret) - 2)

os.environ["MQTT_PASSWORD"] = "example-secret"   # demo only
token = require_secret("MQTT_PASSWORD")
print("loaded secret:", mask(token))

输出:

loaded secret: ex************

示例 99: 验证传入的传感器数据

存储前检查类型和范围。

from dataclasses import dataclass

@dataclass
class Sample:
    device: str
    temp_c: float

def validate(raw):
    errors = []
    if not isinstance(raw.get("device"), str) or not raw["device"]:
        errors.append("device is required")
    temp = raw.get("temp_c")
    if not isinstance(temp, (int, float)):
        errors.append("temp_c must be a number")
    elif not -40 <= temp <= 85:
        errors.append("temp_c out of sensor range")
    if errors:
        return None, errors
    return Sample(raw["device"], float(temp)), []

print(validate({"device": "node-1", "temp_c": 21.5}))
print(validate({"device": "", "temp_c": 300}))

输出:

(Sample(device='node-1', temp_c=21.5), [])
(None, ['device is required', 'temp_c out of sensor range'])

示例 100: 一个完整的小型流水线

读取、平滑、判断报警并保存结果。

import sqlite3
from collections import deque

def fake_sensor(n):
    base = [21.0, 21.2, 21.1, 35.0, 21.3, 21.4]     # 35.0 is a spike
    return base[n % len(base)]

window = deque(maxlen=3)
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE readings (tick INTEGER, raw REAL, smooth REAL, alarm INTEGER)")

for tick in range(6):
    raw = fake_sensor(tick)
    window.append(raw)
    smooth = sum(window) / len(window)
    alarm = 1 if smooth > 25 else 0
    conn.execute("INSERT INTO readings VALUES (?, ?, ?, ?)", (tick, raw, smooth, alarm))
    print(f"tick {tick}: raw {raw:5.1f} smooth {smooth:5.2f} alarm {alarm}")

alarms = conn.execute("SELECT COUNT(*) FROM readings WHERE alarm = 1").fetchone()[0]
print("alarm ticks:", alarms)
conn.close()

输出:

tick 0: raw  21.0 smooth 21.00 alarm 0
tick 1: raw  21.2 smooth 21.10 alarm 0
tick 2: raw  21.1 smooth 21.10 alarm 0
tick 3: raw  35.0 smooth 25.77 alarm 1
tick 4: raw  21.3 smooth 25.80 alarm 1
tick 5: raw  21.4 smooth 25.90 alarm 1
alarm ticks: 3

练习项目：房间监测器

构建一个运用各节思路的监测器。先自己编写，然后与示例比较。在连接硬件之前，先在电脑上测试纯 Python 的部分。

1. 编写一个函数，用两个校准点把原始读数换算为温度，并用三个数值进行测试。

2. 用最近五个值的移动平均平滑读数，并为阈值为 24 和 22 的风扇加入滞回。

3. 把每个读数保存到 SQLite 表中，读取十个数值后打印每分钟的平均值。

4. 每 60 秒把最新读数发布到 MQTT 主题；代理服务器离线时，把读数放入队列。

5. 创建一个 systemd 服务，使监测器在开机时启动，并用 journalctl 读取它的日志。

下一步

接下来，学习 Home Assistant 或 Node-RED 等物联网平台，加入安全的设备身份和无线升级，并学习面向微控制器的 Raspberry Pi Pico。先用你自己的接线测试每个硬件示例，并查阅 Raspberry Pi 官方文档，确认你所用型号的引脚编号。

总结

当树莓派能够可靠地读取传感器、处理数据、保存数据、通过网络发送数据，并且无需人工干预地持续运行时，它就成为了真正的物联网设备。通过 100 个示例，本文涵盖了 Python 基础、判断与循环、函数、数据结构、文件与格式、错误处理与日志、传感器数据处理、GPIO、I2C、SPI、1-Wire 与摄像头、套接字与 HTTP、MQTT、Flask、SQLite、systemd 与 cron，以及完整的流水线。在实验中构建房间监测器，每次只改动一处，以了解各部分的行为。