During PTQ, an error message "Found duplicate qco types!" occurs (quantization is successful)

Issue Type

Others

Source

pip (model-compression-toolkit)

MCT Version

2.1.0

OS Platform and Distribution

Ubuntu 22.04.4 LTS

Python version

3.9.19

Describe the issue

When quantizing SSDLiteMobilenetV3 using MCT, the following error occurred repeatedly.

"ERROR: Model Compression Toolkit: Found duplicate qco types!"

Specifically, it seems to occur in the following function:

"mct.ptq.pytorch_post_training_quantization"

However, despite the error, the quantization was successful, the mAP value was as expected, and it does not seem to affect the completed quantized model.

The script that generated the error is a Python script rewritten from the following Notebook:

https://github.com/sony/model_optimization/blob/main/tutorials/notebooks/mct_features_notebooks/pytorch/example_pytorch_ssdlite_mobilenetv3_object_detection.ipynb

The operating environment for the script is as follows: ・GPU used ・model-compression-toolkit 2.1.0 ・pytorch 2.0.1

Expected behaviour

It doesn't seem to affect the model, but it would be better if no errors occurred.

Code to reproduce the issue

#Sorry, it's a bit long, but here is the reproducible script. The error log is output in the following "mct.ptq.pytorch_post_training_quantization".
#-----
#
import torch
import torchvision
from torchvision.models.detection.ssdlite import SSDLite320_MobileNet_V3_Large_Weights
from torchvision.models.detection.anchor_utils import ImageList
import model_compression_toolkit as mct
from pycocotools.coco import COCO
from pycocotools.cocoeval import COCOeval

device = 'cuda' if torch.cuda.is_available() else 'cpu'

image_size = (320, 320)
model = torchvision.models.detection.ssdlite320_mobilenet_v3_large(weights=SSDLite320_MobileNet_V3_Large_Weights.DEFAULT)
# mAP=0.2131 (float)
# mAP=0.2007 (quantized)

model.eval()
model = model.to(device)

print('device : %s' % (device))

print('model loaded')

def format_results(outputs, img_ids):
    detections = []

    # Process model outputs and convert to detection format
    for idx, output in enumerate(outputs):
        image_id = img_ids[idx]  # Adjust according to your batch size and indexing
        scores = output['scores'].cpu().numpy()
        labels = output['labels'].cpu().numpy()
        boxes = output['boxes'].cpu().numpy()

        for score, label, box in zip(scores, labels, boxes):
            detection = {
                "image_id": image_id,
                "category_id": label,
                "bbox": [box[0], box[1], box[2] - box[0], box[3] - box[1]],
                "score": score
            }
            detections.append(detection)

    return detections

class CocoEval:
    def __init__(self, path2json):

        # Load ground truth annotations
        self.coco_gt = COCO(path2json)

        # A list of reformatted model outputs
        self.all_detections = []

    def add_batch_detections(self, outputs, targets):

        # Collect and format results from the batch
        img_ids, _outs = [], []
        for t, o in zip(targets, outputs):
            if len(t) > 0:
                img_ids.append(t[0]['image_id'])
                _outs.append(o)

        batch_detections = format_results(_outs, img_ids)  # Implement this function

        self.all_detections.extend(batch_detections)

    def result(self):
        # Initialize COCO evaluation object
        self.coco_dt = self.coco_gt.loadRes(self.all_detections)
        coco_eval = COCOeval(self.coco_gt, self.coco_dt, 'bbox')

        # Run evaluation
        coco_eval.evaluate()
        coco_eval.accumulate()
        coco_eval.summarize()

        # Print mAP results
        print("mAP: {:.4f}".format(coco_eval.stats[0]))

        return coco_eval.stats

    def reset(self):
        self.all_detections = []

EVAL_DATASET_FOLDER = './coco/val2017'
EVAL_DATASET_ANNOTATION_FILE = './coco/annotations/instances_val2017.json'

def collate_fn(batch_input):
    images = [b[0] for b in batch_input]
    targets = [b[1] for b in batch_input]
    return images, targets

# Initialize the COCO evaluation DataLoader
coco_eval = torchvision.datasets.CocoDetection(root=EVAL_DATASET_FOLDER,
                                               annFile=EVAL_DATASET_ANNOTATION_FILE,
                                               transform=torchvision.transforms.ToTensor())
batch_size = 50
data_loader = torch.utils.data.DataLoader(coco_eval, batch_size=batch_size, shuffle=False,
                                          num_workers=0, collate_fn=collate_fn)

# Initialize the evaluation metric object
coco_metric = CocoEval(EVAL_DATASET_ANNOTATION_FILE)

class SDD4Quant(torch.nn.Module):
    def __init__(self, in_sdd, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # Save the float model under self.base as a module of the model. Later we'll only run "backbone" & "head"
        self.add_module("base", in_sdd)

    # Forward pass of the model to be quantized. This code is copied from the float model forward function (removed the preprocess and postprocess code)
    def forward(self, x):
        features = self.base.backbone(x)

        features = list(features.values())

        # compute the ssd heads outputs using the features
        head_outputs = self.base.head(features)
        return head_outputs

model4quant = SDD4Quant(model)

def preprocess(image, targets):
    # need to save the original image sizes before resize for the postprocess part
    targets = {'gt': targets, 'img_size': list(image.size[::-1])}
    image = model.transform([torchvision.transforms.ToTensor()(image)])[0].tensors[0, ...]
    return image, targets

# Define the postprocess, which is the code copied from the float model forward code. These layers will not be quantized.
class PostProcess:
    def __init__(self):
        self.features = [torch.zeros((1, 1, s, s)) for s in [20, 10, 5, 3, 2, 1]]

    def __call__(self, head_outputs, image_list, original_image_sizes):
        anchors = [a.to(device) for a in model.anchor_generator(image_list, self.features)]

        # The MCT flattens the outputs of the head to a list, so need to change it to a dictionary as the psotprocess functions expect.
        if not isinstance(head_outputs, dict):
            if head_outputs[0].shape[-1] == 4:
                head_outputs = {"bbox_regression": head_outputs[0],
                                "cls_logits": head_outputs[1]}
            else:
                head_outputs = {"bbox_regression": head_outputs[1],
                                "cls_logits": head_outputs[0]}

        # Float model postprocess functions that handle box regression and NMS
        detections = model.postprocess_detections(head_outputs, anchors, image_list.image_sizes)
        detections = model.transform.postprocess(detections, image_list.image_sizes, original_image_sizes)
        return detections

postprocess = PostProcess()

def train_collate_fn(batch_input):
    # collating images for the quantized model should return a single tensor: [B, C, H, W]
    images = torch.stack([b[0] for b in batch_input])
    targets = [b[1] for b in batch_input]
    return images, targets

coco_eval = torchvision.datasets.CocoDetection(root=EVAL_DATASET_FOLDER, annFile=EVAL_DATASET_ANNOTATION_FILE,
                                               transforms=preprocess)
eval_loader = torch.utils.data.DataLoader(coco_eval, batch_size=50, shuffle=False, num_workers=0,
                                          collate_fn=train_collate_fn)

def get_representative_dataset(n_iter):

    def representative_dataset():
        ds_iter = iter(eval_loader)
        for _ in range(n_iter):
            yield [next(ds_iter)[0]]

    return representative_dataset

# Get representative dataset generator
representative_dataset_gen = get_representative_dataset(20)

quant_model, _ = mct.ptq.pytorch_post_training_quantization(model4quant,
                                                            representative_dataset_gen)

print('quant_model is Ready')

Log output

ssd_mobilenetv3_mct_script_log.txt

The error log will be around line 50.

sony / model_optimization