Amazon Kinesis Video Streams で受け取った動画を定期的に自分のS3に保存する

## 背景
自宅のプランター菜園を常時観察できるようにRaspberry Piに接続したカメラで常時撮影してKinesis Video Streams (KVS) に送信しているのですが、放っておくと所定の時間で動画は削除されてしまいます。勿体ないので保管して見返したり、後々機械学習に利用したりできるようにしたいと思い、S3に定期的に保管する方法を検証してみました。
![IMG_3835.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/534832/318cb1a5-2be6-2c8b-da07-e08f02ceccd1.jpeg)

KVSは流れてきた動画をS3に保管する機能を持っていないため、この部分は作りこむ必要があります。

## 構成
最終的にアーキテクチャはこのようになりました。
![AutomatedGetClip.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/534832/4da8ad8d-2ff7-e6ff-0ceb-0fc57fb7e064.png)

Raspberry PiからKVSにデータを流す方法は下記を参考にしました。具体的な実装手順に関してはこの記事では割愛します。
https://docs.aws.amazon.com/kinesisvideostreams/latest/dg/examples-gstreamer-plugin.html

KVSに流れてきた動画はデフォルトでは24時間保管されています。（最大10年）
全体の流れとしてはこれだけです。
1. KVSから所定の開始時刻／終了時刻の動画を GetClip API でMP4として取り出し S3 に格納する Lambda Function を作成する
1. EventBridge で定期的にこの Lambda Function を呼び出す

1つのMP4のファイルに永続的な動画を保管することはできないため、定期的に区切りながらMP4として保管していく必要があります。動画の取得頻度は GetClip API のレスポンスのペイロードに格納できるサイズに依存します。下記のとおり、開始のタイムスタンプから100MBのサイズまたは200フラグメント以下のどちらか小さいほうになるため、撮影している動画のフラグメントを確認します。
> **Payload**
指定したビデオストリームのメディアクリップを含む従来の MP4 ファイル。出力には、指定された開始タイムスタンプから (最初の) 100 MB 分のフラグメントまたは 200 個のフラグメントが含まれます。詳細については、「Kinesis Video Streams Limits」を参照してください。

https://docs.aws.amazon.com/ja_jp/kinesisvideostreams/latest/dg/API_reader_GetClip.html

フラグメントの長さはKVSのマネジメントコンソールにて、 **ビデオストリーム** - **メディア再生** - **メディア統計** で確認することが出来ます。私の動画は **1.5秒** になっています。

![fragment.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/534832/a5d7c1af-76fe-ece5-9d78-33840f98ea20.png)

200個のフラグメントまで取得できるので、私の動画ではGetClipで取得できる動画の長さは5分が上限になります。
> 1.5秒 x 200 = 300秒 = 5分

さて、これを元に実装していきます。


### 開始時刻のタイムスタンプを Parameter Store に格納
Lambdaを動かす際に問題になるのが、**揮発性**です。要するに何かしらの値をずっとLambdaのおなかに保持することが出来ないので、どこか外に置く必要があります。今回は5分毎に動画を切り出していくので、開始時刻のタイムスタンプを更新しながら保持する必要があります。よくある方法としてはDynamoDBに格納するやり方なのですが、たかだかタイムスタンプ1つのためにわざわざDynamoDBを建てるのも面倒＆もったいないので、Systems Manager の一機能であるParameter Storeを利用してみることにしました。

https://docs.aws.amazon.com/ja_jp/systems-manager/latest/userguide/systems-manager-parameter-store.html

![starttime.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/534832/d8172e23-9071-4adb-3b84-5bfd33300e34.png)

設定内容は下記のとおりです。最後の**値**が動画の切り出し毎に更新されていきます。
| 設定項目 | 値 |
| ----- | ----- |
| 名前 | /iot-farm/latest-video-timestamp |
| 利用枠 | 標準 |
| タイプ | 文字列 |
| データ型 | text |
| 値 | 2023-05-31 00:00:00 |

同様に、切り出す動画の長さ（300秒）を定義するパラメータストアも作っておくことにしました。（意味があるかはわかりませんが。。。）

