import network
import ntptime
import time
import umail
import ubinascii
import urequests as requests
from machine import Pin
from wifi_connect import WiFiConnector # Wi-Fi接続用クラス
from secrets import secrets # 機密情報（Wi-Fi、メール設定など）
# ======= 各種設定 =======
CITY_IDS = {
"東京": "130010",
"大阪": "270000",
"名古屋": "230010",
"福岡": "400010",
"札幌": "016010"
}
SEND_HOUR = 19 # メール送信する時刻（時）
SEND_MINUTE = 0 # メール送信する時刻（分）
MAX_RETRIES = 3 # 通信や送信の最大リトライ回数
TEST_BUTTON_PIN = 3 # テストボタンのGPIOピン番号（GPIO3）
TIMEZONE_OFFSET = 9 * 60 * 60 # JSTの時差（秒）
# ======= テストボタン初期化（内部プルアップ） =======
test_button = Pin(TEST_BUTTON_PIN, Pin.IN, Pin.PULL_UP)
led2 = Pin('LED', Pin.OUT)
# ======= 起動時にNTPで現在時刻を取得 =======
wifi = WiFiConnector(secrets['ssid'], secrets['password'])
wifi.reconnect()
ntptime.host = "time.cloudflare.com"
try:
ntptime.settime()
print("NTP時刻同期成功")
led2.value(1)
time.sleep(2)
led2.value(0)
except:
print("NTP同期失敗")
sta_if = network.WLAN(network.STA_IF)
sta_if.active(False)
# ======= 日本語ヘッダーをエンコードする関数（メール用） =======
def encode_header(header_str):
encoded = ubinascii.b2a_base64(header_str.encode('utf-8')).strip()
return '=?UTF-8?B?' + encoded.decode() + '?='
# ======= 天気予報を取得する関数（都市ID指定） =======
def fetch_weather(city_id):
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 build_weather_message(city_name, weather_json):
forecast = weather_json['forecasts'][1] # 明日の天気
date_label = forecast['dateLabel'] # 「明日」など
weather_telop = forecast['telop'] # 天気の説明
rain_06_12 = forecast['chanceOfRain']['T06_12']
rain_12_18 = forecast['chanceOfRain']['T12_18']
weather_title = weather_json['title']
# 気温情報
temp_min = forecast['temperature']['min']['celsius']
temp_max = forecast['temperature']['max']['celsius']
temp_min_str = temp_min + "℃" if temp_min else "未定"
temp_max_str = temp_max + "℃" if temp_max else "未定"
# メッセージ構築
if rain_06_12 == "--%" and rain_12_18 == "--%":
msg = (
f"<b>{city_name}</b><br>"
f"{weather_title}<br>"
f"{date_label}の天気データは未確定です。<br>"
f"最高気温: {temp_max_str}<br>"
f"最低気温: {temp_min_str}<br><br>"
)
else:
msg = (
f"<b>{city_name}</b><br>"
f"{weather_title}<br>"
f"{date_label}の天気は「{weather_telop}」です。<br>"
f"06～12時の降水確率: {rain_06_12}<br>"
f"12～18時の降水確率: {rain_12_18}<br>"
f"最高気温: {temp_max_str}<br>"
f"最低気温: {temp_min_str}<br><br>"
)
return msg
# ======= メールを送信する関数 =======
def send_email(html_body, 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/html; charset=utf-8\r\n")
smtp.write("Content-Transfer-Encoding: 7bit\r\n")
smtp.write("\r\n")
full_html = f"""
<html>
<body>
<p>明日の複数都市の天気予報です。</p>
{html_body}
<p>現在日時: {date_str}</p>
<hr>
<p style="font-size:small;">RPi Picoより自動送信</p>
</body>
</html>
"""
smtp.write(full_html)
smtp.send()
smtp.quit()
print("✅ HTMLメール送信完了")
led2.value(1)
time.sleep(2)
led2.value(0)
return
except Exception as e:
print(f"[{attempt}] メール送信失敗: {e}")
time.sleep(3)
raise RuntimeError("メール送信に複数回失敗しました。")
# ======= メール送信用メイン関数 =======
def send_weather_email():
wifi = WiFiConnector(secrets['ssid'], secrets['password'])
if not wifi.connect():
wifi.reconnect()
if not wifi.is_connected():
raise RuntimeError('Wi-Fi 接続失敗')
print("Wi-Fi 接続完了")
try:
# 各都市の天気情報を取得してメール本文に追加
all_messages = ""
for city_name, city_id in CITY_IDS.items():
weather_json = fetch_weather(city_id)
msg = build_weather_message(city_name, weather_json)
all_messages += msg
# 現在時刻を取得（JST）
now = time.localtime(time.time() + TIMEZONE_OFFSET)
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(all_messages, date_str)
finally:
sta_if = network.WLAN(network.STA_IF)
sta_if.active(False)
print("Wi-Fi 切断")
# ======= 状態管理用変数 =======
sent_today = False
last_sent_day = -1
# ======= テストボタンの割り込み処理 =======
def handle_test_button(pin):
print("🧪 テストボタンが押されました。メールを送信します。")
try:
send_weather_email()
except Exception as e:
print("❌ テスト送信失敗:", e)
time.sleep(0.1)
test_button.irq(trigger=Pin.IRQ_FALLING, handler=handle_test_button)
# ======= メインループ =======
while True:
now = time.localtime(time.time() + TIMEZONE_OFFSET)
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(60)
つぶやき
コメント