path: root/tools/perf/tests/shell/lib/perf_metric_validation.py

# SPDX-License-Identifier: GPL-2.0
import re
import csv
import json
import argparse
from pathlib import Path
import subprocess


class TestError:
    def __init__(self, metric: list[str], wl: str, value: list[float], low: float, up=float('nan'), description=str()):
        self.metric: list = metric  # multiple metrics in relationship type tests
        self.workloads = [wl]  # multiple workloads possible
        self.collectedValue: list = value
        self.valueLowBound = low
        self.valueUpBound = up
        self.description = description

    def __repr__(self) -> str:
        if len(self.metric) > 1:
            return "\nMetric Relationship Error: \tThe collected value of metric {0}\n\
                \tis {1} in workload(s): {2} \n\
                \tbut expected value range is [{3}, {4}]\n\
                \tRelationship rule description: \'{5}\'".format(self.metric, self.collectedValue, self.workloads,
                                                                 self.valueLowBound, self.valueUpBound, self.description)
        elif len(self.collectedValue) == 0:
            return "\nNo Metric Value Error: \tMetric {0} returns with no value \n\
                    \tworkload(s): {1}".format(self.metric, self.workloads)
        else:
            return "\nWrong Metric Value Error: \tThe collected value of metric {0}\n\
                    \tis {1} in workload(s): {2}\n\
                    \tbut expected value range is [{3}, {4}]"\
                        .format(self.metric, self.collectedValue, self.workloads,
                                self.valueLowBound, self.valueUpBound)


class Validator:
    def __init__(self, rulefname, reportfname='', t=5, debug=False, datafname='', fullrulefname='', workload='true', metrics=''):
        self.rulefname = rulefname
        self.reportfname = reportfname
        self.rules = None
        self.collectlist: str = metrics
        self.metrics = self.__set_metrics(metrics)
        self.skiplist = set()
        self.tolerance = t

        self.workloads = [x for x in workload.split(",") if x]
        self.wlidx = 0  # idx of current workloads
        self.allresults = dict()  # metric results of all workload
        self.alltotalcnt = dict()
        self.allpassedcnt = dict()

        self.results = dict()  # metric results of current workload
        # vars for test pass/failure statistics
        # metrics with no results or negative results, neg result counts failed tests
        self.ignoremetrics = set()
        self.totalcnt = 0
        self.passedcnt = 0
        # vars for errors
        self.errlist = list()

        # vars for Rule Generator
        self.pctgmetrics = set()  # Percentage rule

        # vars for debug
        self.datafname = datafname
        self.debug = debug
        self.fullrulefname = fullrulefname

    def __set_metrics(self, metrics=''):
        if metrics != '':
            return set(metrics.split(","))
        else:
            return set()

    def read_json(self, filename: str) -> dict:
        try:
            with open(Path(filename).resolve(), "r") as f:
                data = json.loads(f.read())
        except OSError as e:
            print(f"Error when reading file {e}")
            sys.exit()

        return data

    def json_dump(self, data, output_file):
        parent = Path(output_file).parent
        if not parent.exists():
            parent.mkdir(parents=True)

        with open(output_file, "w+") as output_file:
            json.dump(data,
                      output_file,
                      ensure_ascii=True,
                      indent=4)

    def get_results(self, idx: int = 0):
        return self.results.get(idx)

    def get_bounds(self, lb, ub, error, alias={}, ridx: int = 0) -> list:
        """
        Get bounds and tolerance from lb, ub, and error.
        If missing lb, use 0.0; missing ub, use float('inf); missing error, use self.tolerance.

        @param lb: str/float, lower bound
        @param ub: str/float, upper bound
        @param error: float/str, error tolerance
        @returns: lower bound, return inf if the lower bound is a metric value and is not collected
                  upper bound, return -1 if the upper bound is a metric value and is not collected
                  tolerance, denormalized base on upper bound value
        """
        # init ubv and lbv to invalid values
        def get_bound_value(bound, initval, ridx):
            val = initval
            if isinstance(bound, int) or isinstance(bound, float):
                val = bound
            elif isinstance(bound, str):
                if bound == '':
                    val = float("inf")
                elif bound in alias:
                    vall = self.get_value(alias[ub], ridx)
                    if vall:
                        val = vall[0]
                elif bound.replace('.', '1').isdigit():
                    val = float(bound)
                else:
                    print("Wrong bound: {0}".format(bound))
            else:
                print("Wrong bound: {0}".format(bound))
            return val

        ubv = get_bound_value(ub, -1, ridx)
        lbv = get_bound_value(lb, float('inf'), ridx)
        t = get_bound_value(error, self.tolerance, ridx)

        # denormalize error threshold
        denormerr = t * ubv / 100 if ubv != 100 and ubv > 0 else t

        return lbv, ubv, denormerr

    def get_value(self, name: str, ridx: int = 0) -> list:
        """
        Get value of the metric from self.results.
        If result of this metric is not provided, the metric name will be added into self.ignoremetics.
        All future test(s) on this metric will fail.

        @param name: name of the metric
        @returns: list with value found in self.results; list is empty when value is not found.
        """
        results = []
        data = self.results[ridx] if ridx in self.results else self.results[0]
        if name not in self.ignoremetrics:
            if name in data:
                results.append(data[name])
            elif name.replace('.', '1').isdigit():
                results.append(float(name))
            else:
                self.ignoremetrics.add(name)
        return results

    def check_bound(self, val, lb, ub, err):
        return True if val <= ub + err and val >= lb - err else False

    # Positive Value Sanity check
    def pos_val_test(self):
        """
        Check if metrics value are non-negative.
        One metric is counted as one test.
        Failure: when metric value is negative or not provided.
        Metrics with negative value will be added into self.ignoremetrics.
        """
        negmetric = dict()
        pcnt = 0
        tcnt = 0
        rerun = list()
        results = self.get_results()
        if not results:
            return
        for name, val in results.items():
            if val < 0:
                negmetric[name] = val
                rerun.append(name)
            else:
                pcnt += 1
            tcnt += 1
        # The first round collect_perf() run these metrics with simple workload
        # "true". We give metrics a second chance with a longer workload if less
        # than 20 metrics failed positive test.
        if len(rerun) > 0 and len(rerun) < 20:
            second_results = dict()
            self.second_test(rerun, second_results)
            for name, val in second_results.items():
                if name not in negmetric:
                    continue
                if val >= 0:
                    del negmetric[name]
                    pcnt += 1

        if len(negmetric.keys()):
            self.ignoremetrics.update(negmetric.keys())
            self.errlist.extend(
                [TestError([m], self.workloads[self.wlidx], negmetric[m], 0) for m in negmetric.keys()])

        return

    def evaluate_formula(self, formula: str, alias: dict, ridx: int = 0):
        """
        Evaluate the value of formula.

        @param formula: the formula to be evaluated
        @param alias: the dict has alias to metric name mapping
        @returns: value of the formula is success; -1 if the one or more metric value not provided
        """
        stack = []
        b = 0
        errs = []
        sign = "+"
        f = str()

        # TODO: support parenthesis?
        for i in range(len(formula)):
            if i+1 == len(formula) or formula[i] in ('+', '-', '*', '/'):