今回は、前回読み取れるようになった温度と湿度を、1 時間に 1 回 Slack へ通知するところまで進めます。
前回までは main.py の中に、センサー読み取り処理がまとまっていました。今回はそこに Slack 通知も入ってくるので、ファイルを分けました。
今回の構成
今回の Python ファイルは、以下の 3 つに分けました。
src/ main.py # 全体の流れ sensor.py # ENV III / SHT30 から温度・湿度を読む slack_notify.py # Wi-Fi 接続と Slack 通知
さらに、Wi-Fi パスワードや Slack の Webhook URL はコードに直接書かず、config.py に分けます。
src/ config.py # 自分の環境用の秘密情報
config.py には、以下のような値を書きます。
WIFI_SSID = "your-wifi-ssid" WIFI_PASSWORD = "your-wifi-password" SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/XXX/YYY/ZZZ"
Wi-Fi パスワードや Slack の Webhook URL は公開してはいけない情報なので、.gitignore に入れて Git 管理しないようにしています。
main.py
main.py は、全体の流れだけを担当します。
from machine import Pin import time from sensor import read_sht30, scan_i2c from slack_notify import build_slack_message, connect_wifi, post_slack_message BTN_PIN = 39 MEASURE_INTERVAL_MS = 60 * 60 * 1000 btn = Pin(BTN_PIN, Pin.IN, Pin.PULL_UP) def main(): print("atom lite ready. reading ENV III and posting to Slack.") connect_wifi() scan_i2c() while True: try: temperature, humidity = read_sht30() print("temperature: {:.2f} C, humidity: {:.2f} %".format(temperature, humidity)) except OSError as error: print("sensor error:", error) else: try: post_slack_message(build_slack_message(temperature, humidity)) print("Slack message sent.") except Exception as error: print("notification error:", error) if btn.value() == 0: scan_i2c() time.sleep_ms(300) time.sleep_ms(MEASURE_INTERVAL_MS) if __name__ == "__main__": main()
ファイルを分ける理由
最初は全部 main.py に書いても動きます。
ただ、処理が増えてくると、1 つのファイルがどんどん長くなります。
今回のプログラムには、大きく 3 種類の処理があります。
- センサーから温度と湿度を読む
- Wi-Fi に接続して Slack に送る
- 1 時間ごとに実行する流れを管理する
これらを分けると、main.py がかなり読みやすくなります。
main.py だけを見ると、
connect_wifi() scan_i2c() temperature, humidity = read_sht30() post_slack_message(build_slack_message(temperature, humidity))
という感じで、何をしているかが上から読めます。
中身の細かい実装は、sensor.py と slack_notify.py に隠れています。
これは Python に限らず、プログラムを大きくしていくときの大事な考え方です。
import で別ファイルの関数を使う
main.py には、次の行があります。
from sensor import read_sht30, scan_i2c from slack_notify import build_slack_message, connect_wifi, post_slack_message
これは、別ファイルで定義した関数を読み込んでいます。
sensor.py にある、
def read_sht30(): ...
を、main.py から使えるようにしているわけです。
Java の import と似ていますが、Python ではファイル名がそのままモジュール名になります。
sensor.py
なら、
from sensor import read_sht30
のように読み込めます。
1 時間に 1 回の設定
MEASURE_INTERVAL_MS = 60 * 60 * 1000
これは、計測間隔をミリ秒で表しています。
MicroPython の time.sleep_ms() は、待ち時間をミリ秒で指定します。
time.sleep_ms(MEASURE_INTERVAL_MS)
そのため、1 時間をミリ秒に直しています。
60 秒 * 60 分 * 1000 ミリ秒
つまり、
60 * 60 * 1000
です。
計算結果は、
3600000
です。
最初から、
MEASURE_INTERVAL_MS = 3600000
と書いても同じです。
ただし、
60 * 60 * 1000
と書いた方が、「1 時間なんだな」と読み取りやすくなります。
try / except / else
今回の main.py では、次のように書いています。
try: temperature, humidity = read_sht30() print("temperature: {:.2f} C, humidity: {:.2f} %".format(temperature, humidity)) except OSError as error: print("sensor error:", error) else: try: post_slack_message(build_slack_message(temperature, humidity)) print("Slack message sent.") except Exception as error: print("notification error:", error)
try は「失敗するかもしれない処理を試す」場所です。
ここでは、まずセンサーから温度と湿度を読みます。
temperature, humidity = read_sht30()
もしここで OSError が起きたら、except に進みます。
except OSError as error: print("sensor error:", error)
OSError は、通信、ファイル、デバイス操作などで失敗したときによく使われる例外です。
SHT30 から読めなかった場合や、I2C 通信に失敗した場合は、ここで「センサーエラー」として表示します。
else は、try の中が成功したときだけ実行されます。
つまり、
else:
post_slack_message(...)
は、
センサー読み取りに成功したときだけ Slack に通知する
という意味です。
ここで else を使うと、「センサー読み取り」と「Slack 通知」の失敗を分けて扱いやすくなります。
センサーが失敗したときは sensor error。
Slack 通知が失敗したときは notification error。
どこで失敗したのかがログから分かりやすくなります。
sensor.py
sensor.py は、ENV III の SHT30 から温度と湿度を読む処理を担当します。
from machine import I2C, Pin import time I2C_SCL_PIN = 32 I2C_SDA_PIN = 26 SHT30_ADDR = 0x44 i2c = I2C(0, scl=Pin(I2C_SCL_PIN), sda=Pin(I2C_SDA_PIN), freq=100000)
ATOM Lite の Grove 端子に ENV III を接続する場合、MicroPython では以下の指定にします。
I2C_SCL_PIN = 32 I2C_SDA_PIN = 26
SHT30 の I2C アドレスは 0x44 です。
SHT30_ADDR = 0x44
このファイルの中では、前回作った crc8()、read_sht30()、scan_i2c() をそのまま使っています。
def read_sht30(): i2c.writeto(SHT30_ADDR, b"\x24\x00") time.sleep_ms(20) data = i2c.readfrom(SHT30_ADDR, 6) ...
read_sht30() は温度と湿度を返します。
return temperature, humidity
そのため、main.py 側では以下のように受け取れます。
temperature, humidity = read_sht30()
Python では、関数が複数の値を返しているように書けます。
実際にはタプルという形で返っています。
return temperature, humidity
は、イメージとしては以下に近いです。
return (temperature, humidity)
受け取る側では、
temperature, humidity = read_sht30()
と書くことで、2 つの変数に分けて受け取れます。
これは Python でよく使う書き方です。
slack_notify.py
slack_notify.py は、Wi-Fi 接続と Slack 通知を担当します。
import network import socket import ssl import time import ujson
それぞれの役割は、ざっくり以下です。
network: Wi-Fi 接続socket: ネットワーク通信ssl: HTTPS 通信time: 待ち時間ujson: JSON 作成
通常の Python では json を使うことが多いですが、MicroPython では軽量版の ujson を使います。
config.py を読み込む
try: from config import SLACK_WEBHOOK_URL, WIFI_PASSWORD, WIFI_SSID except ImportError as error: raise RuntimeError("Create config.py from config.example.py before running.") from error
ここでは、config.py から Wi-Fi と Slack の設定を読み込んでいます。
config.py がないと、ImportError が発生します。
今回の config.py は、なくてもよいファイルではありません。
Slack 通知をするには必須です。
そのため、読み込めなかったらその場で RuntimeError にして止めています。
raise RuntimeError("Create config.py from config.example.py before running.") from error
RuntimeError は、Python に最初から用意されている例外です。
「プログラム実行中に、この状態では続けられない問題が起きた」くらいの意味で使えます。
最後の from error は、元の ImportError も原因として残す書き方です。
これにより、エラーを見たときに、
config.py が読めなかった その結果 RuntimeError にした
というつながりが追いやすくなります。
Wi-Fi に接続する
def connect_wifi(): if not WIFI_SSID or not WIFI_PASSWORD: raise RuntimeError("Wi-Fi settings are missing. Create src/config.py.") wlan = network.WLAN(network.STA_IF) wlan.active(True)
network.WLAN(network.STA_IF) は、Wi-Fi を子機モードで使う設定です。
家庭の Wi-Fi ルーターに接続する側なので、STA_IF を使います。
wlan.active(True) で Wi-Fi 機能を有効にします。
if not wlan.isconnected(): print("connecting to Wi-Fi...") wlan.connect(WIFI_SSID, WIFI_PASSWORD)
まだ接続されていなければ、config.py の SSID とパスワードを使って接続します。
for _ in range(30): if wlan.isconnected(): break time.sleep(1)
ここでは、最大 30 秒待っています。
for _ in range(30): は「30 回繰り返す」という意味です。
_ は、繰り返し回数そのものを使わないときによく使われる変数名です。
1 回ごとに 1 秒待つので、最大 30 秒です。
if not wlan.isconnected(): raise OSError("Wi-Fi connection failed")
30 秒待ってもつながらなかったら、OSError を発生させます。
Slack の Webhook URL を分解する
def parse_https_url(url): prefix = "https://" if not url.startswith(prefix): raise ValueError("Slack webhook URL must start with https://") rest = url[len(prefix):] slash_index = rest.find("/") if slash_index == -1: raise ValueError("Slack webhook URL is invalid") host = rest[:slash_index] path = rest[slash_index:] return host, path
Slack の Webhook URL は、だいたい以下のような形です。
https://hooks.slack.com/services/XXX/YYY/ZZZ
HTTP 通信を自分で組み立てるために、これを host と path に分けます。
host = hooks.slack.com path = /services/XXX/YYY/ZZZ
最初に、
if not url.startswith(prefix):
で、URL が https:// から始まっているか確認しています。
違っていたら、
raise ValueError("Slack webhook URL must start with https://")
でエラーにします。
ValueError は、値の内容がおかしいときによく使われる例外です。
次に、
rest = url[len(prefix):]
で、先頭の https:// を取り除いています。
len(prefix) は、"https://" の長さです。
url[len(prefix):] はスライスで、「https:// の次から最後まで」を取り出しています。
その後、
slash_index = rest.find("/")
で、最初の / の位置を探します。
この / より前が host、この / 以降が path です。
host = rest[:slash_index] path = rest[slash_index:]
ここでもスライスを使っています。
Slack に送るメッセージを作る
def build_slack_message(temperature, humidity): return "🪲 ヘラクレスに最適な環境です\n温度: {:.2f} C\n湿度: {:.2f} %".format( temperature, humidity, )
ここでは、Slack に送る文章を作っています。
\n は改行です。
そのため、Slack には以下のように表示されます。
🪲 ヘラクレスに最適な環境です 温度: 28.58 C 湿度: 50.52 %
{:.2f} は、小数点以下 2 桁で表示するための書き方です。
たとえば、
temperature = 28.581234
なら、
28.58
になります。
Slack に HTTP POST する
def post_slack_message(text): if not SLACK_WEBHOOK_URL: raise RuntimeError("Slack webhook URL is missing. Create src/config.py.") host, path = parse_https_url(SLACK_WEBHOOK_URL) body = ujson.dumps({"text": text}).encode("utf-8")
Slack の Incoming Webhook には、JSON 形式でメッセージを送ります。
Slack に送る本文は、以下のような形です。
{"text": "🪲 ヘラクレスに最適な環境です\n温度: 28.58 C\n湿度: 50.52 %"}
Python の辞書を JSON 文字列に変換しているのが、次の行です。
ujson.dumps({"text": text})
さらに、
.encode("utf-8")
で、文字列をバイト列に変換しています。
ネットワーク通信では、最終的に文字列ではなくバイト列を送る必要があります。
socket と ssl
addr = socket.getaddrinfo(host, 443)[0][-1] sock = socket.socket()
socket.getaddrinfo(host, 443) は、hooks.slack.com の接続先アドレスを調べています。
443 は HTTPS の標準ポート番号です。
sock.connect(addr) sock = ssl.wrap_socket(sock, server_hostname=host)
sock.connect(addr) で Slack のサーバーへ接続します。
その後、ssl.wrap_socket() で HTTPS 通信用に包みます。
Slack の Webhook URL は https:// なので、SSL/TLS が必要です。
HTTP リクエストを組み立てる
request = (
"POST {} HTTP/1.1\r\n"
"Host: {}\r\n"
"Content-Type: application/json\r\n"
"Content-Length: {}\r\n"
"Connection: close\r\n"
"\r\n"
).format(path, host, len(body)).encode("utf-8") + body
ここでは、Slack に送る HTTP リクエストを文字列で組み立てています。
POST は、サーバーにデータを送るための HTTP メソッドです。
Content-Type: application/json は、「送る本文は JSON です」という意味です。
Content-Length は、送る本文の長さです。
len(body)
で、JSON 本文のバイト数を数えています。
\r\n は HTTP の改行です。
普段の Python では改行に \n を使うことが多いですが、HTTP のヘッダーでは \r\n を使います。
最後の空行、
"\r\n"
によって、「ここまでがヘッダーで、ここから本文です」という区切りになります。
その後ろに、JSON の本文である body をつなげています。
+ body
finally で socket を閉じる
try: sock.connect(addr) sock = ssl.wrap_socket(sock, server_hostname=host) ... finally: sock.close()
finally は、成功しても失敗しても最後に必ず実行される場所です。
ネットワーク接続を開いたら、最後に閉じる必要があります。
途中でエラーが起きても、
sock.close()
が実行されるように、finally に書いています。
これは重要です。
接続を開きっぱなしにすると、メモリや通信リソースを無駄に使ってしまう可能性があります。
実機に転送するファイル
今回からファイルが増えたので、ATOM Lite へ転送するファイルも増えます。
python -m mpremote connect COM7 cp src/config.py :config.py python -m mpremote connect COM7 cp src/sensor.py :sensor.py python -m mpremote connect COM7 cp src/slack_notify.py :slack_notify.py python -m mpremote connect COM7 cp src/main.py :main.py
config.py には秘密情報が入っているため、Git 管理しないようにします。
動作確認として PC 上の src/main.py を一時実行する場合は、以下です。
python -m mpremote connect COM7 run src/main.py
ただし、main.py が sensor.py や slack_notify.py を import するため、これらのファイルは先に ATOM Lite 側へ転送しておく必要があります。
今回の到達点
今回できるようになったことは、以下です。
- センサー処理と Slack 通知処理を別ファイルに分けた
- Wi-Fi に接続できるようにした
config.pyで秘密情報を分けた- Slack Incoming Webhook に JSON を POST できるようにした
- 温度と湿度を 1 時間に 1 回 Slack へ送れるようにした
- センサーエラーと通知エラーを分けて表示できるようにした
次は、温度や湿度が適正範囲から外れたときだけ通知したり、メッセージの文面を状態に応じて変えたりできそうです。
run と reset の違い
mpremote には、run と reset というコマンドがあります。
最初はどちらも「実行する」ように見えますが、役割は違います。
python -m mpremote connect COM7 run src/main.py
これは、PC 上の src/main.py を一時的に ATOM Lite へ送って実行するコマンドです。
PC の src/main.py ↓ 一時的に送る ATOM Lite 上で実行
この場合、ATOM Lite 本体に main.py として保存されるわけではありません。
開発中に「今 PC で編集したコードをすぐ試す」ためのコマンドです。
一方で、
python -m mpremote connect COM7 reset
これは、ATOM Lite 本体を再起動するコマンドです。
ATOM Lite を再起動 ↓ 本体内に保存されている boot.py ↓ 本体内に保存されている main.py
reset は PC 上の src/main.py を実行するわけではありません。
ATOM Lite 本体に保存済みの main.py が起動します。
つまり、ざっくり言うと以下です。
run = PC 上のファイルを一時実行する reset = 本体を再起動して、本体内の main.py を自動実行する
開発中に試すだけなら、以下で十分です。
python -m mpremote connect COM7 run src/main.py
ただし今回のようにファイルを分けている場合、main.py が import する sensor.py や slack_notify.py は、先に ATOM Lite 側へ転送しておく必要があります。
python -m mpremote connect COM7 cp src/sensor.py :sensor.py python -m mpremote connect COM7 cp src/slack_notify.py :slack_notify.py python -m mpremote connect COM7 cp src/config.py :config.py
本体へ保存して、電源を入れたら自動で動く状態にするには、main.py も本体へ保存します。
python -m mpremote connect COM7 cp src/main.py :main.py
その後、保存済みのコードで起動するか確認するには、以下を実行します。
python -m mpremote connect COM7 reset
通電開始で main.py が起動する理由
ATOM Lite に電源を入れると、本体に保存済みの main.py が自動で起動します。
これは MicroPython の決まりです。
MicroPython では、ボードが起動すると、決まった名前のファイルを探して実行します。
起動時の流れは、ざっくり以下です。
ATOM Lite に電源が入る ↓ MicroPython ファームウェアが起動する ↓ boot.py があれば実行する ↓ main.py があれば実行する
つまり、main.py というファイル名には特別な意味があります。
PC から以下のように保存すると、
python -m mpremote connect COM7 cp src/main.py :main.py
ATOM Lite 本体の中には、次のファイルが置かれます。
/main.py
MicroPython は起動時にこの /main.py を見つけると、自動で実行します。
そのため、USB 電源を挿したときも、
電源 ON ↓ boot.py ↓ main.py ↓ Slack 通知プログラム開始
という流れになります。
boot.py は、起動直後の初期設定用です。
今回の boot.py では、デバッグ出力を抑えたり、ガベージコレクションを有効にしたりしています。
import gc import esp esp.osdebug(None) gc.enable()
一方、main.py は実際のアプリ本体を置く場所です。
今回のような常時動かしたいプログラムは、ATOM Lite 本体に main.py として保存しておくと、通電開始で自動起動します。
ATOM Lite はバッテリー内蔵ではないので、USB ケーブルを抜くと基本的に電源 OFF になります。
再度 USB 電源や USB 充電器につなぐと電源 ON になり、保存済みの main.py が自動で実行されます。
