OpenVINO™ を使用した YOLOv8 リアルタイム物体検出の変換と最適化#

models_dir = Path("./models") 
models_dir.mkdir(exist_ok=True)

from PIL import Image 
from ultralytics import YOLO 

DET_MODEL_NAME = "yolov8n" 

det_model = YOLO(models_dir / f"{DET_MODEL_NAME}.pt") 
label_map = det_model.model.names 

res = det_model(IMAGE_PATH) 
Image.fromarray(res[0].plot()[:, :, ::-1])

# 物体検出モデル 
det_model_path = models_dir / f"{DET_MODEL_NAME}_openvino_model/{DET_MODEL_NAME}.xml" 
if not det_model_path.exists(): 
    det_model.export(format="openvino", dynamic=True, half=True)

import ipywidgets as widgets 
import openvino as ov 

core = ov.Core() 

device = widgets.Dropdown( 
    options=core.available_devices + ["AUTO"], 
    value="AUTO", 
    description="Device:", 
    disabled=False, 
) 

device

import torch 

core = ov.Core() 

det_ov_model = core.read_model(det_model_path) 

ov_config = {} 
if device.value != "CPU": 
    det_ov_model.reshape({0: [1, 3, 640, 640]}) 
if "GPU" in device.value or ("AUTO" in device.value and "GPU" in core.available_devices): 
    ov_config = {"GPU_DISABLE_WINOGRAD_CONVOLUTION": "YES"} 
det_compiled_model = core.compile_model(det_ov_model, device.value, ov_config) 

def infer(*args):        
    result = det_compiled_model(args) 
    return torch.from_numpy(result[0]) 

det_model.predictor.inference = infer 
det_model.predictor.model.pt = False 

res = det_model(IMAGE_PATH) 
Image.fromarray(res[0].plot()[:, :, ::-1])

from zipfile import ZipFile 

from ultralytics.data.utils import DATASETS_DIR 

DATA_URL = "http://images.cocodataset.org/zips/val2017.zip" 
LABELS_URL = "https://github.com/ultralytics/yolov5/releases/download/v1.0/coco2017labels-segments.zip" 
CFG_URL = 
"https://raw.githubusercontent.com/ultralytics/ultralytics/v8.1.0/ultralytics/cfg/datasets/coco.yaml" 

OUT_DIR = DATASETS_DIR 

DATA_PATH = OUT_DIR / "val2017.zip" 
LABELS_PATH = OUT_DIR / "coco2017labels-segments.zip" 
CFG_PATH = OUT_DIR / "coco.yaml" 

download_file(DATA_URL, DATA_PATH.name, DATA_PATH.parent) 
download_file(LABELS_URL, LABELS_PATH.name, LABELS_PATH.parent) 
download_file(CFG_URL, CFG_PATH.name, CFG_PATH.parent) 

if not (OUT_DIR / "coco/labels").exists(): 
    with ZipFile(LABELS_PATH, "r") as zip_ref: 
        zip_ref.extractall(OUT_DIR) 
    with ZipFile(DATA_PATH, "r") as zip_ref: 
        zip_ref.extractall(OUT_DIR / "coco/images")

import numpy as np
 
from tqdm.notebook import tqdm 
from ultralytics.utils.metrics import ConfusionMatrix 

def test( 
    model: ov.Model, 
    core: ov.Core, 
    data_loader: torch.utils.data.DataLoader, 
    validator, 
    num_samples: int = None, 
): 
    """ 
    OpenVINO YOLOv8 model accuracy validation function.Runs model validation on dataset and returns metrics 
    Parameters: 
        model (Model): OpenVINO model 
        data_loader (torch.utils.data.DataLoader): dataset loader 
        validator: instance of validator class 
        num_samples (int, *optional*, None): validate model only on specified number samples, if provided 
    Returns: 
        stats: (Dict[str, float]) - dictionary with aggregated accuracy metrics statistics, key is metric name, value is metric value 
    """ 
    validator.seen = 0 
    validator.jdict = [] 
    validator.stats = dict(tp=[], conf=[], pred_cls=[], target_cls=[]) 
    validator.batch_i = 1 
    validator.confusion_matrix = ConfusionMatrix(nc=validator.nc) 
    model.reshape({0: [1, 3, -1, -1]}) 
    compiled_model = core.compile_model(model) 
    for batch_i, batch in enumerate(tqdm(data_loader, total=num_samples)): 
        if num_samples is not None and batch_i == num_samples: 
            break 
        batch = validator.preprocess(batch) 
        results = compiled_model(batch["img"]) 
        preds = torch.from_numpy(results[compiled_model.output(0)]) 
        preds = validator.postprocess(preds) 
        validator.update_metrics(preds, batch) 
    stats = validator.get_stats() 
    return stats 

