ラズパイとTypetalkで今年の夏こそ植物の観察日記をやりとげる

by Kohei Ito

2022/07/26

Typetalk / TechBlog

こんにちは、ヌーラボTypetalkチームの伊藤です。

本記事はブログリレーの7/26分になります。

植物の観察日記をつけたことがありますか？

皆さんは、植物の観察日記をつけたことがありますか？

夏休みになると宿題として課される、アレです。

私はいままで一度もこの課題をやり遂げられたことがありません。私の夏休みは常に「植物の観察日記をやり遂げられなかった」という悔しさで幕を閉じます。

20年以上経った現在でも、夏がくると思い出します。このままでは夏を心から楽しむことができません。そこで、この夏をかけて自身のトラウマを克服すべく、植物の観察日記にリベンジすることにしました。

なぜやりとげられないのか？

私なりにやり遂げられない原因を考えました。

毎日の水やりが面倒くなってしまう
植物の絵を描くのがストレス
途中で飽きてしまう

これらは、私自身の問題に繋がります。

“毎日コツコツ” “決まった時間にする” ができない
極めて画力が低い
植物から何も反応がなくやっていると虚しくなる

どうすればやりとげられるのか？

これならばやり遂げられる！という理想の姿を考えました。

土が乾いたなら植物のほうで自ら水をひきこむ
絵を描くのではなく、写真をとる
犬が撫でられたら喜ぶように、植物もなにかリアクションする

最後に、理想を実現するための方法を考えました。

センサーを用いて土の中の水分量を測定し、必要なときに自動で水を汲み上げる
定期的に定点カメラで写真を取り保存する
チャットBOT化し植物を喋らせる

要は、出来るだけ自動化です。

行き着いた先はありがちなハックではありますが、なんとか観察日記をやり遂げて過去の自分を越えたいと思います。

出来上がったもの

よくわからない装置ができました。

妙に縦に長いですね。長く伸びた先にWebカメラがついています。見た目は奇妙ですが、ペットボトルへの水の補充以外はほとんどこの装置で自動化できます。

この装置の機能を説明します。

機能1 : 水やりの定期実行→ 結果に従いTypetalkへ通知

ペットボトルに入った水がポンプによって汲み上がり、植物へと注がれています。水やりの開始は、センサから土の中の水分量を測定 → 水分量が一定に達していなければ一定時間水やりをする、という処理をRaspberry Piにて実行しています。

水やりが成功するとTypetalkへ下記のメッセージが送られます。

感謝の言葉は大事ですね。やる気が爆上がりです。

水やりが常に成功するとは限りません。最も起こり得るのはペットボトルの水が切れていた場合です。このときは下記のメッセージが送られます。

Typetalkでは「@（アット）」を用いた通知機能によって、必ず伝えたい情報を連絡できます。この機能を使い水切れを通知することで、すぐに気付き忘れずにペットボトルへ水の補充を行うことができます。

機能2 : Typetalkへ観察記録を送信

1日に一度、上記の観察記録を写真付きでTypetalkへ送信します。

写真はWebカメラでとったものです。画角を確保するために高さが必要でした。Raspberry PiからWebカメラを起動して撮影しています。

天気や気温の情報はOpen-MeteoのWeather APIから取得しています。緯度と経度から現在地の天気と外気温を取得できるので、その値に従ってそれっぽく絵文字に変換しています。植物は室内に置いている(Raspberry Piがあるので外におけない)ので、観察の記録としては正確とはいえませんが、観察日記っぽい雰囲気がでているのでよしとします。

また、Typetalkには、メッセージに #（ハッシュタグ）を含めることで会話のまとめを作成する機能があります。観察記録にハッシュタグを付与することで、閲覧時に記録だけをまとめて閲覧できるようにしました。

デフォルトだとトピック内のメッセージは下記のように時系列で並んでおり、観察記録以外にも色んなメッセージが飛び交い、賑わっています。

画面右上のアイコンからまとめを選択すると、

下記のように観察記録のみを時系列で閲覧できます。

