# SPDX-License-Identifier: Apache-2.0 # ----------------------------------------------------------------------------- # Copyright 2020-2022 Arm Limited # # Licensed under the Apache License, Version 2.0 (the "License"); you may not # use this file except in compliance with the License. You may obtain a copy # of the License at: # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. # ----------------------------------------------------------------------------- """ A ResultSet stores a set of results about the performance of a TestSet. Each set keeps result Records for each image and block size tested, that store the PSNR and coding time. ResultSets are often backed by a CSV file on disk, and a ResultSet can be compared against a set of reference results created by an earlier test run. """ import csv import enum import numpy as np import os @enum.unique class Result(enum.IntEnum): """ An enumeration of test result status values. Attributes: NOTRUN: The test has not been run. PASS: The test passed. WARN: The test image quality was below the pass threshold but above the fail threshold. FAIL: The test image quality was below the fail threshold. """ NOTRUN = 0 PASS = 1 WARN = 2 FAIL = 3 class ResultSummary(): """ An result summary data container, storing number of results of each type. Attributes: notruns: The number of tests that did not run. passes: The number of tests that passed. warnings: The number of tests that produced a warning. fails: The number of tests that failed. tTimes: Total time speedup vs reference (<1 is slower, >1 is faster). cTimes: Coding time speedup vs reference (<1 is slower, >1 is faster). psnrs: Coding time quality vs reference (<0 is worse, >0 is better). """ def __init__(self): """ Create a new result summary. """ # Pass fail metrics self.notruns = 0 self.passes = 0 self.warnings = 0 self.fails = 0 # Relative results self.tTimesRel = [] self.cTimesRel = [] self.psnrRel = [] # Absolute results self.cTime = [] self.psnr = [] def add_record(self, record): """ Add a record to this summary. Args: record (Record): The Record to add. """ if record.status == Result.PASS: self.passes += 1 elif record.status == Result.WARN: self.warnings += 1 elif record.status == Result.FAIL: self.fails += 1 else: self.notruns += 1 if record.tTimeRel is not None: self.tTimesRel.append(record.tTimeRel) self.cTimesRel.append(record.cTimeRel) self.psnrRel.append(record.psnrRel) self.cTime.append(record.cTime) self.psnr.append(record.psnr) def get_worst_result(self): """ Get the worst result in this set. Returns: Result: The worst test result. """ if self.fails: return Result.FAIL if self.warnings: return Result.WARN if self.passes: return Result.PASS return Result.NOTRUN def __str__(self): # Overall pass/fail results overall = self.get_worst_result().name dat = (overall, self.passes, self.warnings, self.fails) result = ["\nSet Status: %s (Pass: %u | Warn: %u | Fail: %u)" % dat] if (self.tTimesRel): # Performance summaries dat = (np.mean(self.tTimesRel), np.std(self.tTimesRel)) result.append("\nTotal speed: Mean: %+0.3f x Std: %0.2f x" % dat) dat = (np.mean(self.cTimesRel), np.std(self.cTimesRel)) result.append("Coding speed: Mean: %+0.3f x Std: %0.2f x" % dat) dat = (np.mean(self.psnrRel), np.std(self.psnrRel)) result.append("Quality diff: Mean: %+0.3f dB Std: %0.2f dB" % dat) dat = (np.mean(self.cTime), np.std(self.cTime)) result.append("Coding time: Mean: %+0.3f s Std: %0.2f s" % dat) dat = (np.mean(self.psnr), np.std(self.psnr)) result.append("Quality: Mean: %+0.3f dB Std: %0.2f dB" % dat) return "\n".join(result) class Record(): """ A single result record, sotring results for a singel image and block size. Attributes: blkSz: The block size. name: The test image name. psnr: The image quality (PSNR dB) tTime: The total compression time. cTime: The coding compression time. cRate: The coding compression rate. status: The test Result. """ def __init__(self, blkSz, name, psnr, tTime, cTime, cRate): """ Create a result record, initially in the NOTRUN status. Args: blkSz (str): The block size. name (str): The test image name. psnr (float): The image quality PSNR, in dB. tTime (float): The total compression time, in seconds. cTime (float): The coding compression time, in seconds. cRate (float): The coding compression rate, in MPix/s. tTimeRel (float): The relative total time speedup vs reference. cTimeRel (float): The relative coding time speedup vs reference. cRateRel (float): The relative rate speedup vs reference. psnrRel (float): The relative PSNR dB vs reference. """ self.blkSz = blkSz self.name = name self.psnr = psnr self.tTime = tTime self.cTime = cTime self.cRate = cRate self.status = Result.NOTRUN self.tTimeRel = None self.cTimeRel = None self.cRateRel = None self.psnrRel = None def set_status(self, result): """ Set the result status. Args: result (Result): The test result. """ self.status = result def __str__(self): return "'%s' / '%s'" % (self.blkSz, self.name) class ResultSet(): """ A set of results for a TestSet, across one or more block sizes. Attributes: testSet: The test set these results are linked to. records: The list of test results. """ def __init__(self, testSet): """ Create a new empty ResultSet. Args: testSet (TestSet): The test set these results are linked to. """ self.testSet = testSet self.records = [] def add_record(self, record): """ Add a new test record to this result set. Args: record (Record): The test record to add. """ self.records.append(record) def get_record(self, testSet, blkSz, name): """ Get a record matching the arguments. Args: testSet (TestSet): The test set to get results from. blkSz (str): The block size. name (str): The test name. Returns: Record: The test result, if present. Raises: KeyError: No match could be found. """ if testSet != self.testSet: raise KeyError() for record in self.records: if record.blkSz == blkSz and record.name == name: return record raise KeyError() def get_matching_record(self, other): """ Get a record matching the config of another record. Args: other (Record): The pattern result record to match. Returns: Record: The result, if present. Raises: KeyError: No match could be found. """ for record in self.records: if record.blkSz == other.blkSz and record.name == other.name: return record raise KeyError() def get_results_summary(self): """ Get a results summary of all the records in this result set. Returns: ResultSummary: The result summary. """ summary = ResultSummary() for record in self.records: summary.add_record(record) return summary def save_to_file(self, filePath): """ Save this result set to a CSV file. Args: filePath (str): The output file path. """ dirName = os.path.dirname(filePath) if not os.path.exists(dirName): os.makedirs(dirName) with open(filePath, "w", newline="") as csvfile: writer = csv.writer(csvfile) self._save_header(writer) for record in self.records: self._save_record(writer, record) @staticmethod def _save_header(writer): """ Write the header to the CSV file. Args: writer (csv.writer): The CSV writer. """ row = ["Image Set", "Block Size", "Name", "PSNR", "Total Time", "Coding Time", "Coding Rate"] writer.writerow(row) def _save_record(self, writer, record): """ Write a record to the CSV file. Args: writer (csv.writer): The CSV writer. record (Record): The record to write. """ row = [self.testSet, record.blkSz, record.name, "%0.4f" % record.psnr, "%0.4f" % record.tTime, "%0.4f" % record.cTime, "%0.4f" % record.cRate] writer.writerow(row) def load_from_file(self, filePath): """ Load a reference result set from a CSV file on disk. Args: filePath (str): The input file path. """ with open(filePath, "r") as csvfile: reader = csv.reader(csvfile) # Skip the header next(reader) for row in reader: assert row[0] == self.testSet record = Record(row[1], row[2], float(row[3]), float(row[4]), float(row[5]), float(row[6])) self.add_record(record)