def print_stats(stats: np.ndarray, total_images: int, total_objects: int): 
    """ 
    Helper function for printing accuracy statistic 
    Parameters: 
        stats: (Dict[str, float]) - dictionary with aggregated accuracy metrics statistics, key is metric name, value is metric value 
        total_images (int) - number of evaluated images 
        total objects (int) 
    Returns:         None 
    """ 
    print("Boxes:") 
    mp, mr, map50, mean_ap = ( 
        stats["metrics/precision(B)"], 
        stats["metrics/recall(B)"], 
        stats["metrics/mAP50(B)"], 
        stats["metrics/mAP50-95(B)"], 
    ) 
    # 結果をプリント 
    print(" Best mean average:") 
    s = ("%20s" + "%12s" * 6) % ( 
        "Class", 
        "Images", 
        "Labels", 
        "Precision", 
        "Recall", 
        "mAP@.5", 
        "mAP@.5:.95", 
    ) 
    print(s) 
    pf = "%20s" + "%12i" * 2 + "%12.3g" * 4 # print format 
    print(pf % ("all", total_images, total_objects, mp, mr, map50, mean_ap)) 
    if "metrics/precision(M)" in stats: 
        s_mp, s_mr, s_map50, s_mean_ap = ( 
            stats["metrics/precision(M)"], 
            stats["metrics/recall(M)"], 
            stats["metrics/mAP50(M)"], 
            stats["metrics/mAP50-95(M)"],
        ) 
        # 結果をプリント 
        print(" Macro average mean:") 
        s = ("%20s" + "%12s" * 6) % ( 
            "Class", 
            "Images", 
            "Labels", 
            "Precision", 
            "Recall", 
            "mAP@.5", 
            "mAP@.5:.95", 
        ) 
        print(s) 
        pf = "%20s" + "%12i" * 2 + "%12.3g" * 4 # print format 
        print(pf % ("all", total_images, total_objects, s_mp, s_mr, s_map50, s_mean_ap))

from ultralytics.utils import DEFAULT_CFG 
from ultralytics.cfg import get_cfg 
from ultralytics.data.converter import coco80_to_coco91_class 
from ultralytics.data.utils import check_det_dataset 

args = get_cfg(cfg=DEFAULT_CFG) 
args.data = str(CFG_PATH)

det_validator = det_model.task_map[det_model.task]["validator"](args=args)

det_validator.data = check_det_dataset(args.data) 
det_validator.stride = 32 
det_data_loader = det_validator.get_dataloader(OUT_DIR / "coco", 1)

det_validator.is_coco = True 
det_validator.class_map = coco80_to_coco91_class() 
det_validator.names = det_model.model.names 
det_validator.metrics.names = det_validator.names 
det_validator.nc = det_model.model.model[-1].nc

NUM_TEST_SAMPLES = 300

fp_det_stats = test(det_ov_model, core, det_data_loader, det_validator, num_samples=NUM_TEST_SAMPLES)

print_stats(fp_det_stats, det_validator.seen, det_validator.nt_per_class.sum())

import ipywidgets as widgets 

int8_model_det_path = models_dir / f"{DET_MODEL_NAME}_openvino_int8_model/{DET_MODEL_NAME}.xml" 

to_quantize = widgets.Checkbox( 
    value=True, 
    description="Quantization", 
    disabled=False, 
) 

to_quantize

# skip_kernel_extension モジュールを取得 
r = requests.get( 

url="https://raw.githubusercontent.com/openvinotoolkit/openvino_notebooks/latest/utils/skip_kernel_extension.py", 
) 
open("skip_kernel_extension.py", "w").write(r.text) 

%load_ext skip_kernel_extension

%%skip not $to_quantize.value 

