この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。
カフェチームの山本です。
前々回の記事では、クラウド上で動画を処理するために、エッジデバイスから動画ファイルを送信する方法として、Pythonのプログラムを実装しました。(ここでは、予め動画ファイルが作成されていることが前提となっていました)
また、前回の記事では、カメラで撮影した映像をすぐに送信するケースを考え、撮影した画像をリアルタイムにエンコードするために、FFmpegを別プロセスで立ち上げて標準入力で画像を渡す方法を調べました
今回は、前回と同様に、撮影した画像をリアルタイムにエンコードすることを目的として、GStreamerを使用する方法を記載します。Jetson Nano(JetPack)など、FFmpeg(のハードウェアエンコーディング)が使用できず、GStreamerを使える環境において利用できます。
実装したコード
早速結論ですが、以下のようなコードを実装することで、リアルタイムにエンコードできました。
video_writer.py
from enum import Enum
def video_filepath(device_id, producer_time):
return f"temp/{device_id}/{producer_time}.mkv"
class VideoWriterState(Enum):
OPEN = 0
RELEASED = 1
camera.py
OpenCVを使って動画を撮影するスクリプトです。
import cv2
import time
def estimate_capture_time(capture_time_post, time_offset):
capture_time_estimate = capture_time_post - time_offset
return capture_time_estimate
class Camera():
def __init__(self, idx_cam, width, height, fps, time_offset):
self._idx_cam = idx_cam
self._width = width
self._height = height
self._fps = fps
self._time_offset = time_offset
capture = cv2.VideoCapture(idx_cam)
capture.set(cv2.CAP_PROP_FRAME_WIDTH, width)
print(capture.get(cv2.CAP_PROP_FRAME_WIDTH), width)
assert capture.get(cv2.CAP_PROP_FRAME_WIDTH) == width
capture.set(cv2.CAP_PROP_FRAME_HEIGHT, height)
print(capture.get(cv2.CAP_PROP_FRAME_HEIGHT), height)
assert capture.get(cv2.CAP_PROP_FRAME_HEIGHT) == height
capture.set(cv2.CAP_PROP_FPS, fps)
print(capture.get(cv2.CAP_PROP_FPS), fps)
assert capture.get(cv2.CAP_PROP_FPS) == fps
self._capture = capture
def read(self):
# capture_time_pre = time.time()
ret, frame = self._capture.read()
capture_time_post = time.time()
capture_time_estimate = estimate_capture_time(capture_time_post, self._time_offset)
return ret, frame, capture_time_estimate
gstreamer_video_writer.py
OpenCVのVideoWriter経由で、GStreamerを立ち上げます。撮影した映像(画像)を、VideoWriter経由でそのプロセスに送信しつづけることで、リアルタイムにGStreamerで処理することができます。
create_command関数内で、GStreamerのコマンドを作成しています。omxh264encがエンコード(ハードウェアエンコーディング)する箇所です。Jetson NanoやRaspberryで使用できます。
import subprocess
import os
import cv2
from video_writer import video_filepath, VideoWriterState
def create_command(codec, width, height, fps, pixel_format, filename, bitrate=None):
# pixel_format : ex) "bgr24"
# codec : ex) "h264"
# bitrate : ex) "800k"
dimension = f'{width}x{height}'
# filename_os = f'"{os.path.abspath(filename)}"'
if not codec in ["H264", "h264"]:
raise NotImplementedError("Not Supported H264 in gstreamer_video_writer.py")
command = [
f"appsrc",
f"autovideoconvert",
f"omxh264enc" if bitrate is None else f"omxh264enc bitrate={int(bitrate[:-1])*1024}",
f"matroskamux",
f"filesink location={filename} sync=false",
]
command_all = " ! ".join(command)
return command_all
class GstreamerVideoWriter():
def __init__(self, device_id, producer_time, codec_forcc, width, height, fps, pixel_format, bitrate=None):
self._producer_time = producer_time
self._codec_forcc = codec_forcc
self._width = width
self._height = height
self._fps = fps
filename = video_filepath(device_id, producer_time)
if os.path.dirname(filename) != "":
os.makedirs(os.path.dirname(filename), exist_ok=True)
self._filename = filename
command = create_command(codec_forcc, width, height, fps, pixel_format, filename, bitrate)
self._command = command
self._writer = cv2.VideoWriter(command, 0, fps, (width, height))
# print(" ".join(command))
self._frame_count = 0
self._state = VideoWriterState.OPEN
def write(self, frame):
if self._state == VideoWriterState.OPEN:
self._writer.write(frame)
# self._proc.stdin.write(frame.tostring())
self._frame_count += 1
def release(self):
if not self._state == VideoWriterState.RELEASED:
# self._writer.release()
self._writer.release()
self._state = VideoWriterState.RELEASED
return self._filename, self._frame_count
if __name__ == "__main__":
CAMERA_DEVICE_INDEX = 0
CAPTURE_WIDTH, CAPTURE_HEIGHT = 640, 480
CAPTURE_FPS = 10
CAPTURE_TIME_OFFSET = 0.17
CAPTURE_PIXEL_FORMAT = "bgr24"
WRITER_CODEC_FOURCC = "H264"
WRITER_BITRATE = "800k"
import sys
import time
from camera import Camera
CAMERA_DEVICE_INDEX = int(sys.argv[1]) if len(sys.argv) >= 2 else CAMERA_DEVICE_INDEX
writer = GstreamerVideoWriter(CAMERA_DEVICE_INDEX, int(time.time()), WRITER_CODEC_FOURCC, CAPTURE_WIDTH, CAPTURE_HEIGHT, CAPTURE_FPS, CAPTURE_PIXEL_FORMAT, WRITER_BITRATE)
camera = Camera(CAMERA_DEVICE_INDEX, CAPTURE_WIDTH, CAPTURE_HEIGHT, CAPTURE_FPS, CAPTURE_TIME_OFFSET)
n_frame_to_capture = 300
i_frame = 0
while True:
ret, frame, capture_time_estimate = camera.read()
writer.write(frame)
i_frame += 1
if i_frame >= n_frame_to_capture:
break
filename, frame_count = writer.release()
使用するカメラによっては対応していない解像度・FPSがあるので、CAPTURE_WIDTHなどを適宜変更してください。以下のコマンドなどで確認できます。
https://askubuntu.com/questions/214977/how-can-i-find-out-the-supported-webcam-resolutions
lsusb
lsusb -s 00x:00x -v | egrep "Width|Height"
v4l2-ctl --list-formats-ext
補足
今回のプログラムで出力した動画ファイルは、前回と異なり、サムネイルが表示され、動画アプリでも再生できました。
また、前々回のプログラムで動画ファイルをKinesis Video Streamsに送信し、コンソール画面で正しく表示されることを確認できました。
まとめ
撮影した映像をリアルタイムにエンコーディングするために、PythonからOpenCV経由でGStreamerのプロセスを起動し、画像を入力するコードを実装しました。
参考にさせていただいたページ・サイト
https://dev.classmethod.jp/articles/kinesis-video-streams-gstreamer-encode/ https://forums.developer.nvidia.com/t/ffmpeg-using-hardware-gpu-cuda/72312/7