なかなかいいですね。これは観察日記といっても差し支えないです。

全体構成

本システムの構成は下記のようになっています。

装置

Raspberry Piにて

水分測定センサ付き給水ポンプユニット
Webカメラ

を接続し、Pythonのプログラムで制御しています。各プログラムはcronで定期実行しています。

取得したデータは中継サーバのAPIへ送ります。また、各種設定はDBに保存されているので、こちらもAPI経由で取得しています。

中継サーバー以降

Raspberry Piから直接外部サービスへデータを送ってもよかったのですが、

セキュリティ観点からRaspberry PiにグローバルIPを与えたくなかった
あとあと分析してみたくなったときのためにDBを配置したかった

という理由で、中継サーバーを設置する構成としました。

Typetalkへのメッセージ送信は厳密にリアルタイムである必要もなかったので、いったん中継サーバーにメッセージをプールし、定期的にバッチを叩いて送信する形としました。

装置の作り方

せっかくなので、装置の作り方をご紹介します。

用意するもの

水分測定給水ユニット + 4ピン変換コネクタ

給水ポンプと水分測定センサが一体となったIoT製品です。これはM5スタック用の製品なのでそのままでは接続端子が合わずRaspberry Piと繋げません。そこで、4ピン変換コネクタを用います。これにより、ジャンパーピンを挿せるようになり、Raspberry PiのGPIOピンと繋ぐことができます。

Raspberry Pi model 3B + Wifiドングル

今回、各センサーの制御はRaspberry Piで行います。Raspberry Pi model 3B を使いますが、他のモデルでも代用可能です。Wifiモジュールが搭載されているモデルであれば、Wifiドングルは必要ありません。Raspberry Pi Zero Wなどを使用するとさらに省スペース＆省電力で運用でき良いと思います。

A/Dコンバータ

こちらはアナログ値をデジタル値へと置き換えるための変換器です。今回、センサーから送られる水分測定データはアナログ値として送られるため、こちらのA/Dコンバータを用いてデジタル値へ変換しRaspberry Piで受け取れるようにします。I2C通信を用いることで、プログラムから値を取得できます。

ジャンパーピン

上の写真には大量に写っていますが、こんなに必要なわけではありません。

それぞれ以下の本数を使います。

オス/オス : 1本
オス/メス：5本
メス/メス：3本

ブレッドボード

省スペース化のため、今回は小さめのものを使用します。

準備

A/Dコンバータの基盤とピンが分離しているので、半田づけをしてやる必要があります。

あまり上手ではないですが、これでOKです。先にブレッドボードにピンを刺したうえで基盤を重ねると固定できてやりやすいです。半田付けを行う際はしっかりと換気をした部屋で実施しましょう。

配線

あとは各機器を配線していきます。

実物の写真だとわけがわからないので図を用意しました。ブレッドボードを経由して、下記のようにジャンパーピンを配線しています。

今回、プログラム側で扱いたいデータは

土の中の水分量の値
給水ポンプのON/OFF

の２つになります。

1を取得するための配線について補足します。

水分測定センサから取れたアナログの値は4ピン変換コネクタのOUTにて出力されます。こちらをデジタル値に変換するために、A/DコンバータのA0(アナログ入力)に繋ぎます。最終的にプログラム側でI2C通信でデータを受け取るため、A/DコンバータのSCLに対応するRasbery PiのGPIO3ピンと接続しています。

2については、1に比べるとシンプルです。

Raspberry PiからポンプON/OFF信号をGPIO27にて出力する想定で、4ピン変換コネクタのINと繋いでいます。これにより、Raspberry Pi側のプログラムから給水ポンプをON/OFFできます。

これらは後述のRaspberry Piのコードと見比べて頂ければと思います。

プログラム

Raspberry Pi側の処理について説明します。

準備

本記事では、Raspberry piそのもののセットアップ（osインストール/ネットワーク接続など）は省略します。

プログラムを実行するために必要なモジュールをインストールするため、Raspberry Piのターミナルにて下記を実行してください。

