撮影高速化試行

simple_GPIO_rec2_lightweight.py

import cv2
import os
import time
import RPi.GPIO as GPIO
import sys
from picamera2 import Picamera2
import libcamera
import numpy as np

timelog = time.time()

# raw_width           = 640           #2304
# raw_height          = 480           #1296
rec_width           = 320
rec_height          = 240
time_log            = []
time_log2           = []
time_log3           = []
frame_list          = []
meta_data_list      = []
# exposure_time       = 5000              # 1000-100000  defo:5000
exposure_time       = 1000 # 物理シャッターとの干渉を徹底的に避ける
analogue_gain       = 16	            # 1.0-20.0    defo:2.0

buffers             = 1
queue_control       = False

# Bolex
pin_shutter         = 25    # shutter timing picup 

# GPIO設定
GPIO.setmode(GPIO.BCM)
GPIO.setup(pin_shutter,     GPIO.IN, pull_up_down = GPIO.PUD_UP)

number_max_frame    = 32                 #連続撮影可能な最大フレーム数　とりあえず16FPS x 60sec = 960フレーム
record_fps          = 16                #MP4へ変換する際のFPS設定値
share_folder_path   = os.path.expanduser("~/share/")
device_name         = "bolex"
codec               = cv2.VideoWriter_fourcc(*'avc1')
camera              = Picamera2()

# 撮影モードに応じてpicameraのconfigを設定する
def set_camera_mode():
    global rec_width, rec_height, raw_width, raw_height
    camera.stop()

    config  = camera.create_still_configuration(
        main    = {
            "format"    : "BGR888", 
            "size"      : (rec_width, rec_height)
        }, 
#         raw     = {
#             "format"    : "SBGGR10_CSI2P",
#             "size"      : (raw_width, raw_height)
#         },
        buffer_count    = buffers,
        queue           = queue_control,
        controls        = {
            "ExposureTime"          : exposure_time, 
            "AnalogueGain"          : analogue_gain,
#             "FrameDurationLimits"   : (100, 100000)
            "FrameDurationLimits"   : (10000, 10000) # 10ms固定にして可変要素を排除する
        },
#         transform       = libcamera.Transform(hflip=1, vflip=1) # 切り分けのため、処理をスキップする
    )

    camera.configure(config)
    camera.start()

# timerを受けて画像を取得する関数
def shutter(text):
    global time_log, time_log2, time_log3, frame_list, meta_data_list
    
    if len(frame_list) < number_max_frame:
        time_log.append(time.time())
        request = camera.capture_request()
        time_log2.append(time.time())
        frame_list.append(request.make_image("main"))
        meta_data_list.append(request.get_metadata())
        request.release()
        time_log3.append(time.time())
    else:
        movie_save()

# ムービー撮影後、画像を連結してムービーファイルを保存する。
def movie_save():
    global recording_completed, frame_list, meta_data_list
    recording_completed = False
    print("save movie")
    movie_file_path = share_folder_path + device_name + ".mp4"

    video = cv2.VideoWriter(movie_file_path, codec, record_fps, (rec_width, rec_height))

    if not video.isOpened():
        print("video writer can't be opened")
        sys.exit()
    for i in range(len(frame_list)):
        frame = np.array(frame_list[i])
        frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        video.write(frame)
    print("movie rec finished")
    video.release()
    print_log()

def print_log():
    global time_log, time_log2, frame_list, meta_data_list

    print("  frame interval / camera interval // capture request / extract data  (msec)"  )
    for i in range(len(time_log)-1):
        t1 = (time_log[i + 1] - time_log[i]) * 1000
        t2 = (meta_data_list[i + 1]["SensorTimestamp"] - meta_data_list[i]["SensorTimestamp"]) / 1000000
        t3 = (time_log2[i] - time_log[i]) * 1000
        t4 = (time_log3[i] - time_log2[i]) * 1000
        print(f'{t1:16.3f}' + ' / ' + f'{t2:15.3f}' + ' // ' + f'{t3:15.3f}' + ' / ' + f'{t4:12.3f}')

    
    time_log            = []
    time_log2           = []
    frame_list          = []
    meta_data_list      = []