import nncf 
from typing import Dict 

def transform_fn(data_item:Dict):
    """ 
    Quantization transform function.Extracts and preprocess input data from dataloader item for quantization.
    Parameters: 
        data_item: Dict with data item produced by DataLoader during iteration 
    Returns: 
        input_tensor: Input data for quantization 
    """ 
    input_tensor = det_validator.preprocess(data_item)['img'].numpy() 
    return input_tensor 

quantization_dataset = nncf.Dataset(det_data_loader, transform_fn)

%%skip not $to_quantize.value 

ignored_scope = nncf.IgnoredScope( 
    names=[ 
        "__module.model.22.cv3.0.0.conv/aten::_convolution/Convolution", # 後処理サブグラフ内 
            "__module.model.16.conv/aten::_convolution/Convolution", 
            "__module.model.22.cv2.0.0.conv/aten::_convolution/Convolution", 
            "__module.model.6.cv1.conv/aten::_convolution/Convolution", 
            "__module.model.22.cv3.1.1.conv/aten::_convolution/Convolution", 
            "__module.model.21.cv2.conv/aten::_convolution/Convolution", 
            "__module.model.21.m.0.cv1.conv/aten::_convolution/Convolution", 
            "__module.model.22/aten::add/Add_6", 
            "__module.model.22/aten::sub/Subtract", 
            "__module.model.7.conv/aten::_convolution/Convolution", 
            "__module.model.12.cv1.conv/aten::_convolution/Convolution", 
            "__module.model.4.cv1.conv/aten::_convolution/Convolution", 
            "__module.model.22.cv2.2.1.conv/aten::_convolution/Convolution", 
            "__module.model.22.cv2.0.1.conv/aten::_convolution/Convolution", 
            "__module.model.22.dfl.conv/aten::_convolution/Convolution", 
            "__module.model.22.cv3.2.2/aten::_convolution/Convolution", 
            "__module.model.22.cv3.0.2/aten::_convolution/Convolution", 
            "__module.model.15.cv1.conv/aten::_convolution/Convolution", 
            "__module.model.5.conv/aten::_convolution/Convolution", 
            "__module.model.0.conv/aten::_convolution/Convolution" 
    ] 
) 

# 検出モデル 
quantized_det_model = nncf.quantize( 
    det_ov_model, 
    quantization_dataset, 
    preset=nncf.QuantizationPreset.MIXED, 
    ignored_scope=ignored_scope 
)

%%skip not $to_quantize.value 

print(f"Quantized detection model will be saved to {int8_model_det_path}") 
ov.save_model(quantized_det_model, str(int8_model_det_path))

%%skip not $to_quantize.value 

device

%%skip not $to_quantize.value 

ov_config = {} 

if device.value != "CPU": 
    quantized_det_model.reshape({0: [1, 3, 640, 640]}) 
if "GPU" in device.value or ("AUTO" in device.value and "GPU" in core.available_devices): 
    ov_config = {"GPU_DISABLE_WINOGRAD_CONVOLUTION": "YES"} 
quantized_det_compiled_model = core.compile_model(quantized_det_model, device.value, ov_config) 

def infer(*args): 
    result = quantized_det_compiled_model(args) 
    return torch.from_numpy(result[0]) 

det_model.predictor.inference = infer 

res = det_model(IMAGE_PATH) 
display(Image.fromarray(res[0].plot()[:, :, ::-1]))

%%skip not $to_quantize.value 

device

if int8_model_det_path.exists():     # 推論 FP32 モデル model (OpenVINO IR) 
    !benchmark_app -m $det_model_path -d $device.value -api async -shape "[1,3,640,640]"

if int8_model_det_path.exists():
    # 推論 INT8 モデル (OpenVINO IR) 
    !benchmark_app -m $int8_model_det_path -d $device.value -api async -shape "[1,3,640,640]" -t 15

%%skip not $to_quantize.value 

int8_det_stats = test(quantized_det_model, core, det_data_loader, det_validator, num_samples=NUM_TEST_SAMPLES)

%%skip not $to_quantize.value 

print("FP32 model accuracy") 
print_stats(fp_det_stats, det_validator.seen, det_validator.nt_per_class.sum()) 

