diff options
Diffstat (limited to 'resources/tools/presentation/input_data_parser.py')
| -rw-r--r-- | resources/tools/presentation/input_data_parser.py | 1862 |
1 files changed, 0 insertions, 1862 deletions
diff --git a/resources/tools/presentation/input_data_parser.py b/resources/tools/presentation/input_data_parser.py deleted file mode 100644 index 9e61c09d53..0000000000 --- a/resources/tools/presentation/input_data_parser.py +++ /dev/null @@ -1,1862 +0,0 @@ -# Copyright (c) 2023 Cisco and/or its affiliates. -# 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. - -"""Data pre-processing - -- extract data from output.xml files generated by Jenkins jobs and store in - pandas' Series, -- provide access to the data. -- filter the data using tags, -""" - -import re -import copy -import resource -import logging - -from collections import OrderedDict -from os import remove, walk, listdir -from os.path import isfile, isdir, join -from datetime import datetime as dt -from datetime import timedelta -from json import loads -from json.decoder import JSONDecodeError - -import hdrh.histogram -import hdrh.codec -import prettytable -import pandas as pd - -from robot.api import ExecutionResult, ResultVisitor -from robot import errors - -from resources.libraries.python import jumpavg -from input_data_files import download_and_unzip_data_file -from pal_errors import PresentationError - - -# Separator used in file names -SEPARATOR = "__" - - -class ExecutionChecker(ResultVisitor): - """Class to traverse through the test suite structure. - """ - - REGEX_PLR_RATE = re.compile( - r'PLRsearch lower bound::?\s(\d+.\d+).*\n' - r'PLRsearch upper bound::?\s(\d+.\d+)' - ) - REGEX_NDRPDR_RATE = re.compile( - r'NDR_LOWER:\s(\d+.\d+).*\n.*\n' - r'NDR_UPPER:\s(\d+.\d+).*\n' - r'PDR_LOWER:\s(\d+.\d+).*\n.*\n' - r'PDR_UPPER:\s(\d+.\d+)' - ) - REGEX_NDRPDR_GBPS = re.compile( - r'NDR_LOWER:.*,\s(\d+.\d+).*\n.*\n' - r'NDR_UPPER:.*,\s(\d+.\d+).*\n' - r'PDR_LOWER:.*,\s(\d+.\d+).*\n.*\n' - r'PDR_UPPER:.*,\s(\d+.\d+)' - ) - REGEX_PERF_MSG_INFO = re.compile( - r'NDR_LOWER:\s(\d+.\d+)\s.*\s(\d+.\d+)\s.*\n.*\n.*\n' - r'PDR_LOWER:\s(\d+.\d+)\s.*\s(\d+.\d+)\s.*\n.*\n.*\n' - r'Latency at 90% PDR:.*\[\'(.*)\', \'(.*)\'\].*\n' - r'Latency at 50% PDR:.*\[\'(.*)\', \'(.*)\'\].*\n' - r'Latency at 10% PDR:.*\[\'(.*)\', \'(.*)\'\].*\n' - ) - REGEX_CPS_MSG_INFO = re.compile( - r'NDR_LOWER:\s(\d+.\d+)\s.*\s.*\n.*\n.*\n' - r'PDR_LOWER:\s(\d+.\d+)\s.*\s.*\n.*\n.*' - ) - REGEX_PPS_MSG_INFO = re.compile( - r'NDR_LOWER:\s(\d+.\d+)\s.*\s(\d+.\d+)\s.*\n.*\n.*\n' - r'PDR_LOWER:\s(\d+.\d+)\s.*\s(\d+.\d+)\s.*\n.*\n.*' - ) - REGEX_MRR_MSG_INFO = re.compile(r'.*\[(.*)\]') - - REGEX_VSAP_MSG_INFO = re.compile( - r'Transfer Rate: (\d*.\d*).*\n' - r'Latency: (\d*.\d*).*\n' - r'Connection [c|r]ps rate: (\d*).*\n' - r'Total data transferred: (\d*).*\n' - r'Completed requests: (\d*).*\n' - r'Failed requests:\s*(\d*.\d*)' - ) - - # Needed for CPS and PPS tests - REGEX_NDRPDR_LAT_BASE = re.compile( - r'LATENCY.*\[\'(.*)\', \'(.*)\'\]\s\n.*\n.*\n' - r'LATENCY.*\[\'(.*)\', \'(.*)\'\]' - ) - REGEX_NDRPDR_LAT = re.compile( - r'LATENCY.*\[\'(.*)\', \'(.*)\'\]\s\n.*\n.*\n' - r'LATENCY.*\[\'(.*)\', \'(.*)\'\]\s\n.*\n' - r'Latency.*\[\'(.*)\', \'(.*)\'\]\s\n' - r'Latency.*\[\'(.*)\', \'(.*)\'\]\s\n' - r'Latency.*\[\'(.*)\', \'(.*)\'\]\s\n' - r'Latency.*\[\'(.*)\', \'(.*)\'\]' - ) - - REGEX_VERSION_VPP = re.compile( - r"(VPP Version:\s*|VPP version:\s*)(.*)" - ) - REGEX_VERSION_DPDK = re.compile( - r"(DPDK version:\s*|DPDK Version:\s*)(.*)" - ) - REGEX_TCP = re.compile( - r'Total\s(rps|cps|throughput):\s(\d*).*$' - ) - REGEX_MRR = re.compile( - r'MaxReceivedRate_Results\s\[pkts/(\d*)sec\]:\s' - r'tx\s(\d*),\srx\s(\d*)' - ) - REGEX_BMRR = re.compile( - r'.*trial results.*: \[(.*)\]' - ) - REGEX_RECONF_LOSS = re.compile( - r'Packets lost due to reconfig: (\d*)' - ) - REGEX_RECONF_TIME = re.compile( - r'Implied time lost: (\d*.[\de-]*)' - ) - REGEX_TC_TAG = re.compile(r'\d+[tT]\d+[cC]') - - REGEX_TC_NAME_NEW = re.compile(r'-\d+[cC]-') - - REGEX_TC_NUMBER = re.compile(r'tc\d{2}-') - - REGEX_TC_PAPI_CLI = re.compile(r'.*\((\d+.\d+.\d+.\d+.) - (.*)\)') - - REGEX_SH_RUN_HOST = re.compile( - r'hostname=\"(\d{1,3}.\d{1,3}.\d{1,3}.\d{1,3})\",hook=\"(.*)\"' - ) - - def __init__(self, metadata, mapping, ignore, process_oper): - """Initialisation. - - :param metadata: Key-value pairs to be included in "metadata" part of - JSON structure. - :param mapping: Mapping of the old names of test cases to the new - (actual) one. - :param ignore: List of TCs to be ignored. - :param process_oper: If True, operational data (show run, telemetry) is - processed. - :type metadata: dict - :type mapping: dict - :type ignore: list - :type process_oper: bool - """ - - # Mapping of TCs long names - self._mapping = mapping - - # Ignore list - self._ignore = ignore - - # Process operational data - self._process_oper = process_oper - - # Name of currently processed keyword - self._kw_name = None - - # VPP version - self._version = None - - # Timestamp - self._timestamp = None - - # Testbed. The testbed is identified by TG node IP address. - self._testbed = None - - # Number of PAPI History messages found: - # 0 - no message - # 1 - PAPI History of DUT1 - # 2 - PAPI History of DUT2 - self._conf_history_lookup_nr = 0 - - self._sh_run_counter = 0 - self._telemetry_kw_counter = 0 - self._telemetry_msg_counter = 0 - - # Test ID of currently processed test- the lowercase full path to the - # test - self._test_id = None - - # The main data structure - self._data = { - "metadata": dict(), - "suites": dict(), - "tests": dict() - } - - # Save the provided metadata - for key, val in metadata.items(): - self._data["metadata"][key] = val - - @property - def data(self): - """Getter - Data parsed from the XML file. - - :returns: Data parsed from the XML file. - :rtype: dict - """ - return self._data - - def _get_data_from_mrr_test_msg(self, msg): - """Get info from message of MRR performance tests. - - :param msg: Message to be processed. - :type msg: str - :returns: Processed message or original message if a problem occurs. - :rtype: str - """ - - groups = re.search(self.REGEX_MRR_MSG_INFO, msg) - if not groups or groups.lastindex != 1: - return "Test Failed." - - try: - data = groups.group(1).split(", ") - except (AttributeError, IndexError, ValueError, KeyError): - return "Test Failed." - - out_str = "[" - try: - for item in data: - out_str += f"{(float(item) / 1e6):.2f}, " - return out_str[:-2] + "]" - except (AttributeError, IndexError, ValueError, KeyError): - return "Test Failed." - - def _get_data_from_cps_test_msg(self, msg): - """Get info from message of NDRPDR CPS tests. - - :param msg: Message to be processed. - :type msg: str - :returns: Processed message or "Test Failed." if a problem occurs. - :rtype: str - """ - - groups = re.search(self.REGEX_CPS_MSG_INFO, msg) - if not groups or groups.lastindex != 2: - return "Test Failed." - - try: - return ( - f"1. {(float(groups.group(1)) / 1e6):5.2f}\n" - f"2. {(float(groups.group(2)) / 1e6):5.2f}" - ) - except (AttributeError, IndexError, ValueError, KeyError): - return "Test Failed." - - def _get_data_from_pps_test_msg(self, msg): - """Get info from message of NDRPDR PPS tests. - - :param msg: Message to be processed. - :type msg: str - :returns: Processed message or "Test Failed." if a problem occurs. - :rtype: str - """ - - groups = re.search(self.REGEX_PPS_MSG_INFO, msg) - if not groups or groups.lastindex != 4: - return "Test Failed." - - try: - return ( - f"1. {(float(groups.group(1)) / 1e6):5.2f} " - f"{float(groups.group(2)):5.2f}\n" - f"2. {(float(groups.group(3)) / 1e6):5.2f} " - f"{float(groups.group(4)):5.2f}" - ) - except (AttributeError, IndexError, ValueError, KeyError): - return "Test Failed." - - def _get_data_from_perf_test_msg(self, msg): - """Get info from message of NDRPDR performance tests. - - :param msg: Message to be processed. - :type msg: str - :returns: Processed message or "Test Failed." if a problem occurs. - :rtype: str - """ - - groups = re.search(self.REGEX_PERF_MSG_INFO, msg) - if not groups or groups.lastindex != 10: - return "Test Failed." - - try: - data = { - "ndr_low": float(groups.group(1)), - "ndr_low_b": float(groups.group(2)), - "pdr_low": float(groups.group(3)), - "pdr_low_b": float(groups.group(4)), - "pdr_lat_90_1": groups.group(5), - "pdr_lat_90_2": groups.group(6), - "pdr_lat_50_1": groups.group(7), - "pdr_lat_50_2": groups.group(8), - "pdr_lat_10_1": groups.group(9), - "pdr_lat_10_2": groups.group(10), - } - except (AttributeError, IndexError, ValueError, KeyError): - return "Test Failed." - - def _process_lat(in_str_1, in_str_2): - """Extract P50, P90 and P99 latencies or min, avg, max values from - latency string. - - :param in_str_1: Latency string for one direction produced by robot - framework. - :param in_str_2: Latency string for second direction produced by - robot framework. - :type in_str_1: str - :type in_str_2: str - :returns: Processed latency string or None if a problem occurs. - :rtype: tuple - """ - in_list_1 = in_str_1.split('/', 3) - in_list_2 = in_str_2.split('/', 3) - - if len(in_list_1) != 4 and len(in_list_2) != 4: - return None - - in_list_1[3] += "=" * (len(in_list_1[3]) % 4) - try: - hdr_lat_1 = hdrh.histogram.HdrHistogram.decode(in_list_1[3]) - except hdrh.codec.HdrLengthException: - hdr_lat_1 = None - - in_list_2[3] += "=" * (len(in_list_2[3]) % 4) - try: - hdr_lat_2 = hdrh.histogram.HdrHistogram.decode(in_list_2[3]) - except hdrh.codec.HdrLengthException: - hdr_lat_2 = None - - if hdr_lat_1 and hdr_lat_2: - hdr_lat = ( - hdr_lat_1.get_value_at_percentile(50.0), - hdr_lat_1.get_value_at_percentile(90.0), - hdr_lat_1.get_value_at_percentile(99.0), - hdr_lat_2.get_value_at_percentile(50.0), - hdr_lat_2.get_value_at_percentile(90.0), - hdr_lat_2.get_value_at_percentile(99.0) - ) - if all(hdr_lat): - return hdr_lat - - hdr_lat = ( - int(in_list_1[0]), int(in_list_1[1]), int(in_list_1[2]), - int(in_list_2[0]), int(in_list_2[1]), int(in_list_2[2]) - ) - for item in hdr_lat: - if item in (-1, 4294967295, 0): - return None - return hdr_lat - - try: - out_msg = ( - f"1. {(data['ndr_low'] / 1e6):5.2f} " - f"{data['ndr_low_b']:5.2f}" - f"\n2. {(data['pdr_low'] / 1e6):5.2f} " - f"{data['pdr_low_b']:5.2f}" - ) - latency = ( - _process_lat(data['pdr_lat_10_1'], data['pdr_lat_10_2']), - _process_lat(data['pdr_lat_50_1'], data['pdr_lat_50_2']), - _process_lat(data['pdr_lat_90_1'], data['pdr_lat_90_2']) - ) - if all(latency): - max_len = len(str(max((max(item) for item in latency)))) - max_len = 4 if max_len < 4 else max_len - - for idx, lat in enumerate(latency): - if not idx: - out_msg += "\n" - out_msg += ( - f"\n{idx + 3}. " - f"{lat[0]:{max_len}d} " - f"{lat[1]:{max_len}d} " - f"{lat[2]:{max_len}d} " - f"{lat[3]:{max_len}d} " - f"{lat[4]:{max_len}d} " - f"{lat[5]:{max_len}d} " - ) - - return out_msg - - except (AttributeError, IndexError, ValueError, KeyError): - return "Test Failed." - - def _get_testbed(self, msg): - """Called when extraction of testbed IP is required. - The testbed is identified by TG node IP address. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - - if msg.message.count("Setup of TG node") or \ - msg.message.count("Setup of node TG host"): - reg_tg_ip = re.compile( - r'.*TG .* (\d{1,3}.\d{1,3}.\d{1,3}.\d{1,3}).*') - try: - self._testbed = str(re.search(reg_tg_ip, msg.message).group(1)) - except (KeyError, ValueError, IndexError, AttributeError): - pass - finally: - self._data["metadata"]["testbed"] = self._testbed - - def _get_vpp_version(self, msg): - """Called when extraction of VPP version is required. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - - if msg.message.count("VPP version:") or \ - msg.message.count("VPP Version:"): - self._version = str( - re.search(self.REGEX_VERSION_VPP, msg.message).group(2) - ) - self._data["metadata"]["version"] = self._version - - def _get_dpdk_version(self, msg): - """Called when extraction of DPDK version is required. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - - if msg.message.count("DPDK Version:"): - try: - self._version = str(re.search( - self.REGEX_VERSION_DPDK, msg.message).group(2)) - self._data["metadata"]["version"] = self._version - except IndexError: - pass - - def _get_papi_history(self, msg): - """Called when extraction of PAPI command history is required. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - if msg.message.count("PAPI command history:"): - self._conf_history_lookup_nr += 1 - if self._conf_history_lookup_nr == 1: - self._data["tests"][self._test_id]["conf-history"] = str() - text = re.sub( - r"\d{1,3}.\d{1,3}.\d{1,3}.\d{1,3} PAPI command history:", - "", - msg.message, - count=1 - ).replace('"', "'") - self._data["tests"][self._test_id]["conf-history"] += \ - f"**DUT{str(self._conf_history_lookup_nr)}:** {text}" - - def _get_show_run(self, msg): - """Called when extraction of VPP operational data (output of CLI command - Show Runtime) is required. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - - if not msg.message.count("stats runtime"): - return - - # Temporary solution - if self._sh_run_counter > 1: - return - - if "show-run" not in self._data["tests"][self._test_id].keys(): - self._data["tests"][self._test_id]["show-run"] = dict() - - groups = re.search(self.REGEX_TC_PAPI_CLI, msg.message) - if not groups: - return - try: - host = groups.group(1) - except (AttributeError, IndexError): - host = "" - try: - sock = groups.group(2) - except (AttributeError, IndexError): - sock = "" - - dut = "dut{nr}".format( - nr=len(self._data['tests'][self._test_id]['show-run'].keys()) + 1) - - self._data['tests'][self._test_id]['show-run'][dut] = \ - copy.copy( - { - "host": host, - "socket": sock, - "runtime": str(msg.message).replace(' ', ''). - replace('\n', '').replace("'", '"'). - replace('b"', '"').replace('"', '"'). - split(":", 1)[1] - } - ) - - def _get_telemetry(self, msg): - """Called when extraction of VPP telemetry data is required. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - - if self._telemetry_kw_counter > 1: - return - if not msg.message.count("# TYPE vpp_runtime_calls"): - return - - if "telemetry-show-run" not in \ - self._data["tests"][self._test_id].keys(): - self._data["tests"][self._test_id]["telemetry-show-run"] = dict() - - self._telemetry_msg_counter += 1 - groups = re.search(self.REGEX_SH_RUN_HOST, msg.message) - if not groups: - return - try: - host = groups.group(1) - except (AttributeError, IndexError): - host = "" - try: - sock = groups.group(2) - except (AttributeError, IndexError): - sock = "" - runtime = { - "source_type": "node", - "source_id": host, - "msg_type": "metric", - "log_level": "INFO", - "timestamp": msg.timestamp, - "msg": "show_runtime", - "host": host, - "socket": sock, - "data": list() - } - for line in msg.message.splitlines(): - if not line.startswith("vpp_runtime_"): - continue - try: - params, value, timestamp = line.rsplit(" ", maxsplit=2) - cut = params.index("{") - name = params[:cut].split("_", maxsplit=2)[-1] - labels = eval( - "dict" + params[cut:].replace('{', '(').replace('}', ')') - ) - labels["graph_node"] = labels.pop("name") - runtime["data"].append( - { - "name": name, - "value": value, - "timestamp": timestamp, - "labels": labels - } - ) - except (TypeError, ValueError, IndexError): - continue - self._data['tests'][self._test_id]['telemetry-show-run']\ - [f"dut{self._telemetry_msg_counter}"] = copy.copy( - { - "host": host, - "socket": sock, - "runtime": runtime - } - ) - - def _get_ndrpdr_throughput(self, msg): - """Get NDR_LOWER, NDR_UPPER, PDR_LOWER and PDR_UPPER from the test - message. - - :param msg: The test message to be parsed. - :type msg: str - :returns: Parsed data as a dict and the status (PASS/FAIL). - :rtype: tuple(dict, str) - """ - - throughput = { - "NDR": {"LOWER": -1.0, "UPPER": -1.0}, - "PDR": {"LOWER": -1.0, "UPPER": -1.0} - } - status = "FAIL" - groups = re.search(self.REGEX_NDRPDR_RATE, msg) - - if groups is not None: - try: - throughput["NDR"]["LOWER"] = float(groups.group(1)) - throughput["NDR"]["UPPER"] = float(groups.group(2)) - throughput["PDR"]["LOWER"] = float(groups.group(3)) - throughput["PDR"]["UPPER"] = float(groups.group(4)) - status = "PASS" - except (IndexError, ValueError): - pass - - return throughput, status - - def _get_ndrpdr_throughput_gbps(self, msg): - """Get NDR_LOWER, NDR_UPPER, PDR_LOWER and PDR_UPPER in Gbps from the - test message. - - :param msg: The test message to be parsed. - :type msg: str - :returns: Parsed data as a dict and the status (PASS/FAIL). - :rtype: tuple(dict, str) - """ - - gbps = { - "NDR": {"LOWER": -1.0, "UPPER": -1.0}, - "PDR": {"LOWER": -1.0, "UPPER": -1.0} - } - status = "FAIL" - groups = re.search(self.REGEX_NDRPDR_GBPS, msg) - - if groups is not None: - try: - gbps["NDR"]["LOWER"] = float(groups.group(1)) - gbps["NDR"]["UPPER"] = float(groups.group(2)) - gbps["PDR"]["LOWER"] = float(groups.group(3)) - gbps["PDR"]["UPPER"] = float(groups.group(4)) - status = "PASS" - except (IndexError, ValueError): - pass - - return gbps, status - - def _get_plr_throughput(self, msg): - """Get PLRsearch lower bound and PLRsearch upper bound from the test - message. - - :param msg: The test message to be parsed. - :type msg: str - :returns: Parsed data as a dict and the status (PASS/FAIL). - :rtype: tuple(dict, str) - """ - - throughput = { - "LOWER": -1.0, - "UPPER": -1.0 - } - status = "FAIL" - groups = re.search(self.REGEX_PLR_RATE, msg) - - if groups is not None: - try: - throughput["LOWER"] = float(groups.group(1)) - throughput["UPPER"] = float(groups.group(2)) - status = "PASS" - except (IndexError, ValueError): - pass - - return throughput, status - - def _get_ndrpdr_latency(self, msg): - """Get LATENCY from the test message. - - :param msg: The test message to be parsed. - :type msg: str - :returns: Parsed data as a dict and the status (PASS/FAIL). - :rtype: tuple(dict, str) - """ - latency_default = { - "min": -1.0, - "avg": -1.0, - "max": -1.0, - "hdrh": "" - } - latency = { - "NDR": { - "direction1": copy.copy(latency_default), - "direction2": copy.copy(latency_default) - }, - "PDR": { - "direction1": copy.copy(latency_default), - "direction2": copy.copy(latency_default) - }, - "LAT0": { - "direction1": copy.copy(latency_default), - "direction2": copy.copy(latency_default) - }, - "PDR10": { - "direction1": copy.copy(latency_default), - "direction2": copy.copy(latency_default) - }, - "PDR50": { - "direction1": copy.copy(latency_default), - "direction2": copy.copy(latency_default) - }, - "PDR90": { - "direction1": copy.copy(latency_default), - "direction2": copy.copy(latency_default) - }, - } - - groups = re.search(self.REGEX_NDRPDR_LAT, msg) - if groups is None: - groups = re.search(self.REGEX_NDRPDR_LAT_BASE, msg) - if groups is None: - return latency, "FAIL" - - def process_latency(in_str): - """Return object with parsed latency values. - - TODO: Define class for the return type. - - :param in_str: Input string, min/avg/max/hdrh format. - :type in_str: str - :returns: Dict with corresponding keys, except hdrh float values. - :rtype dict: - :throws IndexError: If in_str does not have enough substrings. - :throws ValueError: If a substring does not convert to float. - """ - in_list = in_str.split('/', 3) - - rval = { - "min": float(in_list[0]), - "avg": float(in_list[1]), - "max": float(in_list[2]), - "hdrh": "" - } - - if len(in_list) == 4: - rval["hdrh"] = str(in_list[3]) - - return rval - - try: - latency["NDR"]["direction1"] = process_latency(groups.group(1)) - latency["NDR"]["direction2"] = process_latency(groups.group(2)) - latency["PDR"]["direction1"] = process_latency(groups.group(3)) - latency["PDR"]["direction2"] = process_latency(groups.group(4)) - if groups.lastindex == 4: - return latency, "PASS" - except (IndexError, ValueError): - pass - - try: - latency["PDR90"]["direction1"] = process_latency(groups.group(5)) - latency["PDR90"]["direction2"] = process_latency(groups.group(6)) - latency["PDR50"]["direction1"] = process_latency(groups.group(7)) - latency["PDR50"]["direction2"] = process_latency(groups.group(8)) - latency["PDR10"]["direction1"] = process_latency(groups.group(9)) - latency["PDR10"]["direction2"] = process_latency(groups.group(10)) - latency["LAT0"]["direction1"] = process_latency(groups.group(11)) - latency["LAT0"]["direction2"] = process_latency(groups.group(12)) - if groups.lastindex == 12: - return latency, "PASS" - except (IndexError, ValueError): - pass - - return latency, "FAIL" - - @staticmethod - def _get_hoststack_data(msg, tags): - """Get data from the hoststack test message. - - :param msg: The test message to be parsed. - :param tags: Test tags. - :type msg: str - :type tags: list - :returns: Parsed data as a JSON dict and the status (PASS/FAIL). - :rtype: tuple(dict, str) - """ - result = dict() - status = "FAIL" - - msg = msg.replace("'", '"').replace(" ", "") - if "LDPRELOAD" in tags: - try: - result = loads(msg) - status = "PASS" - except JSONDecodeError: - pass - elif "VPPECHO" in tags: - try: - msg_lst = msg.replace("}{", "} {").split(" ") - result = dict( - client=loads(msg_lst[0]), - server=loads(msg_lst[1]) - ) - status = "PASS" - except (JSONDecodeError, IndexError): - pass - - return result, status - - def _get_vsap_data(self, msg, tags): - """Get data from the vsap test message. - - :param msg: The test message to be parsed. - :param tags: Test tags. - :type msg: str - :type tags: list - :returns: Parsed data as a JSON dict and the status (PASS/FAIL). - :rtype: tuple(dict, str) - """ - result = dict() - status = "FAIL" - - groups = re.search(self.REGEX_VSAP_MSG_INFO, msg) - if groups is not None: - try: - result["transfer-rate"] = float(groups.group(1)) * 1e3 - result["latency"] = float(groups.group(2)) - result["completed-requests"] = int(groups.group(5)) - result["failed-requests"] = int(groups.group(6)) - result["bytes-transferred"] = int(groups.group(4)) - if "TCP_CPS" in tags: - result["cps"] = float(groups.group(3)) - elif "TCP_RPS" in tags: - result["rps"] = float(groups.group(3)) - else: - return result, status - status = "PASS" - except (IndexError, ValueError) as err: - logging.warning(err) - return result, status - - def visit_suite(self, suite): - """Implements traversing through the suite and its direct children. - - :param suite: Suite to process. - :type suite: Suite - :returns: Nothing. - """ - if self.start_suite(suite) is not False: - suite.setup.visit(self) - suite.suites.visit(self) - suite.tests.visit(self) - suite.teardown.visit(self) - self.end_suite(suite) - - def start_suite(self, suite): - """Called when suite starts. - - :param suite: Suite to process. - :type suite: Suite - :returns: Nothing. - """ - try: - parent_name = suite.parent.name - except AttributeError: - return - - self._data["suites"][suite.longname.lower().replace('"', "'").\ - replace(" ", "_")] = { - "name": suite.name.lower(), - "doc": suite.doc, - "parent": parent_name, - "level": len(suite.longname.split(".")) - } - - def visit_test(self, test): - """Implements traversing through the test. - - :param test: Test to process. - :type test: Test - :returns: Nothing. - """ - if self.start_test(test) is not False: - test.setup.visit(self) - test.body.visit(self) - test.teardown.visit(self) - self.end_test(test) - - def start_test(self, test): - """Called when test starts. - - :param test: Test to process. - :type test: Test - :returns: Nothing. - """ - - self._sh_run_counter = 0 - self._telemetry_kw_counter = 0 - self._telemetry_msg_counter = 0 - - longname_orig = test.longname.lower() - - # Check the ignore list - if longname_orig in self._ignore: - return - - tags = [str(tag) for tag in test.tags] - test_result = dict() - - # Change the TC long name and name if defined in the mapping table - longname = self._mapping.get(longname_orig, None) - if longname is not None: - name = longname.split('.')[-1] - else: - longname = longname_orig - name = test.name.lower() - - # Remove TC number from the TC long name (backward compatibility): - self._test_id = re.sub(self.REGEX_TC_NUMBER, "", longname) - # Remove TC number from the TC name (not needed): - test_result["name"] = re.sub(self.REGEX_TC_NUMBER, "", name) - - test_result["parent"] = test.parent.name.lower() - test_result["tags"] = tags - test_result["doc"] = test.doc - test_result["type"] = "" - test_result["status"] = test.status - test_result["starttime"] = test.starttime - test_result["endtime"] = test.endtime - - if test.status == "PASS": - if "NDRPDR" in tags: - if "TCP_PPS" in tags or "UDP_PPS" in tags: - test_result["msg"] = self._get_data_from_pps_test_msg( - test.message) - elif "TCP_CPS" in tags or "UDP_CPS" in tags: - test_result["msg"] = self._get_data_from_cps_test_msg( - test.message) - else: - test_result["msg"] = self._get_data_from_perf_test_msg( - test.message) - elif "MRR" in tags or "FRMOBL" in tags or "BMRR" in tags: - test_result["msg"] = self._get_data_from_mrr_test_msg( - test.message) - else: - test_result["msg"] = test.message - else: - test_result["msg"] = test.message - - if "PERFTEST" in tags and "TREX" not in tags: - # Replace info about cores (e.g. -1c-) with the info about threads - # and cores (e.g. -1t1c-) in the long test case names and in the - # test case names if necessary. - tag_count = 0 - tag_tc = str() - for tag in test_result["tags"]: - groups = re.search(self.REGEX_TC_TAG, tag) - if groups: - tag_count += 1 - tag_tc = tag - - if tag_count == 1: - self._test_id = re.sub( - self.REGEX_TC_NAME_NEW, f"-{tag_tc.lower()}-", - self._test_id, count=1 - ) - test_result["name"] = re.sub( - self.REGEX_TC_NAME_NEW, f"-{tag_tc.lower()}-", - test_result["name"], count=1 - ) - else: - test_result["status"] = "FAIL" - self._data["tests"][self._test_id] = test_result - logging.debug( - f"The test {self._test_id} has no or more than one " - f"multi-threading tags.\n" - f"Tags: {test_result['tags']}" - ) - return - - if "DEVICETEST" in tags: - test_result["type"] = "DEVICETEST" - elif "NDRPDR" in tags: - if "TCP_CPS" in tags or "UDP_CPS" in tags: - test_result["type"] = "CPS" - else: - test_result["type"] = "NDRPDR" - if test.status == "PASS": - test_result["throughput"], test_result["status"] = \ - self._get_ndrpdr_throughput(test.message) - test_result["gbps"], test_result["status"] = \ - self._get_ndrpdr_throughput_gbps(test.message) - test_result["latency"], test_result["status"] = \ - self._get_ndrpdr_latency(test.message) - elif "MRR" in tags or "FRMOBL" in tags or "BMRR" in tags: - if "MRR" in tags: - test_result["type"] = "MRR" - else: - test_result["type"] = "BMRR" - if test.status == "PASS": - test_result["result"] = dict() - groups = re.search(self.REGEX_BMRR, test.message) - if groups is not None: - items_str = groups.group(1) - items_float = [ - float(item.strip().replace("'", "")) - for item in items_str.split(",") - ] - # Use whole list in CSIT-1180. - stats = jumpavg.AvgStdevStats.for_runs(items_float) - test_result["result"]["samples"] = items_float - test_result["result"]["receive-rate"] = stats.avg - test_result["result"]["receive-stdev"] = stats.stdev - else: - groups = re.search(self.REGEX_MRR, test.message) - test_result["result"]["receive-rate"] = \ - float(groups.group(3)) / float(groups.group(1)) - elif "SOAK" in tags: - test_result["type"] = "SOAK" - if test.status == "PASS": - test_result["throughput"], test_result["status"] = \ - self._get_plr_throughput(test.message) - elif "LDP_NGINX" in tags: - test_result["type"] = "LDP_NGINX" - test_result["result"], test_result["status"] = \ - self._get_vsap_data(test.message, tags) - elif "HOSTSTACK" in tags: - test_result["type"] = "HOSTSTACK" - if test.status == "PASS": - test_result["result"], test_result["status"] = \ - self._get_hoststack_data(test.message, tags) - elif "RECONF" in tags: - test_result["type"] = "RECONF" - if test.status == "PASS": - test_result["result"] = None - try: - grps_loss = re.search(self.REGEX_RECONF_LOSS, test.message) - grps_time = re.search(self.REGEX_RECONF_TIME, test.message) - test_result["result"] = { - "loss": int(grps_loss.group(1)), - "time": float(grps_time.group(1)) - } - except (AttributeError, IndexError, ValueError, TypeError): - test_result["status"] = "FAIL" - else: - test_result["status"] = "FAIL" - - self._data["tests"][self._test_id] = test_result - - def visit_keyword(self, kw): - """Implements traversing through the keyword and its child keywords. - - :param keyword: Keyword to process. - :type keyword: Keyword - :returns: Nothing. - """ - if self.start_keyword(kw) is not False: - if hasattr(kw, "body"): - kw.body.visit(self) - kw.teardown.visit(self) - self.end_keyword(kw) - - def start_keyword(self, keyword): - """Called when keyword starts. Default implementation does nothing. - - :param keyword: Keyword to process. - :type keyword: Keyword - :returns: Nothing. - """ - self._kw_name = keyword.name - - def end_keyword(self, keyword): - """Called when keyword ends. Default implementation does nothing. - - :param keyword: Keyword to process. - :type keyword: Keyword - :returns: Nothing. - """ - _ = keyword - self._kw_name = None - - def visit_message(self, msg): - """Implements visiting the message. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - if self.start_message(msg) is not False: - self.end_message(msg) - - def start_message(self, msg): - """Called when message starts. Get required information from messages: - - VPP version. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - if self._kw_name is None: - return - elif self._kw_name.count("Run Telemetry On All Duts"): - if self._process_oper: - self._telemetry_kw_counter += 1 - self._get_telemetry(msg) - elif self._kw_name.count("Show Runtime On All Duts"): - if self._process_oper: - self._sh_run_counter += 1 - self._get_show_run(msg) - elif self._kw_name.count("Show Vpp Version On All Duts"): - if not self._version: - self._get_vpp_version(msg) - elif self._kw_name.count("Install Dpdk Framework On All Duts"): - if not self._version: - self._get_dpdk_version(msg) - elif self._kw_name.count("Setup Framework"): - if not self._testbed: - self._get_testbed(msg) - elif self._kw_name.count("Show Papi History On All Duts"): - self._conf_history_lookup_nr = 0 - self._get_papi_history(msg) - - -class InputData: - """Input data - - The data is extracted from output.xml files generated by Jenkins jobs and - stored in pandas' DataFrames. - - The data structure: - - job name - - build number - - metadata - (as described in ExecutionChecker documentation) - - suites - (as described in ExecutionChecker documentation) - - tests - (as described in ExecutionChecker documentation) - """ - - def __init__(self, spec, for_output): - """Initialization. - - :param spec: Specification. - :param for_output: Output to be generated from downloaded data. - :type spec: Specification - :type for_output: str - """ - - # Specification: - self._cfg = spec - - self._for_output = for_output - - # Data store: - self._input_data = pd.Series(dtype="float64") - - @property - def data(self): - """Getter - Input data. - - :returns: Input data - :rtype: pandas.Series - """ - return self._input_data - - def metadata(self, job, build): - """Getter - metadata - - :param job: Job which metadata we want. - :param build: Build which metadata we want. - :type job: str - :type build: str - :returns: Metadata - :rtype: pandas.Series - """ - return self.data[job][build]["metadata"] - - def suites(self, job, build): - """Getter - suites - - :param job: Job which suites we want. - :param build: Build which suites we want. - :type job: str - :type build: str - :returns: Suites. - :rtype: pandas.Series - """ - return self.data[job][str(build)]["suites"] - - def tests(self, job, build): - """Getter - tests - - :param job: Job which tests we want. - :param build: Build which tests we want. - :type job: str - :type build: str - :returns: Tests. - :rtype: pandas.Series - """ - return self.data[job][build]["tests"] - - def _parse_tests(self, job, build): - """Process data from robot output.xml file and return JSON structured - data. - - :param job: The name of job which build output data will be processed. - :param build: The build which output data will be processed. - :type job: str - :type build: dict - :returns: JSON data structure. - :rtype: dict - """ - - metadata = { - "job": job, - "build": build - } - - with open(build["file-name"], 'r') as data_file: - try: - result = ExecutionResult(data_file) - except errors.DataError as err: - logging.error( - f"Error occurred while parsing output.xml: {repr(err)}" - ) - return None - - process_oper = False - if "-vpp-perf-report-coverage-" in job: - process_oper = True - # elif "-vpp-perf-report-iterative-" in job: - # # Exceptions for TBs where we do not have coverage data: - # for item in ("-2n-icx", ): - # if item in job: - # process_oper = True - # break - checker = ExecutionChecker( - metadata, self._cfg.mapping, self._cfg.ignore, process_oper - ) - result.visit(checker) - - checker.data["metadata"]["tests_total"] = \ - result.statistics.total.total - checker.data["metadata"]["tests_passed"] = \ - result.statistics.total.passed - checker.data["metadata"]["tests_failed"] = \ - result.statistics.total.failed - checker.data["metadata"]["elapsedtime"] = result.suite.elapsedtime - checker.data["metadata"]["generated"] = result.suite.endtime[:14] - - return checker.data - - def _download_and_parse_build(self, job, build, repeat, pid=10000): - """Download and parse the input data file. - - :param pid: PID of the process executing this method. - :param job: Name of the Jenkins job which generated the processed input - file. - :param build: Information about the Jenkins build which generated the - processed input file. - :param repeat: Repeat the download specified number of times if not - successful. - :type pid: int - :type job: str - :type build: dict - :type repeat: int - """ - - logging.info(f"Processing the job/build: {job}: {build['build']}") - - state = "failed" - success = False - data = None - do_repeat = repeat - while do_repeat: - success = download_and_unzip_data_file(self._cfg, job, build, pid) - if success: - break - do_repeat -= 1 - if not success: - logging.error( - f"It is not possible to download the input data file from the " - f"job {job}, build {build['build']}, or it is damaged. " - f"Skipped." - ) - if success: - logging.info(f" Processing data from build {build['build']}") - data = self._parse_tests(job, build) - if data is None: - logging.error( - f"Input data file from the job {job}, build " - f"{build['build']} is damaged. Skipped." - ) - else: - state = "processed" - - try: - remove(build["file-name"]) - except OSError as err: - logging.error( - f"Cannot remove the file {build['file-name']}: {repr(err)}" - ) - - # If the time-period is defined in the specification file, remove all - # files which are outside the time period. - is_last = False - timeperiod = self._cfg.environment.get("time-period", None) - if timeperiod and data: - now = dt.utcnow() - timeperiod = timedelta(int(timeperiod)) - metadata = data.get("metadata", None) - if metadata: - generated = metadata.get("generated", None) - if generated: - generated = dt.strptime(generated, "%Y%m%d %H:%M") - if (now - generated) > timeperiod: - # Remove the data and the file: - state = "removed" - data = None - is_last = True - logging.info( - f" The build {job}/{build['build']} is " - f"outdated, will be removed." - ) - return { - "data": data, - "state": state, - "job": job, - "build": build, - "last": is_last - } - - def download_and_parse_data(self, repeat=1): - """Download the input data files, parse input data from input files and - store in pandas' Series. - - :param repeat: Repeat the download specified number of times if not - successful. - :type repeat: int - """ - - logging.info("Downloading and parsing input files ...") - - for job, builds in self._cfg.input.items(): - for build in builds: - - result = self._download_and_parse_build(job, build, repeat) - if result["last"]: - break - build_nr = result["build"]["build"] - - if result["data"]: - data = result["data"] - build_data = pd.Series({ - "metadata": pd.Series( - list(data["metadata"].values()), - index=list(data["metadata"].keys()) - ), - "suites": pd.Series( - list(data["suites"].values()), - index=list(data["suites"].keys()) - ), - "tests": pd.Series( - list(data["tests"].values()), - index=list(data["tests"].keys()) - ) - }) - - if self._input_data.get(job, None) is None: - self._input_data[job] = pd.Series(dtype="float64") - self._input_data[job][str(build_nr)] = build_data - self._cfg.set_input_file_name( - job, build_nr, result["build"]["file-name"] - ) - self._cfg.set_input_state(job, build_nr, result["state"]) - - mem_alloc = \ - resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1000 - logging.info(f"Memory allocation: {mem_alloc:.0f}MB") - - logging.info("Done.") - - msg = f"Successful downloads from the sources:\n" - for source in self._cfg.environment["data-sources"]: - if source["successful-downloads"]: - msg += ( - f"{source['url']}/{source['path']}/" - f"{source['file-name']}: " - f"{source['successful-downloads']}\n" - ) - logging.info(msg) - - def process_local_file(self, local_file, job="local", build_nr=1, - replace=True): - """Process local XML file given as a command-line parameter. - - :param local_file: The file to process. - :param job: Job name. - :param build_nr: Build number. - :param replace: If True, the information about jobs and builds is - replaced by the new one, otherwise the new jobs and builds are - added. - :type local_file: str - :type job: str - :type build_nr: int - :type replace: bool - :raises: PresentationError if an error occurs. - """ - if not isfile(local_file): - raise PresentationError(f"The file {local_file} does not exist.") - - try: - build_nr = int(local_file.split("/")[-1].split(".")[0]) - except (IndexError, ValueError): - pass - - build = { - "build": build_nr, - "status": "failed", - "file-name": local_file - } - if replace: - self._cfg.input = dict() - self._cfg.add_build(job, build) - - logging.info(f"Processing {job}: {build_nr:2d}: {local_file}") - data = self._parse_tests(job, build) - if data is None: - raise PresentationError( - f"Error occurred while parsing the file {local_file}" - ) - - build_data = pd.Series({ - "metadata": pd.Series( - list(data["metadata"].values()), - index=list(data["metadata"].keys()) - ), - "suites": pd.Series( - list(data["suites"].values()), - index=list(data["suites"].keys()) - ), - "tests": pd.Series( - list(data["tests"].values()), - index=list(data["tests"].keys()) - ) - }) - - if self._input_data.get(job, None) is None: - self._input_data[job] = pd.Series(dtype="float64") - self._input_data[job][str(build_nr)] = build_data - - self._cfg.set_input_state(job, build_nr, "processed") - - def process_local_directory(self, local_dir, replace=True): - """Process local directory with XML file(s). The directory is processed - as a 'job' and the XML files in it as builds. - If the given directory contains only sub-directories, these - sub-directories processed as jobs and corresponding XML files as builds - of their job. - - :param local_dir: Local directory to process. - :param replace: If True, the information about jobs and builds is - replaced by the new one, otherwise the new jobs and builds are - added. - :type local_dir: str - :type replace: bool - """ - if not isdir(local_dir): - raise PresentationError( - f"The directory {local_dir} does not exist." - ) - - # Check if the given directory includes only files, or only directories - _, dirnames, filenames = next(walk(local_dir)) - - if filenames and not dirnames: - filenames.sort() - # local_builds: - # key: dir (job) name, value: list of file names (builds) - local_builds = { - local_dir: [join(local_dir, name) for name in filenames] - } - - elif dirnames and not filenames: - dirnames.sort() - # local_builds: - # key: dir (job) name, value: list of file names (builds) - local_builds = dict() - for dirname in dirnames: - builds = [ - join(local_dir, dirname, name) - for name in listdir(join(local_dir, dirname)) - if isfile(join(local_dir, dirname, name)) - ] - if builds: - local_builds[dirname] = sorted(builds) - - elif not filenames and not dirnames: - raise PresentationError(f"The directory {local_dir} is empty.") - else: - raise PresentationError( - f"The directory {local_dir} can include only files or only " - f"directories, not both.\nThe directory {local_dir} includes " - f"file(s):\n{filenames}\nand directories:\n{dirnames}" - ) - - if replace: - self._cfg.input = dict() - - for job, files in local_builds.items(): - for idx, local_file in enumerate(files): - self.process_local_file(local_file, job, idx + 1, replace=False) - - @staticmethod - def _end_of_tag(tag_filter, start=0, closer="'"): - """Return the index of character in the string which is the end of tag. - - :param tag_filter: The string where the end of tag is being searched. - :param start: The index where the searching is stated. - :param closer: The character which is the tag closer. - :type tag_filter: str - :type start: int - :type closer: str - :returns: The index of the tag closer. - :rtype: int - """ - try: - idx_opener = tag_filter.index(closer, start) - return tag_filter.index(closer, idx_opener + 1) - except ValueError: - return None - - @staticmethod - def _condition(tag_filter): - """Create a conditional statement from the given tag filter. - - :param tag_filter: Filter based on tags from the element specification. - :type tag_filter: str - :returns: Conditional statement which can be evaluated. - :rtype: str - """ - index = 0 - while True: - index = InputData._end_of_tag(tag_filter, index) - if index is None: - return tag_filter - index += 1 - tag_filter = tag_filter[:index] + " in tags" + tag_filter[index:] - - def filter_data(self, element, params=None, data=None, data_set="tests", - continue_on_error=False): - """Filter required data from the given jobs and builds. - - The output data structure is: - - job 1 - - build 1 - - test (or suite) 1 ID: - - param 1 - - param 2 - ... - - param n - ... - - test (or suite) n ID: - ... - ... - - build n - ... - - job n - - :param element: Element which will use the filtered data. - :param params: Parameters which will be included in the output. If None, - all parameters are included. - :param data: If not None, this data is used instead of data specified - in the element. - :param data_set: The set of data to be filtered: tests, suites, - metadata. - :param continue_on_error: Continue if there is error while reading the - data. The Item will be empty then - :type element: pandas.Series - :type params: list - :type data: dict - :type data_set: str - :type continue_on_error: bool - :returns: Filtered data. - :rtype pandas.Series - """ - - try: - if data_set == "suites": - cond = "True" - elif element["filter"] in ("all", "template"): - cond = "True" - else: - cond = InputData._condition(element["filter"]) - logging.debug(f" Filter: {cond}") - except KeyError: - logging.error(" No filter defined.") - return None - - if params is None: - params = element.get("parameters", None) - if params: - params.extend(("type", "status")) - - data_to_filter = data if data else element["data"] - data = pd.Series(dtype="float64") - try: - for job, builds in data_to_filter.items(): - data[job] = pd.Series(dtype="float64") - for build in builds: - data[job][str(build)] = pd.Series(dtype="float64") - try: - data_dict = dict( - self.data[job][str(build)][data_set].items()) - except KeyError: - if continue_on_error: - continue - return None - - for test_id, test_data in data_dict.items(): - if eval(cond, {"tags": test_data.get("tags", "")}): - data[job][str(build)][test_id] = \ - pd.Series(dtype="float64") - if params is None: - for param, val in test_data.items(): - data[job][str(build)][test_id][param] = val - else: - for param in params: - try: - data[job][str(build)][test_id][param] =\ - test_data[param] - except KeyError: - data[job][str(build)][test_id][param] =\ - "No Data" - return data - - except (KeyError, IndexError, ValueError) as err: - logging.error( - f"Missing mandatory parameter in the element specification: " - f"{repr(err)}" - ) - return None - except AttributeError as err: - logging.error(repr(err)) - return None - except SyntaxError as err: - logging.error( - f"The filter {cond} is not correct. Check if all tags are " - f"enclosed by apostrophes.\n{repr(err)}" - ) - return None - - def filter_tests_by_name(self, element, params=None, data_set="tests", - continue_on_error=False): - """Filter required data from the given jobs and builds. - - The output data structure is: - - job 1 - - build 1 - - test (or suite) 1 ID: - - param 1 - - param 2 - ... - - param n - ... - - test (or suite) n ID: - ... - ... - - build n - ... - - job n - - :param element: Element which will use the filtered data. - :param params: Parameters which will be included in the output. If None, - all parameters are included. - :param data_set: The set of data to be filtered: tests, suites, - metadata. - :param continue_on_error: Continue if there is error while reading the - data. The Item will be empty then - :type element: pandas.Series - :type params: list - :type data_set: str - :type continue_on_error: bool - :returns: Filtered data. - :rtype pandas.Series - """ - - include = element.get("include", None) - if not include: - logging.warning("No tests to include, skipping the element.") - return None - - if params is None: - params = element.get("parameters", None) - if params and "type" not in params: - params.append("type") - - cores = element.get("core", None) - if cores: - tests = list() - for core in cores: - for test in include: - tests.append(test.format(core=core)) - else: - tests = include - - data = pd.Series(dtype="float64") - try: - for job, builds in element["data"].items(): - data[job] = pd.Series(dtype="float64") - for build in builds: - data[job][str(build)] = pd.Series(dtype="float64") - for test in tests: - try: - reg_ex = re.compile(str(test).lower()) - for test_id in self.data[job][ - str(build)][data_set].keys(): - if re.match(reg_ex, str(test_id).lower()): - test_data = self.data[job][ - str(build)][data_set][test_id] - data[job][str(build)][test_id] = \ - pd.Series(dtype="float64") - if params is None: - for param, val in test_data.items(): - data[job][str(build)][test_id]\ - [param] = val - else: - for param in params: - try: - data[job][str(build)][ - test_id][param] = \ - test_data[param] - except KeyError: - data[job][str(build)][ - test_id][param] = "No Data" - except KeyError as err: - if continue_on_error: - logging.debug(repr(err)) - continue - logging.error(repr(err)) - return None - return data - - except (KeyError, IndexError, ValueError) as err: - logging.error( - f"Missing mandatory parameter in the element " - f"specification: {repr(err)}" - ) - return None - except AttributeError as err: - logging.error(repr(err)) - return None - - @staticmethod - def merge_data(data): - """Merge data from more jobs and builds to a simple data structure. - - The output data structure is: - - - test (suite) 1 ID: - - param 1 - - param 2 - ... - - param n - ... - - test (suite) n ID: - ... - - :param data: Data to merge. - :type data: pandas.Series - :returns: Merged data. - :rtype: pandas.Series - """ - - logging.info(" Merging data ...") - - merged_data = pd.Series(dtype="float64") - for builds in data.values: - for item in builds.values: - for item_id, item_data in item.items(): - merged_data[item_id] = item_data - return merged_data - - def print_all_oper_data(self): - """Print all operational data to console. - """ - - for job in self._input_data.values: - for build in job.values: - for test_id, test_data in build["tests"].items(): - print(f"{test_id}") - if test_data.get("show-run", None) is None: - continue - for dut_name, data in test_data["show-run"].items(): - if data.get("runtime", None) is None: - continue - runtime = loads(data["runtime"]) - try: - threads_nr = len(runtime[0]["clocks"]) - except (IndexError, KeyError): - continue - threads = OrderedDict( - {idx: list() for idx in range(threads_nr)}) - for item in runtime: - for idx in range(threads_nr): - if item["vectors"][idx] > 0: - clocks = item["clocks"][idx] / \ - item["vectors"][idx] - elif item["calls"][idx] > 0: - clocks = item["clocks"][idx] / \ - item["calls"][idx] - elif item["suspends"][idx] > 0: - clocks = item["clocks"][idx] / \ - item["suspends"][idx] - else: - clocks = 0.0 - - if item["calls"][idx] > 0: - vectors_call = item["vectors"][idx] / \ - item["calls"][idx] - else: - vectors_call = 0.0 - - if int(item["calls"][idx]) + int( - item["vectors"][idx]) + \ - int(item["suspends"][idx]): - threads[idx].append([ - item["name"], - item["calls"][idx], - item["vectors"][idx], - item["suspends"][idx], - clocks, - vectors_call - ]) - - print(f"Host IP: {data.get('host', '')}, " - f"Socket: {data.get('socket', '')}") - for thread_nr, thread in threads.items(): - txt_table = prettytable.PrettyTable( - ( - "Name", - "Nr of Vectors", - "Nr of Packets", - "Suspends", - "Cycles per Packet", - "Average Vector Size" - ) - ) - avg = 0.0 - for row in thread: - txt_table.add_row(row) - avg += row[-1] - if len(thread) == 0: - avg = "" - else: - avg = f", Average Vector Size per Node: " \ - f"{(avg / len(thread)):.2f}" - th_name = "main" if thread_nr == 0 \ - else f"worker_{thread_nr}" - print(f"{dut_name}, {th_name}{avg}") - txt_table.float_format = ".2" - txt_table.align = "r" - txt_table.align["Name"] = "l" - print(f"{txt_table.get_string()}\n") |