Raspberry Pi Pico Wで作るワイヤレス環境センサーノード：MicroPython実装

部品と用品

Raspberry Pi Pico W	オンボードWiFi (CYW43439) 搭載	1
BME280 温湿度・気圧センサー (I2C)	1
SCD41 CO2センサー (I2C)	高精度NDIR、低消費電力	1
リチウムイオン電池 18650 + 充電保護基板 (TP4056)	バッテリー駆動化用	1
3.3V LDOレギュレーター (HT7833等)	電池から3.3V生成	1
防水ケース・ソーラーパネル (オプション)	屋外設置用	1式

必要な工具・環境

• Thonny IDE / VS Code + Pico拡張
• MicroPythonファームウェア (最新版)
• WiFi環境（2.4GHz）

このプロジェクトについて

Raspberry Pi Pico Wの低消費電力性能とWiFiを活かし、バッテリー駆動で長期間動作するワイヤレス環境センサーノードをMicroPythonで実装します。深い睡眠（Deep Sleep）モードを活用して消費電力を最小化し、定期的にセンサー値を読み取ってMQTTまたはHTTPでサーバーに送信。屋内外の環境モニタリングに最適です。

電力管理戦略

• アクティブ時: ~50mA（センサー測定 + WiFi接続 + 送信 約10-15秒）
• Deep Sleep時: ~1-2mA（RTCアラームでウェイクアップ）
• 測定間隔: 5分〜1時間（用途に応じて調整）
• 電池寿命試算: 18650 3000mAhで、5分間隔なら約3ヶ月、1時間間隔なら1年以上

MicroPythonコード

main.py（メインプログラム）

import machine
import network
import urequests
import ujson
import time
import gc
from machine import Pin, I2C, deepsleep, RTC
from bme280 import BME280
from scd41 import SCD41

# ===== 設定 =====
WIFI_SSID = "YOUR_SSID"
WIFI_PASSWORD = "YOUR_PASSWORD"
MQTT_BROKER = "mqtt.example.com"
MQTT_PORT = 1883
MQTT_TOPIC = "sensors/pico_w_01"
MEASURE_INTERVAL_SEC = 300  # 5分
DEVICE_ID = "pico_w_01"

# ピン定義
I2C_SDA = 4
I2C_SCL = 5
LED_PIN = "LED"  # オンボードLED

# ===== グローバル =====
i2c = I2C(0, sda=Pin(I2C_SDA), scl=Pin(I2C_SCL), freq=100000)
led = Pin(LED_PIN, Pin.OUT)
rtc = RTC()

def connect_wifi():
    """WiFi接続（タイムアウト付き）"""
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    wlan.connect(WIFI_SSID, WIFI_PASSWORD)
    
    timeout = 20
    while not wlan.isconnected() and timeout > 0:
        time.sleep(1)
        timeout -= 1
    
    if wlan.isconnected():
        print("WiFi connected:", wlan.ifconfig())
        return wlan
    else:
        print("WiFi connection failed")
        return None

def read_sensors():
    """全センサー読み取り"""
    data = {"device_id": DEVICE_ID, "timestamp": time.time()}
    
    # BME280
    try:
        bme = BME280(i2c=i2c)
        data["temperature"] = round(bme.temperature, 2)
        data["humidity"] = round(bme.humidity, 2)
        data["pressure"] = round(bme.pressure, 2)
    except Exception as e:
        print("BME280 error:", e)
        data["temperature"] = data["humidity"] = data["pressure"] = None
    
    # SCD41
    try:
        scd41 = SCD41(i2c)
        scd41.start_periodic_measurement()
        time.sleep(5)  # 初回測定待ち
        if scd41.data_ready:
            co2, temp, hum = scd41.read_measurement()
            data["co2"] = co2
            # BME280優先だが、SCD41の温湿度もバックアップとして記録
            data["scd41_temp"] = round(temp, 2)
            data["scd41_humidity"] = round(hum, 2)
    except Exception as e:
        print("SCD41 error:", e)
        data["co2"] = None
    
    # バッテリー電圧（ADCで測定、分圧回路必要）
    # data["battery"] = read_battery_voltage()
    
    return data

def send_mqtt(data):
    """MQTTで送信"""
    try:
        import umqtt.simple as mqtt
        client = mqtt.MQTTClient(DEVICE_ID, MQTT_BROKER, port=MQTT_PORT)
        client.connect()
        client.publish(MQTT_TOPIC, ujson.dumps(data))
        client.disconnect()
        print("MQTT sent:", data)
        return True
    except Exception as e:
        print("MQTT error:", e)
        return False

def send_http(data):
    """HTTP POSTで送信（フォールバック）"""
    try:
        url = "https://your-server.com/api/sensor-data"
        headers = {"Content-Type": "application/json"}
        resp = urequests.post(url, data=ujson.dumps(data), headers=headers, timeout=10)
        print("HTTP status:", resp.status_code)
        resp.close()
        return True
    except Exception as e:
        print("HTTP error:", e)
        return False

def blink_led(times=3, delay_ms=100):
    """ステータス表示用LED点滅"""
    for _ in range(times):
        led.on()
        time.sleep_ms(delay_ms)
        led.off()
        time.sleep_ms(delay_ms)