### KVS から動画を切り出してS3に格納する Lambda Function

Lambdaでは下記の処理を作成していきます

#### Parameter Store から値を取得する
Parameter Store からは **get_parameter** というAPIをたたくことでレスポンス内の *Value* に値が格納されて取り出すことが出来ます。楽ちんですね。

https://docs.aws.amazon.com/systems-manager/latest/APIReference/API_GetParameter.html

```Python
# Get starttime value from Parameter Store
ssm = boto3.client('ssm')
str_starttime = ssm.get_parameter(
    Name='/iot-farm/latest-video-timestamp'
    )['Parameter']['Value']
    
str_duration = ssm.get_parameter(
    Name='/iot-farm/video-duration-seconds'
    )['Parameter']['Value']
```

#### 動画を切り出す
KVSから動画を切り出すにはGetClip APIを利用することになりますが、これ単体ではデータは取り出せず、あらかじめKVSのエンドポイントを **GetDataEndpoint** API で取得する必要があります。これを忘れると、 *UnknowOperation* の例外が発生します。

> GetClip API の使用の前提条件として、GetDataEndpoint を使用してエンドポイントを取得し、APINameパラメーターに GET_CLIP を指定する必要があります。

https://docs.aws.amazon.com/ja_jp/kinesisvideostreams/latest/dg/API_reader_GetClip.html

```Python
# KVSのエンドポイントを取得
kv = boto3.client('kinesisvideo')
kv_response = kv.get_data_endpoint(
    StreamName=stream_name,
    APIName='GET_CLIP'
    )
```

そのあと、このエンドポイントを用いてboto3のクライアントを初期化して *get_clip* を使って動画を切り出しますが、このときの動画クリップの開始時刻と終了時刻の指定は共に **datetime** 型です。そのため、下記のようにstrptime メソッドを使って文字列型をdatetime型に変換する必要があります。また、終了時刻はtimedelta メソッドを利用して導き出しました。

```Python
starttime = datetime.datetime.strptime(str_starttime, "%Y-%m-%d %H:%M:%S")
endtime = starttime + datetime.timedelta(seconds=int(str_duration))
```

作成した開始時刻と終了時刻を用いてget_clipを呼び出します。stream_nameはKVSのストリーム名です。

```Python
kvs = boto3.client('kinesis-video-archived-media', endpoint_url=kv_response['DataEndpoint'])
response = kvs.get_clip(
    StreamName=stream_name,
    ClipFragmentSelector={
        'FragmentSelectorType': 'SERVER_TIMESTAMP',
        'TimestampRange': {
            'StartTimestamp': starttime,
            'EndTimestamp': endtime
        }
    }
)
```

#### S3 への格納
切り出した動画をS3 に保管します。一旦Lambdaの/tmpに保管してからアップロードする方法もありますが、二度手間になるので、今回はペイロードをインメモリでそのままアップロードするようにしました。動画の実体は response 内の *Payload* に格納されているため、*read()* メソッドでバイナリで取り出し、そのままS3にput_objectしてあげます。Lambdaはデフォルトで128MBのメモリが搭載されていますが、実行前に念のため確認してください。

```Python
bucket_name = "YOUR-BUCKET-NAME"
body = response['Payload'].read()
key = "video/" + starttime.strftime("%Y-%m-%d_%H-%M-%S") + ".mp4"

s3 = boto3.client('s3')
put_response = s3.put_object(
    Body=body,
    Bucket=bucket_name,
    Key=key
    )
```

#### Parameter Store の更新
最後に、切り出した動画の続きから切り出せるように開始時刻を新しくしたパラメータで更新します。**PutParameter** APIを呼び出すだけなので簡単です。その際、終了時刻をstrftimeメソッドで文字列に変換して格納します。

https://docs.aws.amazon.com/systems-manager/latest/APIReference/API_PutParameter.html

```Python
ssm_put_response = ssm.put_parameter(
    Name='/iot-farm/latest-video-timestamp',
    Value=endtime.strftime("%Y-%m-%d %H:%M:%S"),
    Overwrite=True
    )
```

