ラズパイとAWS IoTを使った見守りシステム自作で学んだこと（５）paho.mqttでセンサーデータをラズパイからAWS IoTにPublishからの続きです。

今回はセンサー検知またはカメラの動体検知をトリガーとして、トリガーの前後１５秒トータル３０秒程度のイベント録画をAmazon S3にuploadする仕組みを説明します。

下図の着色部分に該当します

冒頭に言うのも何ですが、過去に何度かcodeのupdateを重ねて簡潔にしたつもりでいたのに本稿は非常に分かり難い解説になりました。執筆時期にTLを賑わせていた、いわゆるク〇コードなんだなと反省しきりです。記事をお読みいただく方には予めご了承願います。

実行環境

リポジトリーのREADMEに沿って「ラズパイとAWS IoTを使った見守りシステム自作で学んだこと」全１０回の投稿の中で
（２）見守りProjectの設計（名前の設定）
（３）ラズパイ物理構成とパッケージ導入
 （７）ラズパイの自動起動の設定方法
の内容を構築した環境下で実行できます。

mp4ファイルをS3にuploadする際にaws-python-sdk(boto3)を利用しますので、（１）概要の説明で説明したようにAWS IAM userとIAM policyの設定、コマンドラインからのaws configureの実行が必要になります。

なお、IAM policyは予めリポジトリーに準備してあるので、sam deployの説明する際に(8) template.yamlで定義するAWSリソースの説明とsam deployで使い方を説明します。

録画の仕組み

イベント録画の仕組みを説明します。

まず下図のemr_rec.py部分が録画機能になります。

イベントがない時は01.avi～05.aviのファイル名でひたすら上書きを繰り返します。

event発生を検知したらトリガー直後の最後のaviファイルは約１５秒に延長して録画します。

最後の１５秒延長録画が終わると右側のemr_gen.pyからメソッドを呼び出して、タイムスタンプ昇順結合、mp4変換、最後にS3にuploadします。

使用するファイル

treeに示した4つのPython codesを使います。sensing.pyは（５）paho.mqttでセンサーデータをラズパイからAWS IoTにPublishと共用になるので説明を省略します。

/home/user
・
・
┣ parameters.py
┣ sensing.py
┣ emr_rec.py
┣ emr_gen
┣ ・・・

動作

全体の動作は図のようなシーケンスになります。
少しでも分かり易くしたくてシーケンス図を追加しましたが、分かり易くなった気がしません。時間をかけてク〇コードの証明をした気がして残念です涙

以下、個別のモジュールを説明します。

parameters.py

このファイルは/home/userに配置した一連のPythonファイルを実行するのに必要な変数を提供します。

１６行目と１８行目はsensing.py同様に（５）と共用になります。

DETECT_PIN = 21

SENSOR_NO = int(os.environ['SENSOR_NO'])

これ以降がイベント録画用のparametersです。

まず、S3へのuploadに必要なIAM userのCredentials、region、S3Bucket名、イベント録画トリガーを環境変数から読み込みます。

環境変数はラズパイ起動時に設定していて、環境変数設定手順は(7) ラズパイの自動起動の設定方法で説明します。

#####################################################################
#environment for cam system
#####################################################################
ACCESS_KEY = os.environ['ACCESS_KEY']
SECRET_KEY = os.environ['SECRET_KEY']
REGION = os.environ['REGION']
S3BUCKET = os.environ['S3BUCKET']
TRIGGER_SELECT = int(os.environ['TRIGGER_SELECT'])

mp4はS3Bucketにupload後にモノ別(ラズパイ設置場所別)にweb pageから閲覧できるように仕分けします。その仕分け先が３２行目に読み込むPrefixになります。

#####################################################################
#Select S3 prefix as distination of image
#####################################################################
PREFIX_IN =  os.environ['PREFIX_IN']

次はカメラの解像度、日時表示テキストの先頭pixel、fps、日時表示テキストのfont sizeを定義したリストです。

#resolution
#####################################
# 0: 176×144
# 1: 320×240
# 2: 640×480
# 3: 800×600
# 4: 1280×960
#####################################
res = 2 #Default resolution setting 2
resos = ([176, 144, 135, 25, 0.7],
         [320, 240, 230, 20, 1.3], 
         [640, 480, 470, 15, 2], 
         [800, 600, 585, 10, 2.2], 
         [1280, 720, 680, 5, 3.5])#Resolution / recording rate / caption position / frame rate / font size

