| |
| import json |
| import os |
|
|
| import util.misc as utils |
|
|
| try: |
| from panopticapi.evaluation import pq_compute |
| except ImportError: |
| pass |
|
|
|
|
| class PanopticEvaluator(object): |
| def __init__(self, ann_file, ann_folder, output_dir="panoptic_eval"): |
| self.gt_json = ann_file |
| self.gt_folder = ann_folder |
| if utils.is_main_process(): |
| if not os.path.exists(output_dir): |
| os.mkdir(output_dir) |
| self.output_dir = output_dir |
| self.predictions = [] |
|
|
| def update(self, predictions): |
| for p in predictions: |
| with open(os.path.join(self.output_dir, p["file_name"]), "wb") as f: |
| f.write(p.pop("png_string")) |
|
|
| self.predictions += predictions |
|
|
| def synchronize_between_processes(self): |
| all_predictions = utils.all_gather(self.predictions) |
| merged_predictions = [] |
| for p in all_predictions: |
| merged_predictions += p |
| self.predictions = merged_predictions |
|
|
| def summarize(self): |
| if utils.is_main_process(): |
| json_data = {"annotations": self.predictions} |
| predictions_json = os.path.join(self.output_dir, "predictions.json") |
| with open(predictions_json, "w") as f: |
| f.write(json.dumps(json_data)) |
| return pq_compute(self.gt_json, predictions_json, gt_folder=self.gt_folder, pred_folder=self.output_dir) |
| return None |
|
|