# メイン
if __name__ == "__main__":

    set_camera_mode()
    
    # シャッター動作検出時のコールバック関数
    GPIO.add_event_detect(pin_shutter,  GPIO.RISING,    callback = shutter, bouncetime = 1)

    while(True):
        time.sleep(10)

割り込みコールバックを使わない検証用コード→安定しているように見える。

import cv2
import os
import time
import RPi.GPIO as GPIO
import sys
from picamera2 import Picamera2
import libcamera
import numpy as np
from threading import Thread
import gpiozero

timelog = time.time()

# raw_width           = 640           #2304
# raw_height          = 480           #1296
rec_width           = 320
rec_height          = 240
time_log            = []
time_log2           = []
time_log3           = []
frame_list          = []
meta_data_list      = []
# exposure_time       = 5000              # 1000-100000  defo:5000
exposure_time       = 1000 # \u7269\u7406\u30b7\u30e3\u30c3\u30bf\u30fc\u3068\u306e\u5e72\u6e09\u3092\u5fb9\u5e95\u7684\u306b\u907f\u3051\u308b
analogue_gain       = 16	            # 1.0-20.0    defo:2.0

buffers             = 1
queue_control       = False

# Bolex
pin_shutter         = 25    # shutter timing picup 
pin_trigger         = 21

trigger = gpiozero.LED(pin_trigger)
button = gpiozero.Button(pin_shutter, pull_up=True)

# GPIO\u8a2d\u5b9a
#GPIO.setmode(GPIO.BCM)
#GPIO.setup(pin_shutter,     GPIO.IN, pull_up_down = GPIO.PUD_UP)

number_max_frame    = 32                 #\u9023\u7d9a\u64ae\u5f71\u53ef\u80fd\u306a\u6700\u5927\u30d5\u30ec\u30fc\u30e0\u6570\u3000\u3068\u308a\u3042\u3048\u305a16FPS x 60sec = 960\u30d5\u30ec\u30fc\u30e0
record_fps          = 16                #MP4\u3078\u5909\u63db\u3059\u308b\u969b\u306eFPS\u8a2d\u5b9a\u5024
#share_folder_path   = os.path.expanduser("~/share/")
device_name         = "bolex"
codec               = cv2.VideoWriter_fourcc(*'avc1')
camera              = Picamera2()

# \u64ae\u5f71\u30e2\u30fc\u30c9\u306b\u5fdc\u3058\u3066picamera\u306econfig\u3092\u8a2d\u5b9a\u3059\u308b
def set_camera_mode():
    global rec_width, rec_height, raw_width, raw_height
    camera.stop()

    config  = camera.create_still_configuration(
        main    = {
            "format"    : "BGR888", 
            "size"      : (rec_width, rec_height)
        }, 
#         raw     = {
#             "format"    : "SBGGR10_CSI2P",
#             "size"      : (raw_width, raw_height)
#         },
        buffer_count    = buffers,
        queue           = queue_control,
        controls        = {
            "ExposureTime"          : exposure_time, 
            "AnalogueGain"          : analogue_gain,
#             "FrameDurationLimits"   : (100, 100000)
            "FrameDurationLimits"   : (10000, 10000) # 10ms\u56fa\u5b9a\u306b\u3057\u3066\u53ef\u5909\u8981\u7d20\u3092\u6392\u9664\u3059\u308b
        },
#         transform       = libcamera.Transform(hflip=1, vflip=1) # \u5207\u308a\u5206\u3051\u306e\u305f\u3081\u3001\u51e6\u7406\u3092\u30b9\u30ad\u30c3\u30d7\u3059\u308b
    )

    camera.configure(config)
    camera.start()

# timer\u3092\u53d7\u3051\u3066\u753b\u50cf\u3092\u53d6\u5f97\u3059\u308b\u95a2\u6570
def shutter():
    global time_log, time_log2, time_log3, frame_list, meta_data_list
    
    if len(frame_list) < number_max_frame:
        time_log.append(time.time())
        request = camera.capture_request()
        time_log2.append(time.time())
        frame_list.append(request.make_image("main"))
        meta_data_list.append(request.get_metadata())
        request.release()
        time_log3.append(time.time())
    else:
        movie_save()