初期設定は2：640×480にしてあります。

その下のthdとratioはカメラで動体検知する際の閾値です。一般的に使われている二値化して差分を取るだけの仕組みですので、撮影環境が変わる場合は現場合わせで閾値の変更が必要な場合があります。

#image Threshold when setting differential motion detection as trigger
thd = 30 #Threshold for bit judgment on 256-gradation gray scale
ratio = 0.1 #Threshold for motion detection judgment (area ratio exceeding bit judgment threshold to resolution)

parameters.pyの一番最後に、01.avi～05.aviの録画時間をinterval、最後の録画時間をend_intervalとして設定しています。初期値はinterval：4秒、end_interval：14.15秒に設定しています。

#Video tmp file recording interval
interval = datetime.timedelta(seconds=4) #Record for 4 seconds if the sensor does not detect during the last recording loop
end_interval = datetime.timedelta(seconds=14, microseconds=150000) #Record for 14 seconds if the sensor detects it during the last recording loop

emr_rec.py

連続録画を実行する主プロセスのcodeです。

openCVと標準ライブラリー以外に自作のパッケージ3つをimportします。

import cv2
import os
import time
import datetime
import subprocess
import parameters as para
from sensing import Sensor
from emr_gen import Emr_gen

sensing.pyとemr_gen.pyのクラスはインスタンス化しておきます。

senser = Sensor()
gen = Emr_gen()

初期化の際にparameters.pyで設定した初期値を読み込み

class Emr_rec:
    def __init__(self):
        self.res = para.res
        self.thd = para.thd
        self.ratio = para.ratio
        self.resos = para.resos
        self.interval = para.interval
        self.end_interval = para.end_interval
        self.output_dirpath = './tmp/'
        self.emr_dirpath = './emr/'
        self.cam_No = 'emr_' + para.PREFIX_IN + '_'

aviの中間ファイルの出力先(output_dirpath = ./tmp)と、結合後aviと変換後mp4の出力先(emr_dirpath = ./emr)を作り

        # directory for saving
        os.makedirs(self.output_dirpath, exist_ok=True)
        os.makedirs(self.emr_dirpath, exist_ok=True)

解像度に紐づくパラメーターを読み出してクラス変数に格納します。

        for i in range(5):
            if self.res == i: 
                self.width = self.resos[i][0]
                self.heigth = self.resos[i][1]
                self.label_pos = self.resos[i][2]
                self.fps = self.resos[i][3]
                self.f_size = self.resos[i][4]
                self.chd_pix = self.resos[i][0] * self.resos[i][1] * self.ratio
                break

        self.fourcc = cv2.VideoWriter_fourcc(*'mp4v')#('M','J','P','G')

exec()

実質的なemr_rec.pyの実行メソッドです。
まずVideoCaptureインスタンスを作ります。
このcapを使って、whileループの中で01.avi～05.aviを一回撮影する単位ループを繰り返します。単位ループは後述のset_loop()で定義します。

    def exec(self):
        #映像入力用のvideoCapture 作成
        device_id = -1
        self.cap = cv2.VideoCapture(device_id)

        #Cameraをオープンできなかったらreboot
        if True==self.cap.isOpened(): 
            self.cap.set(3, self.width)
            self.cap.set(4, self.heigth)
            self.cap.set(5, self.fps)
        else:
            time.sleep(90)
            subprocess.call(["sudo","reboot"])

        #録画関数呼び出し
        while True:
            self.set_loop()

            # q キーを押したら終了する。 
            if cv2.waitKey(1) == 13:
                self.cap.release()
                cv2.destroyAllWindows()

set_loop()

set_loop()では初期設定で４秒録画×５回を1単位として録画をします。

    def set_loop(self):
        emr_trigger = 0

        for loop_count in (range(5)):

ファイル名で01.aviから05.aviまでforループの中で5回VideoWriterを作成します。

            file_name = '0' + str(loop_count+1)

            writer = cv2.VideoWriter(self.output_dirpath + file_name + '.mp4', self.fourcc, self.fps, (self.width, self.heigth))

