目的は、DHT22測定・送信を安定化し、落ちた時や送信失敗時に原因を追えるようにすることです。

1. dht22.py
/home/pi/script/dht22.py

import time
import traceback
import os
import sys
import board
import adafruit_dht
import requests

SERVER_URL = "https://fideafact.ddns.net/sensor"
INTERVAL = 600
LOCKFILE = "/tmp/dht22.lock"

# 多重起動防止
def check_lock():
    if os.path.exists(LOCKFILE):
        with open(LOCKFILE) as f:
            old_pid = f.read().strip()
        if old_pid and os.path.exists(f"/proc/{old_pid}"):
            print(f"既に起動中 (PID {old_pid})、終了します", flush=True)
            sys.exit(1)
    with open(LOCKFILE, "w") as f:
        f.write(str(os.getpid()))

def cleanup_lock():
    try:
        os.remove(LOCKFILE)
    except OSError:
        pass

check_lock()

def create_dht():
    return adafruit_dht.DHT22(board.D4, use_pulseio=False)

dht = create_dht()
error_count = 0
MAX_RETRY = 3

try:
    while True:
        try:
            print("測定開始", flush=True)

            try:
                temp = dht.temperature
                humidity = dht.humidity
                error_count = 0
            except RuntimeError as e:
                print(f"読み取りエラー（継続）: {e}", flush=True)
                error_count += 1
            except OSError as e:
                print(f"DHT/GPIOエラー（継続）: {type(e).__name__}: {e}", flush=True)
                error_count += 1
            else:
                if temp is None or humidity is None:
                    print("Noneのため送信しない", flush=True)
                else:
                    print(f"温度: {temp:.1f}°C  湿度: {humidity:.1f}%", flush=True)
                    payload = {"temperature": temp, "humidity": humidity}
                    try:
                        r = requests.post(SERVER_URL, json=payload, timeout=10)
                        print(f"送信結果: HTTP {r.status_code}", flush=True)
                        print(f"応答本文: {r.text[:300]}", flush=True)
                        r.raise_for_status()
                    except requests.RequestException as e:
                        print(f"HTTP送信エラー（継続）: {type(e).__name__}: {e}", flush=True)

            # 連続エラー時は即座に再初期化（待機せず）
            if error_count >= MAX_RETRY:
                print(f"連続{error_count}回エラー → DHT再初期化", flush=True)
                try:
                    dht.exit()
                except Exception:
                    pass
                time.sleep(3)
                dht = create_dht()
                error_count = 0

        except Exception:
            print("想定外エラー:", flush=True)
            traceback.print_exc()

        finally:
            print(f"{INTERVAL}秒待機", flush=True)
            time.sleep(INTERVAL)

finally:
    cleanup_lock()
    try:
        dht.exit()
    except Exception:
        pass

主なポイント：

• INTERVAL = 900 で15分間隔 :好きな秒数でOK
• flush=True でログをすぐjournalに出す
• DHT22/GPIOエラーとHTTP送信エラーを分けて表示
• 1回の測定失敗・送信失敗でプログラムを終了しない
• None の温湿度は送信しない
• 想定外エラーは traceback を出して調査可能にする

上記のコードでもOKですが、更にUpgardeさせたコードはこちら

Ver.2

/home/pi/script/dht22.py

import time
import traceback
import os
import sys
import board
import adafruit_dht
import requests

SERVER_URL = "https://fideafact.ddns.net/sensor"

SEND_INTERVAL = 300          # 5分ごとに送信
SAMPLE_COUNT = 12            # 1分間で12回測定
SAMPLE_INTERVAL = 5          # 5秒ごとに測定
LOCKFILE = "/tmp/dht22.lock"

MAX_RETRY = 3

# 明らかに変な値を除外する範囲
TEMP_MIN = -40.0
TEMP_MAX = 80.0
HUM_MIN = 0.0
HUM_MAX = 100.0


# 多重起動防止
def check_lock():
    if os.path.exists(LOCKFILE):
        with open(LOCKFILE) as f:
            old_pid = f.read().strip()
        if old_pid and os.path.exists(f"/proc/{old_pid}"):
            print(f"既に起動中 (PID {old_pid})、終了します", flush=True)
            sys.exit(1)

    with open(LOCKFILE, "w") as f:
        f.write(str(os.getpid()))


