Axel Andersson 816553f1bc Added powertrain-build entrypoint
The entrypoint will make it easier to install powertrain-build to an
isolated venv without having to call python -m powertrain_build.

Change-Id: I3850c97d17707f9bc03640bd1d997508637d97ba
2024-10-30 10:33:42 +01:00

633 lines
25 KiB
Python

# Copyright 2024 Volvo Car Corporation
# Licensed under Apache 2.0.
""" script for running running signal consistency check on specific models. Git HEAD or local.
"""
import argparse
import csv
import os
import sys
from collections import defaultdict
from os.path import join
from pathlib import Path
from typing import List, Optional
import git
from powertrain_build.build_proj_config import BuildProjConfig
from powertrain_build.feature_configs import FeatureConfigs
from powertrain_build.lib import helper_functions, logger
from powertrain_build.signal_incons_html_rep_all import SigConsHtmlReportAll
from powertrain_build.signal_interfaces import CsvSignalInterfaces, YamlSignalInterfaces
from powertrain_build.unit_configs import UnitConfigs
from powertrain_build.user_defined_types import UserDefinedTypes
LOGGER = logger.create_logger(__file__)
EXIT_CODE_OK = 0
EXIT_CODE_UNPRODUCED = 1
EXIT_CODE_MISSING_CONSUMER = 2
# Files define in <mdl>.Unconsumed.csv is consumed by mdl or/and interface.
EXIT_CODE_INCORRECT_CSV = 4
EXIT_CODE_NEVER_ACTIVE_SIGNALS = 8
UNCONSUMED_SIGNAL_FILE_PATTERN = "*_Unconsumed_Sig.csv"
MODEL_ROOT_DIR = "Models"
REPO_ROOT = helper_functions.get_repo_root()
MISSING_CONSUMER_CONSOLE_MESSAGE = (
"\n===============Unused signals===============\n{signals}"
)
UNPRODUCED_SIGNALS_CONSOLE_MESSAGE = (
"\n=============Unproduced signals=============\n{signals}"
)
UNCONSUMED_CSV_CONSOLE_MESSAGE = "\n=========Signals defined in unconsumed.csv-s that are consumed=========\n{signals}"
NEVER_ACTIVE_SIGNALS_CONSOLE_MESSAGE = (
"\n============Never active signals=============\n{signals}"
)
INDEX_FILE = "Reports/Index_SigCheck_All.html"
TEMPLATE = """<!DOCTYPE html>
<!--[if IE 8]><html class="no-js lt-ie9" lang="en" > <![endif]-->
<!--[if gt IE 8]><!--> <html class="no-js" lang="en" > <!--<![endif]-->
<body class="wy-body-for-nav">
<a class="icon icon-home"> Signal Consistency Report
</a>
<ul>
<li class="toctree-l1">
<a class="reference internal" href="SigCheckAll_intro.html">Introduction</a>
</li>
{project_rows}
</ul>
</body>
</html>"""
PARSER_HELP = "Run signal inconsistency check."
def gen_sig_incons_index_file(project_list):
"""Generate Index_SigCheck_All.html."""
# TODO Remove function and add as method in powertrain_build/signal_incons_html_rep_all.py
rows = ""
project_row = """<li class="toctree-l1"><a class="reference internal"
href="SigCheckAll_{project}.html">SigCheckAll {project}</a></li>\n"""
for prj in project_list:
rows += project_row.format(project=prj)
os.makedirs(os.path.dirname(INDEX_FILE), exist_ok=True)
with open(INDEX_FILE, "w", encoding="utf-8") as f_h:
f_h.write(TEMPLATE.format(project_rows=rows))
def configure_parser(parser: argparse.ArgumentParser):
"""Parse the arguments sent to the script."""
parser.add_argument(
"-m",
"--models",
nargs="+",
help="Arguments for running in local repo. "
"Name of models to run test on Ex: VcDepExt VcAcCtrl. "
"If not supplied, changed files in current commit will be used. "
"Reports are stored in local repo",
)
parser.add_argument(
"-r", "--report", help="Create report for all projects", action="store_true"
)
parser.set_defaults(func=run_signal_inconsistency_check)
def get_project_configs():
"""Wrapper for creating project configs for all projects.
Returns (dict):
Project configs.
"""
root_path = Path().resolve()
project_config_files = [
str(f) for f in root_path.glob("Projects/**/ProjectCfg.json")
]
prj_cfgs = {}
for project_config_file in project_config_files:
LOGGER.debug("Get project config for %s", project_config_file)
project_config = BuildProjConfig(os.path.normpath(project_config_file))
prj_cfgs.update({project_config.name: project_config})
return prj_cfgs
def check_inconsistency(signal_ifs, never_active_signals):
"""Check signal inconsistency.
Args:
signal_ifs (CsvSignalInterfaces): class holding signal interface information.
never_active_signals (dict): Dict of projects mapped to units mapped to "NEVER_ACTIVE" signals.
Returns:
inconsistency result (dict)
"""
signal_inconsistency_results = {}
for key, signal_if in signal_ifs.items():
partial_result = signal_if.check_config()
signal_inconsistency_results.update({key: partial_result})
signal_inconsistency_results[key].update(
{"never_active_signals": never_active_signals[key]}
)
return signal_inconsistency_results
def get_signal_interfaces(prj_cfgs, models):
"""Wrapper for creating signal interface dict for all configs.
Args:
Project config.
Returns:
Signal interface dict.
"""
signal_ifs = {}
per_unit_cfgs = {}
for prj, prj_cfg in prj_cfgs.items():
LOGGER.info("Parsing interface %s", prj)
feature_cfg = FeatureConfigs(prj_cfg)
unit_cfg = UnitConfigs(prj_cfg, feature_cfg)
user_defined_types = UserDefinedTypes(prj_cfg, unit_cfg)
if prj_cfg.has_yaml_interface:
signal_if = YamlSignalInterfaces(
prj_cfg, unit_cfg, feature_cfg, user_defined_types, model_names=models
)
else:
signal_if = CsvSignalInterfaces(prj_cfg, unit_cfg, models)
signal_ifs.update({prj: signal_if})
per_unit_cfgs.update({prj: unit_cfg.get_per_unit_cfg()})
return signal_ifs, per_unit_cfgs
def get_never_active_signals(prj_cfgs, models):
"""Functions for getting all signals marked as "NEVER_ACTIVE" in all projects.
Args:
prj_cfgs (list(BuildProjConfig)): List of project configurations to look for never active signals in.
models (list): List of Simulink models to check.
Returns:
never_active_signals (dict): Dict of projects mapped to units mapped to its "NEVER_ACTIVE" signals.
"""
never_active_signals = {}
for prj, prj_cfg in prj_cfgs.items():
LOGGER.info("searching for never active signals in %s", prj)
feature_cfg = FeatureConfigs(prj_cfg)
unit_cfg = UnitConfigs(prj_cfg, feature_cfg)
per_unit_cfg_total = unit_cfg.get_per_unit_cfg_total()
never_active_signals[prj] = {}
for unit, unit_data in per_unit_cfg_total.items():
if unit not in models:
continue
inactive_inports = []
inactive_outports = []
for inport, inport_data in unit_data["inports"].items():
if "NEVER_ACTIVE" in inport_data["configs"]:
inactive_inports.append(inport)
for outport, outport_data in unit_data["outports"].items():
if "NEVER_ACTIVE" in outport_data["configs"]:
inactive_outports.append(outport)
if inactive_inports or inactive_outports:
never_active_signals[prj].update(
{unit: inactive_inports + inactive_outports}
)
return never_active_signals
class SignalInconsistency:
"""Class for running signal consistency check on specific models. Git HEAD or local."""
def __init__(self, args):
"""Models in local repo."""
self.never_active_signals = {}
self.signal_inconsistency_results = {}
self.per_unit_cfgs = {}
self.signal_ifs = {}
self.repo = git.Repo()
if args.models:
self.models = args.models
else:
self.ci_url = args.ci_url
mdl_file_paths = self.get_changed_models()
self.models = self._get_model_names(mdl_file_paths)
@staticmethod
def _get_model_names(mdl_paths):
"""Get Change Model names."""
mdl_index = 1
return [os.path.split(i)[mdl_index].replace(".mdl", "") for i in mdl_paths]
def _create_reports(self):
"""Create reports for all projects."""
sig_report = SigConsHtmlReportAll(self.signal_inconsistency_results)
sig_report.generate_report_file_signal_check(
join(REPO_ROOT, "Reports", "SigCheckAll")
)
def _sort_internal_inconsistency_by_model(self):
"""Sort inconsistency by model."""
sig = self.signal_inconsistency_results
no_producer_int = {}
missing_consumer_int = {}
never_active_int = {}
for project, val in sig.items():
int_dict = val["sigs"]["int"]
for mdl, data in int_dict.items():
if data.get("missing"):
if mdl not in no_producer_int:
no_producer_int[mdl] = self._to_tuple_list(
project, data["missing"]
)
else:
no_producer_int[mdl] += self._to_tuple_list(
project, data["missing"]
)
if data.get("unused"):
if mdl not in missing_consumer_int:
missing_consumer_int[mdl] = self._to_tuple_list(
project, data["unused"]
)
else:
missing_consumer_int[mdl] += self._to_tuple_list(
project, data["unused"]
)
for mdl, never_active_signals in val["never_active_signals"].items():
never_active_int[mdl] = [
(signal, {project}) for signal in never_active_signals
]
return (
self._merge_tuple_list(no_producer_int),
self._merge_tuple_list(missing_consumer_int),
self._merge_tuple_list(never_active_int),
)
def _analyse_inconsistency(self):
"""Collect signals.
Args:
no_producer_int (dict): Consumed signals but not produced.
missing_consumer_int (dict): Produced signals but not consumed.
"""
exit_code = EXIT_CODE_OK
log_message = ""
(
no_producer_int,
missing_consumer_int,
never_active_int,
) = self._sort_internal_inconsistency_by_model()
used_model_inports = self.aggregate_model_inports()
used_external_outports = self.aggregate_supplier_inports()
exit_code_unproduced, unproduced_message = self.check_unproduced_signals(
no_producer_int
)
(
exit_code_missing_consumer,
missing_consumer_message,
) = self.check_missing_consumer_signals(missing_consumer_int)
exit_code_used_inports, used_inports_message = self.check_unconsumed_files(
used_model_inports, used_external_outports
)
exit_code_never_active, never_active_message = self.check_never_active_signals(
never_active_int
)
log_message += unproduced_message
log_message += missing_consumer_message
log_message += used_inports_message
log_message += never_active_message
exit_code += exit_code_unproduced
exit_code += exit_code_missing_consumer
exit_code += exit_code_used_inports
exit_code += exit_code_never_active
return exit_code, log_message
def get_changed_models(self):
"""Get changed models in current commit."""
changed_files_tmp = self.repo.git.diff(
"--diff-filter=d", "--name-only", "HEAD~1"
)
changed_files = changed_files_tmp.splitlines()
changed_models = [
m for m in changed_files if m.endswith(".mdl") or m.endswith(".slx")
]
return changed_models
def check_unproduced_signals(self, unproduced_signals):
"""Check if models expects signals that is not produced.
Args:
unproduced_signals (dict):
mdl:{signals:{projects...}...}...
Returns:
exit_code (int):
Number of unproduced signals in checked models.
console message (string):
Shows if models has unproduced signals.
"""
error_message = ""
exit_code = EXIT_CODE_OK
models_with_unproduced_signals = {
m for m in unproduced_signals.keys() if m in self.models
}
for mdl in models_with_unproduced_signals:
error_message += self._format_console_output(mdl, unproduced_signals[mdl])
if mdl in unproduced_signals and unproduced_signals[mdl]:
exit_code = EXIT_CODE_UNPRODUCED
return exit_code, UNPRODUCED_SIGNALS_CONSOLE_MESSAGE.format(
signals=error_message
)
def check_missing_consumer_signals(self, missing_consumer_signals):
"""Check if models expect signals that is not produced.
Args:
missing_consumer_signals (dict):
mdl:{signals:{projects...}...}...
Returns:
exit_code (int):
Number of unproduced signals in checked models.
console message (string):
Shows if models has unproduced signals.
"""
error_message = ""
exit_code = EXIT_CODE_OK
unconsumed_skip_list = self.fetch_signals_to_skip()
models_producing_not_consumed_signals = {
m for m in missing_consumer_signals.keys() if m in self.models
}
for mdl in models_producing_not_consumed_signals:
sigs = missing_consumer_signals[mdl]
for k in set(sigs.keys()):
if mdl in unconsumed_skip_list and k in unconsumed_skip_list[mdl]:
del sigs[k]
error_message += self._format_console_output(
mdl, missing_consumer_signals[mdl]
)
if mdl in missing_consumer_signals and missing_consumer_signals[mdl]:
exit_code = EXIT_CODE_MISSING_CONSUMER
return exit_code, MISSING_CONSUMER_CONSOLE_MESSAGE.format(signals=error_message)
def check_unconsumed_files(self, inports, ifs_outports):
"""Check that signals define in unconsumed.csv files are not being consumed
Args:
inports (set): all models inports in all projects.
ifs_outports (set): all outports, all projects and all interfaces.
Returns:
exit_code (int):
console message (string): Console error message (if test fail)
"""
error_message_int = ""
error_message_ext = ""
error_message = ""
exit_code = EXIT_CODE_OK
unconsumed_skip_list = self.fetch_signals_to_skip()
for producing_mdl, signals in unconsumed_skip_list.items():
signal_intersect_int = signals & inports
signal_intersect_ext = signals & ifs_outports
if signal_intersect_int:
exit_code, error_message_int = self.get_intersect_exit_code(
producing_mdl, signal_intersect_int
)
if signal_intersect_ext:
exit_code, error_message_int = self.get_intersect_exit_code(
producing_mdl, signal_intersect_ext, False
)
error_message += error_message_int
error_message += error_message_ext
return exit_code, UNCONSUMED_CSV_CONSOLE_MESSAGE.format(signals=error_message)
def check_never_active_signals(self, never_active_signals):
"""Check if there are any never active signals.
Produce corresponding error code and message.
Args:
never_active_signals (dict): Dict mapping models to never active signals in projects.
Returns:
exit_code (int):
console message (string): Console error message (if test fail)
"""
error_message = ""
exit_code = EXIT_CODE_OK
if never_active_signals:
exit_code = EXIT_CODE_NEVER_ACTIVE_SIGNALS
for mdl, signals in never_active_signals.items():
error_message += self._format_console_output(mdl, signals)
return exit_code, NEVER_ACTIVE_SIGNALS_CONSOLE_MESSAGE.format(
signals=error_message
)
def get_intersect_exit_code(
self, producing_mdl, signal_intersect, int_signals=True
):
"""Determine if Unconsumed.csv and used inport signals overlap per project.
Return exit code and which consumer and project is cause of failure.
Args:
producing_mdl (String): Model name.
signal_intersect (set): Unconsumed.csv signals and used inports that overlap.
int_signals (bool):
True: Set exit code for internally overlaping signals (Models)
False: Set exit code for external overlaping signals (Supplier)
Returns:
exit_code (int): Fail or succes
error_message (String): Consumers.
"""
exit_code = EXIT_CODE_OK
console_message = "{producer}\n {signal} is consumed by {consumer} in {prj}\n"
error_message = ""
for sig in signal_intersect:
if int_signals:
consumers = self.get_consumer_int(sig)
else:
consumers = self.get_consumer_ext(sig)
for unit, prj in consumers:
if self.mdl_is_producer_in_prj(producing_mdl, prj, sig):
error_message += console_message.format(
producer=producing_mdl, signal=sig, consumer=unit, prj=prj
)
exit_code = EXIT_CODE_INCORRECT_CSV
return exit_code, error_message
def mdl_is_producer_in_prj(self, mdl, prj, signal):
"""Check that defined signal in <mdl>unconsumed.csv is produced in project"""
try:
return signal in self.per_unit_cfgs[prj][mdl]["outports"].keys()
except KeyError:
return False
def get_consumer_int(self, signal):
"""Find internal consumers"""
signal_consumers = []
for prj, unit_cfgs in self.per_unit_cfgs.items():
for unit, unit_cfg in unit_cfgs.items():
if signal in unit_cfg["inports"].keys():
signal_consumers.append((unit, prj))
return signal_consumers
def get_consumer_ext(self, signal):
"""Find external consumers"""
signal_consumers = []
for prj, if_output in self.signal_ifs.items():
for sig_type, signals in if_output.get_external_outputs(prj).items():
if signal in signals.keys():
signal_consumers.append((sig_type, prj))
return signal_consumers
@staticmethod
def _format_console_output(mdl, signal_dict):
out = ""
for signal, projects in signal_dict.items():
out += " " + str(signal) + str(projects) + "\n"
return f"\n{mdl}\n{out} \n"
@staticmethod
def _to_tuple_list(project, signal_project_dict):
"""Create list of tuples.
NOTE: signal_project_dict is no longer a signal to project mapping.
It was when PyBuild handled multiple projects at once.
Args:
project (str): project name.
signal_project_dict (dict): signal1: {}, signal2: {}...
Returns:
list: [(signal1, {project}), (signal2, {project})...]
"""
return [(signal, {project}) for signal in signal_project_dict.keys()]
@staticmethod
def _merge_tuple_list(t_list):
"""Merged list of tuples.
Args (list(tuple...)):
[(A,1), (A,2), (C,1)]
Returns (list(tuple...)):
[(A,[1,2]), (C,1)]
"""
model_confs = {}
for mdl, sig_prj_pair in t_list.items():
model_conf = defaultdict(list)
for signal, prj in sig_prj_pair:
model_conf[signal].append(prj)
model_confs[mdl] = model_conf
return model_confs
def fetch_signals_to_skip(self, model_root_dir=MODEL_ROOT_DIR):
"""fetch Unconsumed signals from csv files.
Returns:
unconsumed_signals (dict):
{model: {sig1, sig2...}, model2: {sig1, sig2...}}
"""
unconsumed_sig = {mdl: set() for mdl in self.models}
# "mdl_name" = "mdl folder name". '..\\..\\mdl_name\\pybuild_cfg\\mdl_Unconsumed_Sig.csv'
model_dir_index = -3
signal_col_index = 0
for file_name in Path(model_root_dir).glob(
"**/" + UNCONSUMED_SIGNAL_FILE_PATTERN
):
mdl_name = str(file_name).split(os.sep)[model_dir_index]
if mdl_name in self.models:
with file_name.open() as csv_file:
csv_reader = csv.reader(csv_file)
next(csv_reader) # Skip header.
for row in csv_reader:
unconsumed_sig[mdl_name].add(row[signal_col_index])
return unconsumed_sig
@staticmethod
def _print_console_log_info(exit_code):
"""Prints explanatory message to console.
NOTE: The exit_codes_messages variable needs the messages to be in increasing order,
according to the exit code constans above.
"""
missing_producer = (
"Model/Models is configured to consume signals that are not produced. "
"Remove signal or fix/add producer."
)
missing_consumer = (
"Model/Models creates signals that are not consumed. "
"Remove or add to <mdl>_Unconsumed.csv."
)
incorrect_csv = (
"Consumed signals is defined in <mdl>_Unconsumed.csv. "
"Remove signals or fix/remove consumer."
)
never_active_signals = (
"Never active signals will not appear in generated .c file. "
"Remove corresponding part in Simulink model, signals probablty lead to terminators."
)
exit_codes_messages = [
missing_producer,
missing_consumer,
incorrect_csv,
never_active_signals,
]
exit_code_bit_field = [
bool(int(b)) for b in bin(exit_code)[2:]
] # [2:] gets rid of 0b part
for idx, value in enumerate(exit_code_bit_field):
if value:
LOGGER.info(exit_codes_messages[idx])
def aggregate_model_inports(self):
"""Aggregate inport signals from models"""
used_inports = set()
for _, unit_cfg in self.per_unit_cfgs.items():
for _, inp in unit_cfg.items():
used_inports = used_inports.union(set(inp["inports"].keys()))
return used_inports
def aggregate_supplier_inports(self):
"""Aggregate outports from signal interface (supplier inports)"""
external_outports = set()
for if_output in self.signal_ifs.values():
for signals in if_output.get_external_outputs().values():
external_outports = external_outports.union(set(signals.keys()))
return external_outports
def run(self, args):
"""Run signal inconsistency check."""
exit_code = EXIT_CODE_OK
if self.models:
prj_cfs = get_project_configs()
self.signal_ifs, self.per_unit_cfgs = get_signal_interfaces(
prj_cfs, self.models
)
self.never_active_signals = get_never_active_signals(prj_cfs, self.models)
LOGGER.info("Start check signal inconsistencies for: %s", self.models)
self.signal_inconsistency_results = check_inconsistency(
self.signal_ifs, self.never_active_signals
)
exit_code, log_output = self._analyse_inconsistency()
LOGGER.info("Finished check signal inconsistencies %s", log_output)
self._print_console_log_info(exit_code)
if args.report:
LOGGER.info(
"Start generating interface inconsistencies html-report for all projects"
)
gen_sig_incons_index_file(
list(self.signal_inconsistency_results.keys())
)
self._create_reports()
LOGGER.info("Finished - generating inconsistencies html-reports")
else:
LOGGER.info("No models in change")
return exit_code
def run_signal_inconsistency_check(args: argparse.Namespace) -> int:
"""Create Signal Inconsistency instance and run checks."""
sig_in = SignalInconsistency(args)
return sig_in.run(args)
def main(argv: Optional[List[str]] = None) -> int:
"""Create Signal Inconsistency instance and run checks."""
parser = argparse.ArgumentParser(description=PARSER_HELP)
configure_parser(parser)
args = parser.parse_args(argv)
return args.func(args)
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))