このwriterを使ってintervalで設定した秒数の録画をvideo_writer_sensor()またはvideo_writer_diff()で実行します。

set_loop()が５回１セットの単位ループを定義するのに対して、video_writer_sensor()とvideo_writer_diff()はinterval１回分の録画を実行するメソッドになります。

いずれのメソッドもintervalだけ録画する最中にeventを検知した場合には1をイベントを検知しなかったら0を返します。emr_triggerで0/1を受けます。

            if self.trigger_select == 0:
                emr_trigger = self.video_writer_sensor(emr_trigger, writer) #センサーtrigger時はこちらをアンコメント
            elif self.trigger_select == 1:
                emr_trigger = self.video_writer_diff(emr_trigger, writer) #image差分動体検知trigger時はこちらをアンコメント

video_writer_sensor()はGPIO21経由でセンサー検知した信号をトリガーにしたい場合、video_writer_diff()はカメラの動体検知をトリガーにしたい場合に使います。

イベント録画トリガーにどちらのメソッドを使うかは（２）のパラメーター対話入力で選択し、環境変数として設定され、emr_rec.pyの初期化時にparameters.pyで環境変数から読み込まれます。

emr_triggerに１が返った場合、次の録画をした後にemr_gen.pyからcombine_file()を呼び出してファイルを結合します。

            if emr_trigger ==1: #直前の録画ループ中にセンサーが検知してたら、録画後にtmpフォルダーの動画を結合する
                if loop_count == 4: file_name = '01'
                else: file_name = '0' + str(loop_count+2)

                writer = cv2.VideoWriter(self.output_dirpath + file_name + '.mp4', self.fourcc, self.fps, (self.width, self.heigth))

                if self.trigger_select == 0:
                    emr_trigger = self.video_writer_sensor(emr_trigger, writer) #センサーtrigger時はこちらをアンコメント
                elif self.trigger_select == 1:
                    emr_trigger = self.video_writer_diff(emr_trigger, writer) #image差分動体検知trigger時はこちらをアンコメント

                emr_filename = self.cam_No + datetime.datetime.now().strftime('%Y%m%d%H%M')
                gen.combine_file(emr_filename, self.width, self.heigth, self.fps)
                emr_trigger = 0
                writer.release()
                return

            if cv2.waitKey(1) == (13):
                self.cap.release()
                cv2.destroyAllWindows()
                return True  # q キーを押したら終了する。

video_writer_diff()

録画の一時保管directoryへの書込みと、カメラ画像の二値化差分による動体検知でトリガー値を返します。set_loop()の中で呼び出します。

    def video_writer_diff(self, emr_trigger, writer):
        trigger_change = 0
        
        t0=datetime.datetime.now()
        t1=datetime.datetime.now()
        delta = t1 - t0
        
        if emr_trigger == 1: interval = self.end_interval
        else: interval = self.interval

        while delta <= interval:

            _, frame = self.cap.read() # 1フレームずつ取得する            
            _, frame2 = self.cap.read()
            _, frame3 = self.cap.read()

            cv2.putText(frame, t1.strftime('%Y/%m/%d/%H:%M:%S'), (0, self.label_pos), cv2.FONT_HERSHEY_COMPLEX_SMALL, self.f_size, (255, 255, 255), 1, cv2.LINE_AA) #self.label_pos
            writer.write(frame)
            cv2.putText(frame2, t1.strftime('%Y/%m/%d/%H:%M:%S'), (0, self.label_pos), cv2.FONT_HERSHEY_COMPLEX_SMALL, self.f_size, (255, 255, 255), 1, cv2.LINE_AA)
            writer.write(frame2)
            cv2.putText(frame3, t1.strftime('%Y/%m/%d/%H:%M:%S'), (0, self.label_pos), cv2.FONT_HERSHEY_COMPLEX_SMALL, self.f_size, (255, 255, 255), 1, cv2.LINE_AA)
            writer.write(frame3)

３枚の連続する画像から２組の差分データを作り、差分データのANDに対して閾値を設定して物体の動き有り無しを判断しています。

