Raspberry Pi Pico Wで作る！毎日19時に天気予報を自動メール送信するIoTシステム

はじめに

「Raspberry Pi Pico W」を使って、毎日決まった時刻に自動で東京の天気予報をメール送信するIoTシステムを作ってみました。毎晩19時になると「明日の天気」をGmail経由で知らせてくれる――まるでお天気お知らせ係のように働いてくれます。

システムの概要
Wi-Fi接続の仕組み
実行するとどうなる？

システムの概要

今回のシステムは以下の流れで動きます：

Raspberry Pi Pico W がWi-Fiに接続
NTP（ネット時刻同期） で正しい時刻を取得
つくみじま天気API から東京の天気予報を取得
Gmail経由でメール送信
毎日19:00ちょうどに動作し、1日1回だけ送信

これによって、ユーザーは毎日決まった時間に「明日の天気予報」を自動で受け取れます。

Wi-Fi接続の仕組み

まずはWi-Fiに接続する処理。
「wifi_connect.py」に外部関数として分離し、再接続機能も備えています。

# wifi_connect.py

import network
import time

class WiFiConnector:
    def __init__(self, ssid, password, timeout=20, max_retries=3):
        """
        Initialize the WiFiConnector with credentials and settings.
        
        Parameters:
            ssid (str): Wi-Fi network SSID
            password (str): Wi-Fi password
            timeout (int): Timeout in seconds per connection attempt
            max_retries (int): Number of reconnection retries
        """
        self.ssid = ssid
        self.password = password
        self.timeout = timeout
        self.max_retries = max_retries
        self.wlan = network.WLAN(network.STA_IF)

    def connect(self):
        """Attempts to connect to Wi-Fi once. Returns True if successful."""
        try:
            self.wlan.active(True)

            if not self.wlan.isconnected():
                print(f'Connecting to {self.ssid}...')
                self.wlan.connect(self.ssid, self.password)

                for _ in range(self.timeout):
                    if self.wlan.isconnected():
                        break
                    time.sleep(1)

            if self.wlan.isconnected():
                print('Connected successfully!')
                print('IP address:', self.wlan.ifconfig()[0])
                return True
            else:
                print('Connection failed: timeout.')
                return False

        except Exception as e:
            print('Error during Wi-Fi connection:', str(e))
            return False

    def reconnect(self):
        """
        Tries to reconnect to Wi-Fi, retrying up to max_retries times.

        Returns:
            bool: True if reconnection succeeded, False otherwise.
        """
        print(f'Reconnecting (max {self.max_retries} retries)...')
        for attempt in range(1, self.max_retries + 1):
            print(f'Attempt {attempt}...')
            if self.connect():
                return True
            time.sleep(2)  # wait before retrying
        print('Reconnection failed.')
        return False

    def disconnect(self):
        """Disconnects from the Wi-Fi network."""
        if self.wlan.isconnected():
            self.wlan.disconnect()
            print('Disconnected from Wi-Fi.')

    def is_connected(self):
        """Returns True if connected to Wi-Fi."""
        return self.wlan.isconnected()

    def ip_address(self):
        """Returns the current IP address, or None if not connected."""
        if self.wlan.isconnected():
            return self.wlan.ifconfig()[0]
        return None

秘密情報を分離する（secrets.py）

Wi-FiのSSIDやパスワード、Gmailの送信設定などは別ファイルに保存。セキュリティ的にも安心です。

# secrets.py

secrets = {
'ssid': 'SSIDXXX',
'password': 'PASWORDXXX',
'sender_email': 'XXXXXXX@gmail.com',
'sender_name': 'RaspberryPiPico',
'sender_app_password': 'XXXX XXXX XXXX XXXX',
'recipient_email': 'YYYYYYY@gmail.com',
'email_subject': 'Email from RPi Pico',
}

MicroPython 用の SMTP クライアントライブラリ「uMail（MicroMail）」 のコードです、このファイルをmain.pyと同じフォルダに配置してください。

uMail/umail.py at master · shawwwn/uMail

A lightweight, scalable SMTP client for sending email in MicroPython - shawwwn/uMail

# uMail (MicroMail) for MicroPython
# Copyright (c) 2018 Shawwwn <shawwwn1@gmail.com>
# License: MIT
import socket

