2023年10月20日

Raspberry Pi + AWS + LINENotifyAPIでサーバーレスな植物の成長を毎日観察できるシステムを作ってみた

AWS, Raspberry Pi, IT, IT×農業

author | yuki.fushinuki

Raspberry Pi + AWS + LINENotifyAPIでサーバーレスな植物の成長を毎日観察できるシステムを作ってみた

はじめに

はじめまして、アジアクエストの伏貫と申します。普段はJava,PHP,JavaScriptなどを使いフルスタックに開発をやっています。さて今回、以前より注目されているIT×農業につながる試作を行ったので記事にしてみました。

ゴール

・Raspberry Pi + カメラモジュールで撮影した野菜の画像をタイムラプスにして毎日LINEで動画が送られてくるシステム
を作成することが今回の目的です。

必要なもの

・Raspberry Pi
カメラモジュールを付けられるものが必須です。
自分は「ResberryPI 3 A+」を使いました。

・マウス/キーボード
Raspberry Piの初期設定時、必要になります。
ノートPCなどしか持っておらず、外部キーボードやマウスが無い方などは注意です。

・カメラモジュール
Raspberry Piに付けられる簡易カメラです。
例えばこのようなものです。
https://www.amazon.co.jp/dp/B08Q34FKFY

・AWSアカウント
画像の保存、動画の作成、LINEへの送信などをAWS上で行います。
https://aws.amazon.com/jp/

・LINE Developerアカウント
LINEでBOTを作る際、登録が必要となります。
https://developers.line.biz/ja/

※今回、事前にLINEでBOTを作成しておく必要があります。
詳しくは以下の記事を参照してください。（私が執筆しました）
https://qiita.com/oimo23/items/11a0c38bbe1e297fb611#collision-line%E3%81%AEdevelopers%E8%A8%AD%E5%AE%9A%E7%B7%A8

・監視する植物
自分はきゅうりにしました。

この記事で扱わないもの

・Raspberry Piの基本的な使い方
・Raspberry Piの初期設定
・LINE BOTの作成の仕方
・AWSアカウントの作成方法・基本的な使い方

この辺りは長くなりそうなのと、ググると情報がたくさん出てくるので割愛します。

Raspberry Pi → S3への画像アップロード

Raspberry Piで、PythonとPiCameraというライブラリを用いて、写真を撮影するコードを書きます。

まず、必要なライブラリをインストールします。

$ sudo apt-get update
$ pip3 install boto3 picamera

Pythonでカメラモジュールを使って写真を撮るコードを書きます。具体的には以下のようなものになります。

import os
import time
from picamera import PiCamera
import boto3

def take_photo():
    if not os.path.exists('images'):
        os.makedirs('images')

    timestamp = time.strftime("%Y%m%d-%H%M%S")
    date = time.strftime("%Y%m%d")
    img_path = f'images/{timestamp}.jpg'

    with PiCamera() as camera:
        camera.resolution = (1024, 768)
        camera.start_preview()
        time.sleep(2)  # Camera warm-up time
        camera.capture(img_path)
        print(f'Photo taken and saved as {img_path}')
        upload_to_s3(img_path, date)

def upload_to_s3(file_path, date):
    # name of the bucket
    BUCKET_NAME = '***'

    # name of the file on S3, set as 'images/YYYYMMDD/{timestamp}.jpg'
    S3_NAME = f'images/{date}/' + os.path.basename(file_path)

    s3 = boto3.client('s3')
    s3.upload_file(file_path, BUCKET_NAME, S3_NAME)

    print(f'Uploaded {file_path} to {BUCKET_NAME}/{S3_NAME}')
    
if __name__ == '__main__':
    take_photo()

ここで「BUCKET_NAME」に指定したのと同名のS3バケットを自分のAWSアカウントで作成してください。

また、Pythonのコードから上記のS3バケットが置いてある自身のAWSアカウントにアクセスするには、
基本的にはAWSのシークレットキーが必要になります。

しかし、このコードの中にはシークレットキーに関する記述を書いていません。
これは「~/.aws/credentials」内の「default」という名前のプロファイルを事前に設定することで、自動でそれを読みに行ってくれるというAWS SDKの仕組みがあり省略することが可能となっています。

私はこの設定を事前に行なっているため、このコードで自分のバケットにアクセス出来ます。

詳しくはAWSの公式ドキュメントをご覧ください。
https://docs.aws.amazon.com/ja_jp/cli/latest/userguide/cli-configure-files.html