作成当初は単純な２枚の連続画像の差分のみで検知していましたが、安定しないので複数の差分データによる判定に変更しました。

            #グレースケールに変換
            gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            gray2 = cv2.cvtColor(frame2, cv2.COLOR_BGR2GRAY)
            gray3 = cv2.cvtColor(frame3, cv2.COLOR_BGR2GRAY)
    
            diff1 = cv2.absdiff(gray2,gray)
            diff2 = cv2.absdiff(gray3,gray2)

            diff_and = cv2.bitwise_and(diff1, diff2)

            th = cv2.threshold(diff_and, self.thd, 255, cv2.THRESH_BINARY)[1]

parameters.pyで設定した閾値を超えたらフラグに1を立てて返します。

            if cv2.countNonZero(th) > self.chd_pix:
                trigger_change = 1

            t1=datetime.datetime.now()
            delta = t1 - t0

        if trigger_change == 1: emr_trigger = 1

        writer.release()

        return emr_trigger

video_writer_sensor()

録画の一時保管directoryへの書込みと、センサー検知結果でトリガー値を返します。set_loop()の中で呼び出します。

    def video_writer_sensor(self, emr_trigger, writer):
        trigger_change = 0
        
        t0=datetime.datetime.now()
        t1=datetime.datetime.now()
        delta = t1 - t0
        
        if emr_trigger == 1: interval = self.end_interval
        else: interval = self.interval

        while delta <= interval:
            ret, frame = self.cap.read() # 1フレームずつ取得する
            #if not ret:
            #    return  # 映像取得に失敗
            cv2.putText(frame, t1.strftime('%Y/%m/%d/%H:%M:%S'), (0, self.label_pos), cv2.FONT_HERSHEY_COMPLEX_SMALL, self.f_size, (255, 255, 255), 1, cv2.LINE_AA)
            #cv2.imshow('frame', frame) #モニター表示はコメントアウト
            writer.write(frame)  # フレームを書き込む。
            t1=datetime.datetime.now()
            delta = t1 - t0

sensing.pyのdetect_counter()で物体を検知したらフラグに1を立てて返します。

            tr_signal = senser.detect_counter() # センサーの物体検知有無を確認
            if tr_signal == 1: 
                trigger_change = 1 #センサーが検知した場合に返り値1を戻すためのフラグを立てる
            
            if cv2.waitKey(1) & 0xFF == ord('q'):break  # q キーを押したら終了する。

        if trigger_change == 1: emr_trigger = 1

        writer.release()

        return emr_trigger

main関数

emr_rec.pyはコンソールから呼び出して実行します。自動起動することを想定して、ラズパイがカメラデバイスを認識するのに１０秒待機時間を取ってあります。

if __name__ == '__main__':
    time.sleep(10)
    
    rec = Emr_rec()
    rec.exec()

emr_gen()

5つのaviを結合・変換してS3にuploadする副プロセスのcodeです。

自作パッケージの中でparameters.pyをimportします。

import os
import time
import cv2
import subprocess
import boto3
import parameters as para

初期化の際にparameters.pyで設定したAWS関連の設定を読み込みます。タイムスタンプ昇順に並べ替えたファイル名を格納するリストもクラスの変数として定義します。

class Emr_gen:

    def __init__(self):
        self.ACCESS_KEY = para.ACCESS_KEY
        self.SECRET_KEY = para.SECRET_KEY
        self.REGION = para.REGION
        self.s3_bucket_name = para.S3BUCKET
        self.s3_prefix = para.PREFIX_IN 
        self.emergency_dirpath = './emr/'
        self.filepath_timesorted = []

combine_file()

まず、./tmpに保存された5つのaviファイルのタイムスタンプを取得して[タイムスタンプ, ファイル名]のリストを作ります。

    #元ファイル(1.avi〜5.avi)をタイムスタンプ順に結合する関数
    def combine_file(self, emr_filename, width, heigth, fps): 

        for i in range(1,6):  #1.avi〜5.aviのタイムスタンプを取得してリストに格納
            filepath = "./tmp/0"+str(i)+".mp4" #読みだすfile名
            
            try:
                mddt = time.ctime(os.path.getmtime(filepath)) #fileのタイムスタンプを取得
                self.filepath_timesorted.append([mddt,filepath]) #データ格納順は[タイムスタンプ,filepath(i)]
            except FileNotFoundError:
                break