def cleanup_lock():
    try:
        os.remove(LOCKFILE)
    except OSError:
        pass


def create_dht():
    return adafruit_dht.DHT22(board.D4, use_pulseio=False)


def is_valid_reading(temp, humidity):
    if temp is None or humidity is None:
        return False

    if not (TEMP_MIN <= temp <= TEMP_MAX):
        return False

    if not (HUM_MIN <= humidity <= HUM_MAX):
        return False

    return True


def trimmed_average(values):
    """
    上下の値を捨てて平均する。
    データ数が多ければ上下2個、少なければ上下1個を捨てる。
    """
    if not values:
        return None

    values = sorted(values)
    n = len(values)

    if n >= 8:
        trim = 2
    elif n >= 4:
        trim = 1
    else:
        trim = 0

    trimmed = values[trim:n - trim] if trim > 0 else values

    if not trimmed:
        return None

    return sum(trimmed) / len(trimmed)


def measure_for_one_minute(dht):
    global_error_count = 0
    readings = []

    print("1分間の測定開始", flush=True)

    for i in range(SAMPLE_COUNT):
        try:
            temp = dht.temperature
            humidity = dht.humidity

            if is_valid_reading(temp, humidity):
                readings.append({
                    "temperature": temp,
                    "humidity": humidity
                })
                print(
                    f"{i + 1}/{SAMPLE_COUNT}: 温度 {temp:.1f}°C 湿度 {humidity:.1f}%",
                    flush=True
                )
            else:
                print(
                    f"{i + 1}/{SAMPLE_COUNT}: 異常値のため除外 temp={temp}, humidity={humidity}",
                    flush=True
                )

        except RuntimeError as e:
            print(f"{i + 1}/{SAMPLE_COUNT}: 読み取りエラー（継続）: {e}", flush=True)
            global_error_count += 1

        except OSError as e:
            print(f"{i + 1}/{SAMPLE_COUNT}: DHT/GPIOエラー（継続）: {type(e).__name__}: {e}", flush=True)
            global_error_count += 1

        if i < SAMPLE_COUNT - 1:
            time.sleep(SAMPLE_INTERVAL)

    temperatures = [r["temperature"] for r in readings]
    humidities = [r["humidity"] for r in readings]

    avg_temp = trimmed_average(temperatures)
    avg_humidity = trimmed_average(humidities)

    return avg_temp, avg_humidity, len(readings), global_error_count


check_lock()

dht = create_dht()
error_count = 0

try:
    while True:
        cycle_start = time.time()

        try:
            avg_temp, avg_humidity, valid_count, read_errors = measure_for_one_minute(dht)
            error_count += read_errors

            if avg_temp is None or avg_humidity is None:
                print("有効な測定値が不足しているため送信しません", flush=True)
                error_count += 1
            else:
                print(
                    f"平均値: 温度 {avg_temp:.1f}°C 湿度 {avg_humidity:.1f}% "
                    f"有効測定数 {valid_count}/{SAMPLE_COUNT}",
                    flush=True
                )

                payload = {
                    "temperature": round(avg_temp, 2),
                    "humidity": round(avg_humidity, 2),
                    "sample_count": valid_count
                }

                try:
                    r = requests.post(SERVER_URL, json=payload, timeout=10)
                    print(f"送信結果: HTTP {r.status_code}", flush=True)
                    print(f"応答本文: {r.text[:300]}", flush=True)
                    r.raise_for_status()

                    error_count = 0

                except requests.RequestException as e:
                    print(f"HTTP送信エラー（継続）: {type(e).__name__}: {e}", flush=True)

            if error_count >= MAX_RETRY:
                print(f"連続{error_count}回エラー → DHT再初期化", flush=True)
                try:
                    dht.exit()
                except Exception:
                    pass

                time.sleep(3)
                dht = create_dht()
                error_count = 0

        except Exception:
            print("想定外エラー:", flush=True)
            traceback.print_exc()

        finally:
            elapsed = time.time() - cycle_start
            wait_time = max(0, SEND_INTERVAL - elapsed)

            print(f"{wait_time:.0f}秒待機", flush=True)
            time.sleep(wait_time)

finally:
    cleanup_lock()
    try:
        dht.exit()
    except Exception:
        pass

以下のように変更。

• 5分ごとに送信
• 送信前に 1分間で12回測定
• 測定間隔は 5秒
• None、範囲外、読み取りエラーは除外
• 残った値から上下を捨てる
• 平均値を送信

