355 lines
10 KiB
Python
355 lines
10 KiB
Python
# 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)
|