|
| 1 | +# Copyright (c) 2023 Graphcore Ltd. All rights reserved. |
| 2 | +import yacs |
| 3 | +import torch |
| 4 | +import poptorch |
| 5 | +import numpy as np |
| 6 | +from PIL import Image |
| 7 | +from ruamel import yaml |
| 8 | +from run import ipu_options |
| 9 | +from typing import NamedTuple |
| 10 | +from typing import List, Tuple |
| 11 | +from poptorch import inferenceModel |
| 12 | +from models.yolov4_p5 import Yolov4P5 |
| 13 | +from utils.config import get_cfg_defaults |
| 14 | +from torchvision.transforms import Compose |
| 15 | +from utils.visualization import plotting_tool |
| 16 | +from utils.postprocessing import post_processing |
| 17 | +from utils.tools import load_and_fuse_pretrained_weights |
| 18 | +from utils.preprocessing import ResizeImage, Pad, ToNumpy, ToTensor |
| 19 | + |
| 20 | + |
| 21 | +class YoloOutput(NamedTuple): |
| 22 | + x_top_left: float |
| 23 | + y_top_left: float |
| 24 | + width: float |
| 25 | + height: float |
| 26 | + detection_confidence: float |
| 27 | + detected_class_id: float |
| 28 | + detected_class_name: str |
| 29 | + |
| 30 | + |
| 31 | +class YOLOv4InferencePipeline: |
| 32 | + def __init__(self, checkpoint_path: str): |
| 33 | + self.cfg = get_cfg_defaults() |
| 34 | + self.config = "configs/inference-yolov4p5.yaml" |
| 35 | + self.cfg.merge_from_file(self.config) |
| 36 | + self.cfg.freeze() |
| 37 | + self.checkpoint = checkpoint_path |
| 38 | + self.model = self.prepare_inference_yolo_v4_model_for_ipu() |
| 39 | + self.class_names = yaml.safe_load(open("configs/class_name.yaml"))["class_names"] |
| 40 | + |
| 41 | + def __call__(self, image: Image) -> Tuple[List[YoloOutput], List[np.array]]: |
| 42 | + transformed_image, size = self.preprocess_image(image) |
| 43 | + y = self.model(transformed_image) |
| 44 | + dummy_label = [] |
| 45 | + processed_batch = post_processing(self.cfg, y, [size], dummy_label) |
| 46 | + batch_as_list = processed_batch[0][0].tolist() |
| 47 | + named_output = [ |
| 48 | + YoloOutput._make(obj[0:5] + [obj[-1]] + [self.class_names[int(obj[-1])]]) for obj in batch_as_list |
| 49 | + ] |
| 50 | + |
| 51 | + return named_output |
| 52 | + |
| 53 | + def preprocess_image(self, img: Image) -> Tuple[torch.Tensor, torch.Tensor]: |
| 54 | + height, width = img.size |
| 55 | + print("original image dimensions:\nh:", height, "w:", width) |
| 56 | + |
| 57 | + img_conv = img.convert("RGB") |
| 58 | + |
| 59 | + # Change the data type of the dataloader depending of the options |
| 60 | + if self.cfg.model.uint_io: |
| 61 | + image_type = "uint" |
| 62 | + elif not self.cfg.model.ipu or not self.cfg.model.half: |
| 63 | + image_type = "float" |
| 64 | + else: |
| 65 | + image_type = "half" |
| 66 | + |
| 67 | + size = torch.as_tensor(img_conv.size) |
| 68 | + resize_img_mthd = ResizeImage(self.cfg.model.image_size) |
| 69 | + print("img_size: ", self.cfg.model.image_size) |
| 70 | + pad_mthd = Pad(self.cfg.model.image_size) |
| 71 | + image_to_tensor_mthd = Compose([ToNumpy(), ToTensor(int(self.cfg.dataset.max_bbox_per_scale), image_type)]) |
| 72 | + dummy_label = np.array([[1.0, 1.0, 1.0, 1.0, 1.0]]) |
| 73 | + transformed_image, transformed_labels = resize_img_mthd((img_conv, dummy_label)) |
| 74 | + transformed_image, transformed_labels = pad_mthd((transformed_image, transformed_labels)) |
| 75 | + transformed_image, transformed_labels = image_to_tensor_mthd((transformed_image, transformed_labels)) |
| 76 | + transformed_image, _ = torch.unsqueeze(transformed_image, dim=0), torch.unsqueeze(transformed_labels, dim=0) |
| 77 | + |
| 78 | + return transformed_image, size |
| 79 | + |
| 80 | + def prepare_inference_yolo_v4_model_for_ipu(self) -> poptorch.PoplarExecutor: |
| 81 | + mode = "inference" |
| 82 | + original_model = Yolov4P5 |
| 83 | + model = original_model(self.cfg) |
| 84 | + |
| 85 | + # Insert the pipeline splits if using pipeline |
| 86 | + if self.cfg.model.pipeline_splits: |
| 87 | + named_layers = {name: layer for name, layer in model.named_modules()} |
| 88 | + for ipu_idx, split in enumerate(self.cfg.model.pipeline_splits): |
| 89 | + named_layers[split] = poptorch.BeginBlock(ipu_id=ipu_idx + 1, layer_to_call=named_layers[split]) |
| 90 | + |
| 91 | + model = load_and_fuse_pretrained_weights(model, self.checkpoint) |
| 92 | + model.optimize_for_inference() |
| 93 | + model.eval() |
| 94 | + |
| 95 | + # Create the specific ipu options if self.cfg.model.ipu |
| 96 | + ipu_opts = ipu_options(self.cfg, model, mode) if self.cfg.model.ipu else None |
| 97 | + |
| 98 | + model = inferenceModel(model, ipu_opts) |
| 99 | + |
| 100 | + return model |
| 101 | + |
| 102 | + def plot_img(self, processed_batch: List[YoloOutput], original_img: Image) -> List[str]: |
| 103 | + list_of_predictions = [torch.Tensor([list(p)[0:6] for p in processed_batch])] |
| 104 | + img_path = plotting_tool(self.cfg, list_of_predictions, [original_img]) |
| 105 | + return img_path |
| 106 | + |
| 107 | + def detach(self): |
| 108 | + if self.model.isAttachedToDevice(): |
| 109 | + self.model.detachFromDevice() |
0 commit comments