| # Copyright lowRISC contributors. | |
| # Licensed under the Apache License, Version 2.0, see LICENSE for details. | |
| # SPDX-License-Identifier: Apache-2.0 | |
| import logging as log | |
| import pprint | |
| import random | |
| import shlex | |
| from pathlib import Path | |
| from typing import List | |
| from JobTime import JobTime | |
| from LauncherFactory import get_launcher | |
| from sim_utils import (get_cov_summary_table, get_job_runtime, | |
| get_simulated_time) | |
| from tabulate import tabulate | |
| from utils import (VERBOSE, clean_odirs, find_and_substitute_wildcards, | |
| rm_path, subst_wildcards) | |
| class Deploy(): | |
| """ | |
| Abstraction to create and maintain a runnable job (builds, runs, etc.). | |
| """ | |
| # Indicate the target for each sub-class. | |
| target = "none" | |
| # List of variable names that are to be treated as "list of commands". | |
| # This tells '_construct_cmd' that these vars are lists that need to | |
| # be joined with '&&' instead of a space. | |
| cmds_list_vars = [] | |
| # Represents the weight with which a job of this target is scheduled. These | |
| # initial weights set for each of the targets below are roughly inversely | |
| # proportional to their average runtimes. These are subject to change in | |
| # future. Lower the runtime, the higher chance the it gets scheduled. It is | |
| # useful to customize this only in case of targets that may coexist at a | |
| # time. | |
| # TODO: Allow these to be set in the HJson. | |
| weight = 1 | |
| def __str__(self): | |
| return (pprint.pformat(self.__dict__) | |
| if log.getLogger().isEnabledFor(VERBOSE) else self.full_name) | |
| def __init__(self, sim_cfg): | |
| assert self.target is not None | |
| # Cross ref the whole cfg object for ease. | |
| self.sim_cfg = sim_cfg | |
| # A list of jobs on which this job depends. | |
| self.dependencies = [] | |
| # Indicates whether running this job requires all dependencies to pass. | |
| # If this flag is set to False, any passing dependency will trigger | |
| # this current job to run | |
| self.needs_all_dependencies_passing = True | |
| # Declare attributes that need to be extracted from the HJSon cfg. | |
| self._define_attrs() | |
| # Set class instance attributes. | |
| self._set_attrs() | |
| # Check if all attributes that are needed are set. | |
| self._check_attrs() | |
| # Do variable substitutions. | |
| self._subst_vars() | |
| # List of vars required to be exported to sub-shell, as a dict. | |
| self.exports = self._process_exports() | |
| # Construct the job's command. | |
| self.cmd = self._construct_cmd() | |
| # Launcher instance created later using create_launcher() method. | |
| self.launcher = None | |
| # Job's wall clock time (a.k.a CPU time, or runtime). | |
| self.job_runtime = JobTime() | |
| def _define_attrs(self): | |
| """Defines the attributes this instance needs to have. | |
| These attributes are extracted from the Mode object / HJson config with | |
| which this instance is created. There are two types of attributes - | |
| one contributes to the generation of the command directly; the other | |
| provides supplementary information pertaining to the job, such as | |
| patterns that determine whether it passed or failed. These are | |
| represented as dicts, whose values indicate in boolean whether the | |
| extraction was successful. | |
| """ | |
| # These attributes are explicitly used to construct the job command. | |
| self.mandatory_cmd_attrs = {} | |
| # These attributes may indirectly contribute to the construction of the | |
| # command (through substitution vars) or other things such as pass / | |
| # fail patterns. | |
| self.mandatory_misc_attrs = { | |
| "build_mode": False, | |
| "dry_run": False, | |
| "exports": False, | |
| "flow_makefile": False, | |
| "name": False, | |
| } | |
| # Function to parse a dict and extract the mandatory cmd and misc attrs. | |
| def _extract_attrs(self, ddict): | |
| """Extracts the attributes from the supplied dict. | |
| 'ddict' is typically either the Mode object or the entire config | |
| object's dict. It is used to retrieve the instance attributes defined | |
| in 'mandatory_cmd_attrs' and 'mandatory_misc_attrs'. | |
| """ | |
| ddict_keys = ddict.keys() | |
| for key in self.mandatory_cmd_attrs.keys(): | |
| if self.mandatory_cmd_attrs[key] is False: | |
| if key in ddict_keys: | |
| setattr(self, key, ddict[key]) | |
| self.mandatory_cmd_attrs[key] = True | |
| for key in self.mandatory_misc_attrs.keys(): | |
| if self.mandatory_misc_attrs[key] is False: | |
| if key in ddict_keys: | |
| setattr(self, key, ddict[key]) | |
| self.mandatory_misc_attrs[key] = True | |
| def _set_attrs(self): | |
| """Sets additional attributes. | |
| Invokes '_extract_attrs()' to read in all the necessary instance | |
| attributes. Based on those, some additional instance attributes may | |
| be derived. Those are set by this method. | |
| """ | |
| self._extract_attrs(self.sim_cfg.__dict__) | |
| # Enable GUI mode. | |
| self.gui = self.sim_cfg.gui | |
| # Output directory where the artifacts go (used by the launcher). | |
| self.odir = getattr(self, self.target + "_dir") | |
| # Qualified name disambiguates the instance name with other instances | |
| # of the same class (example: 'uart_smoke' reseeded multiple times | |
| # needs to be disambiguated using the index -> '0.uart_smoke'. | |
| self.qual_name = self.name | |
| # Full name disambiguates across multiple cfg being run (example: | |
| # 'aes:default', 'uart:default' builds. | |
| self.full_name = self.sim_cfg.name + ":" + self.qual_name | |
| # Job name is used to group the job by cfg and target. The scratch path | |
| # directory name is assumed to be uniquified, in case there are more | |
| # than one sim_cfgs with the same name. | |
| self.job_name = "{}_{}".format( | |
| Path(self.sim_cfg.scratch_path).name, self.target) | |
| # Input directories (other than self) this job depends on. | |
| self.input_dirs = [] | |
| # Directories touched by this job. These directories are marked | |
| # because they are used by dependent jobs as input. | |
| self.output_dirs = [self.odir] | |
| # Pass and fail patterns. | |
| self.pass_patterns = [] | |
| self.fail_patterns = [] | |
| def _check_attrs(self): | |
| """Checks if all required class attributes are set. | |
| Invoked in __init__() after all attributes are extracted and set. | |
| """ | |
| for attr in self.mandatory_cmd_attrs.keys(): | |
| if self.mandatory_cmd_attrs[attr] is False: | |
| raise AttributeError("Attribute {!r} not found for " | |
| "{!r}.".format(attr, self.name)) | |
| for attr in self.mandatory_misc_attrs.keys(): | |
| if self.mandatory_misc_attrs[attr] is False: | |
| raise AttributeError("Attribute {!r} not found for " | |
| "{!r}.".format(attr, self.name)) | |
| def _subst_vars(self, ignored_subst_vars=[]): | |
| """Recursively search and replace substitution variables. | |
| First pass: search within self dict. We ignore errors since some | |
| substitions may be available in the second pass. Second pass: search | |
| the entire sim_cfg object.""" | |
| self.__dict__ = find_and_substitute_wildcards(self.__dict__, | |
| self.__dict__, | |
| ignored_subst_vars, True) | |
| self.__dict__ = find_and_substitute_wildcards(self.__dict__, | |
| self.sim_cfg.__dict__, | |
| ignored_subst_vars, | |
| False) | |
| def _process_exports(self): | |
| """Convert 'exports' as a list of dicts in the HJson to a dict. | |
| Exports is a list of key-value pairs that are to be exported to the | |
| subprocess' environment so that the tools can lookup those options. | |
| DVSim limits how the data is presented in the HJson (the value of a | |
| HJson member cannot be an object). This method converts a list of dicts | |
| into a dict variable, which makes it easy to merge the list of exports | |
| with the subprocess' env where the ASIC tool is invoked. | |
| """ | |
| return {k: str(v) for item in self.exports for k, v in item.items()} | |
| def _construct_cmd(self): | |
| """Construct the command that will eventually be launched.""" | |
| cmd = "make -f {} {}".format(self.flow_makefile, self.target) | |
| if self.dry_run is True: | |
| cmd += " -n" | |
| for attr in sorted(self.mandatory_cmd_attrs.keys()): | |
| value = getattr(self, attr) | |
| if type(value) is list: | |
| # Join attributes that are list of commands with '&&' to chain | |
| # them together when executed as a Make target's recipe. | |
| separator = " && " if attr in self.cmds_list_vars else " " | |
| value = separator.join(item.strip() for item in value) | |
| if type(value) is bool: | |
| value = int(value) | |
| if type(value) is str: | |
| value = shlex.quote(value.strip()) | |
| cmd += " {}={}".format(attr, value) | |
| return cmd | |
| def is_equivalent_job(self, item): | |
| """Checks if job that would be dispatched with 'item' is equivalent to | |
| 'self'. | |
| Determines if 'item' and 'self' would behave exactly the same way when | |
| deployed. If so, then there is no point in keeping both. The caller can | |
| choose to discard 'item' and pick 'self' instead. To do so, we check | |
| the final resolved 'cmd' & the exports. The 'name' field will be unique | |
| to 'item' and 'self', so we take that out of the comparison. | |
| """ | |
| if type(self) != type(item): | |
| return False | |
| # Check if the cmd field is identical. | |
| item_cmd = item.cmd.replace(item.name, self.name) | |
| if self.cmd != item_cmd: | |
| return False | |
| # Check if exports have identical set of keys. | |
| if self.exports.keys() != item.exports.keys(): | |
| return False | |
| # Check if exports have identical values. | |
| for key, val in self.exports.items(): | |
| item_val = item.exports[key] | |
| if type(item_val) is str: | |
| item_val = item_val.replace(item.name, self.name) | |
| if val != item_val: | |
| return False | |
| log.log(VERBOSE, "Deploy job \"%s\" is equivalent to \"%s\"", | |
| item.name, self.name) | |
| return True | |
| def pre_launch(self): | |
| """Callback to perform additional pre-launch activities. | |
| This is invoked by launcher::_pre_launch(). | |
| """ | |
| pass | |
| def post_finish(self, status): | |
| """Callback to perform additional post-finish activities. | |
| This is invoked by launcher::_post_finish(). | |
| """ | |
| pass | |
| def get_log_path(self): | |
| """Returns the log file path.""" | |
| return "{}/{}.log".format(self.odir, self.target) | |
| def get_timeout_mins(self): | |
| """Returns the timeout in minutes.""" | |
| return None | |
| def extract_info_from_log(self, log_text: List): | |
| """Extracts information pertaining to the job from its log. | |
| This method parses the log text after the job has completed, for the | |
| extraction of information pertaining to the job's performance. This | |
| base class method extracts the job's runtime (i.e. the wall clock time) | |
| as reported by the tool. The tool reported runtime is the most accurate | |
| since it is devoid of the delays incurred due to infrastructure and | |
| setup overhead. | |
| The extended classes may override this method to extract other pieces | |
| of information from the log. | |
| `log_text` is the job's log file contents as a list of lines. | |
| """ | |
| try: | |
| time, unit = get_job_runtime(log_text, self.sim_cfg.tool) | |
| self.job_runtime.set(time, unit) | |
| except RuntimeError as e: | |
| log.warning(f"{self.full_name}: {e} Using dvsim-maintained " | |
| "job_runtime instead.") | |
| self.job_runtime.set(self.launcher.job_runtime_secs, "s") | |
| def create_launcher(self): | |
| """Creates the launcher instance. | |
| Note that the launcher instance for ALL jobs in the same job group must | |
| be created before the Scheduler starts to dispatch one by one. | |
| """ | |
| # Retain the handle to self for lookup & callbacks. | |
| self.launcher = get_launcher(self) | |
| class CompileSim(Deploy): | |
| """Abstraction for building the simulation executable.""" | |
| target = "build" | |
| cmds_list_vars = ["pre_build_cmds", "post_build_cmds"] | |
| weight = 5 | |
| def __init__(self, build_mode, sim_cfg): | |
| self.build_mode_obj = build_mode | |
| self.seed = sim_cfg.build_seed | |
| super().__init__(sim_cfg) | |
| def _define_attrs(self): | |
| super()._define_attrs() | |
| self.mandatory_cmd_attrs.update({ | |
| # tool srcs | |
| "proj_root": False, | |
| # Flist gen | |
| "sv_flist_gen_cmd": False, | |
| "sv_flist_gen_dir": False, | |
| "sv_flist_gen_opts": False, | |
| # Build | |
| "pre_build_cmds": False, | |
| "build_cmd": False, | |
| "build_dir": False, | |
| "build_opts": False, | |
| "post_build_cmds": False, | |
| }) | |
| self.mandatory_misc_attrs.update({ | |
| "build_fail_patterns": False, | |
| "build_pass_patterns": False, | |
| "build_timeout_mins": False, | |
| "cov_db_dir": False, | |
| }) | |
| def _set_attrs(self): | |
| super()._extract_attrs(self.build_mode_obj.__dict__) | |
| super()._set_attrs() | |
| # Dont run the compile job in GUI mode. | |
| self.gui = False | |
| # 'build_mode' is used as a substitution variable in the HJson. | |
| self.build_mode = self.name | |
| self.job_name += f"_{self.build_mode}" | |
| if self.sim_cfg.cov: | |
| self.output_dirs += [self.cov_db_dir] | |
| self.pass_patterns = self.build_pass_patterns | |
| self.fail_patterns = self.build_fail_patterns | |
| if self.sim_cfg.args.build_timeout_mins is not None: | |
| self.build_timeout_mins = self.sim_cfg.args.build_timeout_mins | |
| if self.build_timeout_mins: | |
| log.debug("Timeout for job \"%s\" is %d minutes.", self.name, | |
| self.build_timeout_mins) | |
| def pre_launch(self): | |
| # Delete old coverage database directories before building again. We | |
| # need to do this because the build directory is not 'renewed'. | |
| rm_path(self.cov_db_dir) | |
| def get_timeout_mins(self): | |
| """Returns the timeout in minutes. | |
| Limit build jobs to 60 minutes if the timeout is not set. | |
| """ | |
| return self.build_timeout_mins if self.build_timeout_mins is not None else 60 | |
| class CompileOneShot(Deploy): | |
| """Abstraction for building the design (used by non-DV flows).""" | |
| target = "build" | |
| def __init__(self, build_mode, sim_cfg): | |
| self.build_mode_obj = build_mode | |
| super().__init__(sim_cfg) | |
| def _define_attrs(self): | |
| super()._define_attrs() | |
| self.mandatory_cmd_attrs.update({ | |
| # tool srcs | |
| "proj_root": False, | |
| # Flist gen | |
| "sv_flist_gen_cmd": False, | |
| "sv_flist_gen_dir": False, | |
| "sv_flist_gen_opts": False, | |
| # Build | |
| "build_dir": False, | |
| "build_cmd": False, | |
| "build_opts": False, | |
| "build_log": False, | |
| "build_timeout_mins": False, | |
| "post_build_cmds": False, | |
| "pre_build_cmds": False, | |
| # Report processing | |
| "report_cmd": False, | |
| "report_opts": False | |
| }) | |
| self.mandatory_misc_attrs.update({ | |
| "build_fail_patterns": False, | |
| "build_pass_patterns": False | |
| }) | |
| def _set_attrs(self): | |
| super()._extract_attrs(self.build_mode_obj.__dict__) | |
| super()._set_attrs() | |
| # 'build_mode' is used as a substitution variable in the HJson. | |
| self.build_mode = self.name | |
| self.job_name += f"_{self.build_mode}" | |
| self.fail_patterns = self.build_fail_patterns | |
| self.pass_patterns = self.build_pass_patterns | |
| if self.sim_cfg.args.build_timeout_mins is not None: | |
| self.build_timeout_mins = self.sim_cfg.args.build_timeout_mins | |
| if self.build_timeout_mins: | |
| log.debug("Timeout for job \"%s\" is %d minutes.", self.name, | |
| self.build_timeout_mins) | |
| def get_timeout_mins(self): | |
| """Returns the timeout in minutes. | |
| Limit build jobs to 60 minutes if the timeout is not set. | |
| """ | |
| return self.build_timeout_mins if self.build_timeout_mins is not None else 60 | |
| class RunTest(Deploy): | |
| """Abstraction for running tests. This is one per seed for each test.""" | |
| # Initial seed values when running tests (if available). | |
| target = "run" | |
| seeds = [] | |
| fixed_seed = None | |
| cmds_list_vars = ["pre_run_cmds", "post_run_cmds"] | |
| def __init__(self, index, test, build_job, sim_cfg): | |
| self.test_obj = test | |
| self.index = index | |
| self.build_seed = sim_cfg.build_seed | |
| self.seed = RunTest.get_seed() | |
| self.simulated_time = JobTime() | |
| super().__init__(sim_cfg) | |
| if build_job is not None: | |
| self.dependencies.append(build_job) | |
| # We did something wrong if build_mode is not the same as the build_job | |
| # arg's name. | |
| assert self.build_mode == build_job.name | |
| def _define_attrs(self): | |
| super()._define_attrs() | |
| self.mandatory_cmd_attrs.update({ | |
| # tool srcs | |
| "proj_root": False, | |
| "uvm_test": False, | |
| "uvm_test_seq": False, | |
| "sw_images": False, | |
| "sw_build_device": False, | |
| "sw_build_cmd": False, | |
| "sw_build_opts": False, | |
| "run_dir": False, | |
| "pre_run_cmds": False, | |
| "run_cmd": False, | |
| "run_opts": False, | |
| "post_run_cmds": False, | |
| "build_seed": True, # Already set in the constructor. | |
| "seed": True, # Already set in the constructor. | |
| }) | |
| self.mandatory_misc_attrs.update({ | |
| "cov_db_dir": False, | |
| "cov_db_test_dir": False, | |
| "run_dir_name": False, | |
| "run_fail_patterns": False, | |
| "run_pass_patterns": False, | |
| "run_timeout_mins": False, | |
| "run_timeout_multiplier": False, | |
| }) | |
| def _set_attrs(self): | |
| super()._extract_attrs(self.test_obj.__dict__) | |
| super()._set_attrs() | |
| # 'test' is used as a substitution variable in the HJson. | |
| self.test = self.name | |
| self.build_mode = self.test_obj.build_mode.name | |
| self.qual_name = self.run_dir_name + "." + str(self.seed) | |
| self.full_name = self.sim_cfg.name + ":" + self.qual_name | |
| self.job_name += f"_{self.build_mode}" | |
| if self.sim_cfg.cov: | |
| self.output_dirs += [self.cov_db_dir] | |
| # In GUI mode, the log file is not updated; hence, nothing to check. | |
| if not self.gui: | |
| self.pass_patterns = self.run_pass_patterns | |
| self.fail_patterns = self.run_fail_patterns | |
| if self.sim_cfg.args.run_timeout_mins is not None: | |
| self.run_timeout_mins = self.sim_cfg.args.run_timeout_mins | |
| if self.sim_cfg.args.run_timeout_multiplier is not None: | |
| self.run_timeout_multiplier = ( | |
| self.sim_cfg.args.run_timeout_multiplier) | |
| if self.run_timeout_mins and self.run_timeout_multiplier: | |
| self.run_timeout_mins = int(self.run_timeout_mins * | |
| self.run_timeout_multiplier) | |
| if self.run_timeout_multiplier: | |
| log.debug("Timeout multiplier for job \"%s\" is %f.", | |
| self.full_name, self.run_timeout_multiplier) | |
| if self.run_timeout_mins: | |
| log.debug("Timeout for job \"%s\" is %d minutes.", self.full_name, | |
| self.run_timeout_mins) | |
| def pre_launch(self): | |
| self.launcher.renew_odir = True | |
| def post_finish(self, status): | |
| if status != 'P': | |
| # Delete the coverage data if available. | |
| rm_path(self.cov_db_test_dir) | |
| @staticmethod | |
| def get_seed(): | |
| # If --seeds option is passed, then those custom seeds are consumed | |
| # first. If --fixed-seed <val> is also passed, the subsequent tests | |
| # (once the custom seeds are consumed) will be run with the fixed seed. | |
| if not RunTest.seeds: | |
| if RunTest.fixed_seed: | |
| return RunTest.fixed_seed | |
| for i in range(1000): | |
| seed = random.getrandbits(32) | |
| RunTest.seeds.append(seed) | |
| return RunTest.seeds.pop(0) | |
| def get_timeout_mins(self): | |
| """Returns the timeout in minutes. | |
| Limit run jobs to 60 minutes if the timeout is not set. | |
| """ | |
| return self.run_timeout_mins if self.run_timeout_mins is not None else 60 | |
| def extract_info_from_log(self, log_text: List): | |
| """Extracts the time the design was simulated for, from the log.""" | |
| super().extract_info_from_log(log_text) | |
| try: | |
| time, unit = get_simulated_time(log_text, self.sim_cfg.tool) | |
| self.simulated_time.set(time, unit) | |
| except RuntimeError as e: | |
| log.debug(f"{self.full_name}: {e}") | |
| class CovUnr(Deploy): | |
| """Abstraction for coverage UNR flow.""" | |
| target = "cov_unr" | |
| def __init__(self, sim_cfg): | |
| super().__init__(sim_cfg) | |
| def _define_attrs(self): | |
| super()._define_attrs() | |
| self.mandatory_cmd_attrs.update({ | |
| # tool srcs | |
| "proj_root": False, | |
| # Need to generate filelist based on build mode | |
| "sv_flist_gen_cmd": False, | |
| "sv_flist_gen_dir": False, | |
| "sv_flist_gen_opts": False, | |
| "build_dir": False, | |
| "cov_unr_build_cmd": False, | |
| "cov_unr_build_opts": False, | |
| "cov_unr_run_cmd": False, | |
| "cov_unr_run_opts": False | |
| }) | |
| self.mandatory_misc_attrs.update({ | |
| "cov_unr_dir": False, | |
| "cov_merge_db_dir": False, | |
| "build_fail_patterns": False | |
| }) | |
| def _set_attrs(self): | |
| super()._set_attrs() | |
| self.qual_name = self.target | |
| self.full_name = self.sim_cfg.name + ":" + self.qual_name | |
| self.input_dirs += [self.cov_merge_db_dir] | |
| # Reuse the build_fail_patterns set in the HJson. | |
| self.fail_patterns = self.build_fail_patterns | |
| class CovMerge(Deploy): | |
| """Abstraction for merging coverage databases.""" | |
| target = "cov_merge" | |
| weight = 10 | |
| def __init__(self, run_items, sim_cfg): | |
| # Construct the cov_db_dirs right away from the run_items. This is a | |
| # special variable used in the HJson. The coverage associated with | |
| # the primary build mode needs to be first in the list. | |
| self.cov_db_dirs = [] | |
| for run in run_items: | |
| if run.cov_db_dir not in self.cov_db_dirs: | |
| if sim_cfg.primary_build_mode == run.build_mode: | |
| self.cov_db_dirs.insert(0, run.cov_db_dir) | |
| else: | |
| self.cov_db_dirs.append(run.cov_db_dir) | |
| # Early lookup the cov_merge_db_dir, which is a mandatory misc | |
| # attribute anyway. We need it to compute additional cov db dirs. | |
| self.cov_merge_db_dir = subst_wildcards("{cov_merge_db_dir}", | |
| sim_cfg.__dict__) | |
| # Prune previous merged cov directories, keeping past 7 dbs. | |
| prev_cov_db_dirs = clean_odirs(odir=self.cov_merge_db_dir, max_odirs=7) | |
| # If the --cov-merge-previous command line switch is passed, then | |
| # merge coverage with the previous runs. | |
| if sim_cfg.cov_merge_previous: | |
| self.cov_db_dirs += [str(item) for item in prev_cov_db_dirs] | |
| super().__init__(sim_cfg) | |
| self.dependencies += run_items | |
| # Run coverage merge even if one test passes. | |
| self.needs_all_dependencies_passing = False | |
| # Append cov_db_dirs to the list of exports. | |
| self.exports["cov_db_dirs"] = shlex.quote(" ".join(self.cov_db_dirs)) | |
| def _define_attrs(self): | |
| super()._define_attrs() | |
| self.mandatory_cmd_attrs.update({ | |
| "cov_merge_cmd": False, | |
| "cov_merge_opts": False | |
| }) | |
| self.mandatory_misc_attrs.update({ | |
| "cov_merge_dir": False, | |
| "cov_merge_db_dir": False | |
| }) | |
| def _set_attrs(self): | |
| super()._set_attrs() | |
| self.qual_name = self.target | |
| self.full_name = self.sim_cfg.name + ":" + self.qual_name | |
| # For merging coverage db, the precise output dir is set in the HJson. | |
| self.odir = self.cov_merge_db_dir | |
| self.input_dirs += self.cov_db_dirs | |
| self.output_dirs = [self.odir] | |
| class CovReport(Deploy): | |
| """Abstraction for coverage report generation. """ | |
| target = "cov_report" | |
| weight = 10 | |
| def __init__(self, merge_job, sim_cfg): | |
| super().__init__(sim_cfg) | |
| self.dependencies.append(merge_job) | |
| def _define_attrs(self): | |
| super()._define_attrs() | |
| self.mandatory_cmd_attrs.update({ | |
| "cov_report_cmd": False, | |
| "cov_report_opts": False | |
| }) | |
| self.mandatory_misc_attrs.update({ | |
| "cov_report_dir": False, | |
| "cov_merge_db_dir": False, | |
| "cov_report_txt": False | |
| }) | |
| def _set_attrs(self): | |
| super()._set_attrs() | |
| self.qual_name = self.target | |
| self.full_name = self.sim_cfg.name + ":" + self.qual_name | |
| # Keep track of coverage results, once the job is finished. | |
| self.cov_total = "" | |
| self.cov_results = "" | |
| self.cov_results_dict = dict() | |
| def post_finish(self, status): | |
| """Extract the coverage results summary for the dashboard. | |
| If the extraction fails, an appropriate exception is raised, which must | |
| be caught by the caller to mark the job as a failure. | |
| """ | |
| if self.dry_run or status != 'P': | |
| return | |
| results, self.cov_total = get_cov_summary_table( | |
| self.cov_report_txt, self.sim_cfg.tool) | |
| colalign = (("center", ) * len(results[0])) | |
| self.cov_results = tabulate(results, | |
| headers="firstrow", | |
| tablefmt="pipe", | |
| colalign=colalign) | |
| for tup in zip(*results): | |
| self.cov_results_dict[tup[0]] = tup[1] | |
| class CovAnalyze(Deploy): | |
| """Abstraction for running the coverage analysis tool.""" | |
| target = "cov_analyze" | |
| def __init__(self, sim_cfg): | |
| # Enforce GUI mode for coverage analysis. | |
| sim_cfg.gui = True | |
| super().__init__(sim_cfg) | |
| def _define_attrs(self): | |
| super()._define_attrs() | |
| self.mandatory_cmd_attrs.update({ | |
| # tool srcs | |
| "proj_root": False, | |
| "cov_analyze_cmd": False, | |
| "cov_analyze_opts": False | |
| }) | |
| self.mandatory_misc_attrs.update({ | |
| "cov_analyze_dir": False, | |
| "cov_merge_db_dir": False | |
| }) | |
| def _set_attrs(self): | |
| super()._set_attrs() | |
| self.qual_name = self.target | |
| self.full_name = self.sim_cfg.name + ":" + self.qual_name | |
| self.input_dirs += [self.cov_merge_db_dir] |