Python で作る Raspberry Pi と IoT: 初心者からプロまでの 100 の例

Raspberry Pi は、センサーの読み取り、リレーの制御、ネットワーク経由のデータ送信ができる小型コンピューターで、IoT（モノのインターネット）プロジェクトの基盤としてよく使われます。この記事では、最初の Python プログラムから完全なセンサーのパイプラインまで、100 の例で進みます。Python の基礎、センサーデータの処理、GPIO によるピン制御、I2C・SPI・1-Wire でのセンサー読み取り、HTTP と MQTT によるデータ送信、小さな Web ダッシュボード、SQLite への保存、起動時に自動で動くサービスとしての運用を学びます。正直な注記: パソコンで実行できる例は実際に実行し、その実際の出力をコードの後ろに示しています。それ以外の例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行い、出力は示していません。

必要なもの

Raspberry Pi OS が動く Raspberry Pi 3、4、5 のいずれか、microSD カード、電源、ネットワーク接続。Raspberry Pi OS には Python 3.10 以降が含まれています。ハードウェアの例には、LED、抵抗、押しボタン、ジャンパー線、いくつかのセンサーが必要です。実行できる例の多くは、Python 3.10 以降が入ったどのパソコンでも動きます。実行可能な例はそのように検証しました。

例を実行する方法

例を main.py として保存し、python3 main.py で実行します。実行可能と示した例は、コンピューターで実際に実行した出力を示しています。bash の例は Pi の設定ファイルであり、ここでは実行していません。

トピック 1: Pi で Python を始める

Python は読みやすく、ほぼすべてのセンサー向けのライブラリがあるため、Raspberry Pi プロジェクトで一般的に使われる言語です。最初の例では、出力、変数、型、計算を扱います。

最初のプログラムを実行する

拡張子 .py でファイルを保存し、Pi で 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: 条件分岐と繰り返し

Pi のプロジェクトは入力に反応します。条件は何が起きるかを決め、ループは作業を繰り返します。例えば数秒ごとにセンサーを読むような処理です。

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: 機器と読み取り値のためのデータ構造

Pi のプロジェクトでは、読み取り値のリスト、機器の設定の辞書、重複しない名前の集合を扱います。適切な構造を選ぶと、コードは短く速くなります。

リストと内包表記

リストは変更できる順序付きの並びです。リスト内包表記は、一行で古いリストから新しいリストを作ります。例えば、摂氏の値のリストを華氏に変換する場合です。

名前付きの値のための辞書

辞書はキーを値に対応づけ、一台の機器の設定や、項目付きの一回の読み取り値に向いています。欠けているかもしれないキーには 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 は多くの IoT プロジェクトで使われる形式で、標準ライブラリはこの三つを読み書きできます。

ファイルの読み書き

with ブロックでファイルを開けば、自動的に閉じられます。テキストには encoding='utf-8' を使います。pathlib の Path オブジェクトはどの OS でも動き、ファイル名の結合も簡単です。

CSV と JSON

csv モジュールはヘッダー付きの行を書き出し、表計算ソフトやデータツールが読めます。json モジュールは辞書を Web サービスや 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 はエラーの有無にかかわらず実行されるので、出力を切ったりファイルを閉じたりするのに適した場所です。

print の代わりに logging

logging モジュールは、INFO や WARNING などのレベル付きでメッセージを記録し、表示するレベルを選べます。Pi に画面がない状態で動かすときは、ログをファイルに送る必要があります。

防御的な検査と時間の計測

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 ピンのヘッダーで、Pi は出力を切り替え、入力を読めます。例では LED、ボタン、PWM による調光、サーボ、距離センサー、リレーを扱います。電源を入れる前に、必ず配線をピン配置図と照合してください。

出力、入力、プルアップ抵抗

ピンを出力に設定すると、抵抗を介して LED を駆動できます。入力に設定すると、ボタンを読めます。プルアップ抵抗は、ボタンがグラウンドに接続されるまで入力を High に保ち、ピンが浮いた状態になるのを防ぎます。

PWM とサーボ

パルス幅変調は、ピンを素早くオンとオフに切り替えます。オンの時間の割合が明るさやサーボの位置を決めます。gpiozero ライブラリはタイミングの処理のほとんどを隠してくれます。

リレー、センサー、安全な停止

リレーは、ポンプのような大電力の機器を切り替えるので、Pi は小さな制御コイルを駆動するだけで済みます。出力は必ず 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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 52: gpiozero による LED

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)

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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)

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

トピック 9: I2C、SPI、1-Wire、カメラのセンサー

多くのセンサーはバス経由で Pi と通信します。I2C は二本の線とアドレスを使い、SPI は四本の線とチップセレクトを使い、1-Wire はデータ線が一本です。カメラモジュールは独自のライブラリを持つ特別なケースです。

I2C: スキャンとレジスタの読み取り

I2C の各機器にはアドレスがあります。まずバスをスキャンして、何が接続されているかを見つけます。チップ ID のレジスタを読んでセンサーを確認し、その後で制御レジスタに書き込み、データのバイトを読みます。

SPI とアナログ入力

Pi にはアナログ入力がないので、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)

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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)

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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)

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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")

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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")

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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: 静止画の撮影