# \u30e0\u30fc\u30d3\u30fc\u64ae\u5f71\u5f8c\u3001\u753b\u50cf\u3092\u9023\u7d50\u3057\u3066\u30e0\u30fc\u30d3\u30fc\u30d5\u30a1\u30a4\u30eb\u3092\u4fdd\u5b58\u3059\u308b\u3002
def movie_save():
    #global recording_completed, frame_list, meta_data_list
    #recording_completed = False
    #print("save movie")
    #movie_file_path = share_folder_path + device_name + ".mp4"

    #video = cv2.VideoWriter(movie_file_path, codec, record_fps, (rec_width, rec_height))

    #if not video.isOpened():
    #    print("video writer can't be opened")
    #    sys.exit()
    #for i in range(len(frame_list)):
    #    frame = np.array(frame_list[i])
    #    frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
    #    video.write(frame)
    #print("movie rec finished")
    #video.release()
    print_log()

def print_log():
    global time_log, time_log2, frame_list, meta_data_list

    print("  frame interval / camera interval // capture request / extract data  (msec)"  )
    for i in range(len(time_log)-1):
        t1 = (time_log[i + 1] - time_log[i]) * 1000
        t2 = (meta_data_list[i + 1]["SensorTimestamp"] - meta_data_list[i]["SensorTimestamp"]) / 1000000
        t3 = (time_log2[i] - time_log[i]) * 1000
        t4 = (time_log3[i] - time_log2[i]) * 1000
        print(f'{t1:16.3f}' + ' / ' + f'{t2:15.3f}' + ' // ' + f'{t3:15.3f}' + ' / ' + f'{t4:12.3f}')

    
    time_log            = []
    time_log2           = []
    frame_list          = []
    meta_data_list      = []

def trigger_th():
    while True:
        trigger.on()
        time.sleep(0.01)
        trigger.off()
        time.sleep(0.05)

# \u30e1\u30a4\u30f3
if __name__ == "__main__":

    set_camera_mode()
    
    # \u30b7\u30e3\u30c3\u30bf\u30fc\u52d5\u4f5c\u691c\u51fa\u6642\u306e\u30b3\u30fc\u30eb\u30d0\u30c3\u30af\u95a2\u6570
    #GPIO.add_event_detect(pin_shutter,  GPIO.RISING,    callback = shutter, bouncetime = 1)

    t = Thread(target=trigger_th, args=[])
    t.start()

    while True:
        if button.value:
            shutter()

  frame interval / camera interval // capture request / extract data  (msec)
          43.770 /          29.948 //          41.442 /        2.284
          27.449 /          29.949 //          25.534 /        1.874
          32.626 /          29.947 //          30.408 /        2.174
          27.269 /          29.949 //          25.349 /        1.879
          32.143 /          29.947 //          30.265 /        1.837
          27.714 /          29.949 //          25.863 /        1.811
          32.206 /          29.947 //          30.284 /        1.881
          27.604 /          19.965 //          25.760 /        1.804
          22.301 /          29.949 //          20.411 /        1.848
          33.348 /          29.947 //          25.799 /        1.899
          26.864 /          29.949 //          24.936 /        1.888
          38.727 /          29.947 //          25.435 /        1.900

結構、安定した感じ！？

検証用コードを踏まえての専用ハードウェア適用版。

import cv2
import os
import time
#import RPi.GPIO as GPIO
import sys
from picamera2 import Picamera2
import libcamera
import numpy as np
import gpiozero

timelog = time.time()

# raw_width           = 640           #2304
# raw_height          = 480           #1296
rec_width           = 320
rec_height          = 240
time_log            = []
time_log2           = []
time_log3           = []
frame_list          = []
meta_data_list      = []
# exposure_time       = 5000              # 1000-100000  defo:5000
exposure_time       = 1000 # 物理シャッターとの干渉を徹底的に避ける
analogue_gain       = 16	            # 1.0-20.0    defo:2.0

