import json import math import os import shutil from code.base import ( BaseEvaluation, EvalResult, SemanticSegmentationEvalConfig, ) import cv2 import numpy as np import yaml from hat.evaluation.semantic_segmentation import evaluate from hat.visualize.semantic_segmentation import ( base64_to_img, plot_cm, render_image, ) from PIL import Image as Img from aidisdk.experiment import Image, Table class SemanticSegmentationEval(BaseEvaluation): def __init__(self, run_config): super().__init__(run_config) def preprocess(self) -> SemanticSegmentationEvalConfig: return super().semantic_segmentation_preprocess() def evaluate( self, eval_config: SemanticSegmentationEvalConfig ) -> EvalResult: if os.path.exists("outputs"): shutil.rmtree("outputs") os.makedirs("outputs", exist_ok=True) if eval_config.images_json: results = evaluate( images_dir=eval_config.images_dir, gt_dir=eval_config.labels_dir, pred_dir=eval_config.prediction_dir, config_file=eval_config.setting, outfile_dir="outputs", images_json=eval_config.images_json, ) else: results = evaluate( images_dir=eval_config.images_dir, gt_dir=eval_config.labels_dir, pred_dir=eval_config.prediction_dir, config_file=eval_config.setting, outfile_dir="outputs", ) summary = {} for result in results: summary[result["name"]] = result["value"] tables = [] analysis_images = [] result = json.load(open("outputs/all.json", "rb")) if result.get("number") is None: table1 = Table( name="all", columns=["mean_iou", "mean_acc"], data=[ { "mean_iou": "%.4f" % result["mean_iou"], "mean_acc": "%.4f" % result["mean_acc"], } ], ) tables.append(table1) data = [] for classinfo in result["classes"]: new_dict_data = { "name": "%s" % classinfo["name"], "iou": "%.4f" % classinfo["iou"], "acc": "%.4f" % classinfo["acc"], } data.append(new_dict_data) table2 = Table( name="classes", columns=["name", "iou", "acc"], data=data, ) tables.append(table2) else: table1 = Table( name="all", columns=["mean_iou", "mean_acc", "number"], data=[ { "mean_iou": "%.4f" % result["mean_iou"], "mean_acc": "%.4f" % result["mean_acc"], "number": "%s" % result["number"], } ], ) tables.append(table1) data = [] for classinfo in result["classes"]: new_dict_data = { "name": "%s" % classinfo["name"], "number": "%s" % classinfo["number"], "iou": "%.4f" % classinfo["iou"], "acc": "%.4f" % classinfo["acc"], } data.append(new_dict_data) table2 = Table( name="classes", columns=["name", "number", "iou", "acc"], data=data, ) tables.append(table2) if "freespace" in result.keys(): table3 = Table( name="freespace", columns=["mean_iou_freespace", "mean_acc_freespace"], data=[ { "mean_iou_freespace": "%.4f" % result["freespace"]["mean_iou_freespace"] if "mean_iou_freespace" in result["freespace"].keys() else result["freespace"]["freespace_iou"], "mean_acc_freespace": "%.4f" % result["freespace"]["mean_acc_freespace"] if "mean_acc_freespace" in result["freespace"].keys() else result["freespace"]["freespace_acc"], } ], ) tables.append(table3) if "margin_res" in result["freespace"].keys(): arr_acc = list() arr_precison_05 = list() arr_precison_07 = list() margin_classes = [" "] for idx, margin in enumerate( result["freespace"]["margin_res"] ): margin_class = list(margin.keys())[0] margin_classes = margin_classes + [margin_class] arr_acc.append( result["freespace"]["margin_res"][idx][margin_class][ "acc" ] ) arr_precison_05.append( result["freespace"]["margin_res"][idx][margin_class][ "precison_0.5" ] ) arr_precison_07.append( result["freespace"]["margin_res"][idx][margin_class][ "precison_0.7" ] ) acc_list = ["acc"] + arr_acc p_05_list = ["precison_0.5"] + arr_precison_05 p_07_list = ["precison_0.7"] + arr_precison_07 data = [] new_dict_data1 = {} for idx, value in enumerate(acc_list): new_dict_data1[margin_classes[idx]] = value data.append(new_dict_data1) new_dict_data2 = {} for idx, value in enumerate(p_05_list): new_dict_data2[margin_classes[idx]] = value data.append(new_dict_data2) new_dict_data3 = {} for idx, value in enumerate(p_07_list): new_dict_data3[margin_classes[idx]] = value data.append(new_dict_data3) table4 = Table( name="freespace margin res", columns=margin_classes, data=data, ) tables.append(table4) if "margin_result" in result["freespace"].keys(): arr_acc = list() arr_precison_05 = list() arr_precison_07 = list() margin_classes = [" "] for idx, margin in enumerate( result["freespace"]["margin_result"] ): margin_class = list(margin.keys())[0] margin_classes = margin_classes + [margin_class] arr_acc.append( result["freespace"]["margin_result"][idx][ margin_class ]["acc_point"] ) arr_precison_05.append( result["freespace"]["margin_result"][idx][ margin_class ]["precison_0.5_image"] ) arr_precison_07.append( result["freespace"]["margin_result"][idx][ margin_class ]["precison_0.7_image"] ) acc_list = ["acc_point"] + arr_acc p_05_list = ["precison_0.5_image"] + arr_precison_05 p_07_list = ["precison_0.7_image"] + arr_precison_07 data = [] new_dict_data1 = {} for idx, value in enumerate(acc_list): new_dict_data1[margin_classes[idx]] = value data.append(new_dict_data1) new_dict_data2 = {} for idx, value in enumerate(p_05_list): new_dict_data2[margin_classes[idx]] = value data.append(new_dict_data2) new_dict_data3 = {} for idx, value in enumerate(p_07_list): new_dict_data3[margin_classes[idx]] = value data.append(new_dict_data3) table4 = Table( name="freespace margin res", columns=margin_classes, data=data, ) tables.append(table4) for key in result["freespace"].keys(): if key.startswith("categories_result_vertical"): titles = [ " ", "precision", "recall", "F1", "num_gt", "num_pred", ] data = [] for item in result["freespace"][key]: new_dict_data = {} for idx, _value in enumerate(titles): if idx == 0: new_dict_data[titles[idx]] = list(item.keys())[ 0 ] else: v = item[list(item.keys())[0]][titles[idx]] if type(v) == float and math.isnan(v): v = "nan" new_dict_data[titles[idx]] = v data.append(new_dict_data) table5 = Table( name=key, columns=titles, data=data, ) tables.append(table5) if "freespace_distance_statistic" in result["freespace"].keys(): titles = [ " ", "max", "min", "mean", "var", "std", "90 percentile", "95 percentile", "98 percentile", ] data = [] for item in result["freespace"][ "freespace_distance_statistic" ]: new_dict_data = {} for idx, _value in enumerate(titles): if idx == 0: new_dict_data[titles[idx]] = list(item.keys())[0] else: v = item[list(item.keys())[0]][titles[idx]] if type(v) == float and math.isnan(v): v = "nan" new_dict_data[titles[idx]] = v data.append(new_dict_data) table6 = Table( name="freespace_distance_statistic", columns=titles, data=data, ) tables.append(table6) classnames = [classinfo["name"] for classinfo in result["classes"]] confusion_matrix = np.array(result["confusion_matrix"]) confusion_matrix_img_io = plot_cm(classnames, confusion_matrix) confusion_matrix_img = Img.open(confusion_matrix_img_io) image_name = "outputs/confusion_matrix.png" confusion_matrix_img.save(image_name) image = Image(image_name) image.add_slice(data_or_path=image_name) analysis_images.append(image) samples = [] if os.path.exists("outputs/samples"): shutil.rmtree("outputs/samples") os.makedirs("outputs/samples", exist_ok=True) with open(self.run_config.setting_file_name, "r") as fid: config_dict = yaml.load(fid, Loader=yaml.FullLoader) add_vis_info = config_dict.get("add_vis", None) labels_info = json.load(open("outputs/labels.json", "rb"))["labels"] eval_file_io = open("outputs/images.json", "rb") for line in eval_file_io: sample_eval = json.loads(line) image_name = sample_eval["image_name"] image_file_path = os.path.join(eval_config.images_dir, image_name) with open(image_file_path.encode("utf-8"), "rb") as f: image_content = f.read() npar = np.frombuffer(image_content, dtype="uint8") image = cv2.imdecode(npar, 1) if "b64_diff" in sample_eval: image_map = ( base64_to_img(sample_eval["b64_image"]) if sample_eval.get("b64_image", None) is not None else None ) diff_map = base64_to_img(sample_eval["b64_diff"]) gt_label_map = base64_to_img(sample_eval["b64_gt_label"]) pred_label_map = base64_to_img(sample_eval["b64_pred_label"]) else: image_npar = np.frombuffer( open( os.path.join("outputs/res_image", image_name), "rb" ).read(), dtype=np.uint8, ) diff_npar = np.frombuffer( open( os.path.join( "outputs/res_diff", os.path.splitext(image_name)[0] + ".png", ), "rb", ).read(), dtype=np.uint8, ) gt_label_npar = np.frombuffer( open( os.path.join( "outputs/res_gt_label", os.path.splitext(image_name)[0] + ".png", ), "rb", ).read(), dtype=np.uint8, ) pred_label_npar = np.frombuffer( open( os.path.join( "outputs/res_pred_label", os.path.splitext(image_name)[0] + ".png", ), "rb", ).read(), dtype=np.uint8, ) image_map = cv2.imdecode(image_npar, cv2.IMREAD_UNCHANGED) diff_map = cv2.imdecode(diff_npar, cv2.IMREAD_UNCHANGED) gt_label_map = cv2.imdecode( gt_label_npar, cv2.IMREAD_UNCHANGED ) pred_label_map = cv2.imdecode( pred_label_npar, cv2.IMREAD_UNCHANGED ) gt_freespace_line, pred_freespace_line = sample_eval.get( "gt_freespace_line" ), sample_eval.get("pred_freespace_line") rendered_image = render_image( image, diff_map, gt_label_map, pred_label_map, gt_freespace_line, pred_freespace_line, labels_info, image_map, add_vis_info=add_vis_info, ) output_file_path = os.path.join("outputs/samples/", image_name) cv2.imwrite(output_file_path, rendered_image) image = Image(image_name) image.add_slice(data_or_path=output_file_path) samples.append(image) images = samples + analysis_images return EvalResult( summary=summary, tables=tables, plots=None, images=images, )