app.py

try:
    import detectron2
except:
    import os 
    os.system('pip install git+https://github.com/facebookresearch/detectron2.git')
import torch
from detectron2.utils.logger import setup_logger
setup_logger()

from detectron2.config import get_cfg
import detectron2.data.transforms as T
from detectron2.checkpoint import DetectionCheckpointer
from detectron2.modeling import build_model
from detectron2.utils.visualizer import Visualizer
from detectron2.data.detection_utils import read_image
from detectron2.data import MetadataCatalog

import numpy as np
import cv2
from PIL import Image
import random
import gradio as gr
import pandas as pd
import matplotlib.pyplot as plt
import io
from pickle import load

torch.manual_seed(0)
np.random.seed(0)
random.seed(0)

from base_cam import EigenCAM
from pytorch_grad_cam.utils.model_targets import FasterRCNNBoxScoreTarget

class Detectron2Monitor():
    def __init__(self, label_list, label_dict, config_file, model_file):
        self.label_list = label_list
        self.cfg = self._setup_cfg(config_file, model_file)
        self.model = build_model(self.cfg)
        self.model.eval()
        checkpointer = DetectionCheckpointer(self.model)
        checkpointer.load(self.cfg.MODEL.WEIGHTS)
        self.monitors_dict = self._load_monitor()
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.class_dict = label_dict
        
    def _setup_cfg(self, config_file, model_file):  
        cfg = get_cfg()
        cfg.merge_from_file(config_file)
        cfg.MODEL.WEIGHTS = model_file
        cfg.MODEL.ROI_HEADS.SCORE_THRESH_TEST = 0.5
        if torch.cuda.is_available():
            cfg.MODEL.DEVICE = "cuda"
        else:
            cfg.MODEL.DEVICE = "cpu"
        cfg.freeze()
        return cfg
    
    def _get_input_dict(self, original_image):
        height, width = original_image.shape[:2]
        transform_gen = T.ResizeShortestEdge(
        [self.cfg.INPUT.MIN_SIZE_TEST, self.cfg.INPUT.MIN_SIZE_TEST], self.cfg.INPUT.MAX_SIZE_TEST
        )
        image = transform_gen.get_transform(original_image).apply_image(original_image)
        image = torch.as_tensor(image.astype("float32").transpose(2, 0, 1))
        inputs = {"image": image, "height": height, "width": width}
        return inputs
    
    def _postprocess_cam(self, raw_cam, img_width, img_height):
        cam_orig = np.sum(raw_cam, axis=0)  # [H,W]
        cam_orig = np.maximum(cam_orig, 0)  # ReLU
        cam_orig -= np.min(cam_orig)
        cam_orig /= np.max(cam_orig)
        cam = cv2.resize(cam_orig, (img_width, img_height))
        return cam

    def _inference(self, inputs):
        with torch.no_grad():
            images = self.model.preprocess_image(inputs)  
            features = self.model.backbone(images.tensor)  
            proposals, _ = self.model.proposal_generator(images, features, None)  # RPN

            features_ = [features[f] for f in self.model.roi_heads.box_in_features]
            box_features = self.model.roi_heads.box_pooler(features_, [x.proposal_boxes for x in proposals])
            box_features = self.model.roi_heads.box_head(box_features)  # features of all 1k candidates
            predictions = self.model.roi_heads.box_predictor(box_features)
            pred_instances, pred_inds = self.model.roi_heads.box_predictor.inference(predictions, proposals)
            pred_instances = self.model.roi_heads.forward_with_given_boxes(features, pred_instances)

            # output boxes, masks, scores, etc
            pred_instances = self.model._postprocess(pred_instances, inputs, images.image_sizes)  # scale box to orig size
            # features of the proposed boxes
            feats = box_features[pred_inds]
        return pred_instances, feats 
     
    def _load_monitor(self):
        with open("monitors_dict.pkl", 'rb') as f:
            monitors_dict = load(f)
        return monitors_dict
    
    def _fasterrcnn_reshape_transform(self, x):
        target_size = x['p6'].size()[-2 : ]
        activations = []
        for key, value in x.items():
            activations.append(torch.nn.functional.interpolate(torch.abs(value), target_size, mode='bilinear'))
        activations = torch.cat(activations, axis=1)
        return activations

    def get_output(self, img):
        image = read_image(img, format="BGR")
        input_image_dict = [self._get_input_dict(image)]
        
        pred_instances, feats = self._inference(input_image_dict)
        feats = feats.cpu().detach().numpy()
        detections = pred_instances[0]["instances"].to("cpu")
        cls_idxs = detections.pred_classes.detach().numpy()
        # get labels from class indices
        labels = [self.class_dict[i] for i in cls_idxs]
        # count values in labels, and return a dictionary
        labels_count_dict = dict((i, labels.count(i)) for i in labels)
        v = Visualizer(image[..., ::-1], MetadataCatalog.get("bdd_dataset"), scale=1)
        v = v.draw_instance_predictions(detections)
        img_detection = v.get_image()
        df = pd.DataFrame(list(labels_count_dict.items()), columns=['Object', 'Count'])
        verdicts = []
        for label, feat in zip(labels, feats):
            verdict = self.monitors_dict[label].make_verdicts(feat[np.newaxis,:])[0] if label in self.monitors_dict else True
            verdicts.append(verdict)
        detections_ood = detections[[i for i, x in enumerate(verdicts) if not x]]
        detections_ood.pred_classes = torch.tensor([10]*len(detections_ood.pred_classes))
        v = Visualizer(image[..., ::-1], MetadataCatalog.get("bdd_dataset"), scale=1)
        v = v.draw_instance_predictions(detections_ood)
        img_ood = v.get_image()
        pred_bboxes = detections.pred_boxes.tensor.numpy().astype(np.int32)
        target_layers = [self.model.backbone]
        targets = [FasterRCNNBoxScoreTarget(labels=labels, bounding_boxes=pred_bboxes)]
        cam = EigenCAM(self.model,
                        target_layers, 
                        use_cuda=False,
                        reshape_transform=self._fasterrcnn_reshape_transform)
        grayscale_cam = cam(input_image_dict, targets)
        cam = self._postprocess_cam(grayscale_cam, input_image_dict[0]["width"], input_image_dict[0]["height"])
        plt.rcParams["figure.figsize"] = (30,10)
        plt.imshow(img_detection[..., ::-1], interpolation='none')
        plt.imshow(cam, cmap='jet', alpha=0.5)
        plt.axis("off")
        img_buff = io.BytesIO()
        plt.savefig(img_buff, format='png', bbox_inches='tight', pad_inches=0)
        img_cam = Image.open(img_buff)
        image_dict = {}
        image_dict["image"] = image
        image_dict["cam"] = img_cam
        image_dict["detection"] = img_detection
        image_dict["verdict"] = img_ood
        return image_dict, df

