Closed kou12345 closed 1 year ago
以下は、モデルの推論処理をバッチ処理化するためのソースコードの例です。Model
クラスの infer_batch
メソッドが追加されています。
import os
import torch
import random
import cv2
import onnxruntime as ort
import numpy as np
from PIL import Image
from function.letterbox import letterbox
from function.draw import draw
class Model:
def __init__(self, model_type, model_name, providers):
# ... (省略)
self.batch_size = 4 # バッチサイズを設定
def infer(self, frame):
# 単一のフレームに対する推論処理
def infer_batch(self, frames):
# バッチ処理による推論処理
# バッチサイズに分割
num_frames = len(frames)
num_batches = (num_frames + self.batch_size - 1) // self.batch_size
results = []
for batch_idx in range(num_batches):
start_idx = batch_idx * self.batch_size
end_idx = min(start_idx + self.batch_size, num_frames)
batch_frames = frames[start_idx:end_idx]
batch_images = []
ratios = []
dwdhs = []
for frame in batch_frames:
copy_frame = frame.copy()
# Preprocessing frame (type: numpy -> type: tensor)
copy_frame, ratio, dwdh = letterbox(copy_frame, auto=False)
copy_frame = copy_frame.transpose((2, 0, 1))
copy_frame = np.expand_dims(copy_frame, 0)
copy_frame = np.ascontiguousarray(copy_frame)
copy_frame = copy_frame.astype(np.float32)
copy_frame /= 255
batch_images.append(copy_frame)
ratios.append(ratio)
dwdhs.append(dwdh)
# バッチの推論を行う
batch_images = np.concatenate(batch_images, axis=0)
inp = {self.inname[0]: batch_images}
outputs = self.model.run(self.outname, inp)[0]
# バッチ結果を格納
for idx, frame in enumerate(batch_frames):
ratio = ratios[idx]
dwdh = dwdhs[idx]
output = outputs[idx]
result_frame = draw(frame, output, ratio, dwdh, self.label_names, self.colors)
results.append(result_frame)
return results
上記の例では、infer_batch
メソッドが追加されています。このメソッドは、複数のフレームをまとめて推論するバッチ処理を行います。frames
パラメータには複数のフレームがリストとして渡されます。
バッチサイズは self.batch_size
で設定されており、バッチ処理のためにフレームを適切なサイズで分割しています。各バッチに対して推論を行い、結果を results
リストに格納しています。
このようにバッチ処
理を行うことで、複数のフレームをまとめて効率的に推論できます。ただし、バッチサイズを大きくしすぎるとメモリ使用量が増えるため、適切なバッチサイズを選択する必要があります。また、バッチ処理には一定のオーバーヘッドがありますが、フレーム数が増えるほど効果が現れます。最適なバッチサイズは、環境やモデルの特性によって異なる場合があるため、試行錯誤して調整することをおすすめします。
はい、非同期処理を使用して画像の表示処理を軽くすることができます。Tkinterでは非同期処理を実現するためのモジュールとしてasyncio
を使用することが一般的です。以下に修正例を示します。
import tkinter as tk
import asyncio
from tkinter import Tk, ttk
import cv2
import numpy as np
from PIL import Image, ImageOps, ImageTk # 画像データ用
from function.infer import Model
class ImageRecognition(ttk.Frame):
"""カメラ映像と認識後画像を扱うカメラ"""
def __init__(self, master: Tk | None = None):
super().__init__(master)
if master is not None:
self.master: Tk = master
# ウィンドウのサイズ
window_width: int = self.master.winfo_width()
print(f"ウィンドウ 幅:{window_width}")
window_height: int = self.master.winfo_height()
print(f"ウィンドウ 高さ:{window_height}")
# キャンバスのサイズ
canvas_width: int = window_width // 2
canvas_height: int = window_height
# カメラ映像のキャンバス
self.camera_image_canvas: tk.Canvas = tk.Canvas(
self.master, width=canvas_width, height=canvas_height
)
self.camera_image_canvas.pack(side="left", expand=True, fill="both")
# 認識後画像のキャンバス
self.infer_image_canvas: tk.Canvas = tk.Canvas(
self.master, width=canvas_width, height=canvas_height
)
self.infer_image_canvas.pack(side="right", expand=True, fill="both")
# カメラを起動する
self.capture: cv2.VideoCapture = cv2.VideoCapture(0)
# self.display_image()
self.camera_image_canvas_display_id: str = ""
self.infer_image_canvas_display_id: str = ""
self.model: Model = Model("Yolo v7", "sugarcane", "CPUExecutionProvider")
# 非同期処理用のイベントループを作成
self.loop = asyncio.get_event_loop()
# バックグラウンドタスクを実行
self.loop.create_task(self.display_image())
async def display_image(self):
"""画像をCanvasに非同期に表示する"""
while True:
# フレーム画像を取得
is_success, frame = self.capture.read()
# print(frame)
# BGRからRGBへ変換
cv_image: np.ndarray = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
# NumpyのndarrayからPillowのImageへ変換
pil_image = Image.fromarray(cv_image)
# Canvasのサイズを取得
canvas_width: int = self.camera_image_canvas.winfo_width()
canvas_height: int = self.camera_image_canvas.winfo_height()
# 画像のアスペクト比(縦横比)を崩さずに指定したサイズ(キャンバスのサイ
ズ)全体に画像をリサイズする
pil_image = ImageOps.pad(pil_image, (canvas_width, canvas_height))
# PIL.ImageからPhotoImageへ変換する
self.photo_image = ImageTk.PhotoImage(image=pil_image)
# 画像の描画
# カメラ映像のcanvasに画像を描画
self.camera_image_canvas.create_image(
canvas_width / 2, # 画像表示位置(Canvasの中心)
canvas_height / 2,
image=self.photo_image, # 表示画像データ
)
# 推論処理
if self.model.is_batch_ready():
infer_results = self.model.infer()
await self.display_infer_image(infer_results)
# 少し待機
await asyncio.sleep(0.01)
async def display_infer_image(self, infer_results):
"""認識後画像を非同期に表示する"""
for infer_frame in infer_results:
# BGRからRGBへ変換
cv_image2: np.ndarray = cv2.cvtColor(infer_frame, cv2.COLOR_BGR2RGB)
# NumpyのndarrayからPillowのImageへ変換
pil_image2 = Image.fromarray(cv_image2)
# Canvasのサイズを取得
canvas_width: int = self.infer_image_canvas.winfo_width()
canvas_height: int = self.infer_image_canvas.winfo_height()
# 画像のアスペクト比(縦横比)を崩さずに指定したサイズ(キャンバスのサイズ)全体に画像をリサイズする
pil_image2 = ImageOps.pad(pil_image2, (canvas_width, canvas_height))
# PIL.ImageからPhotoImageへ変換する
self.photo_image2 = ImageTk.PhotoImage(image=pil_image2)
# 認識後画像のcanvasに画像を描画
self.infer_image_canvas.create_image(
canvas_width / 2,
canvas_height / 2,
image=self.photo_image2,
)
def display_exit(self):
"""アプリを終了する"""
self.loop.stop()
self.master.destroy()
修正点の概要は以下の通りです。
display_image()
メソッドとdisplay_infer_image()
メソッドを非同期関数に変更し、asyncio
モジュールを使用して非同期処理を実現します。display_image()
メソッド内で、フレーム画像の取得とカメラ映像の表示を継続的に行うためにwhile
ループを使用します。また、推論処理の実行と認識後画像の表示は非同期関数display_infer_image()
を呼び出すことで行います。display_image()
メソッド内でawait asyncio.sleep(0.01)
を挿入し、少しの間隔を置いて非同期処理を継続することで、ループが無駄に高速になるのを防ぎます。
__init__()
メソッド内でasyncio.get_event_loop()
を使用してイベントループを取得し、loop.create_task()
を使用してdisplay_image()
メソッドを非同期に実行します。display_exit()
メソッド内でイベントループの停止とウィンドウの破棄を行います。これにより、画像の表示と推論処理が非同期に実行されるため、処理が軽くなります。
ChatGPT