DepthAnythingV2 と OpenVINO による深度推定#

from PIL import Image 

import requests 

r = requests.get( 

url="https://raw.githubusercontent.com/openvinotoolkit/openvino_notebooks/latest/utils/notebook_utils.py", 
) 

open("notebook_utils.py", "w").write(r.text) 
from notebook_utils import download_file 

download_file( 
    "https://github.com/openvinotoolkit/openvino_notebooks/assets/29454499/3f779fc1-c1b2-4dec-915a-64dae510a2bb", 
    "furseal.png", 
) 

Image.open("furseal.png").resize((600, 400))

from huggingface_hub import hf_hub_download 

encoder = "vits" 
model_type = "Small" 
model_id = f"depth_anything_v2_{encoder}" 

model_path = hf_hub_download(repo_id=f"depth-anything/Depth-Anything-V2-{model_type}", filename=f"{model_id}.pth", repo_type="model")

import cv2 
import torch 
import torch.nn.functional as F 

from depth_anything_v2.dpt import DepthAnythingV2 

model = DepthAnythingV2(encoder=encoder, features=64, out_channels=[48, 96, 192, 384]) 
model.load_state_dict(torch.load(model_path, map_location="cpu")) 
model.eval() 

raw_img = cv2.imread("furseal.png") 
image, (h, w) = model.image2tensor(raw_img) 
image = image.to("cpu").to(torch.float32) 

with torch.no_grad(): 
    depth = model.forward(image) 

depth = F.interpolate(depth[:, None], (h, w), mode="bilinear", align_corners=True)[0, 0] 

output = depth.cpu().numpy()

from matplotlib import pyplot as plt 
import numpy as np 
import cv2 

def get_depth_map(output, w, h): 
    depth = cv2.resize(output, (w, h)) 

    depth = (depth - depth.min()) / (depth.max() - depth.min()) * 255.0 
    depth = depth.astype(np.uint8) 

    depth = cv2.applyColorMap(depth, cv2.COLORMAP_INFERNO) 

    return depth

h, w = raw_img.shape[:-1] 
res_depth = get_depth_map(output, w, h) 
plt.imshow(res_depth[:, :, ::-1])

import ipywidgets as widgets 

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

device

compiled_model = core.compile_model(OV_DEPTH_ANYTHING_PATH, device.value)

from depth_anything_v2.util.transform import Resize, NormalizeImage, PrepareForNet 
from torchvision.transforms import Compose 

transform = Compose( 
    [ 
        Resize( 
            width=518, 
            height=518, 
            resize_target=False, 
            ensure_multiple_of=14, 
            resize_method="lower_bound", 
            image_interpolation_method=cv2.INTER_CUBIC, 
        ), 
        NormalizeImage(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]), 
        PrepareForNet(), 
    ] 
)

h, w = raw_img.shape[:-1] 

image = cv2.cvtColor(raw_img, cv2.COLOR_BGR2RGB) / 255.0 
image = transform({"image": image})["image"] 
image = torch.from_numpy(image).unsqueeze(0) 

res = compiled_model(image)[0]

depth_color = get_depth_map(res[0], w, h)

plt.imshow(depth_color[:, :, ::-1])

download_file( 

"https://storage.openvinotoolkit.org/repositories/openvino_notebooks/data/data/video/Coco%20Walking%20in%20Berkeley.mp4", 
    "./Coco Walking in Berkeley.mp4", 
) 

VIDEO_FILE = "./Coco Walking in Berkeley.mp4" 
# 処理する入力ビデオの秒数。ビデオ全体を処理するには、 
# `NUM_SECONDS` を 0 に設定
NUM_SECONDS = 4 
# 入力ビデオのすべてのフレームを処理するには、`ADVANCE_FRAMES` を 1 に設定 
# 1 フレームごとに処理するには、`ADVANCE_FRAMES` を 2 に設定。これにより、 
# ビデオの処理にかかる時間が短縮されます。
ADVANCE_FRAMES = 2 
# 結果のビデオのサイズを縮小するには、`SCALE_OUTPUT` を設定 
# `SCALE_OUTPUT` が 0.5 の場合、結果ビデオの幅と高さは 
# 入力ビデオの幅と高さの半分になります。
SCALE_OUTPUT = 0.5 
# ビデオのエンコードに使用する形式。'vp09` は低速ですが、 
# ほとんどのシステムで動作します# FFMPEG がインストールされている場合は、`THEO` エンコードを試してください。
# FOURCC = cv2.VideoWriter_fourcc(*"THEO") 
FOURCC = cv2.VideoWriter_fourcc(*"vp09") 

# 入力ビデオと結果ビデオの Path オブジェクトを作成 
output_directory = Path("output") 
output_directory.mkdir(exist_ok=True) 
result_video_path = output_directory / f"{Path(VIDEO_FILE).stem}_depth_anything.mp4"

cap = cv2.VideoCapture(str(VIDEO_FILE)) 
ret, image = cap.read() 
if not ret: 
    raise ValueError(f"The video at {VIDEO_FILE} cannot be read.") 