config_file = "vanilla.yaml"
model_file = "model_final_vos_resnet_bdd.pth"
label_dict = {
            0: 'pedestrian',
            1: 'rider',
            2: 'car',
            3: 'truck',
            4: 'bus',
            5: 'train',
            6: 'motor',
            7: 'bike',
            8: 'traffic light',
            9: 'traffic sign',
            10: 'OOD'
        }
label_list = list(label_dict.values())
MetadataCatalog.get("bdd_dataset").set(thing_classes=label_list)
extractor = Detectron2Monitor(config_file=config_file, label_list=label_list, label_dict=label_dict, model_file=model_file)

# %%
def inference_gd(file):
    image_dict, df = extractor.get_output(file)
    return image_dict["detection"], df, image_dict["verdict"], image_dict["cam"]


examples = ["examples/0.jpg", "examples/1.jpg", "examples/2.jpg", "examples/3.jpg"]
with gr.Blocks(theme='gradio/monochrome') as demo:
    gr.Markdown("# Runtime Monitoring Object Detection")
    gr.Markdown(
        """This interactive demo is based on the box abstraction-based monitors for Faster R-CNN model. The model is trained using [Detectron2](https://github.com/facebookresearch/detectron2) library on the in-distribution dataset [Berkeley DeepDrive-100k](https://www.bdd100k.com/), which contains objects within autonomous driving domain. The monitors are constructed by abstraction of extracted feature from the training data. The demo showcases the monitors' capacity to reject problematic detections due to out-of-distribution(OOD) objects. 

To utilize the demo, upload an image and click on *"Infer"* to view the following results: 

- **Detection**: outputs of Object Detector
- **Detection summary**: a summary of the detection outputs
- **Verdict**: verdicts from Monitors (problematic detections caused by out-of-distribution(OOD) objects will be identified as OOD objects)
- **Explainable AI**: visual explanation generated by [grad-cam](https://github.com/jacobgil/pytorch-grad-cam) library which is based on Class Activation Mapping(CAM) method.

You can also select an image from the cached **Examples** to quickly try out. Without clicking *"Infer"*, the cached outputs will be loaded automatically. 

In case the output image seems too small, simply right-click on the image, and choose “Open image in new tab” to visualize it in full size.
    """
    )
    with gr.Row().style(equal_height=True):
        with gr.Column():
            image = gr.Image(type="filepath", label="Input")
            button = gr.Button("Infer")
            
        with gr.Column():
            with gr.Tab("Detection"):
                detection = gr.Image(label="Output")
            with gr.Tab("Detection summary"):
                df = gr.Dataframe(label="Detection summary")
            with gr.Tab("Verdict"):
                verdict = gr.Image(label="Output")
            with gr.Tab("Explainable AI"):
                cam = gr.Image(label="Output")
    examples_block = gr.Examples(inputs=image, examples=examples, fn=inference_gd, outputs=[detection, df, verdict, cam], cache_examples=True)
    button.click(fn=inference_gd, inputs=image, outputs=[detection, df, verdict, cam])

demo.launch()