# Copyright (c) OpenMMLab. All rights reserved.
import csv
import os
import os.path as osp
from typing import List, Sequence

import numpy as np
import torch
from mmengine.dist.utils import get_rank
from mmengine.evaluator import BaseMetric

from mmpretrain.registry import METRICS


@METRICS.register_module()
class ShapeBiasMetric(BaseMetric):
    """Evaluate the model on ``cue_conflict`` dataset.

    This module will evaluate the model on an OOD dataset, cue_conflict, in
    order to measure the shape bias of the model. In addition to compuate the
    Top-1 accuracy, this module also generate a csv file to record the
    detailed prediction results, such that this csv file can be used to
    generate the shape bias curve.

    Args:
        csv_dir (str): The directory to save the csv file.
        model_name (str): The name of the csv file. Please note that the
            model name should be an unique identifier.
        dataset_name (str): The name of the dataset. Default: 'cue_conflict'.
    """

    # mapping several classes from ImageNet-1K to the same category
    airplane_indices = [404]
    bear_indices = [294, 295, 296, 297]
    bicycle_indices = [444, 671]
    bird_indices = [
        8, 10, 11, 12, 13, 14, 15, 16, 18, 19, 20, 22, 23, 24, 80, 81, 82, 83,
        87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 98, 99, 100, 127, 128, 129,
        130, 131, 132, 133, 135, 136, 137, 138, 139, 140, 141, 142, 143, 144,
        145
    ]
    boat_indices = [472, 554, 625, 814, 914]
    bottle_indices = [440, 720, 737, 898, 899, 901, 907]
    car_indices = [436, 511, 817]
    cat_indices = [281, 282, 283, 284, 285, 286]
    chair_indices = [423, 559, 765, 857]
    clock_indices = [409, 530, 892]
    dog_indices = [
        152, 153, 154, 155, 156, 157, 158, 159, 160, 161, 162, 163, 164, 165,
        166, 167, 168, 169, 170, 171, 172, 173, 174, 175, 176, 177, 178, 179,
        180, 181, 182, 183, 184, 185, 186, 187, 188, 189, 190, 191, 193, 194,
        195, 196, 197, 198, 199, 200, 201, 202, 203, 205, 206, 207, 208, 209,
        210, 211, 212, 213, 214, 215, 216, 217, 218, 219, 220, 221, 222, 223,
        224, 225, 226, 228, 229, 230, 231, 232, 233, 234, 235, 236, 237, 238,
        239, 240, 241, 243, 244, 245, 246, 247, 248, 249, 250, 252, 253, 254,
        255, 256, 257, 259, 261, 262, 263, 265, 266, 267, 268
    ]
    elephant_indices = [385, 386]
    keyboard_indices = [508, 878]
    knife_indices = [499]
    oven_indices = [766]
    truck_indices = [555, 569, 656, 675, 717, 734, 864, 867]

    def __init__(self,
                 csv_dir: str,
                 model_name: str,
                 dataset_name: str = 'cue_conflict',
                 **kwargs) -> None:
        super().__init__(**kwargs)

        self.categories = sorted([
            'knife', 'keyboard', 'elephant', 'bicycle', 'airplane', 'clock',
            'oven', 'chair', 'bear', 'boat', 'cat', 'bottle', 'truck', 'car',
            'bird', 'dog'
        ])
        self.csv_dir = csv_dir
        self.model_name = model_name
        self.dataset_name = dataset_name
        if get_rank() == 0:
            self.csv_path = self.create_csv()

    def process(self, data_batch, data_samples: Sequence[dict]) -> None:
        """Process one batch of data samples.

        The processed results should be stored in ``self.results``, which will
        be used to computed the metrics when all batches have been processed.

        Args:
            data_batch: A batch of data from the dataloader.
            data_samples (Sequence[dict]): A batch of outputs from the model.
        """
        for data_sample in data_samples:
            result = dict()
            if 'pred_score' in data_sample:
                result['pred_score'] = data_sample['pred_score'].cpu()
            else:
                result['pred_label'] = data_sample['pred_label'].cpu()
            result['gt_label'] = data_sample['gt_label'].cpu()
            result['gt_category'] = data_sample['img_path'].split('/')[-2]
            result['img_name'] = data_sample['img_path'].split('/')[-1]

            aggregated_category_probabilities = []
            # get the prediction for each category of current instance
            for category in self.categories:
                category_indices = getattr(self, f'{category}_indices')
                category_probabilities = torch.gather(
                    result['pred_score'], 0,
                    torch.tensor(category_indices)).mean()
                aggregated_category_probabilities.append(
                    category_probabilities)
            # sort the probabilities in descending order
            pred_indices = torch.stack(aggregated_category_probabilities
                                       ).argsort(descending=True).numpy()
            result['pred_category'] = np.take(self.categories, pred_indices)

            # Save the result to `self.results`.
            self.results.append(result)

    def create_csv(self) -> str:
        """Create a csv file to store the results."""
        session_name = 'session-1'
        csv_path = osp.join(
            self.csv_dir, self.dataset_name + '_' + self.model_name + '_' +
            session_name + '.csv')
        if osp.exists(csv_path):
            os.remove(csv_path)
        directory = osp.dirname(csv_path)
        if not osp.exists(directory):
            os.makedirs(directory, exist_ok=True)
        with open(csv_path, 'w') as f:
            writer = csv.writer(f)
            writer.writerow([
                'subj', 'session', 'trial', 'rt', 'object_response',
                'category', 'condition', 'imagename'
            ])
        return csv_path

    def dump_results_to_csv(self, results: List[dict]) -> None:
        """Dump the results to a csv file.

        Args:
            results (List[dict]): A list of results.
        """
        for i, result in enumerate(results):
            img_name = result['img_name']
            category = result['gt_category']
            condition = 'NaN'
            with open(self.csv_path, 'a') as f:
                writer = csv.writer(f)
                writer.writerow([
                    self.model_name, 1, i + 1, 'NaN',
                    result['pred_category'][0], category, condition, img_name
                ])

    def compute_metrics(self, results: List[dict]) -> dict:
        """Compute the metrics from the results.

        Args:
            results (List[dict]): A list of results.

        Returns:
            dict: A dict of metrics.
        """
        if get_rank() == 0:
            self.dump_results_to_csv(results)
        metrics = dict()
        metrics['accuracy/top1'] = np.mean([
            result['pred_category'][0] == result['gt_category']
            for result in results
        ])

        return metrics