DEFAULT_TIMEOUT = 10 # sec
LOCAL_DOMAIN = '127.0.0.1'
CMD_EHLO = 'EHLO'
CMD_STARTTLS = 'STARTTLS'
CMD_AUTH = 'AUTH'
CMD_MAIL = 'MAIL'
AUTH_PLAIN = 'PLAIN'
AUTH_LOGIN = 'LOGIN'

class SMTP:
    def cmd(self, cmd_str):
        sock = self._sock;
        sock.write('%s\r\n' % cmd_str)
        resp = []
        next = True
        while next:
            code = sock.read(3)
            next = sock.read(1) == b'-'
            resp.append(sock.readline().strip().decode())
        return int(code), resp

    def __init__(self, host, port, ssl=False, username=None, password=None):
        import ssl
        self.username = username
        addr = socket.getaddrinfo(host, port)[0][-1]
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(DEFAULT_TIMEOUT)
        sock.connect(addr)
        if ssl:
            sock = ssl.wrap_socket(sock)
        code = int(sock.read(3))
        sock.readline()
        assert code==220, 'cant connect to server %d, %s' % (code, resp)
        self._sock = sock

        code, resp = self.cmd(CMD_EHLO + ' ' + LOCAL_DOMAIN)
        assert code==250, '%d' % code
        if not ssl and CMD_STARTTLS in resp:
            code, resp = self.cmd(CMD_STARTTLS)
            assert code==220, 'start tls failed %d, %s' % (code, resp)
            self._sock = ssl.wrap_socket(sock)

        if username and password:
            self.login(username, password)

    def login(self, username, password):
        self.username = username
        code, resp = self.cmd(CMD_EHLO + ' ' + LOCAL_DOMAIN)
        assert code==250, '%d, %s' % (code, resp)

        auths = None
        for feature in resp:
            if feature[:4].upper() == CMD_AUTH:
                auths = feature[4:].strip('=').upper().split()
        assert auths!=None, "no auth method"

        from ubinascii import b2a_base64 as b64
        if AUTH_PLAIN in auths:
            cren = b64("\0%s\0%s" % (username, password))[:-1].decode()
            code, resp = self.cmd('%s %s %s' % (CMD_AUTH, AUTH_PLAIN, cren))
        elif AUTH_LOGIN in auths:
            code, resp = self.cmd("%s %s %s" % (CMD_AUTH, AUTH_LOGIN, b64(username)[:-1].decode()))
            assert code==334, 'wrong username %d, %s' % (code, resp)
            code, resp = self.cmd(b64(password)[:-1].decode())
        else:
            raise Exception("auth(%s) not supported " % ', '.join(auths))

        assert code==235 or code==503, 'auth error %d, %s' % (code, resp)
        return code, resp

    def to(self, addrs, mail_from=None):
        mail_from = self.username if mail_from==None else mail_from
        code, resp = self.cmd('MAIL FROM: <%s>' % mail_from)
        assert code==250, 'sender refused %d, %s' % (code, resp)

        if isinstance(addrs, str):
            addrs = [addrs]
        count = 0
        for addr in addrs:
            code, resp = self.cmd('RCPT TO: <%s>' % addr)
            if code!=250 and code!=251:
                print('%s refused, %s' % (addr, resp))
                count += 1
        assert count!=len(addrs), 'recipient refused, %d, %s' % (code, resp)

        code, resp = self.cmd('DATA')
        assert code==354, 'data refused, %d, %s' % (code, resp)
        return code, resp

    def write(self, content):
        self._sock.write(content)

    def send(self, content=''):
        if content:
            self.write(content)
        self._sock.write('\r\n.\r\n') # the five letter sequence marked for ending
        line = self._sock.readline()
        return (int(line[:3]), line[4:].strip().decode())

    def quit(self):
        self.cmd("QUIT")
        self._sock.close()

メイン処理（main.py）

いよいよメイン。
このプログラムは毎日19:00に天気予報を取得し、Gmail経由で自動送信します。

ポイントは次のとおり：

メール送信：umailライブラリを利用し、Gmail SMTP（SSL）経由

NTP同期：CloudflareのNTPサーバーを利用

天気API：つくみじま天気API（無料で使える気象庁データラッパー）

# main.py

import network
import ntptime
import time
import umail
import ubinascii
import urequests as requests

from wifi_connect import WiFiConnector  # WiFi接続の外部関数
from secrets import secrets  # プライベート情報

# 設定
CITY_ID = "130010"  # 東京
SEND_HOUR = 19
SEND_MINUTE = 0
MAX_RETRIES = 3  # リトライ最大回数