input_fps = cap.get(cv2.CAP_PROP_FPS) 
input_video_frame_height, input_video_frame_width = image.shape[:2] 

target_fps = input_fps / ADVANCE_FRAMES 
target_frame_height = int(input_video_frame_height * SCALE_OUTPUT) 
target_frame_width = int(input_video_frame_width * SCALE_OUTPUT) 

cap.release() 
print(f"The input video has a frame width of {input_video_frame_width}, " f"frame height of {input_video_frame_height} and runs at {input_fps:.2f} fps") 
print( 
    "The output video will be scaled with a factor " 
    f"{SCALE_OUTPUT}, have width {target_frame_width}, " 
    f" height {target_frame_height}, and run at {target_fps:.2f} fps" 
)

def normalize_minmax(data): 
    """Normalizes the values in `data` between 0 and 1""" 
    return (data - data.min()) / (data.max() - data.min()) 

def convert_result_to_image(result, colormap="viridis"): 
    """ 
    Convert network result of floating point numbers to an RGB image with 
    integer values from 0-255 by applying a colormap.`result` is expected to be a single network result in 1,H,W shape 
    `colormap` is a matplotlib colormap.See https://matplotlib.org/stable/tutorials/colors/colormaps.html 
    """ 

    result = result.squeeze(0) 
    result = normalize_minmax(result) 
    result = result * 255 
    result = result.astype(np.uint8) 
    result = cv2.applyColorMap(result, cv2.COLORMAP_INFERNO)[:, :, ::-1] 
    return result 

def to_rgb(image_data) -> np.ndarray: 
    """ 
    Convert image_data from BGR to RGB 
    """ 
    return cv2.cvtColor(image_data, cv2.COLOR_BGR2RGB)

import time 
from IPython.display import ( 
    HTML, 
    FileLink, 
    Pretty, 
    ProgressBar,
    Video, 
    clear_output, 
    display, 
) 

def process_video(compiled_model, video_file, result_video_path):
    # 変数を初期化 
    input_video_frame_nr = 0 
    start_time = time.perf_counter() 
    total_inference_duration = 0 

    # 入力ビデオを開く 
    cap = cv2.VideoCapture(str(video_file)) 

    # 結果ビデオを作成 
    out_video = cv2.VideoWriter( 
        str(result_video_path), 
        FOURCC, 
        target_fps, 
        (target_frame_width * 2, target_frame_height),
    ) 

    num_frames = int(NUM_SECONDS * input_fps) 
    total_frames = cap.get(cv2.CAP_PROP_FRAME_COUNT) if num_frames == 0 else num_frames 
    progress_bar = ProgressBar(total=total_frames) 
    progress_bar.display() 

    try: 
        while cap.isOpened(): 
            ret, image = cap.read() 
            if not ret: 
                cap.release() 
                break 

            if input_video_frame_nr >= total_frames: 
                break 

            h, w = image.shape[:-1] 
            input_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) / 255.0 
            input_image = transform({"image": input_image})["image"] 
            # 画像をネットワーク入力形状 NCHW に合わせて再形成 
            input_image = np.expand_dims(input_image, 0) 

            # 推論を実行 
            inference_start_time = time.perf_counter() 
            result = compiled_model(input_image)[0] 
            inference_stop_time = time.perf_counter() 
            inference_duration = inference_stop_time - inference_start_time 
            total_inference_duration += inference_duration 

            if input_video_frame_nr % (10 * ADVANCE_FRAMES) == 0: 
                clear_output(wait=True) 
                progress_bar.display() 
                # input_video_frame_nr // ADVANCE_FRAMES は、 
                # ネットワークによって処理されたフレームの数を示します 
                display( 
                    Pretty( 
                        f"Processed frame {input_video_frame_nr // ADVANCE_FRAMES}" 
                        f"/{total_frames // ADVANCE_FRAMES}." 
                        f"Inference time per frame: {inference_duration:.2f} seconds " 
                        f"({1/inference_duration:.2f} FPS)" 
                    ) 
                ) 
            # ネットワーク結果を RGB 画像に変換 
            result_frame = to_rgb(convert_result_to_image(result)) 
            # 画像と結果をターゲットのフレーム形状に合わせてサイズを変更 
            result_frame = cv2.resize(result_frame, (target_frame_width, target_frame_height)) 
            image = cv2.resize(image, (target_frame_width, target_frame_height)) 
            # 画像と結果を並べて表示 
            stacked_frame = np.hstack((image, result_frame)) 
            # ビデオにフレームを保存 
            out_video.write(stacked_frame) 

            input_video_frame_nr = input_video_frame_nr + ADVANCE_FRAMES 
            cap.set(1, input_video_frame_nr) 

            progress_bar.progress = input_video_frame_nr 
            progress_bar.update() 

    except KeyboardInterrupt: 
        print("Processing interrupted.") 
    finally: 
        clear_output() 
        processed_frames = num_frames // ADVANCE_FRAMES 
        out_video.release() 
        cap.release() 
        end_time = time.perf_counter() 
        duration = end_time - start_time 

        print( 
            f"Processed {processed_frames} frames in {duration:.2f} seconds." 
            f"Total FPS (including video processing): {processed_frames/duration:.2f}." 
            f"Inference FPS: {processed_frames/total_inference_duration:.2f} " 
        ) 
        print(f"Video saved to '{str(result_video_path)}'.") 
    return stacked_frame