高頻度測定なら、DHT22より Sensirion SHT45 がおすすめです。

おすすめ順

センサー	おすすめ度	特徴
SHT45	◎	高精度、I2C接続、Raspberry Piで扱いやすい。測定自体は数ms級、湿度応答は約4秒
SHT31 / SHT35	○	入手性がよく、実績が多い。最大10回/秒モードあり。ただし湿度応答は約6〜8秒
SHT85	○	高精度で交換しやすい形状。業務寄り。湿度応答は約8秒
BME280	△	温湿度＋気圧が取れる。安価で便利だが、温湿度の精度重視ならSHT系の方が無難
DHT22	×	低速。2秒以上あける必要があり、高頻度測定には不向き

結論:

Raspberry Pi Zero 2 Wで使うなら、まずは SHT45搭載のI2Cモジュール（価格は高い） を選ぶのが一番よいです。AdafruitなどのSHT45ブレイクアウトはRaspberry Pi用Pythonライブラリもあり、DHT22から置き換えやすいです。ただし注意点があります。温度は比較的速く追従しますが、湿度は空気中の水分がセンサー内部に拡散する時間があるため、1秒に何十回読んでも実際の変化追従は数秒単位です。つまり実用上は 1秒ごと、または2秒ごと の記録で十分です。

DHT22の代替としては：

DHT22 -> SHT45
DATA 1本 -> I2C SDA/SCL
読み取り間隔 5分 -> 1〜2秒も可能
精度・安定性アップ

Raspberry Piへの接続例：

SHT45 VIN -> 3.3V
SHT45 GND -> GND
SHT45 SDA -> GPIO2 / SDA
SHT45 SCL -> GPIO3 / SCL

DHT22なら SAMPLE_INTERVAL = 5 はかなり安全です。もっと細かくしたい場合でも、DHT22では 2秒未満 にはしない方がよいです。

2. dht22.service
/etc/systemd/system/dht22.service

[Unit]
Description=DHT22 Sensor

[Service]
ExecStart=/usr/bin/python3 /home/pi/dht22.py
Restart=on-failure
RestartSec=10

# 多重起動防止
Type=simple

# 停止時に確実にプロセスを殺す
KillMode=control-group
KillSignal=SIGTERM
TimeoutStopSec=15

[Install]
WantedBy=multi-user.target

主なポイント：

• network-online.target でWi-Fi/DNSが使える状態に近づいてから起動
• Restart=always で落ちたら自動再起動
• RestartSec=10 で10秒待って再起動
• PYTHONUNBUFFERED=1 でログを即時出力
• User=pi でpiユーザーとして実行

3. 反映コマンド

sudo systemctl daemon-reload sudo systemctl restart dht22.service

4. ログ永続化
再起動前のログも残すために実施済み。

sudo mkdir -p /var/log/journal sudo systemctl restart systemd-journald

5. 確認コマンド

現在状態：

systemctl status dht22.service

直近ログ：

journalctl -u dht22.service -n 30 --no-pager -l

15分後の動作確認例：

journalctl -u dht22.service --since "2026-06-08 06:48:00" --no-pager -l

期待ログ：

測定開始 温度: 23.3°C 湿度: 71.0% 送信完了: HTTP 200 900秒待機

今回分かったこと
落ちた直接ログは、journal永続化前だったため残っていませんでした。
ただし、現在はログ永続化済みなので、次回もし再起動や異常終了が起きても原因を追いやすくなっています。

また、起動直後に以下のDNS失敗が出ていました。

Failed to resolve 'fideafact.ddns.net'

そのため、dht22.service を network-online.target 待ちに修正しました。今は HTTP 200 が返っており、測定・送信は正常です。

---

Part2は受信側になります。

---

追記

GPOIエラーで、センサーが読み取れない状態になった。
/home/pi/script/dht22.pyを修正（掲載は修正済コード）
/etc/systemd/system/dht22.service（掲載は修正済コード）
Pi Zeroシリーズはそもそもpulseioが不安定なことで知られており、use_pulseio=False が推奨設定です。

確認・反映手順

# スクリプトを編集
nano /home/pi/script/dht22.py

# サービス再起動　編集後に再起動させ反映させる
sudo systemctl restart dht22.service

# ログ確認
journalctl -u dht22.service -f