buffers             = 1
queue_control       = False

# Bolex
pin_shutter         = 25    # shutter timing picup 

trigger = gpiozero.Button(pin_shutter, pull_up=True)

# GPIO設定
#GPIO.setmode(GPIO.BCM)
#GPIO.setup(pin_shutter,     GPIO.IN, pull_up_down = GPIO.PUD_UP)

number_max_frame    = 32                 #連続撮影可能な最大フレーム数　とりあえず16FPS x 60sec = 960フレーム
record_fps          = 16                #MP4へ変換する際のFPS設定値
share_folder_path   = os.path.expanduser("~/share/")
device_name         = "bolex"
codec               = cv2.VideoWriter_fourcc(*'avc1')
camera              = Picamera2()

# 撮影モードに応じてpicameraのconfigを設定する
def set_camera_mode():
    global rec_width, rec_height, raw_width, raw_height
    camera.stop()

    config  = camera.create_still_configuration(
        main    = {
            "format"    : "BGR888", 
            "size"      : (rec_width, rec_height)
        }, 
#         raw     = {
#             "format"    : "SBGGR10_CSI2P",
#             "size"      : (raw_width, raw_height)
#         },
        buffer_count    = buffers,
        queue           = queue_control,
        controls        = {
            "ExposureTime"          : exposure_time, 
            "AnalogueGain"          : analogue_gain,
#             "FrameDurationLimits"   : (100, 100000)
            "FrameDurationLimits"   : (10000, 10000) # 10ms固定にして可変要素を排除する
        },
#         transform       = libcamera.Transform(hflip=1, vflip=1) # 切り分けのため、処理をスキップする
    )

    camera.configure(config)
    camera.start()

# timerを受けて画像を取得する関数
def shutter(text):
    global time_log, time_log2, time_log3, frame_list, meta_data_list
    
    if len(frame_list) < number_max_frame:
        time_log.append(time.time())
        request = camera.capture_request()
        time_log2.append(time.time())
        frame_list.append(request.make_image("main"))
        meta_data_list.append(request.get_metadata())
        request.release()
        time_log3.append(time.time())
    else:
        movie_save()

# ムービー撮影後、画像を連結してムービーファイルを保存する。
def movie_save():
    global recording_completed, frame_list, meta_data_list
    recording_completed = False
    print("save movie")
    movie_file_path = share_folder_path + device_name + ".mp4"

    video = cv2.VideoWriter(movie_file_path, codec, record_fps, (rec_width, rec_height))

    if not video.isOpened():
        print("video writer can't be opened")
        sys.exit()
    for i in range(len(frame_list)):
        frame = np.array(frame_list[i])
        frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        video.write(frame)
    print("movie rec finished")
    video.release()
    print_log()

def print_log():
    global time_log, time_log2, frame_list, meta_data_list

    print("  frame interval / camera interval // capture request / extract data  (msec)"  )
    for i in range(len(time_log)-1):
        t1 = (time_log[i + 1] - time_log[i]) * 1000
        t2 = (meta_data_list[i + 1]["SensorTimestamp"] - meta_data_list[i]["SensorTimestamp"]) / 1000000
        t3 = (time_log2[i] - time_log[i]) * 1000
        t4 = (time_log3[i] - time_log2[i]) * 1000
        print(f'{t1:16.3f}' + ' / ' + f'{t2:15.3f}' + ' // ' + f'{t3:15.3f}' + ' / ' + f'{t4:12.3f}')

    
    time_log            = []
    time_log2           = []
    frame_list          = []
    meta_data_list      = []

# メイン
if __name__ == "__main__":

    set_camera_mode()
    
    # シャッター動作検出時のコールバック関数
    # GPIO.add_event_detect(pin_shutter,  GPIO.RISING,    callback = shutter, bouncetime = 1)

    #while(True):
    #   time.sleep(10)

    while True:
        if trigger.value:
            shutter("dummy")

raspberrypi/picamera2#787