xref: /third_party/astc-encoder/Test/testlib/resultset.py

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