# Wi-Fi接続
wifi = WiFiConnector(secrets['ssid'], secrets['password'])
if not wifi.connect():
    wifi.reconnect()
if not wifi.is_connected():
    raise RuntimeError('Wi-Fi 接続失敗')
print("Wi-Fi 接続完了")

# NTP同期
ntptime.host = "time.cloudflare.com"
try:
    ntptime.settime()
    print("NTP時刻同期成功")
except:
    print("NTP同期失敗")
    raise

# ヘッダーエンコード（日本語対応）
def encode_header(header_str):
    encoded = ubinascii.b2a_base64(header_str.encode('utf-8')).strip()
    return '=?UTF-8?B?' + encoded.decode() + '?='

# 天気情報取得（リトライ付き）
def fetch_weather():
    url = f"https://weather.tsukumijima.net/api/forecast/city/{CITY_ID}"
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            response = requests.get(url)
            weather_json = response.json()
            response.close()
            return weather_json
        except Exception as e:
            print(f"[{attempt}] 天気取得失敗: {e}")
            time.sleep(3)
    raise RuntimeError("天気取得に複数回失敗しました。")

# メール送信（リトライ付き）
def send_email(message, date_str):
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            smtp = umail.SMTP('smtp.gmail.com', 465, ssl=True)
            smtp.login(secrets['sender_email'], secrets['sender_app_password'])
            smtp.to(secrets['recipient_email'])

            smtp.write("From: {} <{}>\r\n".format(encode_header(secrets['sender_name']), secrets['sender_email']))
            smtp.write("Subject: {}\r\n".format(encode_header(secrets['email_subject'])))
            smtp.write("Content-Type: text/plain; charset=utf-8\r\n")
            smtp.write("Content-Transfer-Encoding: 7bit\r\n")
            smtp.write("\r\n")
            smtp.write("こんにちは！明日の天気予報です。\r\n")
            smtp.write(message + "\r\n")
            smtp.write("現在日時: " + date_str + "\r\n")
            smtp.write("--\r\nRPi Picoより自動送信\r\n")

            smtp.send()
            smtp.quit()
            print("✅ メール送信完了")
            return
        except Exception as e:
            print(f"[{attempt}] メール送信失敗: {e}")
            time.sleep(3)
    raise RuntimeError("メール送信に複数回失敗しました。")

# メイン処理（天気＋メール）
def send_weather_email():
    weather_json = fetch_weather()
    forecast = weather_json['forecasts'][1]
    date_label = forecast['dateLabel']
    weather_telop = forecast['telop']
    rain_probability = forecast['chanceOfRain']['T12_18']
    weather_title = weather_json['title']

    if rain_probability == "--%":
        weather_msg = f"{weather_title}\n{date_label}の天気データは未確定です。"
    else:
        weather_msg = f"{weather_title}\n{date_label}の天気は「{weather_telop}」です。\n12時～18時の降水確率は {rain_probability} です。"

    now = time.localtime(time.time() + 9 * 60 * 60)
    date_str = "{0}/{1:02d}/{2:02d} {3:02d}:{4:02d}:{5:02d}".format(now[0], now[1], now[2], now[3], now[4], now[5])
    send_email(weather_msg, date_str)

# 送信フラグ管理
sent_today = False
last_sent_day = -1

# 定時送信用ループ
while True:
    now = time.localtime(time.time() + 9 * 60 * 60)
    hour = now[3]
    minute = now[4]
    today = now[2]

    if hour == SEND_HOUR and minute == SEND_MINUTE and not sent_today:
        try:
            send_weather_email()
        except Exception as e:
            print("❌ 処理失敗:", e)
        sent_today = True
        last_sent_day = today

    if today != last_sent_day:
        sent_today = False

    time.sleep(30)

実行するとどうなる？

毎日19:00になると、自動でこんなメールが届きます。

📩 受信メール例

こんにちは！明日の天気予報です。

東京地方の天気予報
明日の天気は「晴れ時々曇り」です。
12時～18時の降水確率は 20% です。

現在日時: 2025/08/16 19:00:00
--
RPi Picoより自動送信

Wi-Fi接続したRaspberry Pi Pico Wからメール送信 | CoTechWorks

概要組込デバイスからの通知はLINEやslackが流行っているようですが、環境によってはメール送信の方が都合がいいこともあるかと思います。ここではGmailの送信サーバーを使っ...