$ sudo pip3 install adafruit-circuitpython-ads1x15

また、Raspberry Piの設定でI2C通信をONにする必要があります。

メニューから、[設定]＞[Raspberry Piの設定]を開き、 [インターフェイス]タブにて移動してI2CをONにします。

自動水やり機能

import busio
import board
import RPi.GPIO as GPIO
import datetime
import time
import adafruit_ads1x15.ads1015 as ADS
from adafruit_ads1x15.analog_in import AnalogIn
import pprint
import json
import requests
from dotenv import load_dotenv
import os
 
 
load_dotenv()
 
HOST_NAME = os.getenv('API_SERVER_HOST_NAME')
PORT = os.getenv('API_PORT')
PLANT_ID = os.getenv('PLANT_ID')
GPIO_PUMP_OUT_PIN = 27
 
 
def main():
 
   # APIから土の水分の必要量と十分量を取得
   need_pump,complete_pump = get_pump_setting()
 
   # ラズパイの初期化
   moist = setup_pi()
 
   # 土の水分が必要量以上であればスキップ
   if need_pump < moist.voltage:
       print('no need to pump now')
       return
 
   # 給水のタイムアウト値を10秒とする
   base_time = time.time()
   timeout = base_time + 10
 
   # 開始時点の水分量を保持
   start_moist = moist.voltage
   is_bottle_empty = False
 
   # 給水開始
   pump_on()
 
   # 十分な量に達するまで実行
   while complete_pump <= moist.voltage:
 
       # 1秒ごとに値を計測（デバッグ確認用）
       if time.time() - base_time >= 1:
           recordtime = '{0:%Y-%m-%d %H:%M:%S.%f}'.format(datetime.datetime.now())
           data = [recordtime, moist.voltage]
           print(data)
           base_time = time.time()
 
       # 最大時間を超過し
       if time.time() > timeout:
           # スタート地点の水分量と差が少なかったら
           if abs(start_moist - moist.voltage) < 0.03:
               # 水切れと判断
               is_bottle_empty = True
               print("water bottle is empty")
           break
 
   # 給水終了
   pump_off()
 
   # 水切れ？
   if is_bottle_empty:
       message = " @kohei_ito_smn おみずないよー :angry: :anger: "
   else:
       message = "おみずごちそうさま :blush: "
       # 給水ログの登録
       send_pump_log()
 
   # 通知登録
   send_notification(message)
 
# ラズベリーパイの立ち上げ
def setup_pi():
   GPIO.setup(GPIO_PUMP_OUT_PIN, GPIO.OUT)
   i2c = busio.I2C(board.SCL, board.SDA)
   ads = ADS.ADS1015(i2c)
   moist = AnalogIn(ads, ADS.P0)
   return moist
 
# APIから水やり開始/完了の閾値を取
def get_pump_setting():
   res = requests.get(f'http://{HOST_NAME}:{PORT}/api/pump_settings/?plant_id={PLANT_ID}')
   if res.status_code != 200: raise Exception('ERROR')
   json = res.json()
   need_pump = json['need_pump']
   complete_pump = json['complete_pump']
   return need_pump, complete_pump
 
# 水やりポンプON
def pump_on():
   GPIO.output(GPIO_PUMP_OUT_PIN, True)
   print("start pumping")
 
# 水やりポンプOFF
def pump_off():
   GPIO.output(GPIO_PUMP_OUT_PIN, False)
   print("stop pumping")
 
 
def send_notification(message: str):
   res = requests.post(
       f'http://{HOST_NAME}:{PORT}/api/notifications',
           json.dumps({
           'new_notification':{
               'plant_id': PLANT_ID,
               'service_type': "TYPETALK",
               "notified_to_service": "false",
               "snapshot_id": None,
               "message": message
               }
           })
   )
   if res.status_code != 201: 
       pprint(res)
       raise Exception('send_notification failed')
 
def send_pump_log():
   res = requests.post(
       f'http://{HOST_NAME}:{PORT}/api/pump_logs',
           json.dumps({
           'new_pump_log':{
               'plant_id': PLANT_ID
               }
           })
   )
   if res.status_code != 201: 
       pprint(res)
       raise Exception('send_pump_log failed')
 
