| 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, | |
| ) | |