print("INT8 model accuracy") 
print_stats(int8_det_stats, det_validator.seen, det_validator.nt_per_class.sum())

import collections 
import time 
from IPython import display 

# 物体検出を実行するメイン処理関数 
def run_object_detection( 
    source=0, 
    flip=False, 
    use_popup=False, 
    skip_first_frames=0, 
    model=det_model, 
    device=device.value, 
): 
    player = None 
    ov_config = {} 
    if device != "CPU": 
        model.reshape({0: [1, 3, 640, 640]}) 
    if "GPU" in device or ("AUTO" in device and "GPU" in core.available_devices): 
        ov_config = {"GPU_DISABLE_WINOGRAD_CONVOLUTION": "YES"} 
    compiled_model = core.compile_model(model, device, ov_config) 

    def infer(*args): 
        result = compiled_model(args) 
        return torch.from_numpy(result[0]) 

    det_model.predictor.inference = infer 

    try: # ターゲット fps で再生するビデオプレーヤーを作成 
        player = VideoPlayer(source=source, flip=flip, fps=30, skip_first_frames=skip_first_frames) 
        # キャプチャーを開始 
        player.start() 
        if use_popup: 
            title = "Press ESC to Exit" 
            cv2.namedWindow(winname=title, flags=cv2.WINDOW_GUI_NORMAL | cv2.WINDOW_AUTOSIZE) 

        processing_times = collections.deque() 
        while True:
            # フレームをグラブ 
            frame = player.next() 
            if frame is None: 
                print("Source ended") 
                break 
            # フレームがフル HD より大きい場合は、サイズを縮小してパフォーマンスを向上させます 
            scale = 1280 / max(frame.shape) 
            if scale < 1: 
                frame = cv2.resize( 
                    src=frame, 
                    dsize=None, 
                    fx=scale, 
                    fy=scale, 
                    interpolation=cv2.INTER_AREA, 
                ) 
            # 結果を取得 
            input_image = np.array(frame) 

            start_time = time.time() 
            detections = det_model(input_image) 
            stop_time = time.time() 
            frame = detections[0].plot() 

            processing_times.append(stop_time - start_time) 
            # 最後の 200 フレームの処理時間を使用 
            if len(processing_times) > 200: 
                processing_times.popleft() 

            _, f_width = frame.shape[:2] 
            # 平均処理時間 [ms] 
            processing_time = np.mean(processing_times) * 1000 
            fps = 1000 / processing_time 
            cv2.putText( 
                img=frame, 
                text=f"Inference time: {processing_time:.1f}ms ({fps:.1f} FPS)", 
                org=(20, 40), 
                fontFace=cv2.FONT_HERSHEY_COMPLEX, 
                fontScale=f_width / 1000, 
                color=(0, 0, 255), 
                thickness=1, 
                lineType=cv2.LINE_AA, 
            )
 
            # ちらつきがある場合はこの回避策を使用 
            if use_popup: 
                cv2.imshow(winname=title, mat=frame) 
                key = cv2.waitKey(1) 
                # escape = 27 
                if key == 27: 
                    break 
                else: 
                    # numpy 配列を jpg にエンコード
                    _, encoded_img = cv2.imencode(ext=".jpg", img=frame, params=[cv2.IMWRITE_JPEG_QUALITY, 100]) 
                    # IPython イメージを作成 
                    i = display.Image(data=encoded_img) 
                    # このノートブックに画像を表示 
                    display.clear_output(wait=True) 
                    display.display(i) 

# ctrl-c 
except KeyboardInterrupt: 
    print("Interrupted") 
# 異なるエラー 
except RuntimeError as e: 
    print(e) 
finally: 
    if player is not None: 
        # キャプチャーを停止 
        Player.stop().stop() 
    if use_popup: 
        cv2.destroyAllWindows()

WEBCAM_INFERENCE = False 

if WEBCAM_INFERENCE: 
    VIDEO_SOURCE = 0 # ウェブカメラ 
else: 
    VIDEO_SOURCE = "https://storage.openvinotoolkit.org/repositories/openvino_notebooks/data/data/video/people.mp4"

device

run_object_detection( 
    source=VIDEO_SOURCE, 
    flip=True, 
    use_popup=False, 
    model=det_ov_model, 
    device=device.value, 
)