Pi のカメラモジュールで写真を撮ります。

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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

トピック 10: ソケットと HTTP によるネットワーク

IoT 機器は他のコンピューターと通信しなければなりません。ソケットは低レベルの基盤、HTTP は Web サービスと API の仕組みです。堅牢なコードはタイムアウト、再試行、オフラインの期間を扱います。

TCP と UDP のソケット

TCP ソケットは信頼できる接続で、UDP は接続なしで単一のメッセージを送ります。ポート 0 にバインドすると、OS が空きポートを選ぶので、例同士が衝突しません。

タイムアウト付きの 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)

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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)

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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)

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 74: ステータスの Web サーバー

Pi で小さなステータスページを配信します。

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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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")

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

トピック 11: MQTT: 発行と購読

MQTT は IoT の標準的なメッセージングプロトコルです。機器はトピックに値を発行し、ブローカーがそれを配信し、任意の数の購読者が受け取ります。小さな機器にも十分軽く、弱いネットワークにも強い設計です。

トピック、発行、購読

トピックは home/kitchen/temperature のようなパスです。発行はブローカーを通してトピックにメッセージを送ります。購読はワイルドカードを使えます。+ は一つの階層に一致し、# はその下のすべてに一致します。

QoS と保持メッセージ

QoS 0 は送りっぱなし、QoS 1 は少なくとも一回の配信を保証し、QoS 2 は正確に一回配信します。保持メッセージはブローカーが保存するので、新しい購読者はすぐに最後の値を受け取れます。

Last Will、TLS、ディスカバリー

Last Will のメッセージは、機器が予期せず切断されたときにブローカーが発行するもので、オフラインの機器を検知できます。インターネット上のブローカーには 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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 79: MQTT: 切断検知のための last will

機器が切断されたら全員に知らせます。

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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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()

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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 による Web ダッシュボードと API

Flask は小さな Python の Web フレームワークです。数行で状態ページを配信し、JSON を返し、他の機器からの読み取り値を受け付けられるので、スマートフォンやブラウザから Pi を見ることができます。

ルートとレスポンス

@app.get('/') のようなルートのデコレーターは、URL を関数に結びつけます。関数はテキスト、ページ、または jsonify による JSON を返します。ローカルネットワークからの接続を受け付けるには 0.0.0.0 のアドレスを使います。

検証付きのデータ受け付け

POST のルートはリクエストから JSON を読みます。何かを保存する前に型と範囲を検証し、データが誤っていれば 400 のステータスを返します。送信側が何を直せばよいかわかるようにするためです。

バックグラウンドのスレッドとダッシュボード

バックグラウンドのスレッドは、Web サーバーがリクエストに応答している間にセンサーを読みます。共有データは単純で安全に保ちましょう。小さな HTML テンプレートで、最新の読み取り値をネットワーク上の誰でも見られるページにできます。

例 83: Flask の hello エンドポイント

Pi で最初の Web ルートを提供します。

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)

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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)

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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)

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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)

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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)

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

トピック 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: データの保持期間

古い行を削除して、Pi の容量を小さく保ちます。

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 サービス

Pi の起動時にロガーを自動で動かします。

# /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

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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

この例は Raspberry Pi、ネットワークサービス、またはシステムサービスが必要なため、構文チェックのみを行いました。出力は示していません。

例 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 カードがいっぱいにならないためのローテーションするログ、コードの外に置く秘密情報、受信データの検証、そしてセンサーから警報までの完全なパイプラインです。

ローテーションするログとハートビート

ローテーションするファイルハンドラーは、各ログファイルのサイズを制限し、古いものをいくつか残します。タイムスタンプ付きのハートビートファイルは、ウォッチドッグがプログラムの生存を確認できるようにします。

秘密情報と検証

パスワードやトークンは環境変数に入れ、決して表示しません。保存や警報に使う前に、すべての受信値の型と範囲を検証します。センサーやネットワークは何でも送ってくる可能性があるからです。

パイプラインを組み立てる

パイプラインはセンサーを読み、窓を使って値を平滑化し、しきい値と比べ、警報フラグを立て、各段階の結果を保存します。各段階は小さくテスト可能で、これがプロフェッショナルな IoT コードの構成方法です。

例 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 のような IoT プラットフォームを学び、安全な機器の識別情報と無線アップデートを加え、マイコン向けに Raspberry Pi Pico を学びましょう。ハードウェアの例は、まず自分の配線でテストし、使っているモデルのピン番号を Raspberry Pi の公式ドキュメントで確認してください。

まとめ

Raspberry Pi は、センサーを確実に読み取り、データを処理し、保存し、ネットワーク経由で送り、人の手を借りずに動き続けるとき、本物の IoT 機器になります。100 の例を通して、この記事では Python の基礎、判断と繰り返し、関数、データ構造、ファイルと形式、エラーとログ、センサーデータの処理、GPIO、I2C、SPI、1-Wire とカメラ、ソケットと HTTP、MQTT、Flask、SQLite、systemd と cron、そして完全なパイプラインを扱いました。ラボで部屋のモニターを作り、一度に一つずつ変えて、各部分の振る舞いを学びましょう。