evaluate_master/data_quality.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
##################################################################
#
# Copyright (c) 2023 CICV, Inc. All Rights Reserved
#
##################################################################
"""
@Authors:           yangzihao(yangzihao@china-icv.cn)
@Data:              2023/12/19
@Last Modified:     2023/12/19
@Summary:           Data quality evaluate.
"""

import os
import pandas as pd
from collections import Counter


class DataQuality(object):

    def __init__(self, df):
        """
        TODO: 数据跳变
        """
        self.df = df
        self.frame_list = []
        self.frame_diff_list = []
        self.frame_diff_counter = {}

        self.total_frame_count = 0
        self.max_frame_number = 0
        self.frame_loss_count = 0
        self.frame_loss_rate = 0
        self.frame_loss_max = 0
        self.frame_loss_max_start = []
        # self.quality_detect()
        self.result = ""

    def quality_detect(self):
        self.frame_extract()
        if self.frame_list:
            self.cal_total_frame_count()
            self.cal_max_frame_number()
            self.cal_frame_diff()
            self.cal_frame_loss_count()
            self.cal_frame_loss_rate()
            self.cal_frame_loss_max()
            self.cal_frame_diff_counter()
            self.result_print()
        else:
            self.result = "No data in this file."
            print("No data in this file.")

    def frame_extract(self):
        self.df = self.df.dropna(subset="simFrame")
        self.frame_list = list(set(self.df['simFrame']))
        # self.frame_list = list(set(self.df['frame_ID']))
        self.frame_list.sort()

    def cal_frame_diff(self):
        maxx = -1
        for i in range(1, len(self.frame_list)):
            diff = self.frame_list[i] - self.frame_list[i - 1] - 1
            if diff > maxx and diff != 0:
                maxx = diff
                self.frame_loss_max_start = [self.frame_list[i - 1], self.frame_list[i]]
            self.frame_diff_list.append(int(diff))

    def cal_total_frame_count(self):
        self.total_frame_count = int(self.frame_list[-1] - self.frame_list[0])

    def cal_max_frame_number(self):
        self.max_frame_number = int(self.frame_list[-1])

    def cal_frame_loss_count(self):
        self.frame_loss_count = sum(self.frame_diff_list)

    def cal_frame_loss_rate(self):
        self.frame_loss_rate = round(self.frame_loss_count / self.frame_list[-1], 4)

    def cal_frame_loss_max(self):
        self.frame_loss_max = max(self.frame_diff_list) if self.frame_diff_list else 0

    def cal_frame_diff_counter(self):
        self.frame_diff_counter = dict(Counter(self.frame_diff_list))
        self.frame_diff_counter = dict(sorted(self.frame_diff_counter.items(), reverse=True))
        del self.frame_diff_counter[0]  # 删除不丢帧次数统计

    def result_print(self):
        self.result += f"丢帧率: {self.frame_loss_rate* 100:.2f}%, "
        self.result += f"总帧数: {self.total_frame_count}, "
        self.result += f"丢帧数量: {self.frame_loss_count}, "
        # self.result += f"此文件最大帧数: {self.max_frame_number}, "
        self.result += f"最长丢帧数量: {self.frame_loss_max}, "
        self.result += f"最长丢帧时起始帧: {self.frame_loss_max_start}, "
        self.result += f"丢帧数及次数统计: {self.frame_diff_counter}."

        print("此文件总帧数:", self.total_frame_count)
        print("此文件最大帧数:", self.max_frame_number)
        print("此文件丢帧数量:", self.frame_loss_count)
        print("此文件丢帧率:", f"{self.frame_loss_rate * 100:.2f}%")
        print("此文件最长丢帧数量:", self.frame_loss_max)
        print("此文件最长丢帧时起始帧:", self.frame_loss_max_start)
        print("此文件丢帧数及次数统计:", self.frame_diff_counter)


def get_all_files(path):
    file_names = []
    for root, dirs, files in os.walk(path):
        for file in files:
            file_names.append(os.path.join(root, file))
    return file_names


def frame_loss_statistic(path):
    file_names = get_all_files(path)
    frame_loss_dict = dict()
    for file in file_names:
        file_name = file.split('\\')[-1]
        print(f"\n[{file_name}]")
        df = pd.read_csv(file, index_col=False)
        d_q = DataQuality(df)
        d_q.quality_detect()
        # frame_loss_dict[file_name] = d_q.result
        frame_loss_dict[file_name] = dict()
        frame_loss_dict[file_name]["result"] = d_q.result
        frame_loss_dict[file_name]["frame_loss_rate"] = d_q.frame_loss_rate

    return frame_loss_dict


if __name__ == "__main__":
    # path = r"C:\Users\cicv\Desktop\ads_evaluate_V3.2.0\task_1218_\case_1218\data"
    path = r"./task_1228/case0322/data"
    # path = r"C:\Users\cicv\Desktop\ads_evaluate_V3.2.0\task_1221\data1\data"
    # path = r"C:\Users\cicv\Desktop\ADS_evaluate\ads_evaluation_development\adas_evaluation_V3.0\data\data_0920_zhikong_doubleCar"
    # path = r"C:\Users\cicv\Desktop\align\data2"
    #
    frame_loss_statistic(path)