#### コードの全貌
全体のコードはこちらです。

```Python
import json
import boto3
import datetime

def lambda_handler(event, context):
    # Get starttime value from Parameter Store
    ssm = boto3.client('ssm')
    str_starttime = ssm.get_parameter(
        Name='/iot-farm/latest-video-timestamp'
        )['Parameter']['Value']
        
    str_duration = ssm.get_parameter(
        Name='/iot-farm/video-duration-seconds'
        )['Parameter']['Value']
    
    # Parameters    
    stream_name = event['stream_name']
    starttime = datetime.datetime.strptime(str_starttime, "%Y-%m-%d %H:%M:%S")
    endtime = starttime + datetime.timedelta(seconds=int(str_duration))
    
    # KVSのエンドポイントを取得
    kv = boto3.client('kinesisvideo')
    kv_response = kv.get_data_endpoint(
        StreamName=stream_name,
        APIName='GET_CLIP'
        )
    
    kvs = boto3.client('kinesis-video-archived-media', endpoint_url=kv_response['DataEndpoint'])
    response = kvs.get_clip(
        StreamName=stream_name,
        ClipFragmentSelector={
            'FragmentSelectorType': 'SERVER_TIMESTAMP',
            'TimestampRange': {
                'StartTimestamp': starttime,
                'EndTimestamp': endtime
            }
        }
    )
    
    bucket_name = "YOUR-BUCKET-NAME"
    body = response['Payload'].read()
    key = "video/" + starttime.strftime("%Y-%m-%d_%H-%M-%S") + ".mp4"
    
    s3 = boto3.client('s3')
    put_response = s3.put_object(
        Body=body,
        Bucket=bucket_name,
        Key=key
        )
        
    
    ssm_put_response = ssm.put_parameter(
        Name='/iot-farm/latest-video-timestamp',
        Value=endtime.strftime("%Y-%m-%d %H:%M:%S"),
        Overwrite=True
        )
```

### EventBridge Scheduler で5分毎にLambdaが起動するよう設定する
定期的にLambdaをキックするために利用します。

https://docs.aws.amazon.com/ja_jp/eventbridge/latest/userguide/scheduler.html

EventBridge のマネジメントコンソールにて**スケジューラ** をクリックし、**スケジュールを作成**をクリックします。

![event01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/534832/acac79ed-976c-d139-c68b-d9b23c4be68b.png)

スケジュールの設定などをしていきます。繰り返し実行ができるように **定期的なスケジュール** を選択して**rateベースのスケジュール**を選択してください。また、実効間隔(rate式)は *5 minites* に設定します。

![event02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/534832/c7c1a494-0ba0-e93e-849d-f391301987bb.png)

**ターゲットの選択** にて **AWS Lambda** を選択し、上で作成したLambda関数とLambda関数の引数（ペイロード）を指定します。

```JSON
{
    "stream_name": "YOUR-STREAM-NAME", #ストリーム名
    "bucket_name": "YOURMP4BUCKET", #出力先バケット
    "folder": "video/" #バケット内の出力先フォルダー
}
```

![event03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/534832/aa0b821e-acae-8dac-8552-b364cc7f455a.png)

再試行回数を5回程度に減らしておきます。

![event04.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/534832/6a8af4c0-37d5-7c0c-9d32-4ec6eba8325c.png)

最後に**スケジュールを作成**をクリックして完了です。

## 結果
放っておくと、このように5分毎に動画が切り出されてS3に格納されるようになりました。

![s3.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/534832/e3adb6e3-a9d8-e656-20ab-d14b7d518488.jpeg)

また Parameter Store も5分毎にバージョンが上がり続けてます。どこまで上げられるのかわかりませんが、やはりDynamoDBに・・・といずれなるかもしれません。。。

![parameter02.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/534832/5bc199de-82e7-cfe2-5d61-c83d09b66e73.jpeg)


## まとめ
GetClip API + Lambda + EventBridgeの組み合わせで定期的にKVSの動画を切り出して保管できるようになりました。今後KVS自体に機能が実装されていくかもしれませんが、この内容が今現在で困っている方の参考になれば幸いです。