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

$ sudo apt-get update
$ pip3 install boto3 picamera

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()

crontab -e

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

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

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

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

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=${自分のバケット名}

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

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  # レスポンス本文（エラーメッセージなど）を返す
        }