コード内にシークレットキーをハードコードする方法もありますが、間違ってGitHubにpushしてしまったりすると危ないのでオススメしません。

cronにより定期的に実行する

cronを使ってこのプログラムを8:00~19:00まで15分ごとに実行することにします。

crontab -e

するとcrontabの設定ファイルが開かれるので、
末尾に以下のような形で追記します。

...
*/15 8-19 * * * cd ${path/to/code} && /usr/bin/python3 take_photo.py >> /home/${ユーザー名}/cron.log 2>&1

${path/to/code}と${ユーザー名}は適宜読み替えてください。

S3にアップされた画像をタイムラプス動画にする

問題なく上記のcronが動くと、S3バケット内の「images/YYYYMMDD/」内に8:00~19:00の間、
15分ごとに画像が追加されていきます。

この後、方針としては以下のLambda関数2つを作成します。

1.その日のS3内（images/YYYYMMDD/**）の画像をつなげてタイムラプス動画を作成する関数
2.タイムラプス動画をLINEに送信する関数

作成時は以下のような流れになります。
なお、ここからはRaspberry Piではなく自分の作業用のPCで構いません。

今回は、ServerlessFrameworkを使用しました。
Lambda/EventBridge/API Gateway/DynamoDBなどのサーバーレスアーキテクチャの構築をコードベースですることが出来ます。（今回使うのはLambdaとEventBridgeだけです）

まだ入っていない場合はインストールします。

npm install --location=global serverless

プロジェクトの作成

serverless create --template aws-python3 --path plant-monitoring-generate-movie

すると、色々とファイルが自動的に作成されるはずです。
多分、こんな感じになります。

handler.py
node_modules/
package.json
package-lock.json
serverless.yml

続いて、
「serverless-python-requirements」をインストール。

$ cd generate-movie
$ serverless plugin install -n serverless-python-requirements

これを入れると、requirements.txt にインストールしたいライブラリを書くことでデプロイ時にまとめて入れてくれます。
Lambdaでライブラリを使用するのが意外と面倒なので助かります。

requirements.txt

opencv-python-headless==4.6.0.66
numpy==1.23.5

serveless.ymlを編集して、定期的にLambda関数が自動実行されるようにします。
また、環境変数の設定もここでします。

service: plant-monitoring-generate-movie

frameworkVersion: '3'

provider:
  name: aws
  runtime: python3.10
  region: ap-northeast-1
  timeout: 180
  environment:
    BUCKET_NAME: ${env:BUCKET_NAME}

  iam:
    role:
      statements:
        - Effect: Allow
          Action:
              - s3:PutObject
              - s3:GetObject
              - s3:DeleteObject
              - s3:ListBucket
          Resource:
            - arn:aws:s3:::${self:provider.environment.BUCKET_NAME}
            - arn:aws:s3:::${self:provider.environment.BUCKET_NAME}/*
        - Effect: Allow
          Action:
            - logs:CreateLogGroup
            - logs:CreateLogStream
            - logs:PutLogEvents
          Resource:
            - '*'

functions:
  create:
    handler: handler.lambda_handler
    events:
      - schedule: cron(50 9 * * ? *) # JST 18:50 (UTC 09:50) daily

plugins:
  - serverless-python-requirements

custom:
  pythonRequirements:
    dockerizePip: non-linux
    noDeploy: []

環境変数の内容を定義しておきます。

export BUCKET_NAME=${自分のバケット名}

Serverless Frameworkプロジェクト内にある handler.pyを上書きしてコードを書いていきます。

import boto3
import os
import tempfile
import cv2
from datetime import datetime

def create_video(bucket_name, folder_name, video_name, frame_rate=6):
    s3 = boto3.resource('s3')
    bucket = s3.Bucket(bucket_name)
    
    # S3からオブジェクトのリストを取得
    objects = bucket.objects.filter(Prefix=folder_name)

    image_files = []
    for obj in objects:
        # .jpgで終わるものののみを処理
        if obj.key.endswith(".jpg"):
            tmp = tempfile.NamedTemporaryFile(delete=False)
            bucket.download_file(obj.key, tmp.name)
            image_files.append(tmp.name)

    if not image_files:
        print("No image files found.")
        return

    frame = cv2.imread(image_files[0])
    height, width, channels = frame.shape

    # コーデックを決定し VideoWriter オブジェクトを作成する
    video_name_path = os.path.join('/tmp', video_name)
    video = cv2.VideoWriter(video_name_path, cv2.VideoWriter_fourcc(*'MP4V'), frame_rate, (width, height))

    for image_file in image_files:
        frame = cv2.imread(image_file)
        video.write(frame)  # Write out frame to video

    video.release()

    print(f"Video saved as {video_name}")

    # VideoをS3へアップロード
    bucket.upload_file(video_name_path, f'{folder_name}/{video_name}')
    print(f"Video uploaded to S3 as {folder_name}/{video_name}")

def lambda_handler(event, context):
    today = datetime.now().strftime('%Y%m%d')
    folder_name = f'images/{today}'
    bucket_name = os.environ['BUCKET_NAME']
    
    create_video(bucket_name, folder_name, 'output.mp4')

作成できたらデプロイします。

serverless deploy

正常に終了すると、AWSでLambdaを確認すると関数が増えています。

同じ流れで「2.タイムラプス動画をLINEに送信する関数」も作成します。
ほぼ同じ流れなのでコマンドは割愛します。

requirements.txt

requests==2.29.0

service: send-movie-to-line

frameworkVersion: '3'

provider:
  name: aws
  runtime: python3.10
  region: ap-northeast-1
  timeout: 180
  environment:
    BUCKET_NAME: ${env:BUCKET_NAME}
    LINE_ACCESS_TOKEN: ${env:LINE_ACCESS_TOKEN}
    LINE_TO: ${env:LINE_ACCESS_TOKEN}

  iam:
    role:
      statements:
        - Effect: Allow
          Action:
              - s3:PutObject
              - s3:GetObject
              - s3:DeleteObject
              - s3:ListBucket
          Resource:
            - arn:aws:s3:::${self:provider.environment.BUCKET_NAME}
            - arn:aws:s3:::${self:provider.environment.BUCKET_NAME}/*
        - Effect: Allow
          Action:
            - logs:CreateLogGroup
            - logs:CreateLogStream
            - logs:PutLogEvents
          Resource:
            - '*'

functions:
  create:
    handler: handler.lambda_handler
    events:
      - schedule: cron(0 10 * * ? *) # JST 19:00 (UTC 10:00)

plugins:
  - serverless-python-requirements

custom:
  pythonRequirements:
    dockerizePip: non-linux
    noDeploy: []

import boto3
import requests
import json
from datetime import datetime

def lambda_handler(event, context):
    url = "https://api.line.me/v2/bot/message/push"
    access_token = os.environ['LINE_ACCESS_TOKEN']
    line_to = os.environ['LINE_TO']
    
    # Generate a presigned URL for the S3 object
    s3 = boto3.client('s3')
    today = datetime.now().strftime('%Y%m%d')
    object_name = f'images/{today}/output.mp4'
    bucket_name = os.environ['BUCKET_NAME']
    response = s3.generate_presigned_url('get_object',
                                        Params={'Bucket': bucket_name,
                                                'Key': object_name},
                                        ExpiresIn=604800)  # 7 days in seconds
    
    headers = {
        'Content-Type': 'application/json; charset=UTF-8',
        'Authorization': 'Bearer {}'.format(access_token),
    }

    # Create a Flex Message with video preview
    flex_message = {
        "type": "flex",
        "altText": "動画プレビュー",
        "contents": {
            "type": "bubble",
            "hero": {
                "type": "video",
                "url": response,
                "previewUrl": "https://example.com/video_preview.jpg", # サムネイル画像が欲しい時は正しいURLに
                "altContent": {
                    "type": "image",
                    "size": "full",
                    "aspectRatio": "16:9",
                    "aspectMode": "cover",
                    "url": "https://example.com/video_preview.jpg", # サムネイル画像が欲しい時は正しいURLに
                },
                "aspectRatio": "16:9",
            },
            "body": {
                "type": "box",
                "layout": "vertical",
                "contents": [
                    {
                        "type": "text",
                        "text": "新しい動画が利用可能です",
                        "weight": "bold",
                        "size": "md",
                        "wrap": True
                    }
                ]
            }
        }
    }

    data = {
        'to': line_to,
        'messages': [flex_message],
    }

    response = requests.post(url, headers=headers, data=json.dumps(data))

    # check the response
    if response.status_code == 200:
        return {
            'statusCode': 200,
            'body': json.dumps('Request was successful')
        }
    else:
        return {
            'statusCode': response.status_code,
            'body': response.text  # レスポンス本文（エラーメッセージなど）を返す
        }