stacked_frame = process_video(compiled_model, VIDEO_FILE, result_video_path)

def display_video(stacked_frame): 
    video = Video(result_video_path, width=800, embed=True) 
    if not result_video_path.exists(): 
        plt.imshow(stacked_frame) 
        raise ValueError("OpenCV was unable to write the video file.Showing one video frame.") 
    else: 
        print(f"Showing video saved at\n{result_video_path.resolve()}") 
        print("If you cannot see the video in your browser, please click on the " "following link to download the video ") 
        video_link = FileLink(result_video_path) 
        video_link.html_link_str = "<a href='%s' download>%s</a>" 
        display(HTML(video_link._repr_html_())) 
        display(video)

display_video(stacked_frame)

%%skip not $to_quantize.value 

import datasets 

if not OV_DEPTH_ANYTHING_INT8_PATH.exists(): 
    subset_size = 300 
    calibration_data = [] 
    dataset = datasets.load_dataset("Nahrawy/VIDIT-Depth-ControlNet", split="train", streaming=True).shuffle(seed=42).take(subset_size) 
    for batch in dataset: 
        image = np.array(batch["image"])[...,:3] 
        image = image / 255.0 
        image = transform({'image': image})['image'] 
        image = np.expand_dims(image, 0) 
        calibration_data.append(image)

%%skip not $to_quantize.value 

import nncf 

if not OV_DEPTH_ANYTHING_INT8_PATH.exists(): 
    model = core.read_model(OV_DEPTH_ANYTHING_PATH) 
    quantized_model = nncf.quantize( 
        model=model, 
        subset_size=subset_size, 
        model_type=nncf.ModelType.TRANSFORMER, 
        calibration_dataset=nncf.Dataset(calibration_data), 
    ) 
    ov.save_model(quantized_model, OV_DEPTH_ANYTHING_INT8_PATH)

%%skip not $to_quantize.value 

def visualize_results(orig_img:Image.Image, optimized_img:Image.Image):
    """ 
    Helper function for results visualization 

    Parameters: 
        orig_img (Image.Image): generated image using FP16 model 
        optimized_img (Image.Image): generated image using quantized model 
    Returns: 
        fig (matplotlib.pyplot.Figure): matplotlib generated figure contains drawing result 
    """ 
    orig_title = "FP16 model" 
    control_title = "INT8 model" 
    figsize = (20, 20) 
    fig, axs = plt.subplots(1, 2, figsize=figsize, sharex='all', sharey='all') 
    list_axes = list(axs.flat) 
    for a in list_axes: 
        a.set_xticklabels([]) 
        a.set_yticklabels([]) 
        a.get_xaxis().set_visible(False) 
        a.get_yaxis().set_visible(False) 
        a.grid(False) 
    list_axes[0].imshow(np.array(orig_img)) 
    list_axes[1].imshow(np.array(optimized_img)) 
    list_axes[0].set_title(orig_title, fontsize=15) 
    list_axes[1].set_title(control_title, fontsize=15) 

    fig.subplots_adjust(wspace=0.01, hspace=0.01) 
    fig.tight_layout() 
    return fig

%%skip not $to_quantize.value 

image = cv2.cvtColor(cv2.imread('furseal.png'), cv2.COLOR_BGR2RGB) / 255.0 
image = transform({'image': image})['image'] 
image = torch.from_numpy(image).unsqueeze(0) 

int8_compiled_model = core.compile_model(OV_DEPTH_ANYTHING_INT8_PATH, device.value) 
int8_res = int8_compiled_model(image)[0] 
int8_depth_color = get_depth_map(int8_res[0], w, h)

%%skip not $to_quantize.value 

visualize_results(depth_color[:, :, ::-1], int8_depth_color[:, :, ::-1])

%%skip not $to_quantize.value 

int8_result_video_path = output_directory / f"{Path(VIDEO_FILE).stem}_depth_anything_int8.mp4" 
stacked_frame = process_video(int8_compiled_model, VIDEO_FILE, int8_result_video_path) 
display_video(stacked_frame)

import re 

def get_fps(benchmark_output: str): 
    parsed_output = [line for line in benchmark_output if "Throughput:" in line] 
    fps = re.findall(r"\d+\.\d+", parsed_output[0])[0] 
    return fps 

if OV_DEPTH_ANYTHING_INT8_PATH.exists(): 
    benchmark_output = !benchmark_app -m $OV_DEPTH_ANYTHING_PATH -d $device.value -api async 
    original_fps = get_fps(benchmark_output) 
    print(f"FP16 Throughput: {original_fps} FPS") 

    benchmark_output = !benchmark_app -m $OV_DEPTH_ANYTHING_INT8_PATH -d $device.value -api async 
    optimized_fps = get_fps(benchmark_output) 
    print(f"INT8 Throughput: {optimized_fps} FPS") 
    print(f"Speed-up: {float(optimized_fps) / float(original_fps):.2f}")