def main():
    print("=== Pico W Sensor Node Starting ===")
    print("Device ID:", DEVICE_ID)
    print("Interval:", MEASURE_INTERVAL_SEC, "sec")
    
    # WiFi接続
    wlan = connect_wifi()
    if not wlan:
        blink_led(5, 200)  # エラー：高速点滅
        goto_sleep()
    
    # センサー読み取り
    blink_led(1, 500)
    data = read_sensors()
    
    # 送信試行
    success = send_mqtt(data)
    if not success:
        success = send_http(data)
    
    if success:
        blink_led(2, 200)  # 成功：2回点滅
    else:
        blink_led(5, 200)  # 失敗：高速点滅
    
    # メモリ解放
    gc.collect()
    
    # ディープスリープへ
    goto_sleep()

def goto_sleep():
    """RTCアラーム設定してディープスリープ"""
    print("Entering deep sleep for", MEASURE_INTERVAL_SEC, "seconds")
    rtc.alarm(rtc.ALARM0, MEASURE_INTERVAL_SEC * 1000)
    machine.deepsleep()

# エントリーポイント
if __name__ == "__main__":
    main()

bme280.py（BME280ドライバー）

# MicroPython BME280 driver (simplified)
import time
from micropython import const

class BME280:
    def __init__(self, i2c, addr=0x76):
        self.i2c = i2c
        self.addr = addr
        self._load_calibration()
        self.i2c.writeto_mem(self.addr, 0xF2, b'\x01')  # hum oversampling x1
        self.i2c.writeto_mem(self.addr, 0xF4, b'\x27')  # temp/press oversampling x1, normal mode
    
    def _load_calibration(self):
        # キャリブレーションデータ読み込み（省略版）
        pass
    
    @property
    def temperature(self):
        # 温度読み取り・補正計算
        raw = self._read_raw_temp()
        return self._compensate_temp(raw)
    
    @property
    def humidity(self):
        raw = self._read_raw_hum()
        return self._compensate_hum(raw)
    
    @property
    def pressure(self):
        raw = self._read_raw_press()
        return self._compensate_press(raw) / 100.0  # hPa
    
    # 以下、生値読み取り・補正計算メソッド（実装省略）
    # 完全版は https://github.com/robert-hh/BME280-MicroPython 参照

scd41.py（SCD41ドライバー）

# MicroPython SCD41 driver (simplified)
class SCD41:
    def __init__(self, i2c, addr=0x62):
        self.i2c = i2c
        self.addr = addr
        self.stop_periodic_measurement()
        time.sleep(0.5)
    
    def start_periodic_measurement(self):
        self.i2c.writeto(self.addr, b'\x21\xb1')
    
    def stop_periodic_measurement(self):
        self.i2c.writeto(self.addr, b'\x3f\x86')
        time.sleep(0.5)
    
    @property
    def data_ready(self):
        resp = self.i2c.readfrom_mem(self.addr, 0xE4\xB8, 3)
        return (resp[0] & 0x07) != 0
    
    def read_measurement(self):
        self.i2c.writeto(self.addr, b'\xEC\x05')
        time.sleep(0.001)
        data = self.i2c.readfrom(self.addr, 9)
        co2 = (data[0] << 8) | data[1]
        temp = -45 + 175 * ((data[3] << 8) | data[4]) / 65535
        hum = 100 * ((data[6] << 8) | data[7]) / 65535
        return co2, temp, hum

boot.py（起動時設定）

# boot.py - 自動実行設定
import machine
import gc

# CPU周波数を下げて省電力（デフォルト133MHz → 48MHz）
machine.freq(48000000)

# メモリ最適化
gc.collect()
gc.threshold(gc.mem_free() // 4)

print("Boot complete, free mem:", gc.mem_free())

セットアップ手順

1. Pico Wに最新MicroPythonファームウェアを書き込み
2. Thonnyで接続し、bme280.py、scd41.py、boot.py、main.py をアップロード
3. WiFi認証情報とサーバー設定をmain.pyに記入
4. リセットボタン押下で自動起動確認

サーバー側受信例（Python/Flask）

from flask import Flask, request, jsonify
import paho.mqtt.client as mqtt
import json
from datetime import datetime

app = Flask(__name__)

# MQTT購読
def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    print(f"Received: {data}")
    # DB保存処理等

client = mqtt.Client()
client.on_message = on_message
client.connect("localhost", 1883)
client.subscribe("sensors/+")
client.loop_start()

@app.route("/api/sensor-data", methods=["POST"])
def receive_http():
    data = request.get_json()
    print(f"HTTP Received: {data}")
    # DB保存
    return jsonify({"status": "ok"})

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

運用のポイント

• 初回起動時: SCD41は最初の測定まで5秒以上必要。定期測動モード移行後も最初の数回はスキップ推奨
• 時刻同期: RTCはディープスリープ跨ぎで保持されるが、初回起動時はNTPで同期すると確実
• ウォッチドッグ: 万一ハングした場合にリセットするようWDT設定推奨
• OTA更新: 現地メンテナンス不要にするならMicroPython OTA実装を検討

拡張アイデア

• ソーラーパネル＋リチウムイオン充電コントローラーで完全自立化
• 複数ノードをメッシュネットワーク（ESP-NOW等）で中継
• 加速度センサー追加で振動・傾斜検知（土砂崩れ・構造物監視）
• LoRaモジュール追加で長距離・低消費電力通信