作成したリストをタイムスタンプ昇順にsortしてタイムスタンプを削除してファイル名だけのリストにします。

        list_length=len(self.filepath_timesorted) #aviファイル数をカウント(起動直後にトリガーが入ると5.aviまで揃わない)
        self.filepath_timesorted.sort() #ファイル名をタイムスタンプ昇順に並べ替える
    
        for i in range(list_length):
            self.filepath_timesorted[i].pop(0) #リストのタイムスタンプ部分を削除

ファイル結合後に書き込みする為のWriterを作ります。

        fourcc = cv2.VideoWriter_fourcc(*'mp4v')#('M','J','P','G')#(*'mp4v')
        emr_writer = cv2.VideoWriter(self.emergency_dirpath + emr_filename +'.mp4', fourcc, fps, (width, heigth))

タイムスタンプ昇順に./tmpのaviファイルを開き./emrに連続してフレームを出力します。

        for i in range(list_length): #タイムスタンプの古い映像を先頭にファイルを結合する
            file_path="".join(self.filepath_timesorted[i]) #i番目のavi元file、こうしないとリストからfile名がきれいに取り出せない
            cap_emr = cv2.VideoCapture(file_path) #i番目のavi元file
            
            if cap_emr.isOpened() == True: # 最初の1フレームを読み込む
                ret,frame = cap_emr.read()
            else:ret = False

            while ret: # フレームの読み込みに成功している間フレームを書き出して結合を続ける
                emr_writer.write(frame)# 読み込んだフレームを書き込み
                ret,frame = cap_emr.read()# 次のフレームを読み込
            
            subprocess.call(["sudo","rm",file_path]) #i番目のavi元ファイルを削除する

結合が終わったら使ったインスタンスを終了します。
Pythonを勉強し始めた頃ここを理解していなくてハマりました(思い出して汗

        emr_writer.release()
        cap_emr.release()
        self.filepath_timesorted.clear()

結合後のaviをmp4に変換してs3にuploadします。

        self.chenge_codec(emr_filename, fps) #mp4に変換
        self.upload_awsS3(emr_filename) #s3へアップロード
        subprocess.call(["sudo","rm",self.emergency_dirpath + emr_filename + '.mp4']) #元のaviファイルを削除する

chenge_code()

subprocessでffmpegをcallしてaviからmp4に変換します。
変換後に元のaviファイルは削除します。

    def chenge_codec(self, emr_filename, fps):
        subprocess.call(["ffmpeg", "-i", self.emergency_dirpath + emr_filename + '.mp4', "-r", str(fps), self.emergency_dirpath + emr_filename + '-1.mp4'])

upload_awsS3()

boto3.clientを使ってラズパイからS3にmp4変換後のイベント記録映像をuploadします。
upload後に送信したmp4ファイルは削除します。

    def upload_awsS3(self, emr_filename):
        s3 = boto3.client('s3', aws_access_key_id=self.ACCESS_KEY, aws_secret_access_key= self.SECRET_KEY, region_name=self.REGION)
        s3.upload_file(self.emergency_dirpath + emr_filename+"-1.mp4", self.s3_bucket_name, emr_filename+".mp4")
        print("uploaded {0}".format(emr_filename+".mp4"))
        subprocess.call(["sudo", "rm", self.emergency_dirpath + emr_filename + '-1.mp4'])

動作確認

AWS IAM userとpolicyの設定が済んだ状態でラズパイのコンソールからemr_rec.pyを実行します。

python emr_rec.py

センサーまたはカメラで検知をさせてトリガーを入れるとffmpegが実行されてupload完了のメッセージを確認することが出来ます。

次回

(7) ラズパイの自動起動の設定に続きます。

追記

図追加　2022.09.03　

録画とuploadのシーケンス図を追加しました

修正　2022.09.04

リポジトリー修正箇所の解説を修正しました。

set_parameters.pyの対話入力項目にイベント録画のトリガーを追加し、emr_rec.pyでは読み込んだ環境変数をemrrec.set_loop()で識別してトリガータイプを設定する仕様に変更しました。