if __name__ == '__main__':
   try:
       main()
   except Exception as e:
       pprint('catch exception', e)
   finally:
       GPIO.cleanup()

メイン処理

センサーから取得した土中の水分量が閾値(need_pump)を下回っていた場合は水やりが必要と判断し給水を開始します。

給水開始後は、

土中の水分量が十分な量(complete_pump)に達していた場合
水やりが最大時間(TIMEOUT)に達していた場合

のいずれかに当てはまると終了します。

TIMEOUT到達で終了した上に、最終的な土中の水分量が開始時点とほとんど変わっていなかった場合は水が切れていたと判定しています。

写真撮影機能

from shutil import ExecError
import cv2
import numpy as np
from datetime import datetime
import pprint
import json
import requests
from dotenv import load_dotenv
import os
from logging import getLogger
 
logger = getLogger(__name__)
load_dotenv()
 
# 環境変数の取得
DEV_ID = int(os.getenv('DEV_ID_WEB_CAM'))
HOST_NAME = os.getenv('API_SERVER_HOST_NAME')
PORT = os.getenv('API_PORT')
PLANT_ID = os.getenv('PLANT_ID')
UPLOAD_IMG_TMP_DIR = os.getenv('UPLOAD_IMG_TMP_DIR')
 
# パラメータ
IMG_WIDTH = 800
IMG_HEIGHT = 600
 
def main():
 
   try:
 
       # Webカメラののデバイス番号を指定(/dev/video*)
       cap = cv2.VideoCapture(DEV_ID)
 
       # 解像度の指定
       cap.set(cv2.CAP_PROP_FRAME_WIDTH, IMG_WIDTH)
       cap.set(cv2.CAP_PROP_FRAME_HEIGHT, IMG_HEIGHT)
 
      
       # カメラのフレームの読み込み
       read_complete, frame = cap.read()
       if not read_complete: raise Exception('frame read failed')
 
       # 明るさ補正
       frame_adjusted = adjust_brightness(frame, alpha=0.8, beta=70.0)
 
       # 画像出力
       img_file_name = f'{datetime.now().strftime("%Y%m%d_%H%M%S")}.jpg'
       img_file_path = f'{UPLOAD_IMG_TMP_DIR}/{img_file_name}'
       write_complete = cv2.imwrite(img_file_path, frame_adjusted)
       if not write_complete: raise Exception('image write failed')
 
       # 画像 + メタデータをAPIへ送信
       upload_image(img_file_path)
       send_image_meta_info(img_file_name)
 
   except Exception as e:
       pprint('catch exception:', e)
   finally:
       cap.release()
       cv2.destroyAllWindows()
   return
 
def adjust_brightness(img, alpha=1.0, beta=0.0):
   dst = alpha * img + beta
   return np.clip(dst, 0, 255).astype(np.uint8)
 
def upload_image(img_file_path: str):
   file = {'upload_file': open(img_file_path, 'rb')}
   res = requests.post(
       f'http://{HOST_NAME}:{PORT}/api/snapshots/upload/image',
       files=file)
   if res.status_code != 200:
       pprint(res)
       raise Exception('upload image failed:')
 
def send_image_meta_info(img_file_name: str):
   res = requests.post(
       f'http://{HOST_NAME}:{PORT}/api/snapshots/upload/meta',
       json.dumps({
           'new_snapshot': {
               'plant_id': PLANT_ID,
               'image_file': img_file_name
           }
       }),
       headers={'Content-Type': 'application/json'})
   if res.status_code != 201:
       pprint(res)
       raise Exception('send image meta info failed')
   res_json = res.json()
   image_id = res_json["id"]
   return image_id
 
if __name__ == "__main__":
   main()

複雑なことはしていません。

OpenCVを用いて、カメラデバイスへアクセスしフレームを取得→画像を出力しています。そのままだと写真が暗かったので、申し訳程度に光量の補正を行っています。

