Pythonで一定時間だけ処理したい

Question

前提・実現したいこと

あらかじめ指定した処理時間で、if name == 'main':の処理を打ち切りたいのですがどうすれば良いですか?

発生している問題・エラーメッセージ

エラーメッセージ

該当のソースコード

if __name__ == '__main__':
    # 定数定義
    ESC_KEY = 27     # Escキー
    INTERVAL= 33     # 待ち時間
    FRAME_RATE = 30  # fps
    r=0 
    #カウンタ用演算子
    ORG_WINDOW_NAME = "org"
   
     
    DEVICE_ID = 0

    # 分類器の指定
    cascade_file = "C:/Users/wakimoto/Anaconda2/Library/etc/haarcascades/haarcascade_frontalface_alt.xml"
    cascade = cv2.CascadeClassifier(cascade_file)
    

    # カメラ映像取得
    cap = cv2.VideoCapture(DEVICE_ID)
    #カメラ映像録画の用意　たぶん2フレーム/秒が限界　
    fourcc = cv2.VideoWriter_fourcc(*'WMV1')
    out = cv2.VideoWriter('output.wmv',fourcc, 2.0, (640,480))


    # 初期フレームの読込
    end_flag, c_frame = cap.read()
    height, width, channels = c_frame.shape

    # ウィンドウの準備
    cv2.namedWindow(ORG_WINDOW_NAME)
    

    # 変換処理ループ
    while end_flag == True:
        
        ret, frame = cap.read()
        color = (0, 0, 255)
        
        # 画像の取得と顔の検出
        img = c_frame
        img_gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
         #検出器の最小サイズを決める
        face_list = cascade.detectMultiScale(img_gray, minSize=(1, 1) )    
        
        # フレーム表示
        cv2.imshow(ORG_WINDOW_NAME, c_frame)
        
        frame = cv2.flip(frame,0)

        # write the flipped frame
       


#物体認識（顔認識）の実行
#image – CV_8U 型の行列．ここに格納されている画像中から物体が検出されます
#objects – 矩形を要素とするベクトル．それぞれの矩形は，検出した物体を含みます
#scaleFactor – 各画像スケールにおける縮小量を表します
#minNeighbors – 物体候補となる矩形は，最低でもこの数だけの近傍矩形を含む必要があります
#flags – このパラメータは，新しいカスケードでは利用されません．古いカスケードに対しては，cvHaarDetectObjects 関数の場合と同じ意味を持ちます
#minSize – 物体が取り得る最小サイズ．これよりも小さい物体は無視されます
        facerect = cascade.detectMultiScale(img_gray, scaleFactor=1.1, minNeighbors=1, minSize=(1, 1))
       
        #facerect = cascade.detectMultiScale(image_gray, scaleFactor=1.1, minNeighbors=3, minSize=(10, 10), flags = cv2.cv.CV_HAAR_SCALE_IMAGE)



        if len(facerect) > 0:
            #検出した顔を囲む矩形の作成
            for rect in facerect:
             cv2.rectangle(img, tuple(rect[0:2]),tuple(rect[0:2]+rect[2:4]), color, thickness=2)
            #フレームごとに枠の分だけカウントが増える
             r +=1
        
        
        # フレーム表示(枠つけた後)
        cv2.imshow(ORG_WINDOW_NAME, c_frame)
       
        #フレーム書き込み
        out.write(c_frame)
        # Escキーで終了
        key = cv2.waitKey(INTERVAL)
        if key == ESC_KEY:
            break

        # 次のフレーム読み込み
        end_flag, c_frame = cap.read()
        
    # 終了処理
    cv2.destroyAllWindows()
    out.release()
    cap.release()

    #全フレーム分の検出数がprintされる 
    print(r)
   

試したこと

補足情報（FW/ツールのバージョンなど）

引用したコードはWebカメラで顔検出した数を数えるコードです。

Answer 1

from os import _exit as exit
from threading import Timer
from time import sleep


