植木への水やり自体はスマートプラグで別途タイマー制御して、貯水タンクからポンプで吸い上げるようにしています。しかし、このタンクが見えない場所にあるので、タンク自体の水が枯渇しそうな場合に、タンクへの水の補充を LINE 通知する機能が欲しくなりました。
ちょうど良い製品が無かったので Pico 2 W で作成することにしました。はじめての Pico なので sozorablog さんにご教示いただきながら、試行錯誤で以下の配線にしました。
ご参考として main.py の Python コードも載せておきます。5 秒おきにセンサーの値を読み出して、変化があったら LINE Messaging API で通知しているだけです。2025/3/31 に LINE Notify API がサービス終了してしまったので Messaging API を使用します。
同じ通知でも対処せずに放置したら再送したい (水不足のメッセージが来ないから大丈夫と思っていたら監視システムが止まっていたとかは困るな) とか考えて、少し余計な制御を入れています。
消費電力の節約のため、メッセージ送信する時だけ WiFi に接続して、送信後は切断します。Deep Sleep が Pico 2 では、まだ使用できないようなので Deep Sleep 対応はファームウェアの更新を待ちたいと思います。
WiFi アクセスポイントが停止しているような場合は例外がスローされるので再試行されます。しかし、動作テスト中に ENOMEM (メモリ不足) の OS エラーが出ることがありました。このような場合は、再試行では回復できないケースが多いです。このため、例外を捕捉した回数を累積して、上限に到達したら Pico 自体を強制リセットするようにしています。
また、短時間で再試行が連続して Pico や WiFi アクセスポイントに過度な負担を掛けないように、例外発生時の待機時間は Exponential Backoff で 5 秒, 10 秒, 20 秒 … (最大で約 21 分) のように 2 倍ずつ延ばすようにしています。
尚、センサーの設置条件に依ると思いますが、水面がセンサーの高さとスレスレの場合や、水面が揺れると、短時間で連続して通知が来てしまうので、HISTORY_LENGTH で指定した回数だけ同じ結果が連続で得られた場合だけ、安定したセンサーの結果として採用するようにしています。
main.py import machine import network import ujson import urequests import utime import config # Seconds in a day SECONDS_IN_A_DAY = 60 * 60 * 24 # Detect interval (sec) INTERVAL_SEC = 5 # Number of times to check if the same result is obtained consecutively HISTORY_LENGTH = 5 # Number of times an exception occurs before reset MAX_EXCEPTIONS = 9 # Embeded LED led = machine . Pin ( " LED " , machine . Pin . OUT ) # Wireless LAN wlan = network . WLAN ( network . STA_IF ) # Liquid level sensor sensor = machine . Pin ( 28 , machine . Pin . IN , machine . Pin . PULL_UP ) def console_print ( text ): """ Print to console. Args: text (str): message. """ print ( f " { utime . time () } : { text } " ) def connect_wifi (): """ Connect to wifi network. Raises: RuntimeError: Network connect failure. """ if wlan . status () == 3 : return console_print ( " Connecting ... " ) led . on () wlan . active ( True ) wlan . connect ( config . WLAN_SSID , config . WLAN_PASSWORD ) count = 30 while 0 < count : led . toggle () if wlan . status () < 0 or 3 <= wlan . status (): led . off () wlan_status = wlan . ifconfig () console_print ( f "Connected { wlan_status [ 0 ] } ." ) return count -= 1 utime . sleep ( 0.5 ) led . off () raise RuntimeError ( " Connect failed. " ) def disconnect_wifi (): """ Disconnect from wifi network. """ wlan . disconnect () wlan . active ( False ) console_print ( " Disconnected. " ) def send_message ( message ): """ Send message using LINE Messaging API Args: message (str): message """ connect_wifi () console_print ( message ) response = urequests . post ( config . CHANNEL_ENDPOINT_URL , headers = { " Content-Type " : " application/json " , " Authorization " : f "Bearer { config . CHANNEL_ACCESS_TOKEN } " , }, data = ujson . dumps ( { " to " : config . CHANNEL_USER_ID , " messages " : [{ " type " : " text " , " text " : message }], } ). encode ( " utf-8 " ), ) response . close () disconnect_wifi def get_liquid_level (): """ Detect liquid level Returns: bool: Liquid detect status """ return sensor . value () == 0 # Initialize led . off () results = [] last_message = ( None , 0 ) exceptions = 0 # Loop while True : try : # Get current result current_result = get_liquid_level () led . value ( current_result ) # Update results results . append ( current_result ) results = results [ - HISTORY_LENGTH :] # Check continuity if len ( set ( results )) == 1 : # Select message if current_result : message = ( " タンクは水が入っています " , utime . time () + 7 * SECONDS_IN_A_DAY , ) else : message = ( " タンクの水が不足しています!! " , utime . time () + 1 * SECONDS_IN_A_DAY , ) # Send message if last_message [ 0 ] != message [ 0 ] or last_message [ 1 ] < utime . time (): send_message ( message [ 0 ]) last_message = message # Continue exceptions = 0 utime . sleep ( INTERVAL_SEC ) except KeyboardInterrupt : # Break break except Exception as e : # Error console_print ( e ) # Check force reset exceptions += 1 if MAX_EXCEPTIONS < exceptions : console_print ( " Reset. " ) machine . reset () # Continue retry_factor = 2 ** min ( exceptions - 1 , 8 ) retry_duration = INTERVAL_SEC * retry_factor console_print ( f "Continue after { retry_duration } seconds." ) utime . sleep ( retry_duration ) # Break console_print ( " Break. " ) 別途 config.py というファイルに、以下のように環境固有の情報を定義して main.py と並べて Pico に保存します。
config.py WLAN_SSID = " WiFi の SSID " WLAN_PASSWORD = " WiFi のパスワード " CHANNEL_ENDPOINT_URL = " https://api.line.me/v2/bot/message/push " CHANNEL_ACCESS_TOKEN = " 自分のチャネルアクセストークン " CHANNEL_USER_ID = " 自分の LINE のユーザ ID " Pico 面白い。
