はじめに
AWS Kinesis Video Streamsに配信された映像データをリアルタイムで解析し、人物や物体を検出するシステムを作ってみました！解析にはAmazon Rekognitionを使っています。映像データをそのまま解析させると料金が高額になってしまうので、映像データからフレームを抽出し、画像ファイルを解析することにしました。
今回は、Kinesis Video Streamsで映像を受信できている状態から構築しています。
映像を受信するまでの手順は別の記事にまとめているので参考にしてもらえればと思います！
構築手順
1. IAMロール作成
まずはLambdaの実行ロールと必要なポリシーを設定します。
必要なポリシー：
-
AWSLambdaBasicExecutionRole(Lambda実行) -
AmazonKinesisVideoStreamsReadOnlyAccess(映像取得) -
AmazonRekognitionReadOnlyAccess(解析) -
AmazonS3FullAccess(解析結果保存) -
AmazonSNSFullAccess(メール通知)
2. Lambda関数の作成
- ランタイム:
Python 3.11 - メモリ:
1024MB - タイムアウト:
15秒
- メモリ不足のエラーが出たので128MBから1024MBに増やしました。
- フレーム抽出に使用するOpenCVの依存ライブラリのnumpy 1.24.3との互換性の問題でPython 3.12でエラーが出たのでPython 3.11で設定しました。
3. OpenCVレイヤー構築
LambdaでOpenCVを使用するためにはカスタムレイヤーが必要です。Dockerを使用してLinux環境でOpenCVをビルドした後、S3にアップロードしてレイヤーに設定しています。
4. SNSの作成
トピックとサブスクリプションを設定するだけなので割愛します。
5. Lambda関数の実装
コードをLambda関数にデプロイします。処理の流れは↓のようになっています。
- Kinesis Video Streamsからデータ取得
- OpenCVでフレーム抽出・最適化
- Rekognitionで物体検出
- 結果の可視化処理
- S3保存・SNS通知
全て記載すると長くなるので、要点を絞って紹介します。
フレーム抽出
def extract_frame_lightweight(video_data):
# Lambda一時ディスクに保存
temp_path = '/tmp/temp_video.mp4'
with open(temp_path, 'wb') as f:
f.write(video_data)
# OpenCV設定
cap = cv2.VideoCapture(temp_path)
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 360)
ret, frame = cap.read()
cap.release()
# 解像度調整
frame_resized = cv2.resize(frame, (640, 360),
interpolation=cv2.INTER_LINEAR)
# JPEG圧縮（品質70%）
encode_params = [cv2.IMWRITE_JPEG_QUALITY, 70]
_, buffer = cv2.imencode('.jpg', frame_resized, encode_params)
return {
'frame_bytes': buffer.tobytes(),
'original_frame': frame_resized.copy(),
'file_size_kb': len(buffer.tobytes()) // 1024
}
解析にかかるコストの削減、リアルタイム性を向上させることを目的とした処理です。
映像データからフレームを抽出して、解像度と画質を落としています。
- 解像度削減：1280x720 → 640x360
- JPEG品質調整：70%
- 単一フレーム処理：1フレームのみ抽出
Rekognitionの呼び出し
Rekognitionのパラメータを調整し、処理速度と検出精度を改善します。
def analyze_with_rekognition_optimized(image_bytes):
# 最適化されたRekognition呼び出し
response = rekognition.detect_labels(
Image={'Bytes': image_bytes},
MaxLabels=5,
MinConfidence=80.0,
Features=['GENERAL_LABELS']
)
# 結果の軽量化処理
labels = []
for label in response['Labels'][:3]: # 上位3つのみ
label_info = {
'Name': label['Name'],
'Confidence': round(label['Confidence'], 1)
}
# バウンディングボックス（最初の1つのみ）
if 'Instances' in label and label['Instances']:
instance = label['Instances'][0]
label_info['BoundingBox'] = {
'Left': round(instance['BoundingBox']['Left'], 2),
'Top': round(instance['BoundingBox']['Top'], 2),
'Width': round(instance['BoundingBox']['Width'], 2),
'Height': round(instance['BoundingBox']['Height'], 2)
}
labels.append(label_info)
return {
'labels': labels,
'label_count': len(labels),
'confidence_avg': round(sum(l['Confidence'] for l in labels) / len(labels), 1) if labels else 0
}
検出するラベル数は3つに絞っていて、信頼度が80%以上の結果のみ採用するようにしています。検出された物の内、最初の１つにだけバウンディングボックスが付くようにしました。
6. EventBridgeの設定
LambdaをトリガーするためにEventBridgeを使用しています。とりあえず、スケジュールを指定して定期実行していますが実際に利用するときは何らかのイベントと連動するような形でトリガーする仕組みを作りたいですね。
実行結果
実行されるとS3に解析結果が保存され、メールに記載されたURLから確認することができます。簡単な解析の例になりますが、ペットボトルのコーラが映った映像を解析した例がこちらです！
Beverage：99.7%、Soda：99.7%、Coke：99.6%といったように正しく検知してくれました。
まとめ
Rekognitionを使ってリアルタイムな解析をすることができました！
解析のリアルタイム性についてですが、EventBridgeでLambdaをトリガーしてからS3に解析結果が格納されるまで約2秒でした。Kinesis Video Streamsへの配信遅延も考慮するとシステム全体としての遅延は10秒程です。十分実用可能な範囲じゃないでしょうか。
今後はユースケースに応じた機能拡張も考えてみたいと思います！