From 86b1ae5247b0732b647e62715a9c5304c59a3121 Mon Sep 17 00:00:00 2001 From: vinothk-master Date: Sat, 22 Feb 2025 17:19:56 +0000 Subject: [PATCH 1/4] Reduced the type: ignore comments and allocated the structured dataclass --- odxtools/cli/compare.py | 125 ++++++++++++++++++++-------------------- 1 file changed, 61 insertions(+), 64 deletions(-) diff --git a/odxtools/cli/compare.py b/odxtools/cli/compare.py index 609344c9..855c826b 100644 --- a/odxtools/cli/compare.py +++ b/odxtools/cli/compare.py @@ -3,7 +3,7 @@ import argparse import os -from typing import Any, Dict, List, Optional, Set, Union, cast +from typing import TypedDict, Any, Dict, List, Optional, Set, Union, cast, Tuple from rich import print as rich_print from rich.padding import Padding as RichPadding @@ -23,11 +23,13 @@ from ._parser_utils import SubparsersList from ._print_utils import (extract_service_tabulation_data, print_dl_metrics, print_service_parameters) +from dataclasses import dataclass # name of the tool _odxtools_tool_name_ = "compare" - +""" VariantName = str + VariantType = str NewServices = List[DiagService] DeletedServices = List[DiagService] @@ -41,76 +43,79 @@ DeletedVariants = List[DiagLayer] SpecsChangesVariants = Dict[str, Union[NewVariants, DeletedVariants, SpecsServiceDict]] - - +""" + + +class SpecsServiceDict(TypedDict): + diag_layer: str + diag_layer_type: str + new_services: List[str] # List of new service names + deleted_services: List[DiagService] # List of deleted services + renamed_service: List[Tuple[str, DiagService]] # List of (old_name, DiagService) + changed_name_of_service: List[Tuple[str, DiagService]] # Similar to renamed_service + changed_parameters_of_service: List[Tuple[str, DiagService, List[Union[str, int, float]]]] +class SpecsChangesVariants(TypedDict): + new_variants : List[str] + deleted_variants : List[str] + service_changes : SpecsServiceDict +#class extract_service_tabulation_data(TypedDict): +# table: RichTable + + +@dataclass class Display: - # class with variables and functions to display the result of the comparison - - # TODO - # - Idea: results as json export - # - write results of comparison in json structure - # - use odxlinks to refer to dignostic services / objects if - # changes have already been detected (e.g. in another ecu - # variant / diagnostic layer) - # - print all information about parameter properties (request, - # pos. response & neg. response parameters) for changed diagnostic - # services - param_detailed: bool - obj_detailed: bool - - def __init__(self) -> None: - pass + param_detailed: bool = False + obj_detailed: bool = False def print_dl_changes(self, service_dict: SpecsServiceDict) -> None: if service_dict["new_services"] or service_dict["deleted_services"] or service_dict[ - "changed_name_of_service"][0] or service_dict["changed_parameters_of_service"][0]: - assert isinstance(service_dict["diag_layer"], str) + "changed_name_of_service"] or service_dict["changed_parameters_of_service"]: + # assert isinstance(service_dict["diag_layer"], str) rich_print() rich_print( f"Changed diagnostic services for diagnostic layer '{service_dict['diag_layer']}' ({service_dict['diag_layer_type']}):" ) if service_dict["new_services"]: - assert isinstance(service_dict["new_services"], List) + # assert isinstance(service_dict["new_services"], List) rich_print() rich_print(" [blue]New services[/blue]") rich_print(extract_service_tabulation_data( - service_dict["new_services"])) # type: ignore[arg-type] + service_dict["new_services"])) + if service_dict["deleted_services"]: - assert isinstance(service_dict["deleted_services"], List) + # assert isinstance(service_dict["deleted_services"], List) rich_print() rich_print(" [blue]Deleted services[/blue]") rich_print(extract_service_tabulation_data( - service_dict["deleted_services"])) # type: ignore[arg-type] + service_dict["deleted_services"])) if service_dict["changed_name_of_service"][0]: rich_print() rich_print(" [blue]Renamed services[/blue]") rich_print(extract_service_tabulation_data( - service_dict["changed_name_of_service"][0])) # type: ignore[arg-type] + service_dict["changed_name_of_service"][0])) if service_dict["changed_parameters_of_service"][0]: rich_print() rich_print(" [blue]Services with parameter changes[/blue]") # create table with information about services with parameter changes changed_param_column = [ str(x) for x in service_dict["changed_parameters_of_service"][ - 1] # type: ignore[union-attr] + 1] ] table = extract_service_tabulation_data( - service_dict["changed_parameters_of_service"][0], # type: ignore[arg-type] + service_dict["changed_parameters_of_service"][0], additional_columns=[("Changed Parameters", changed_param_column)]) rich_print(table) for service_idx, service in enumerate( - service_dict["changed_parameters_of_service"][0]): # type: ignore[arg-type] + service_dict["changed_parameters_of_service"][0]): assert isinstance(service, DiagService) rich_print() rich_print( f" Detailed changes of diagnostic service [u cyan]{service.short_name}[/u cyan]" ) # detailed_info in [infotext1, dict1, infotext2, dict2, ...] - info_list = cast( - list, # type: ignore[type-arg] - service_dict["changed_parameters_of_service"])[2][service_idx] + info_list = cast(list, service_dict["changed_parameters_of_service"])[2][service_idx] for detailed_info in info_list: if isinstance(detailed_info, str): rich_print() @@ -165,7 +170,6 @@ class Comparison(Display): db_indicator_1: int db_indicator_2: int - def __init__(self) -> None: pass @@ -280,9 +284,10 @@ def compare_services(self, service1: DiagService, service2: DiagService) -> List[SpecsServiceDict]: # compares request, positive response and negative response parameters of two diagnostic services + information: List[Union[str, Dict[str, Any]]] = [ ] # information = [infotext1, table1, infotext2, table2, ...] - changed_params = "" + changed_params: str = "" # Request if service1.request is not None and service2.request is not None and len( @@ -403,33 +408,25 @@ def compare_services(self, service1: DiagService, str(service2.negative_responses)] }) - return [information, changed_params] # type: ignore[list-item] + return [information, changed_params] def compare_diagnostic_layers(self, dl1: DiagLayer, - dl2: DiagLayer) -> dict: # type: ignore[type-arg] + dl2: DiagLayer)-> SpecsServiceDict: # compares diagnostic services of two diagnostic layers with each other # save changes in dictionary (service_dict) # TODO: add comparison of SingleECUJobs - new_services: NewServices = [] - deleted_services: DeletedServices = [] - renamed_service: RenamedServices = [[], - []] # TODO: implement list of (str, DiagService)-tuples - services_with_param_changes: ServicesWithParamChanges = [ - [], [], [] - ] # TODO: implement list of tuples (str, str, DiagService)-tuples - service_dict: SpecsServiceDict = { "diag_layer": dl1.short_name, "diag_layer_type": dl1.variant_type.value, # list with added diagnostic services [service1, service2, service3, ...] Type: DiagService - "new_services": new_services, + "new_services": [], # list with deleted diagnostic services [service1, service2, service3, ...] Type: DiagService - "deleted_services": deleted_services, + "deleted_services": [], # list with diagnostic services where the service name changed [[services], [old service names]] - "changed_name_of_service": renamed_service, + "changed_name_of_service": [[],[]], # list with diagnostic services where the service parameter changed [[services], [changed_parameters], [information_texts]] - "changed_parameters_of_service": services_with_param_changes + "changed_parameters_of_service": [[],[],[]] } # service_dict["changed_name_of_service"][{0 = services, 1 = old service names}][i] # service_dict["changed_parameters_of_service"][{0 = services, 1 = changed_parameters, 2 = information_texts}][i] @@ -457,7 +454,7 @@ def compare_diagnostic_layers(self, dl1: DiagLayer, if rq_prefix is None or rq_prefix not in dl2_request_prefixes: # TODO: this will not work in cases where the constant # prefix of a request was modified... - service_dict["new_services"].append( # type: ignore[union-attr] + service_dict["new_services"].append( service1) # type: ignore[arg-type] # check whether names of diagnostic services have changed @@ -470,10 +467,10 @@ def compare_diagnostic_layers(self, dl1: DiagLayer, # save information about changes in dictionary # add new service (type: DiagService) - service_dict["changed_name_of_service"][0].append( # type: ignore[union-attr] + service_dict["changed_name_of_service"][0].append( service1) # add old service name (type: String) - service_dict["changed_name_of_service"][1].append( # type: ignore[union-attr] + service_dict["changed_name_of_service"][1].append( service2.short_name) # compare request, pos. response and neg. response parameters of diagnostic services @@ -484,16 +481,16 @@ def compare_diagnostic_layers(self, dl1: DiagLayer, if detailed_information[1]: # check whether string "changed_params" is empty # new service (type: DiagService) service_dict["changed_parameters_of_service"][ - 0].append( # type: ignore[union-attr] + 0].append( service1) # add parameters which have been changed (type: String) service_dict["changed_parameters_of_service"][ - 1].append( # type: ignore[union-attr] - detailed_information[1]) # type: ignore[arg-type] + 1].append( + detailed_information[1]) # add detailed information about changed service parameters (type: list) [infotext1, table1, infotext2, table2, ...] service_dict["changed_parameters_of_service"][ - 2].append( # type: ignore[union-attr] - detailed_information[0]) # type: ignore[arg-type] + 2].append( + detailed_information[0]) for service2_idx, service2 in enumerate(dl2.services): @@ -505,7 +502,7 @@ def compare_diagnostic_layers(self, dl1: DiagLayer, assert isinstance(deleted_list, list) if service2 not in deleted_list: service_dict["deleted_services"].append( # type: ignore[union-attr] - service2) # type: ignore[arg-type] + service2) if service1.short_name == service2.short_name: # compare request, pos. response and neg. response parameters of both diagnostic services @@ -516,22 +513,22 @@ def compare_diagnostic_layers(self, dl1: DiagLayer, if detailed_information[1]: # check whether string "changed_params" is empty # new service (type: DiagService) service_dict["changed_parameters_of_service"][ - 0].append( # type: ignore[union-attr] + 0].append( service1) # add parameters which have been changed (type: String) - service_dict["changed_parameters_of_service"][ # type: ignore[union-attr] - 1].append(detailed_information[1]) # type: ignore[arg-type] + service_dict["changed_parameters_of_service"][ + 1].append(detailed_information[1]) # add detailed information about changed service parameters (type: list) [infotext1, table1, infotext2, table2, ...] service_dict["changed_parameters_of_service"][ # type: ignore[union-attr] 2].append(detailed_information[0]) # type: ignore[arg-type] return service_dict def compare_databases(self, database_new: Database, - database_old: Database) -> dict: # type: ignore[type-arg] + database_old: Database) -> SpecsChangesVariants: # type: ignore[type-arg] # compares two PDX-files with each other - new_variants: NewVariants = [] - deleted_variants: DeletedVariants = [] + new_variants = [] + deleted_variants= [] changes_variants: SpecsChangesVariants = { "new_diagnostic_layers": new_variants, From 38dec1f66e88f9842e2b0e61d7401a23633a21c4 Mon Sep 17 00:00:00 2001 From: vinothk-master Date: Sat, 22 Feb 2025 17:23:16 +0000 Subject: [PATCH 2/4] Reduced the type: ignore comments and allocated the structured dataclass --- odxtools/cli/compare.py | 18 ------------------ 1 file changed, 18 deletions(-) diff --git a/odxtools/cli/compare.py b/odxtools/cli/compare.py index 855c826b..db25e522 100644 --- a/odxtools/cli/compare.py +++ b/odxtools/cli/compare.py @@ -27,23 +27,7 @@ # name of the tool _odxtools_tool_name_ = "compare" -""" -VariantName = str -VariantType = str -NewServices = List[DiagService] -DeletedServices = List[DiagService] -RenamedServices = List[List[Union[str, DiagService]]] -ServicesWithParamChanges = List[List[Union[str, DiagService]]] - -SpecsServiceDict = Dict[str, Union[VariantName, VariantType, NewServices, DeletedServices, - RenamedServices, ServicesWithParamChanges]] - -NewVariants = List[DiagLayer] -DeletedVariants = List[DiagLayer] - -SpecsChangesVariants = Dict[str, Union[NewVariants, DeletedVariants, SpecsServiceDict]] -""" class SpecsServiceDict(TypedDict): @@ -58,8 +42,6 @@ class SpecsChangesVariants(TypedDict): new_variants : List[str] deleted_variants : List[str] service_changes : SpecsServiceDict -#class extract_service_tabulation_data(TypedDict): -# table: RichTable @dataclass From 824a499edb5da01b610a448fd8379156eb23d9e2 Mon Sep 17 00:00:00 2001 From: vinothk-master Date: Sat, 22 Feb 2025 23:57:57 +0000 Subject: [PATCH 3/4] fixed the lint issues --- odxtools/cli/compare.py | 562 ++++++++++++++++++++++++++-------------- 1 file changed, 367 insertions(+), 195 deletions(-) diff --git a/odxtools/cli/compare.py b/odxtools/cli/compare.py index db25e522..d5d64117 100644 --- a/odxtools/cli/compare.py +++ b/odxtools/cli/compare.py @@ -3,7 +3,9 @@ import argparse import os -from typing import TypedDict, Any, Dict, List, Optional, Set, Union, cast, Tuple +from dataclasses import dataclass +from typing import (Any, Dict, List, Optional, Set, Tuple, TypedDict, Union, + cast) from rich import print as rich_print from rich.padding import Padding as RichPadding @@ -23,13 +25,11 @@ from ._parser_utils import SubparsersList from ._print_utils import (extract_service_tabulation_data, print_dl_metrics, print_service_parameters) -from dataclasses import dataclass # name of the tool _odxtools_tool_name_ = "compare" - class SpecsServiceDict(TypedDict): diag_layer: str diag_layer_type: str @@ -37,11 +37,15 @@ class SpecsServiceDict(TypedDict): deleted_services: List[DiagService] # List of deleted services renamed_service: List[Tuple[str, DiagService]] # List of (old_name, DiagService) changed_name_of_service: List[Tuple[str, DiagService]] # Similar to renamed_service - changed_parameters_of_service: List[Tuple[str, DiagService, List[Union[str, int, float]]]] + changed_parameters_of_service: List[ + Tuple[str, DiagService, List[Union[str, int, float]]] + ] + + class SpecsChangesVariants(TypedDict): - new_variants : List[str] - deleted_variants : List[str] - service_changes : SpecsServiceDict + new_variants: List[str] + deleted_variants: List[str] + service_changes: SpecsServiceDict @dataclass @@ -51,53 +55,63 @@ class Display: def print_dl_changes(self, service_dict: SpecsServiceDict) -> None: - if service_dict["new_services"] or service_dict["deleted_services"] or service_dict[ - "changed_name_of_service"] or service_dict["changed_parameters_of_service"]: - # assert isinstance(service_dict["diag_layer"], str) + if ( + service_dict["new_services"] + or service_dict["deleted_services"] + or service_dict["changed_name_of_service"] + or service_dict["changed_parameters_of_service"] + ): + # assert isinstance(service_dict["diag_layer"], str) rich_print() rich_print( f"Changed diagnostic services for diagnostic layer '{service_dict['diag_layer']}' ({service_dict['diag_layer_type']}):" ) if service_dict["new_services"]: - # assert isinstance(service_dict["new_services"], List) + # assert isinstance(service_dict["new_services"], List) rich_print() rich_print(" [blue]New services[/blue]") - rich_print(extract_service_tabulation_data( - service_dict["new_services"])) - + rich_print(extract_service_tabulation_data(service_dict["new_services"])) + if service_dict["deleted_services"]: - # assert isinstance(service_dict["deleted_services"], List) + # assert isinstance(service_dict["deleted_services"], List) rich_print() rich_print(" [blue]Deleted services[/blue]") - rich_print(extract_service_tabulation_data( - service_dict["deleted_services"])) + rich_print( + extract_service_tabulation_data(service_dict["deleted_services"]) + ) if service_dict["changed_name_of_service"][0]: rich_print() rich_print(" [blue]Renamed services[/blue]") - rich_print(extract_service_tabulation_data( - service_dict["changed_name_of_service"][0])) + rich_print( + extract_service_tabulation_data( + service_dict["changed_name_of_service"][0] + ) + ) if service_dict["changed_parameters_of_service"][0]: rich_print() rich_print(" [blue]Services with parameter changes[/blue]") # create table with information about services with parameter changes changed_param_column = [ - str(x) for x in service_dict["changed_parameters_of_service"][ - 1] + str(x) for x in service_dict["changed_parameters_of_service"][1] ] table = extract_service_tabulation_data( - service_dict["changed_parameters_of_service"][0], - additional_columns=[("Changed Parameters", changed_param_column)]) + service_dict["changed_parameters_of_service"][0], + additional_columns=[("Changed Parameters", changed_param_column)], + ) rich_print(table) for service_idx, service in enumerate( - service_dict["changed_parameters_of_service"][0]): + service_dict["changed_parameters_of_service"][0] + ): assert isinstance(service, DiagService) rich_print() rich_print( f" Detailed changes of diagnostic service [u cyan]{service.short_name}[/u cyan]" ) # detailed_info in [infotext1, dict1, infotext2, dict2, ...] - info_list = cast(list, service_dict["changed_parameters_of_service"])[2][service_idx] + info_list = cast(list, service_dict["changed_parameters_of_service"])[ + 2 + ][service_idx] for detailed_info in info_list: if isinstance(detailed_info, str): rich_print() @@ -107,7 +121,8 @@ def print_dl_changes(self, service_dict: SpecsServiceDict) -> None: show_header=True, header_style="bold cyan", border_style="blue", - show_lines=True) + show_lines=True, + ) for header in detailed_info: table.add_column(header) rows = zip(*detailed_info.values()) @@ -124,20 +139,24 @@ def print_database_changes(self, changes_variants: SpecsChangesVariants) -> None # prints result of database comparison (input variable: dictionary: changes_variants) # diagnostic layers - if changes_variants["new_diagnostic_layers"] or changes_variants[ - "deleted_diagnostic_layers"]: + if ( + changes_variants["new_diagnostic_layers"] + or changes_variants["deleted_diagnostic_layers"] + ): rich_print() rich_print("[bright_blue]Changed diagnostic layers[/bright_blue]: ") rich_print(" New diagnostic layers: ") for variant in changes_variants["new_diagnostic_layers"]: assert isinstance(variant, DiagLayer) rich_print( - f" [magenta]{variant.short_name}[/magenta] ({variant.variant_type.value})") + f" [magenta]{variant.short_name}[/magenta] ({variant.variant_type.value})" + ) rich_print(" Deleted diagnostic layers: ") for variant in changes_variants["deleted_diagnostic_layers"]: assert isinstance(variant, DiagLayer) rich_print( - f" [magenta]{variant.short_name}[/magenta] ({variant.variant_type.value})") + f" [magenta]{variant.short_name}[/magenta] ({variant.variant_type.value})" + ) # diagnostic services for _, value in changes_variants.items(): @@ -152,10 +171,13 @@ class Comparison(Display): db_indicator_1: int db_indicator_2: int + def __init__(self) -> None: pass - def compare_parameters(self, param1: Parameter, param2: Parameter) -> Dict[str, Any]: + def compare_parameters( + self, param1: Parameter, param2: Parameter + ) -> Dict[str, Any]: # checks whether properties of param1 and param2 differ # checked properties: Name, Byte Position, Bit Length, Semantic, Parameter Type, Value (Coded, Constant, Default etc.), Data Type, Data Object Property (Name, Physical Data Type, Unit) @@ -163,8 +185,11 @@ def compare_parameters(self, param1: Parameter, param2: Parameter) -> Dict[str, old = [] new = [] - def append_list(property_name: str, new_property_value: Optional[AtomicOdxType], - old_property_value: Optional[AtomicOdxType]) -> None: + def append_list( + property_name: str, + new_property_value: Optional[AtomicOdxType], + old_property_value: Optional[AtomicOdxType], + ) -> None: property.append(property_name) old.append(old_property_value) new.append(new_property_value) @@ -174,35 +199,62 @@ def append_list(property_name: str, new_property_value: Optional[AtomicOdxType], if param1.byte_position != param2.byte_position: append_list("Byte position", param1.byte_position, param2.byte_position) if param1.get_static_bit_length() != param2.get_static_bit_length(): - append_list("Bit Length", param1.get_static_bit_length(), - param2.get_static_bit_length()) + append_list( + "Bit Length", + param1.get_static_bit_length(), + param2.get_static_bit_length(), + ) if param1.semantic != param2.semantic: append_list("Semantic", param1.semantic, param2.semantic) if param1.parameter_type != param2.parameter_type: append_list("Parameter type", param1.parameter_type, param2.parameter_type) - if isinstance(param1, CodedConstParameter) and isinstance(param2, CodedConstParameter): - if param1.diag_coded_type.base_data_type != param2.diag_coded_type.base_data_type: - append_list("Data type", param1.diag_coded_type.base_data_type.name, - param2.diag_coded_type.base_data_type.name) + if isinstance(param1, CodedConstParameter) and isinstance( + param2, CodedConstParameter + ): + if ( + param1.diag_coded_type.base_data_type + != param2.diag_coded_type.base_data_type + ): + append_list( + "Data type", + param1.diag_coded_type.base_data_type.name, + param2.diag_coded_type.base_data_type.name, + ) if param1.coded_value != param2.coded_value: - if isinstance(param1.coded_value, int) and isinstance(param2.coded_value, int): + if isinstance(param1.coded_value, int) and isinstance( + param2.coded_value, int + ): append_list( "Value", f"0x{param1.coded_value:0{(param1.get_static_bit_length() or 0) // 4}X}", - f"0x{param2.coded_value:0{(param2.get_static_bit_length() or 0) // 4}X}") + f"0x{param2.coded_value:0{(param2.get_static_bit_length() or 0) // 4}X}", + ) else: - append_list("Value", f"{param1.coded_value!r}", f"{param2.coded_value!r}") + append_list( + "Value", f"{param1.coded_value!r}", f"{param2.coded_value!r}" + ) - elif isinstance(param1, NrcConstParameter) and isinstance(param2, NrcConstParameter): - if param1.diag_coded_type.base_data_type != param2.diag_coded_type.base_data_type: - append_list("Data type", param1.diag_coded_type.base_data_type.name, - param2.diag_coded_type.base_data_type.name) + elif isinstance(param1, NrcConstParameter) and isinstance( + param2, NrcConstParameter + ): + if ( + param1.diag_coded_type.base_data_type + != param2.diag_coded_type.base_data_type + ): + append_list( + "Data type", + param1.diag_coded_type.base_data_type.name, + param2.diag_coded_type.base_data_type.name, + ) if param1.coded_values != param2.coded_values: - append_list("Values", str(param1.coded_values), str(param2.coded_values)) + append_list( + "Values", str(param1.coded_values), str(param2.coded_values) + ) - elif (dop_1 := getattr(param1, "dop", None)) is not None and (dop_2 := getattr( - param2, "dop", None)) is not None: + elif (dop_1 := getattr(param1, "dop", None)) is not None and ( + dop_2 := getattr(param2, "dop", None) + ) is not None: if dop_1 != dop_2: # TODO: compare INTERNAL-CONSTR, COMPU-INTERNAL-TO-PHYS of DOP @@ -215,93 +267,140 @@ def append_list(property_name: str, new_property_value: Optional[AtomicOdxType], # DOP Unit if getattr(dop_1, "unit", None) and getattr(dop_2, "unit", None): # (properties of unit object: short_name, long_name, description, odx_id, display_name, oid, factor_si_to_unit, offset_si_to_unit, physical_dimension_ref) - if dop_1.unit != dop_2.unit and dop_1.unit.short_name != dop_2.unit.short_name: - append_list(" DOP unit name", dop_1.unit.short_name, dop_2.unit.short_name) - elif dop_1.unit != dop_2.unit and dop_1.unit.display_name != dop_2.unit.display_name: - append_list(" DOP unit display name", dop_1.unit.display_name, - dop_2.unit.display_name) + if ( + dop_1.unit != dop_2.unit + and dop_1.unit.short_name != dop_2.unit.short_name + ): + append_list( + " DOP unit name", + dop_1.unit.short_name, + dop_2.unit.short_name, + ) + elif ( + dop_1.unit != dop_2.unit + and dop_1.unit.display_name != dop_2.unit.display_name + ): + append_list( + " DOP unit display name", + dop_1.unit.display_name, + dop_2.unit.display_name, + ) elif dop_1.unit != dop_2.unit: append_list(" DOP unit object", "", "") if hasattr(dop_1, "physical_type") and hasattr(dop_2, "physical_type"): - if (dop_1.physical_type and dop_2.physical_type and - dop_1.physical_type.base_data_type - != dop_2.physical_type.base_data_type): - append_list(" DOP physical data type", - dop_1.physical_type.base_data_type.name, - dop_2.physical_type.base_data_type.name) - - if (isinstance(param1, PhysicalConstantParameter) and - isinstance(param2, PhysicalConstantParameter) and - param1.physical_constant_value != param2.physical_constant_value): + if ( + dop_1.physical_type + and dop_2.physical_type + and dop_1.physical_type.base_data_type + != dop_2.physical_type.base_data_type + ): + append_list( + " DOP physical data type", + dop_1.physical_type.base_data_type.name, + dop_2.physical_type.base_data_type.name, + ) + + if ( + isinstance(param1, PhysicalConstantParameter) + and isinstance(param2, PhysicalConstantParameter) + and param1.physical_constant_value != param2.physical_constant_value + ): if isinstance(param1.physical_constant_value, int) and isinstance( - param2.physical_constant_value, int): + param2.physical_constant_value, int + ): append_list( "Constant value", f"0x{param1.physical_constant_value:0{(param1.get_static_bit_length() or 0) // 4}X}", - f"0x{param2.physical_constant_value:0{(param2.get_static_bit_length() or 0) // 4}X}" + f"0x{param2.physical_constant_value:0{(param2.get_static_bit_length() or 0) // 4}X}", ) else: - append_list("Constant value", f"{param1.physical_constant_value!r}", - f"{param2.physical_constant_value!r}") + append_list( + "Constant value", + f"{param1.physical_constant_value!r}", + f"{param2.physical_constant_value!r}", + ) - elif (isinstance(param1, ValueParameter) and isinstance(param2, ValueParameter) and - param1.physical_default_value is not None and - param2.physical_default_value is not None and - param1.physical_default_value != param2.physical_default_value): + elif ( + isinstance(param1, ValueParameter) + and isinstance(param2, ValueParameter) + and param1.physical_default_value is not None + and param2.physical_default_value is not None + and param1.physical_default_value != param2.physical_default_value + ): if isinstance(param1.physical_default_value, int) and isinstance( - param2.physical_default_value, int): + param2.physical_default_value, int + ): append_list( "Default value", f"0x{param1.physical_default_value:0{(param1.get_static_bit_length() or 0) // 4}X}", - f"0x{param2.physical_default_value:0{(param2.get_static_bit_length() or 0) // 4}X}" + f"0x{param2.physical_default_value:0{(param2.get_static_bit_length() or 0) // 4}X}", ) else: - append_list("Default value", f"{param1.physical_default_value!r}", - f"{param2.physical_default_value!r}") + append_list( + "Default value", + f"{param1.physical_default_value!r}", + f"{param2.physical_default_value!r}", + ) return {"Property": property, "Old Value": old, "New Value": new} - def compare_services(self, service1: DiagService, - service2: DiagService) -> List[SpecsServiceDict]: + def compare_services( + self, service1: DiagService, service2: DiagService + ) -> List[SpecsServiceDict]: # compares request, positive response and negative response parameters of two diagnostic services - - information: List[Union[str, Dict[str, Any]]] = [ - ] # information = [infotext1, table1, infotext2, table2, ...] + information: List[Union[str, Dict[str, Any]]] = ( + [] + ) # information = [infotext1, table1, infotext2, table2, ...] changed_params: str = "" # Request - if service1.request is not None and service2.request is not None and len( - service1.request.parameters) == len(service2.request.parameters): + if ( + service1.request is not None + and service2.request is not None + and len(service1.request.parameters) == len(service2.request.parameters) + ): for res1_idx, param1 in enumerate(service1.request.parameters): for res2_idx, param2 in enumerate(service2.request.parameters): if res1_idx == res2_idx: # find changed request parameter properties table = self.compare_parameters(param1, param2) - infotext = (f" Properties of request parameter '{param2.short_name}' " - f"that have changed:\n") + infotext = ( + f" Properties of request parameter '{param2.short_name}' " + f"that have changed:\n" + ) # array index starts with 0 -> param[0] is 1. service parameter if table["Property"]: information.append(infotext) information.append(table) - changed_params += f"request parameter '{param2.short_name}',\n" + changed_params += ( + f"request parameter '{param2.short_name}',\n" + ) else: changed_params += "request parameter list, " # infotext - information.append(f"List of request parameters for service '{service2.short_name}' " - f"is not identical.\n") + information.append( + f"List of request parameters for service '{service2.short_name}' " + f"is not identical.\n" + ) # table - param_list1 = [] if service1.request is None else service1.request.parameters - param_list2 = [] if service2.request is None else service2.request.parameters + param_list1 = ( + [] if service1.request is None else service1.request.parameters + ) + param_list2 = ( + [] if service2.request is None else service2.request.parameters + ) - information.append({ - "List": ["Old list", "New list"], - "Values": [f"\\{param_list1}", f"\\{param_list2}"] - }) + information.append( + { + "List": ["Old list", "New list"], + "Values": [f"\\{param_list1}", f"\\{param_list2}"], + } + ) # Positive Responses if len(service1.positive_responses) == len(service2.positive_responses): @@ -310,13 +409,16 @@ def compare_services(self, service1: DiagService, if res1_idx == res2_idx: if len(response1.parameters) == len(response2.parameters): for param1_idx, param1 in enumerate(response1.parameters): - for param2_idx, param2 in enumerate(response2.parameters): + for param2_idx, param2 in enumerate( + response2.parameters + ): if param1_idx == param2_idx: # find changed positive response parameter properties table = self.compare_parameters(param1, param2) infotext = ( f" Properties of positive response parameter '{param2.short_name}' that " - f"have changed:\n") + f"have changed:\n" + ) # array index starts with 0 -> param[0] is first service parameter if table["Property"]: @@ -330,22 +432,31 @@ def compare_services(self, service1: DiagService, f"List of positive response parameters for service '{service2.short_name}' is not identical." ) # table - information.append({ - "List": ["Old list", "New list"], - "Values": [str(response1.parameters), - str(response2.parameters)] - }) + information.append( + { + "List": ["Old list", "New list"], + "Values": [ + str(response1.parameters), + str(response2.parameters), + ], + } + ) else: changed_params += "positive responses list, " # infotext information.append( - f"List of positive responses for service '{service2.short_name}' is not identical.") + f"List of positive responses for service '{service2.short_name}' is not identical." + ) # table - information.append({ - "List": ["Old list", "New list"], - "Values": [str(service1.positive_responses), - str(service2.positive_responses)] - }) + information.append( + { + "List": ["Old list", "New list"], + "Values": [ + str(service1.positive_responses), + str(service2.positive_responses), + ], + } + ) # Negative Responses if len(service1.negative_responses) == len(service2.negative_responses): @@ -354,7 +465,9 @@ def compare_services(self, service1: DiagService, if res1_idx == res2_idx: if len(response1.parameters) == len(response2.parameters): for param1_idx, param1 in enumerate(response1.parameters): - for param2_idx, param2 in enumerate(response2.parameters): + for param2_idx, param2 in enumerate( + response2.parameters + ): if param1_idx == param2_idx: # find changed negative response parameter properties table = self.compare_parameters(param1, param2) @@ -372,11 +485,15 @@ def compare_services(self, service1: DiagService, f"List of positive response parameters for service '{service2.short_name}' is not identical.\n" ) # table - information.append({ - "List": ["Old list", "New list"], - "Values": [str(response1.parameters), - str(response2.parameters)] - }) + information.append( + { + "List": ["Old list", "New list"], + "Values": [ + str(response1.parameters), + str(response2.parameters), + ], + } + ) else: changed_params += "negative responses list, " # infotext @@ -384,16 +501,21 @@ def compare_services(self, service1: DiagService, f"List of positive responses for service '{service2.short_name}' is not identical.\n" ) # table - information.append({ - "List": ["Old list", "New list"], - "Values": [str(service1.negative_responses), - str(service2.negative_responses)] - }) + information.append( + { + "List": ["Old list", "New list"], + "Values": [ + str(service1.negative_responses), + str(service2.negative_responses), + ], + } + ) - return [information, changed_params] + return [information, changed_params] - def compare_diagnostic_layers(self, dl1: DiagLayer, - dl2: DiagLayer)-> SpecsServiceDict: + def compare_diagnostic_layers( + self, dl1: DiagLayer, dl2: DiagLayer + ) -> SpecsServiceDict: # compares diagnostic services of two diagnostic layers with each other # save changes in dictionary (service_dict) # TODO: add comparison of SingleECUJobs @@ -406,9 +528,9 @@ def compare_diagnostic_layers(self, dl1: DiagLayer, # list with deleted diagnostic services [service1, service2, service3, ...] Type: DiagService "deleted_services": [], # list with diagnostic services where the service name changed [[services], [old service names]] - "changed_name_of_service": [[],[]], + "changed_name_of_service": [[], []], # list with diagnostic services where the service parameter changed [[services], [changed_parameters], [information_texts]] - "changed_parameters_of_service": [[],[],[]] + "changed_parameters_of_service": [[], [], []], } # service_dict["changed_name_of_service"][{0 = services, 1 = old service names}][i] # service_dict["changed_parameters_of_service"][{0 = services, 1 = changed_parameters, 2 = information_texts}][i] @@ -418,10 +540,12 @@ def compare_diagnostic_layers(self, dl1: DiagLayer, # extract the constant prefixes for the requests of all # services (used for duck-typed rename detection) dl1_request_prefixes: List[Optional[bytes]] = [ - None if s.request is None else s.request.coded_const_prefix() for s in dl1.services + None if s.request is None else s.request.coded_const_prefix() + for s in dl1.services ] dl2_request_prefixes: List[Optional[bytes]] = [ - None if s.request is None else s.request.coded_const_prefix() for s in dl2.services + None if s.request is None else s.request.coded_const_prefix() + for s in dl2.services ] # compare diagnostic services @@ -436,8 +560,9 @@ def compare_diagnostic_layers(self, dl1: DiagLayer, if rq_prefix is None or rq_prefix not in dl2_request_prefixes: # TODO: this will not work in cases where the constant # prefix of a request was modified... - service_dict["new_services"].append( - service1) # type: ignore[arg-type] + service_dict["new_services"].append( + service1 + ) # type: ignore[arg-type] # check whether names of diagnostic services have changed elif service1 not in dl2.services: @@ -449,42 +574,47 @@ def compare_diagnostic_layers(self, dl1: DiagLayer, # save information about changes in dictionary # add new service (type: DiagService) - service_dict["changed_name_of_service"][0].append( - service1) + service_dict["changed_name_of_service"][0].append(service1) # add old service name (type: String) - service_dict["changed_name_of_service"][1].append( - service2.short_name) + service_dict["changed_name_of_service"][1].append( + service2.short_name + ) # compare request, pos. response and neg. response parameters of diagnostic services detailed_information = self.compare_services(service1, service2) # detailed_information = [[infotext1, table1, infotext2, table2, ...], changed_params] # add information about changed diagnostic service parameters to dicitionary - if detailed_information[1]: # check whether string "changed_params" is empty + if detailed_information[ + 1 + ]: # check whether string "changed_params" is empty # new service (type: DiagService) - service_dict["changed_parameters_of_service"][ - 0].append( - service1) + service_dict["changed_parameters_of_service"][0].append( + service1 + ) # add parameters which have been changed (type: String) - service_dict["changed_parameters_of_service"][ - 1].append( - detailed_information[1]) + service_dict["changed_parameters_of_service"][1].append( + detailed_information[1] + ) # add detailed information about changed service parameters (type: list) [infotext1, table1, infotext2, table2, ...] - service_dict["changed_parameters_of_service"][ - 2].append( - detailed_information[0]) + service_dict["changed_parameters_of_service"][2].append( + detailed_information[0] + ) for service2_idx, service2 in enumerate(dl2.services): # check for deleted diagnostic services - if service2.short_name not in dl1_service_names and dl2_request_prefixes[ - service2_idx] not in dl1_request_prefixes: + if ( + service2.short_name not in dl1_service_names + and dl2_request_prefixes[service2_idx] not in dl1_request_prefixes + ): deleted_list = service_dict["deleted_services"] assert isinstance(deleted_list, list) if service2 not in deleted_list: service_dict["deleted_services"].append( # type: ignore[union-attr] - service2) + service2 + ) if service1.short_name == service2.short_name: # compare request, pos. response and neg. response parameters of both diagnostic services @@ -492,29 +622,36 @@ def compare_diagnostic_layers(self, dl1: DiagLayer, # detailed_information = [[infotext1, table1, infotext2, table2, ...], changed_params] # add information about changed diagnostic service parameters to dicitionary - if detailed_information[1]: # check whether string "changed_params" is empty + if detailed_information[ + 1 + ]: # check whether string "changed_params" is empty # new service (type: DiagService) - service_dict["changed_parameters_of_service"][ - 0].append( - service1) + service_dict["changed_parameters_of_service"][0].append( + service1 + ) # add parameters which have been changed (type: String) - service_dict["changed_parameters_of_service"][ - 1].append(detailed_information[1]) + service_dict["changed_parameters_of_service"][1].append( + detailed_information[1] + ) # add detailed information about changed service parameters (type: list) [infotext1, table1, infotext2, table2, ...] service_dict["changed_parameters_of_service"][ # type: ignore[union-attr] - 2].append(detailed_information[0]) # type: ignore[arg-type] + 2 + ].append( + detailed_information[0] + ) # type: ignore[arg-type] return service_dict - def compare_databases(self, database_new: Database, - database_old: Database) -> SpecsChangesVariants: # type: ignore[type-arg] + def compare_databases( + self, database_new: Database, database_old: Database + ) -> SpecsChangesVariants: # type: ignore[type-arg] # compares two PDX-files with each other new_variants = [] - deleted_variants= [] + deleted_variants = [] changes_variants: SpecsChangesVariants = { "new_diagnostic_layers": new_variants, - "deleted_diagnostic_layers": deleted_variants + "deleted_diagnostic_layers": deleted_variants, } # compare databases @@ -525,17 +662,27 @@ def compare_databases(self, database_new: Database, for _, dl2 in enumerate(database_old.diag_layers): # check for deleted diagnostic layers - if (dl2.short_name not in [dl.short_name for dl in database_new.diag_layers] and - dl2 not in changes_variants["deleted_diagnostic_layers"]): + if ( + dl2.short_name + not in [dl.short_name for dl in database_new.diag_layers] + and dl2 not in changes_variants["deleted_diagnostic_layers"] + ): changes_variants[ - "deleted_diagnostic_layers"].append( # type: ignore[union-attr] - dl2) + "deleted_diagnostic_layers" + ].append( # type: ignore[union-attr] + dl2 + ) - if dl1.short_name == dl2.short_name and dl1.short_name in self.diagnostic_layer_names: + if ( + dl1.short_name == dl2.short_name + and dl1.short_name in self.diagnostic_layer_names + ): # compare diagnostic services of both diagnostic layers # save diagnostic service changes in dictionary (empty if no changes) - service_dict: SpecsServiceDict = self.compare_diagnostic_layers(dl1, dl2) + service_dict: SpecsServiceDict = self.compare_diagnostic_layers( + dl1, dl2 + ) if service_dict: # adds information about diagnostic service changes to return variable (changes_variants) changes_variants.update({dl1.short_name: service_dict}) @@ -546,17 +693,19 @@ def compare_databases(self, database_new: Database, def add_subparser(subparsers: SubparsersList) -> None: parser = subparsers.add_parser( "compare", - description="\n".join([ - "Compares two versions of diagnostic layers or databases with each other. Checks whether diagnostic services and its parameters have changed.", - "", - "Examples:", - " Comparison of two diagnostic layers:", - " odxtools compare ./path/to/database.pdx -v variant1 variant2", - " Comparison of two database versions:", - " odxtools compare ./path/to/database.pdx -db ./path/to/old-database.pdx", - " For more information use:", - " odxtools compare -h", - ]), + description="\n".join( + [ + "Compares two versions of diagnostic layers or databases with each other. Checks whether diagnostic services and its parameters have changed.", + "", + "Examples:", + " Comparison of two diagnostic layers:", + " odxtools compare ./path/to/database.pdx -v variant1 variant2", + " Comparison of two database versions:", + " odxtools compare ./path/to/database.pdx -db ./path/to/old-database.pdx", + " For more information use:", + " odxtools compare -h", + ] + ), help="Compares two versions of diagnostic layers and/or databases with each other. Checks whether diagnostic services and its parameters have changed.", formatter_class=argparse.RawTextHelpFormatter, ) @@ -603,7 +752,9 @@ def run(args: argparse.Namespace) -> None: task = Comparison() task.param_detailed = args.no_details - db_names = [args.pdx_file if isinstance(args.pdx_file, str) else str(args.pdx_file[0])] + db_names = [ + args.pdx_file if isinstance(args.pdx_file, str) else str(args.pdx_file[0]) + ] if args.database and args.variants: # compare specified databases, consider only specified variants @@ -612,7 +763,9 @@ def run(args: argparse.Namespace) -> None: db_names.append(name) if isinstance(name, str) else str(name[0]) task.databases = [load_file(name) for name in db_names] - diag_layer_names = {dl.short_name for db in task.databases for dl in db.diag_layers} + diag_layer_names = { + dl.short_name for db in task.databases for dl in db.diag_layers + } task.diagnostic_layer_names = diag_layer_names.intersection(set(args.variants)) @@ -633,22 +786,32 @@ def run(args: argparse.Namespace) -> None: rich_print(f" (compared to '{os.path.basename(db_names[db_idx + 1])}')") rich_print() - rich_print(f"Overview of diagnostic layers (for {os.path.basename(db_names[0])})") - print_dl_metrics([ - variant for variant in task.databases[0].diag_layers - if variant.short_name in task.diagnostic_layer_names - ]) + rich_print( + f"Overview of diagnostic layers (for {os.path.basename(db_names[0])})" + ) + print_dl_metrics( + [ + variant + for variant in task.databases[0].diag_layers + if variant.short_name in task.diagnostic_layer_names + ] + ) rich_print() rich_print( - f"Overview of diagnostic layers (for {os.path.basename(db_names[db_idx+1])})") - print_dl_metrics([ - variant for variant in task.databases[db_idx + 1].diag_layers - if variant.short_name in task.diagnostic_layer_names - ]) + f"Overview of diagnostic layers (for {os.path.basename(db_names[db_idx+1])})" + ) + print_dl_metrics( + [ + variant + for variant in task.databases[db_idx + 1].diag_layers + if variant.short_name in task.diagnostic_layer_names + ] + ) task.print_database_changes( - task.compare_databases(task.databases[0], task.databases[db_idx + 1])) + task.compare_databases(task.databases[0], task.databases[db_idx + 1]) + ) elif args.database: # compare specified databases, consider all variants @@ -659,9 +822,7 @@ def run(args: argparse.Namespace) -> None: # collect all diagnostic layers from all specified databases task.diagnostic_layer_names = { - dl.short_name - for db in task.databases - for dl in db.diag_layers + dl.short_name for db in task.databases for dl in db.diag_layers } task.db_indicator_1 = 0 @@ -675,16 +836,20 @@ def run(args: argparse.Namespace) -> None: rich_print(f" (compared to '{os.path.basename(db_names[db_idx + 1])}')") rich_print() - rich_print(f"Overview of diagnostic layers (for {os.path.basename(db_names[0])})") + rich_print( + f"Overview of diagnostic layers (for {os.path.basename(db_names[0])})" + ) print_dl_metrics(list(task.databases[0].diag_layers)) rich_print() rich_print( - f"Overview of diagnostic layers (for {os.path.basename(db_names[db_idx+1])})") + f"Overview of diagnostic layers (for {os.path.basename(db_names[db_idx+1])})" + ) print_dl_metrics(list(task.databases[db_idx + 1].diag_layers)) task.print_database_changes( - task.compare_databases(task.databases[0], task.databases[db_idx + 1])) + task.compare_databases(task.databases[0], task.databases[db_idx + 1]) + ) elif args.variants: # no databases specified -> comparison of diagnostic layers @@ -692,11 +857,15 @@ def run(args: argparse.Namespace) -> None: odxdb = _parser_utils.load_file(args) task.databases = [odxdb] - diag_layer_names = {dl.short_name for db in task.databases for dl in db.diag_layers} + diag_layer_names = { + dl.short_name for db in task.databases for dl in db.diag_layers + } task.diagnostic_layer_names = diag_layer_names.intersection(set(args.variants)) task.diagnostic_layers = [ - dl for db in task.databases for dl in db.diag_layers + dl + for db in task.databases + for dl in db.diag_layers if dl.short_name in task.diagnostic_layer_names ] @@ -714,12 +883,15 @@ def run(args: argparse.Namespace) -> None: break rich_print() - rich_print(f"Changes in diagnostic layer '{dl.short_name}' ({dl.variant_type.value})") + rich_print( + f"Changes in diagnostic layer '{dl.short_name}' ({dl.variant_type.value})" + ) rich_print( f" (compared to '{task.diagnostic_layers[db_idx + 1].short_name}' ({task.diagnostic_layers[db_idx + 1].variant_type.value}))" ) task.print_dl_changes( - task.compare_diagnostic_layers(dl, task.diagnostic_layers[db_idx + 1])) + task.compare_diagnostic_layers(dl, task.diagnostic_layers[db_idx + 1]) + ) else: # no databases & no variants specified From 716acc589dc10516b78d0350094eb21fc01f2a66 Mon Sep 17 00:00:00 2001 From: vinothk-master Date: Mon, 24 Feb 2025 21:52:47 +0000 Subject: [PATCH 4/4] fixed the issues : change the name to service_spec, executed the reformat-source.sh etc. --- odxtools/cli/compare.py | 474 ++++++++++++++-------------------------- 1 file changed, 167 insertions(+), 307 deletions(-) diff --git a/odxtools/cli/compare.py b/odxtools/cli/compare.py index d5d64117..80692169 100644 --- a/odxtools/cli/compare.py +++ b/odxtools/cli/compare.py @@ -4,8 +4,7 @@ import argparse import os from dataclasses import dataclass -from typing import (Any, Dict, List, Optional, Set, Tuple, TypedDict, Union, - cast) +from typing import Any, Dict, List, Optional, Set, Tuple, TypedDict, Union, cast from rich import print as rich_print from rich.padding import Padding as RichPadding @@ -30,88 +29,78 @@ _odxtools_tool_name_ = "compare" -class SpecsServiceDict(TypedDict): +@dataclass +class ServiceSpecs: diag_layer: str diag_layer_type: str new_services: List[str] # List of new service names deleted_services: List[DiagService] # List of deleted services renamed_service: List[Tuple[str, DiagService]] # List of (old_name, DiagService) changed_name_of_service: List[Tuple[str, DiagService]] # Similar to renamed_service - changed_parameters_of_service: List[ - Tuple[str, DiagService, List[Union[str, int, float]]] - ] + changed_parameters_of_service: List[Tuple[str, DiagService, List[Union[str, int, float]]]] -class SpecsChangesVariants(TypedDict): +@dataclass +class SpecsChangesVariants: new_variants: List[str] deleted_variants: List[str] - service_changes: SpecsServiceDict + service_changes: ServiceSpecs -@dataclass class Display: param_detailed: bool = False obj_detailed: bool = False - def print_dl_changes(self, service_dict: SpecsServiceDict) -> None: + def __init__(self) -> None: + pass - if ( - service_dict["new_services"] - or service_dict["deleted_services"] - or service_dict["changed_name_of_service"] - or service_dict["changed_parameters_of_service"] - ): - # assert isinstance(service_dict["diag_layer"], str) + def print_dl_changes(self, service_spec: ServiceSpecs) -> None: + + if (service_spec["new_services"] or service_spec["deleted_services"] or + service_spec["changed_name_of_service"] or + service_spec["changed_parameters_of_service"]): rich_print() rich_print( - f"Changed diagnostic services for diagnostic layer '{service_dict['diag_layer']}' ({service_dict['diag_layer_type']}):" + f"Changed diagnostic services for diagnostic layer '{service_spec['diag_layer']}' ({service_spec['diag_layer_type']}):" ) - if service_dict["new_services"]: - # assert isinstance(service_dict["new_services"], List) + if service_spec["new_services"]: rich_print() rich_print(" [blue]New services[/blue]") - rich_print(extract_service_tabulation_data(service_dict["new_services"])) + rich_print(extract_service_tabulation_data(service_spec.new_services)) - if service_dict["deleted_services"]: - # assert isinstance(service_dict["deleted_services"], List) + if service_spec["deleted_services"]: rich_print() rich_print(" [blue]Deleted services[/blue]") - rich_print( - extract_service_tabulation_data(service_dict["deleted_services"]) - ) - if service_dict["changed_name_of_service"][0]: + rich_print(extract_service_tabulation_data(service_spec["deleted_services"])) + if service_spec["changed_name_of_service"][0]: rich_print() rich_print(" [blue]Renamed services[/blue]") - rich_print( - extract_service_tabulation_data( - service_dict["changed_name_of_service"][0] - ) - ) - if service_dict["changed_parameters_of_service"][0]: + rich_print(extract_service_tabulation_data(service_spec["changed_name_of_service"][0])) + + if service_spec["changed_parameters_of_service"][0]: rich_print() rich_print(" [blue]Services with parameter changes[/blue]") # create table with information about services with parameter changes changed_param_column = [ - str(x) for x in service_dict["changed_parameters_of_service"][1] + str(x) for x in service_spec["changed_parameters_of_service"][1] ] table = extract_service_tabulation_data( - service_dict["changed_parameters_of_service"][0], + service_spec["changed_parameters_of_service"][0], additional_columns=[("Changed Parameters", changed_param_column)], ) rich_print(table) - for service_idx, service in enumerate( - service_dict["changed_parameters_of_service"][0] - ): + for service_idx, service in enumerate(service_spec["changed_parameters_of_service"][0]): assert isinstance(service, DiagService) rich_print() rich_print( f" Detailed changes of diagnostic service [u cyan]{service.short_name}[/u cyan]" ) # detailed_info in [infotext1, dict1, infotext2, dict2, ...] - info_list = cast(list, service_dict["changed_parameters_of_service"])[ - 2 - ][service_idx] + info_list = cast( + list, # type: ignore[type-arg] + service_spec["changed_parameters_of_service"])[2][service_idx] + for detailed_info in info_list: if isinstance(detailed_info, str): rich_print() @@ -139,24 +128,20 @@ def print_database_changes(self, changes_variants: SpecsChangesVariants) -> None # prints result of database comparison (input variable: dictionary: changes_variants) # diagnostic layers - if ( - changes_variants["new_diagnostic_layers"] - or changes_variants["deleted_diagnostic_layers"] - ): + if (changes_variants["new_diagnostic_layers"] or + changes_variants["deleted_diagnostic_layers"]): rich_print() rich_print("[bright_blue]Changed diagnostic layers[/bright_blue]: ") rich_print(" New diagnostic layers: ") for variant in changes_variants["new_diagnostic_layers"]: assert isinstance(variant, DiagLayer) rich_print( - f" [magenta]{variant.short_name}[/magenta] ({variant.variant_type.value})" - ) + f" [magenta]{variant.short_name}[/magenta] ({variant.variant_type.value})") rich_print(" Deleted diagnostic layers: ") for variant in changes_variants["deleted_diagnostic_layers"]: assert isinstance(variant, DiagLayer) rich_print( - f" [magenta]{variant.short_name}[/magenta] ({variant.variant_type.value})" - ) + f" [magenta]{variant.short_name}[/magenta] ({variant.variant_type.value})") # diagnostic services for _, value in changes_variants.items(): @@ -175,9 +160,7 @@ class Comparison(Display): def __init__(self) -> None: pass - def compare_parameters( - self, param1: Parameter, param2: Parameter - ) -> Dict[str, Any]: + def compare_parameters(self, param1: Parameter, param2: Parameter) -> Dict[str, Any]: # checks whether properties of param1 and param2 differ # checked properties: Name, Byte Position, Bit Length, Semantic, Parameter Type, Value (Coded, Constant, Default etc.), Data Type, Data Object Property (Name, Physical Data Type, Unit) @@ -209,52 +192,35 @@ def append_list( if param1.parameter_type != param2.parameter_type: append_list("Parameter type", param1.parameter_type, param2.parameter_type) - if isinstance(param1, CodedConstParameter) and isinstance( - param2, CodedConstParameter - ): - if ( - param1.diag_coded_type.base_data_type - != param2.diag_coded_type.base_data_type - ): + if isinstance(param1, CodedConstParameter) and isinstance(param2, CodedConstParameter): + if (param1.diag_coded_type.base_data_type != param2.diag_coded_type.base_data_type): append_list( "Data type", param1.diag_coded_type.base_data_type.name, param2.diag_coded_type.base_data_type.name, ) if param1.coded_value != param2.coded_value: - if isinstance(param1.coded_value, int) and isinstance( - param2.coded_value, int - ): + if isinstance(param1.coded_value, int) and isinstance(param2.coded_value, int): append_list( "Value", f"0x{param1.coded_value:0{(param1.get_static_bit_length() or 0) // 4}X}", f"0x{param2.coded_value:0{(param2.get_static_bit_length() or 0) // 4}X}", ) else: - append_list( - "Value", f"{param1.coded_value!r}", f"{param2.coded_value!r}" - ) + append_list("Value", f"{param1.coded_value!r}", f"{param2.coded_value!r}") - elif isinstance(param1, NrcConstParameter) and isinstance( - param2, NrcConstParameter - ): - if ( - param1.diag_coded_type.base_data_type - != param2.diag_coded_type.base_data_type - ): + elif isinstance(param1, NrcConstParameter) and isinstance(param2, NrcConstParameter): + if (param1.diag_coded_type.base_data_type != param2.diag_coded_type.base_data_type): append_list( "Data type", param1.diag_coded_type.base_data_type.name, param2.diag_coded_type.base_data_type.name, ) if param1.coded_values != param2.coded_values: - append_list( - "Values", str(param1.coded_values), str(param2.coded_values) - ) + append_list("Values", str(param1.coded_values), str(param2.coded_values)) - elif (dop_1 := getattr(param1, "dop", None)) is not None and ( - dop_2 := getattr(param2, "dop", None) - ) is not None: + elif (dop_1 := getattr(param1, "dop", None)) is not None and (dop_2 := getattr( + param2, "dop", None)) is not None: if dop_1 != dop_2: # TODO: compare INTERNAL-CONSTR, COMPU-INTERNAL-TO-PHYS of DOP @@ -267,19 +233,15 @@ def append_list( # DOP Unit if getattr(dop_1, "unit", None) and getattr(dop_2, "unit", None): # (properties of unit object: short_name, long_name, description, odx_id, display_name, oid, factor_si_to_unit, offset_si_to_unit, physical_dimension_ref) - if ( - dop_1.unit != dop_2.unit - and dop_1.unit.short_name != dop_2.unit.short_name - ): + if (dop_1.unit != dop_2.unit and + dop_1.unit.short_name != dop_2.unit.short_name): append_list( " DOP unit name", dop_1.unit.short_name, dop_2.unit.short_name, ) - elif ( - dop_1.unit != dop_2.unit - and dop_1.unit.display_name != dop_2.unit.display_name - ): + elif (dop_1.unit != dop_2.unit and + dop_1.unit.display_name != dop_2.unit.display_name): append_list( " DOP unit display name", dop_1.unit.display_name, @@ -289,26 +251,20 @@ def append_list( append_list(" DOP unit object", "", "") if hasattr(dop_1, "physical_type") and hasattr(dop_2, "physical_type"): - if ( - dop_1.physical_type - and dop_2.physical_type - and dop_1.physical_type.base_data_type - != dop_2.physical_type.base_data_type - ): + if (dop_1.physical_type and dop_2.physical_type and + dop_1.physical_type.base_data_type + != dop_2.physical_type.base_data_type): append_list( " DOP physical data type", dop_1.physical_type.base_data_type.name, dop_2.physical_type.base_data_type.name, ) - if ( - isinstance(param1, PhysicalConstantParameter) - and isinstance(param2, PhysicalConstantParameter) - and param1.physical_constant_value != param2.physical_constant_value - ): + if (isinstance(param1, PhysicalConstantParameter) and + isinstance(param2, PhysicalConstantParameter) and + param1.physical_constant_value != param2.physical_constant_value): if isinstance(param1.physical_constant_value, int) and isinstance( - param2.physical_constant_value, int - ): + param2.physical_constant_value, int): append_list( "Constant value", f"0x{param1.physical_constant_value:0{(param1.get_static_bit_length() or 0) // 4}X}", @@ -321,16 +277,12 @@ def append_list( f"{param2.physical_constant_value!r}", ) - elif ( - isinstance(param1, ValueParameter) - and isinstance(param2, ValueParameter) - and param1.physical_default_value is not None - and param2.physical_default_value is not None - and param1.physical_default_value != param2.physical_default_value - ): + elif (isinstance(param1, ValueParameter) and isinstance(param2, ValueParameter) and + param1.physical_default_value is not None and + param2.physical_default_value is not None and + param1.physical_default_value != param2.physical_default_value): if isinstance(param1.physical_default_value, int) and isinstance( - param2.physical_default_value, int - ): + param2.physical_default_value, int): append_list( "Default value", f"0x{param1.physical_default_value:0{(param1.get_static_bit_length() or 0) // 4}X}", @@ -345,62 +297,44 @@ def append_list( return {"Property": property, "Old Value": old, "New Value": new} - def compare_services( - self, service1: DiagService, service2: DiagService - ) -> List[SpecsServiceDict]: + def compare_services(self, service1: DiagService, service2: DiagService) -> List[ServiceSpecs]: # compares request, positive response and negative response parameters of two diagnostic services information: List[Union[str, Dict[str, Any]]] = ( - [] - ) # information = [infotext1, table1, infotext2, table2, ...] + []) # information = [infotext1, table1, infotext2, table2, ...] changed_params: str = "" # Request - if ( - service1.request is not None - and service2.request is not None - and len(service1.request.parameters) == len(service2.request.parameters) - ): + if (service1.request is not None and service2.request is not None and + len(service1.request.parameters) == len(service2.request.parameters)): for res1_idx, param1 in enumerate(service1.request.parameters): for res2_idx, param2 in enumerate(service2.request.parameters): if res1_idx == res2_idx: # find changed request parameter properties table = self.compare_parameters(param1, param2) - infotext = ( - f" Properties of request parameter '{param2.short_name}' " - f"that have changed:\n" - ) + infotext = (f" Properties of request parameter '{param2.short_name}' " + f"that have changed:\n") # array index starts with 0 -> param[0] is 1. service parameter if table["Property"]: information.append(infotext) information.append(table) - changed_params += ( - f"request parameter '{param2.short_name}',\n" - ) + changed_params += (f"request parameter '{param2.short_name}',\n") else: changed_params += "request parameter list, " # infotext - information.append( - f"List of request parameters for service '{service2.short_name}' " - f"is not identical.\n" - ) + information.append(f"List of request parameters for service '{service2.short_name}' " + f"is not identical.\n") # table - param_list1 = ( - [] if service1.request is None else service1.request.parameters - ) - param_list2 = ( - [] if service2.request is None else service2.request.parameters - ) + param_list1 = ([] if service1.request is None else service1.request.parameters) + param_list2 = ([] if service2.request is None else service2.request.parameters) - information.append( - { - "List": ["Old list", "New list"], - "Values": [f"\\{param_list1}", f"\\{param_list2}"], - } - ) + information.append({ + "List": ["Old list", "New list"], + "Values": [f"\\{param_list1}", f"\\{param_list2}"], + }) # Positive Responses if len(service1.positive_responses) == len(service2.positive_responses): @@ -409,16 +343,13 @@ def compare_services( if res1_idx == res2_idx: if len(response1.parameters) == len(response2.parameters): for param1_idx, param1 in enumerate(response1.parameters): - for param2_idx, param2 in enumerate( - response2.parameters - ): + for param2_idx, param2 in enumerate(response2.parameters): if param1_idx == param2_idx: # find changed positive response parameter properties table = self.compare_parameters(param1, param2) infotext = ( f" Properties of positive response parameter '{param2.short_name}' that " - f"have changed:\n" - ) + f"have changed:\n") # array index starts with 0 -> param[0] is first service parameter if table["Property"]: @@ -432,31 +363,26 @@ def compare_services( f"List of positive response parameters for service '{service2.short_name}' is not identical." ) # table - information.append( - { - "List": ["Old list", "New list"], - "Values": [ - str(response1.parameters), - str(response2.parameters), - ], - } - ) + information.append({ + "List": ["Old list", "New list"], + "Values": [ + str(response1.parameters), + str(response2.parameters), + ], + }) else: changed_params += "positive responses list, " # infotext information.append( - f"List of positive responses for service '{service2.short_name}' is not identical." - ) + f"List of positive responses for service '{service2.short_name}' is not identical.") # table - information.append( - { - "List": ["Old list", "New list"], - "Values": [ - str(service1.positive_responses), - str(service2.positive_responses), - ], - } - ) + information.append({ + "List": ["Old list", "New list"], + "Values": [ + str(service1.positive_responses), + str(service2.positive_responses), + ], + }) # Negative Responses if len(service1.negative_responses) == len(service2.negative_responses): @@ -465,9 +391,7 @@ def compare_services( if res1_idx == res2_idx: if len(response1.parameters) == len(response2.parameters): for param1_idx, param1 in enumerate(response1.parameters): - for param2_idx, param2 in enumerate( - response2.parameters - ): + for param2_idx, param2 in enumerate(response2.parameters): if param1_idx == param2_idx: # find changed negative response parameter properties table = self.compare_parameters(param1, param2) @@ -485,15 +409,13 @@ def compare_services( f"List of positive response parameters for service '{service2.short_name}' is not identical.\n" ) # table - information.append( - { - "List": ["Old list", "New list"], - "Values": [ - str(response1.parameters), - str(response2.parameters), - ], - } - ) + information.append({ + "List": ["Old list", "New list"], + "Values": [ + str(response1.parameters), + str(response2.parameters), + ], + }) else: changed_params += "negative responses list, " # infotext @@ -501,26 +423,22 @@ def compare_services( f"List of positive responses for service '{service2.short_name}' is not identical.\n" ) # table - information.append( - { - "List": ["Old list", "New list"], - "Values": [ - str(service1.negative_responses), - str(service2.negative_responses), - ], - } - ) + information.append({ + "List": ["Old list", "New list"], + "Values": [ + str(service1.negative_responses), + str(service2.negative_responses), + ], + }) return [information, changed_params] - def compare_diagnostic_layers( - self, dl1: DiagLayer, dl2: DiagLayer - ) -> SpecsServiceDict: + def compare_diagnostic_layers(self, dl1: DiagLayer, dl2: DiagLayer) -> ServiceSpecs: # compares diagnostic services of two diagnostic layers with each other # save changes in dictionary (service_dict) # TODO: add comparison of SingleECUJobs - service_dict: SpecsServiceDict = { + service_dict: ServiceSpecs = { "diag_layer": dl1.short_name, "diag_layer_type": dl1.variant_type.value, # list with added diagnostic services [service1, service2, service3, ...] Type: DiagService @@ -540,12 +458,10 @@ def compare_diagnostic_layers( # extract the constant prefixes for the requests of all # services (used for duck-typed rename detection) dl1_request_prefixes: List[Optional[bytes]] = [ - None if s.request is None else s.request.coded_const_prefix() - for s in dl1.services + None if s.request is None else s.request.coded_const_prefix() for s in dl1.services ] dl2_request_prefixes: List[Optional[bytes]] = [ - None if s.request is None else s.request.coded_const_prefix() - for s in dl2.services + None if s.request is None else s.request.coded_const_prefix() for s in dl2.services ] # compare diagnostic services @@ -560,9 +476,7 @@ def compare_diagnostic_layers( if rq_prefix is None or rq_prefix not in dl2_request_prefixes: # TODO: this will not work in cases where the constant # prefix of a request was modified... - service_dict["new_services"].append( - service1 - ) # type: ignore[arg-type] + service_dict["new_services"].append(service1) # type: ignore[arg-type] # check whether names of diagnostic services have changed elif service1 not in dl2.services: @@ -576,45 +490,34 @@ def compare_diagnostic_layers( # add new service (type: DiagService) service_dict["changed_name_of_service"][0].append(service1) # add old service name (type: String) - service_dict["changed_name_of_service"][1].append( - service2.short_name - ) + service_dict["changed_name_of_service"][1].append(service2.short_name) # compare request, pos. response and neg. response parameters of diagnostic services detailed_information = self.compare_services(service1, service2) # detailed_information = [[infotext1, table1, infotext2, table2, ...], changed_params] # add information about changed diagnostic service parameters to dicitionary - if detailed_information[ - 1 - ]: # check whether string "changed_params" is empty + if detailed_information[1]: # check whether string "changed_params" is empty # new service (type: DiagService) - service_dict["changed_parameters_of_service"][0].append( - service1 - ) + service_dict["changed_parameters_of_service"][0].append(service1) # add parameters which have been changed (type: String) service_dict["changed_parameters_of_service"][1].append( - detailed_information[1] - ) + detailed_information[1]) # add detailed information about changed service parameters (type: list) [infotext1, table1, infotext2, table2, ...] service_dict["changed_parameters_of_service"][2].append( - detailed_information[0] - ) + detailed_information[0]) for service2_idx, service2 in enumerate(dl2.services): # check for deleted diagnostic services - if ( - service2.short_name not in dl1_service_names - and dl2_request_prefixes[service2_idx] not in dl1_request_prefixes - ): + if (service2.short_name not in dl1_service_names and + dl2_request_prefixes[service2_idx] not in dl1_request_prefixes): deleted_list = service_dict["deleted_services"] assert isinstance(deleted_list, list) if service2 not in deleted_list: service_dict["deleted_services"].append( # type: ignore[union-attr] - service2 - ) + service2) if service1.short_name == service2.short_name: # compare request, pos. response and neg. response parameters of both diagnostic services @@ -622,28 +525,19 @@ def compare_diagnostic_layers( # detailed_information = [[infotext1, table1, infotext2, table2, ...], changed_params] # add information about changed diagnostic service parameters to dicitionary - if detailed_information[ - 1 - ]: # check whether string "changed_params" is empty + if detailed_information[1]: # check whether string "changed_params" is empty # new service (type: DiagService) - service_dict["changed_parameters_of_service"][0].append( - service1 - ) + service_dict["changed_parameters_of_service"][0].append(service1) # add parameters which have been changed (type: String) service_dict["changed_parameters_of_service"][1].append( - detailed_information[1] - ) + detailed_information[1]) # add detailed information about changed service parameters (type: list) [infotext1, table1, infotext2, table2, ...] service_dict["changed_parameters_of_service"][ # type: ignore[union-attr] - 2 - ].append( - detailed_information[0] - ) # type: ignore[arg-type] + 2].append(detailed_information[0]) # type: ignore[arg-type] return service_dict - def compare_databases( - self, database_new: Database, database_old: Database - ) -> SpecsChangesVariants: # type: ignore[type-arg] + def compare_databases(self, database_new: Database, + database_old: Database) -> SpecsChangesVariants: # type: ignore[type-arg] # compares two PDX-files with each other new_variants = [] @@ -662,27 +556,18 @@ def compare_databases( for _, dl2 in enumerate(database_old.diag_layers): # check for deleted diagnostic layers - if ( - dl2.short_name - not in [dl.short_name for dl in database_new.diag_layers] - and dl2 not in changes_variants["deleted_diagnostic_layers"] - ): + if (dl2.short_name not in [dl.short_name for dl in database_new.diag_layers] and + dl2 not in changes_variants["deleted_diagnostic_layers"]): changes_variants[ - "deleted_diagnostic_layers" - ].append( # type: ignore[union-attr] - dl2 - ) + "deleted_diagnostic_layers"].append( # type: ignore[union-attr] + dl2) - if ( - dl1.short_name == dl2.short_name - and dl1.short_name in self.diagnostic_layer_names - ): + if (dl1.short_name == dl2.short_name and + dl1.short_name in self.diagnostic_layer_names): # compare diagnostic services of both diagnostic layers # save diagnostic service changes in dictionary (empty if no changes) - service_dict: SpecsServiceDict = self.compare_diagnostic_layers( - dl1, dl2 - ) + service_dict: ServiceSpecs = self.compare_diagnostic_layers(dl1, dl2) if service_dict: # adds information about diagnostic service changes to return variable (changes_variants) changes_variants.update({dl1.short_name: service_dict}) @@ -693,19 +578,17 @@ def compare_databases( def add_subparser(subparsers: SubparsersList) -> None: parser = subparsers.add_parser( "compare", - description="\n".join( - [ - "Compares two versions of diagnostic layers or databases with each other. Checks whether diagnostic services and its parameters have changed.", - "", - "Examples:", - " Comparison of two diagnostic layers:", - " odxtools compare ./path/to/database.pdx -v variant1 variant2", - " Comparison of two database versions:", - " odxtools compare ./path/to/database.pdx -db ./path/to/old-database.pdx", - " For more information use:", - " odxtools compare -h", - ] - ), + description="\n".join([ + "Compares two versions of diagnostic layers or databases with each other. Checks whether diagnostic services and its parameters have changed.", + "", + "Examples:", + " Comparison of two diagnostic layers:", + " odxtools compare ./path/to/database.pdx -v variant1 variant2", + " Comparison of two database versions:", + " odxtools compare ./path/to/database.pdx -db ./path/to/old-database.pdx", + " For more information use:", + " odxtools compare -h", + ]), help="Compares two versions of diagnostic layers and/or databases with each other. Checks whether diagnostic services and its parameters have changed.", formatter_class=argparse.RawTextHelpFormatter, ) @@ -752,9 +635,7 @@ def run(args: argparse.Namespace) -> None: task = Comparison() task.param_detailed = args.no_details - db_names = [ - args.pdx_file if isinstance(args.pdx_file, str) else str(args.pdx_file[0]) - ] + db_names = [args.pdx_file if isinstance(args.pdx_file, str) else str(args.pdx_file[0])] if args.database and args.variants: # compare specified databases, consider only specified variants @@ -763,9 +644,7 @@ def run(args: argparse.Namespace) -> None: db_names.append(name) if isinstance(name, str) else str(name[0]) task.databases = [load_file(name) for name in db_names] - diag_layer_names = { - dl.short_name for db in task.databases for dl in db.diag_layers - } + diag_layer_names = {dl.short_name for db in task.databases for dl in db.diag_layers} task.diagnostic_layer_names = diag_layer_names.intersection(set(args.variants)) @@ -786,32 +665,22 @@ def run(args: argparse.Namespace) -> None: rich_print(f" (compared to '{os.path.basename(db_names[db_idx + 1])}')") rich_print() - rich_print( - f"Overview of diagnostic layers (for {os.path.basename(db_names[0])})" - ) - print_dl_metrics( - [ - variant - for variant in task.databases[0].diag_layers - if variant.short_name in task.diagnostic_layer_names - ] - ) + rich_print(f"Overview of diagnostic layers (for {os.path.basename(db_names[0])})") + print_dl_metrics([ + variant for variant in task.databases[0].diag_layers + if variant.short_name in task.diagnostic_layer_names + ]) rich_print() rich_print( - f"Overview of diagnostic layers (for {os.path.basename(db_names[db_idx+1])})" - ) - print_dl_metrics( - [ - variant - for variant in task.databases[db_idx + 1].diag_layers - if variant.short_name in task.diagnostic_layer_names - ] - ) + f"Overview of diagnostic layers (for {os.path.basename(db_names[db_idx+1])})") + print_dl_metrics([ + variant for variant in task.databases[db_idx + 1].diag_layers + if variant.short_name in task.diagnostic_layer_names + ]) task.print_database_changes( - task.compare_databases(task.databases[0], task.databases[db_idx + 1]) - ) + task.compare_databases(task.databases[0], task.databases[db_idx + 1])) elif args.database: # compare specified databases, consider all variants @@ -822,7 +691,9 @@ def run(args: argparse.Namespace) -> None: # collect all diagnostic layers from all specified databases task.diagnostic_layer_names = { - dl.short_name for db in task.databases for dl in db.diag_layers + dl.short_name + for db in task.databases + for dl in db.diag_layers } task.db_indicator_1 = 0 @@ -836,20 +707,16 @@ def run(args: argparse.Namespace) -> None: rich_print(f" (compared to '{os.path.basename(db_names[db_idx + 1])}')") rich_print() - rich_print( - f"Overview of diagnostic layers (for {os.path.basename(db_names[0])})" - ) + rich_print(f"Overview of diagnostic layers (for {os.path.basename(db_names[0])})") print_dl_metrics(list(task.databases[0].diag_layers)) rich_print() rich_print( - f"Overview of diagnostic layers (for {os.path.basename(db_names[db_idx+1])})" - ) + f"Overview of diagnostic layers (for {os.path.basename(db_names[db_idx+1])})") print_dl_metrics(list(task.databases[db_idx + 1].diag_layers)) task.print_database_changes( - task.compare_databases(task.databases[0], task.databases[db_idx + 1]) - ) + task.compare_databases(task.databases[0], task.databases[db_idx + 1])) elif args.variants: # no databases specified -> comparison of diagnostic layers @@ -857,15 +724,11 @@ def run(args: argparse.Namespace) -> None: odxdb = _parser_utils.load_file(args) task.databases = [odxdb] - diag_layer_names = { - dl.short_name for db in task.databases for dl in db.diag_layers - } + diag_layer_names = {dl.short_name for db in task.databases for dl in db.diag_layers} task.diagnostic_layer_names = diag_layer_names.intersection(set(args.variants)) task.diagnostic_layers = [ - dl - for db in task.databases - for dl in db.diag_layers + dl for db in task.databases for dl in db.diag_layers if dl.short_name in task.diagnostic_layer_names ] @@ -883,15 +746,12 @@ def run(args: argparse.Namespace) -> None: break rich_print() - rich_print( - f"Changes in diagnostic layer '{dl.short_name}' ({dl.variant_type.value})" - ) + rich_print(f"Changes in diagnostic layer '{dl.short_name}' ({dl.variant_type.value})") rich_print( f" (compared to '{task.diagnostic_layers[db_idx + 1].short_name}' ({task.diagnostic_layers[db_idx + 1].variant_type.value}))" ) task.print_dl_changes( - task.compare_diagnostic_layers(dl, task.diagnostic_layers[db_idx + 1]) - ) + task.compare_diagnostic_layers(dl, task.diagnostic_layers[db_idx + 1])) else: # no databases & no variants specified