「水やり三年」は伊達じゃない

水やり三年という言葉があります。

これは水やりをマスターするまでに3年は必要だという意味です。自動水やり機能を実装するうえでいくつか試行錯誤したのですが、その中でこの言葉の意味を痛感しました。

私は水やりのタイミングについて、人間が喉がかわいたら水を飲むように、植物も必要なときに水分補給するのがベストではと考えていました。その考えに基づき、水分量が基準を下回ったら水やり開始→一定の水分量まで給水したら停止、それを繰り返すというプログラムを実装し常時起動させていました。

下記のイメージで、上限と下限を設けてその中で土の水分量を推移させる形です。

しかし、テスト運用してみると単純にはいかず、想定外が2つありました。

想定外1 : 植物は夜に渇きを求める

土に含まれる水分量しか見ていないので、土が渇けばいつでも水やりを実行します。

この結果、夜中にも水やりが実行されていました。植物を避けて生きてきた私は知らなかったのですが、妻から「夜に水をやるのはよくないよ」と注意されてだめなことに気付きました。

植物は夜に根が成長するため、夜はあえて水を与えず土が乾いている時間を作るほうがいいとのことでした。

この想定外については、夜中の時間帯には処理を実行させないようにすることで対処できそうでしたが、次の想定外に直面し挫折しました。

想定外2 : 水分測定値にバラツキがある

しばらくテスト運用した中で、水分測定センサーの値が安定しないことがあると気付きました。

瞬間的に突然閾値を下回ることがまれに発生し、意図しない給水が実行されてしまいました。このままでは水のやりすぎによる根腐れの懸念があります。

原因について調査してみたところ、そもそも閾値を設定し土の水分の測定値をもとに水やりのトリガーをかける方式自体に不確定要素が多すぎて厳しいと考え始めました。

【調査結果】

センサーから外れた値が取れてしまうのはよくある（仕方ない）
バラツキを許容した上で、瞬時値で判定するべきでなく、傾向をもとに判断すべき
センサーを土に挿すときの深さや角度で値の取れ方が微妙に変わる→いちいち閾値の微修正が必要
植物の種類や育てる時期、室温によっても閾値の変更が必要

このままでは常に閾値を変更することが必要になりそうでした。

理想をいえば、植物の種類や時期、室温に従い閾値も変動するようにできればいいのでしょうが、それを実行するには圧倒的にデータが足りず、実装しているうちに夏が終わります。（否、この夏をかけても無理です。）

方針変更

私のような植物を一度も育てたことがない人間が、植物にとって何がベストかを考えること自体に無理がありました。方針を変更し、私が真面目に育てるとしたらこうする、を目指すことにしました。

私が真面目に取り組むとしたら、

毎日9:00に起きて水をコップ一杯分注ぎます。

これでいいのです。うまくいかなければまた考えましょう。

下記のイメージになります。

9:00に水やり処理を実行し、10秒間だけ水やりをするという極めてシンプルな実装になりました。

一応、水分測定のデータも引き続き使っています。ただし、いままでの方式のように厳密に閾値を設定し判定するのではなく、ゆるめに設定して異常があったときのセーフティネットくらいの使い方にしました。

水やり三年という言葉の重みを痛感しました。私には三年かけてもできる気がしません。

結局、植物の観察日記をやりとげられたのか？

開発したシステムを用いて、ひまわりを種から育て始めました。

しかし、記事公開の7/26時点では完遂できていません。

絶賛、挑戦続行中の状況になります。

順調にいけば、8月末に開花する予定です。

丁度、夏休みが終わる頃ですね。今年の夏は気持ちよく終えられるでしょうか。自分を変える夏にしたいです。

おまけ

Twitter投稿にも対応しました。

メッセージ通知はTypetalk上でのみですが、日々の観察記録についてはTwitterにもあがっていきますので、もし興味を持たれた方がいれば応援を頂けると励みになります。

ハッシュタグ

#ひまわりの小夏ちゃんの観察日記

以上、最後まで読んでいただきありがとうございました。