if __name__ == "__main__":
    Timer(10, exit, args=(1, )).start()  # 10秒後にos._exit(1)を呼ぶ

    for i in range(20):
        print(i, "秒")
        sleep(1)

from os import getpid
from os import kill
from threading import Timer
from time import sleep
from signal import SIGTERM


if __name__ == "__main__":
    Timer(10, kill, args=(getpid(), SIGTERM)).start()  # 10秒後にos.kill(PID, SIGTERM)を呼ぶ

    for i in range(20):
        print(i, "秒")
        sleep(1)

Comments

[mkgrei]

これは便利ですね。
勉強になります。

[YouheiSakurai]

ありがとうございます。後者のほうがSIGINTやCTRL_C_EVENTで安全に停止できるかもしれないという点でスマートかもしれません。ご参考までに。

Answer 2

ちょっと質問の意図（やりたいこと）が把握できていないかもしれませんが、__main__処理というより変換処理ループにて、指定時間が来たらループを抜けるようにしてもよいかと思います。

処理時間のトータルで打ち切りする場合

import time
if __name__ == '__main__':
    # 略
    MAX_TIME = 100 # 打ち切り時間（トータルで） [sec]

    # 変換処理ループ
    while end_flag == True:
        # 変換処理。略

        # Escキーで終了
        key = cv2.waitKey(INTERVAL)
        if key == ESC_KEY:
            break

        # 打ち切り時間が来た
        if (time.time() - prevTime) >= MAX_TIME:
            break

        # 次のフレーム読み込み
        end_flag, c_frame = cap.read()

    # 終了処理

毎処理で打ち切りする場合

import time
if __name__ == '__main__':
    # 略
    MAX_TIME = 1 # 打ち切り時間（処理毎に） [sec]

    # 変換処理ループ
    while end_flag == True:

        prevTime = time.time() # 処理開始時間

        # 変換処理。略

        # 打ち切り時間までに終わらなかった
        if (time.time() - prevTime) >= MAX_TIME:
            break

        # Escキーで終了
        key = cv2.waitKey(INTERVAL)
        if key == ESC_KEY:
            break

        # 次のフレーム読み込み
        end_flag, c_frame = cap.read()

    # 終了処理

Answer 3

subprocessを許せば、他に制御用スクリプトを作って時間を計るのが一番楽です。

import subprocess
import time

max_time = 1000

st = time.time()

cmd = 'python my_script.py'
run = subprocess.Popen(cmd, shell=True)

while True:
    time.sleep(1)
    return_code = run.poll()
    if return_code == 0:
        break
    if time.time() - st > max_time:
        run.terminate()
        break

---

普通にやるなら。

import time

max_time = 1000

st = time.time()

time_per_loop = -999

while True:
    st0 = time.time()

    #
    # processing
    #

    time_per_loop = max(time_per_loop, time.time()-st0)
    
    remain_time = max_time - (time.time() - st)

    if remain_time < time_per_loop:
        break

Comments

[mochisio]

別のスクリプトを作って、本物と制御用プログラムを2つ同時に動かすという意味ですか?

[mkgrei]

おっしゃるとおりです。

基本的にwhile文を使って時間の経過を管理しますが、本編の中の場合、どのコールでどれほど時間がかかるのかといった解析が必要となります。
Pythonは同時に1つのコールしか実行できませんので、例えば画像の処理に数分かかってしまうのであれば、数分間は制御ができなくなります。
それでも予定時間を推定できれば、次のループに入るべきかどうかを判定できますが、いろいろとコードが入り乱れて面倒です。

制御用のコードを1つ書けば、本編でどんな処理をしているのかにかかわらず途中で処理を打ち止めることができます。

[mkgrei]

一応本編コードにダイレクトに時間を計るものを追記しました。

時間を計るルーチンと普通の処理が入り混じって保守しにくいです。
オブジェクト化してもよいのですが、コードが増えます。