diff --git a/applications/RomApplication/python_scripts/calculate_rom_basis_output_process.py b/applications/RomApplication/python_scripts/calculate_rom_basis_output_process.py index 38fd530ebfb8..594404103d4c 100644 --- a/applications/RomApplication/python_scripts/calculate_rom_basis_output_process.py +++ b/applications/RomApplication/python_scripts/calculate_rom_basis_output_process.py @@ -1,11 +1,11 @@ # Import Python modules -import json import numpy from pathlib import Path # Importing the Kratos Library import KratosMultiphysics from KratosMultiphysics.RomApplication.randomized_singular_value_decomposition import RandomizedSingularValueDecomposition +from KratosMultiphysics.RomApplication import rom_parameters def Factory(settings, model): if not isinstance(settings, KratosMultiphysics.Parameters): @@ -203,9 +203,7 @@ def _PrintRomBasis(self, u, sigma): raise Exception(err_msg) # Creating the ROM JSON file containing or not the modes depending on "self.rom_basis_output_format" - output_filename = self.rom_basis_output_folder / f"{self.rom_basis_output_name}.json" - with output_filename.open('w') as f: - json.dump(rom_basis_dict, f, indent = 4) + rom_parameters.WriteRomParameters(self.rom_basis_output_folder, self.rom_basis_output_name, rom_basis_dict) def ExecuteFinalize(self): diff --git a/applications/RomApplication/python_scripts/hrom_training_utility.py b/applications/RomApplication/python_scripts/hrom_training_utility.py index cef280b07a5e..cd6e513ba44c 100644 --- a/applications/RomApplication/python_scripts/hrom_training_utility.py +++ b/applications/RomApplication/python_scripts/hrom_training_utility.py @@ -10,6 +10,7 @@ import KratosMultiphysics.RomApplication as KratosROM from KratosMultiphysics.RomApplication.empirical_cubature_method import EmpiricalCubatureMethod from KratosMultiphysics.RomApplication.randomized_singular_value_decomposition import RandomizedSingularValueDecomposition +from KratosMultiphysics.RomApplication import rom_parameters class HRomTrainingUtility(object): """Auxiliary utility for the HROM training. @@ -229,9 +230,7 @@ def CreateHRomModelParts(self): if self.hrom_output_format == "numpy": element_ids_list, condition_ids_list = self.__CreateListsWithRomElements() elif self.hrom_output_format == "json": - with (self.rom_basis_output_folder / self.rom_basis_output_name).with_suffix('.json').open('r') as f: - rom_parameters = KratosMultiphysics.Parameters(f.read()) - hrom_info = rom_parameters["elements_and_weights"] + hrom_info = KratosMultiphysics.Parameters(json.dumps(rom_parameters.ReadRomParametersAsVersion1(self.rom_basis_output_folder, self.rom_basis_output_name)["elements_and_weights"])) # Get the weights and fill the HROM computing model part if (self.projection_strategy=="lspg"): @@ -427,12 +426,9 @@ def AppendHRomWeightsToRomParameters(self): np.save(self.rom_basis_output_folder / "HROM_ConditionIds.npy", condition_ids) elif self.hrom_output_format=="json": - with (self.rom_basis_output_folder / self.rom_basis_output_name).with_suffix('.json').open('r') as f: - updated_rom_parameters = json.load(f) - updated_rom_parameters["elements_and_weights"] = hrom_weights # TODO: Rename elements_and_weights to hrom_weights - - with (self.rom_basis_output_folder / self.rom_basis_output_name).with_suffix('.json').open('w') as f: - json.dump(updated_rom_parameters, f, indent=4) + updated_rom_parameters = rom_parameters.ReadRomParametersAsVersion1(self.rom_basis_output_folder, self.rom_basis_output_name) + updated_rom_parameters["elements_and_weights"] = hrom_weights # TODO: Rename elements_and_weights to hrom_weights + rom_parameters.WriteRomParameters(self.rom_basis_output_folder, self.rom_basis_output_name, updated_rom_parameters) if self.echo_level > 0 : KratosMultiphysics.Logger.PrintInfo("HRomTrainingUtility","\'RomParameters.json\' file updated with HROM weights.") diff --git a/applications/RomApplication/python_scripts/petrov_galerkin_training_utility.py b/applications/RomApplication/python_scripts/petrov_galerkin_training_utility.py index 6e0c907eb9d1..a138e52eb910 100644 --- a/applications/RomApplication/python_scripts/petrov_galerkin_training_utility.py +++ b/applications/RomApplication/python_scripts/petrov_galerkin_training_utility.py @@ -1,5 +1,4 @@ # Import Python modules -import json import numpy as np from pathlib import Path from glob import glob @@ -11,6 +10,7 @@ # Import applications import KratosMultiphysics.RomApplication as KratosROM from KratosMultiphysics.RomApplication.randomized_singular_value_decomposition import RandomizedSingularValueDecomposition +from KratosMultiphysics.RomApplication import rom_parameters import KratosMultiphysics.kratos_utilities as kratos_utils class PetrovGalerkinTrainingUtility(object): @@ -151,30 +151,26 @@ def _AppendNewBasisToRomParameters(self, u): # Storing Petrov-Galerkin modes in Numpy format np.save(self.rom_basis_output_folder / "LeftBasisMatrix.npy", u) - with (self.rom_basis_output_folder / self.rom_basis_output_name).with_suffix('.json').open('r') as f: - updated_rom_parameters = json.load(f) - updated_rom_parameters["rom_settings"]["petrov_galerkin_number_of_rom_dofs"] = petrov_galerkin_number_of_rom_dofs - updated_rom_parameters["petrov_galerkin_nodal_modes"] = petrov_galerkin_nodal_modes - - with (self.rom_basis_output_folder / self.rom_basis_output_name).with_suffix('.json').open('w') as f: - json.dump(updated_rom_parameters, f, indent=4) + updated_rom_parameters = rom_parameters.ReadRomParametersAsVersion1(self.rom_basis_output_folder, self.rom_basis_output_name) + updated_rom_parameters["rom_settings"]["petrov_galerkin_number_of_rom_dofs"] = petrov_galerkin_number_of_rom_dofs + updated_rom_parameters["petrov_galerkin_nodal_modes"] = petrov_galerkin_nodal_modes + rom_parameters.WriteRomParameters(self.rom_basis_output_folder, self.rom_basis_output_name, updated_rom_parameters) if self.echo_level > 0 : KratosMultiphysics.Logger.PrintInfo("PetrovGalerkinTrainingUtility","\'RomParameters.json\' file updated with HROM weights.") def __GetGalerkinBasis(self): if self.rom_format == "json": - with open(self.rom_basis_output_folder / self.rom_basis_output_name.with_suffix(".json"), 'r') as f: - galerkin_rom_parameters = json.load(f) - N_Dof_per_node = len(galerkin_rom_parameters["rom_settings"]["nodal_unknowns"]) - N_nodes = len(galerkin_rom_parameters["nodal_modes"]) - N_Dofs = int(N_Dof_per_node*N_nodes) - N_Dofs_rom = galerkin_rom_parameters["rom_settings"]["number_of_rom_dofs"] - u = np.zeros((N_Dofs,N_Dofs_rom)) - counter_in = 0 - for key in galerkin_rom_parameters["nodal_modes"].keys(): - counter_fin = counter_in + N_Dof_per_node - u[counter_in:counter_fin,:] = np.array(galerkin_rom_parameters["nodal_modes"][key]) - counter_in = counter_fin + galerkin_rom_parameters = rom_parameters.ReadRomParametersAsVersion1(self.rom_basis_output_folder, self.rom_basis_output_name) + N_Dof_per_node = len(galerkin_rom_parameters["rom_settings"]["nodal_unknowns"]) + N_nodes = len(galerkin_rom_parameters["nodal_modes"]) + N_Dofs = int(N_Dof_per_node*N_nodes) + N_Dofs_rom = galerkin_rom_parameters["rom_settings"]["number_of_rom_dofs"] + u = np.zeros((N_Dofs,N_Dofs_rom)) + counter_in = 0 + for key in galerkin_rom_parameters["nodal_modes"].keys(): + counter_fin = counter_in + N_Dof_per_node + u[counter_in:counter_fin,:] = np.array(galerkin_rom_parameters["nodal_modes"][key]) + counter_in = counter_fin elif self.rom_format == "numpy": u = np.load(self.rom_basis_output_folder / "RightBasisMatrix.npy") diff --git a/applications/RomApplication/python_scripts/rom_analysis.py b/applications/RomApplication/python_scripts/rom_analysis.py index 10beca4a53e2..c02f389a9a43 100644 --- a/applications/RomApplication/python_scripts/rom_analysis.py +++ b/applications/RomApplication/python_scripts/rom_analysis.py @@ -1,4 +1,5 @@ import importlib +import json import KratosMultiphysics import KratosMultiphysics.RomApplication as KratosROM @@ -6,6 +7,7 @@ from KratosMultiphysics.RomApplication.hrom_training_utility import HRomTrainingUtility from KratosMultiphysics.RomApplication.petrov_galerkin_training_utility import PetrovGalerkinTrainingUtility from KratosMultiphysics.RomApplication.calculate_rom_basis_output_process import CalculateRomBasisOutputProcess +from KratosMultiphysics.RomApplication import rom_parameters import numpy as np from glob import glob @@ -38,8 +40,7 @@ def _CreateSolver(self): self.rom_basis_output_folder = Path(self.rom_basis_output_folder) # Get the ROM settings from the RomParameters.json input file - with open(self.rom_basis_output_folder / self.rom_basis_output_name.with_suffix('.json')) as rom_parameters: - self.rom_parameters = KratosMultiphysics.Parameters(rom_parameters.read()) + self.rom_parameters = KratosMultiphysics.Parameters(json.dumps(rom_parameters.ReadRomParametersAsVersion1(self.rom_basis_output_folder, self.rom_basis_output_name))) # Set the ROM settings in the "solver_settings" of the solver introducing the physics self.project_parameters["solver_settings"].AddValue("rom_settings", self.rom_parameters["rom_settings"]) diff --git a/applications/RomApplication/python_scripts/rom_manager.py b/applications/RomApplication/python_scripts/rom_manager.py index 2ea4668e94d7..2a614b7fcb3c 100644 --- a/applications/RomApplication/python_scripts/rom_manager.py +++ b/applications/RomApplication/python_scripts/rom_manager.py @@ -1,7 +1,6 @@ import numpy as np import importlib import json -from pathlib import Path import types import KratosMultiphysics from KratosMultiphysics.RomApplication.rom_database import RomDatabase @@ -9,6 +8,7 @@ from KratosMultiphysics.RomApplication.calculate_rom_basis_output_process import CalculateRomBasisOutputProcess from KratosMultiphysics.RomApplication.randomized_singular_value_decomposition import RandomizedSingularValueDecomposition from KratosMultiphysics.RomApplication.rom_nn_interface import NN_ROM_Interface +from KratosMultiphysics.RomApplication import rom_parameters class RomManager(object): @@ -721,81 +721,72 @@ def _ChangeRomFlags(self, simulation_to_run = 'ROM'): parameters_file_folder = self.general_rom_manager_parameters["ROM"]["rom_basis_output_folder"].GetString() if self.general_rom_manager_parameters["ROM"].Has("rom_basis_output_folder") else "rom_data" parameters_file_name = self.general_rom_manager_parameters["ROM"]["rom_basis_output_name"].GetString() if self.general_rom_manager_parameters["ROM"].Has("rom_basis_output_name") else "RomParameters" - # Convert to Path objects - parameters_file_folder = Path(parameters_file_folder) - parameters_file_name = Path(parameters_file_name) - - parameters_file_path = parameters_file_folder / parameters_file_name.with_suffix('.json') - - with parameters_file_path.open('r+') as parameter_file: - f=json.load(parameter_file) - f['assembling_strategy'] = self.general_rom_manager_parameters['assembling_strategy'].GetString() if self.general_rom_manager_parameters.Has('assembling_strategy') else 'global' - self._AddHromParametersToRomParameters(f) - if simulation_to_run=='GalerkinROM': - f['projection_strategy']="galerkin" - f['train_hrom']=False - f['run_hrom']=False - f["rom_settings"]['rom_bns_settings'] = self._SetGalerkinBnSParameters() - elif simulation_to_run=='trainHROMGalerkin': - f['train_hrom']=True - f['run_hrom']=False - f["rom_settings"]['rom_bns_settings'] = self._SetGalerkinBnSParameters() - elif simulation_to_run=='runHROMGalerkin': - f['projection_strategy']="galerkin" - f['train_hrom']=False - f['run_hrom']=True - f["rom_settings"]['rom_bns_settings'] = self._SetGalerkinBnSParameters() - elif simulation_to_run == 'lspg': - f['train_hrom'] = False - f['run_hrom'] = False - f['projection_strategy'] = "lspg" - f["rom_settings"]['rom_bns_settings'] = self._SetLSPGBnSParameters() - elif simulation_to_run == 'trainHROMLSPG': - f['train_hrom'] = True - f['run_hrom'] = False - f['projection_strategy'] = "lspg" - f["rom_settings"]['rom_bns_settings'] = self._SetLSPGBnSParameters() - elif simulation_to_run == 'runHROMLSPG': - f['train_hrom'] = False - f['run_hrom'] = True - f['projection_strategy'] = "lspg" - f["rom_settings"]['rom_bns_settings'] = self._SetLSPGBnSParameters() - elif simulation_to_run == 'TrainPG': - f['train_hrom'] = False - f['run_hrom'] = False - f['projection_strategy'] = "lspg" - f["rom_settings"]['rom_bns_settings'] = self._SetLSPGBnSParameters() - f["rom_settings"]['rom_bns_settings']['train_petrov_galerkin'] = True # Override the default - elif simulation_to_run=='PG': - f['train_hrom']=False - f['run_hrom']=False - f['projection_strategy']="petrov_galerkin" - f["rom_settings"]['rom_bns_settings'] = self._SetPetrovGalerkinBnSParameters() - elif simulation_to_run=='trainHROMPetrovGalerkin': - f['train_hrom']=True - f['run_hrom']=False - f['projection_strategy']="petrov_galerkin" - f["rom_settings"]['rom_bns_settings'] = self._SetPetrovGalerkinBnSParameters() - elif simulation_to_run=='runHROMPetrovGalerkin': - f['train_hrom']=False - f['run_hrom']=True - f['projection_strategy']="petrov_galerkin" - f["rom_settings"]['rom_bns_settings'] = self._SetPetrovGalerkinBnSParameters() - elif simulation_to_run=='GalerkinROM_ANN': - f['train_hrom']=False - f['run_hrom']=False - f['projection_strategy']="galerkin_ann" - f["rom_settings"]['rom_bns_settings'] = self._SetGalerkinBnSParameters() - elif simulation_to_run=='lspg_ANN': - f['train_hrom']=False - f['run_hrom']=False - f['projection_strategy']="lspg_ann" - f["rom_settings"]['rom_bns_settings'] = self._SetLSPGBnSParameters() - else: - raise Exception(f'Unknown flag "{simulation_to_run}" change for RomParameters.json') - parameter_file.seek(0) - json.dump(f,parameter_file,indent=4) - parameter_file.truncate() + f = rom_parameters.ReadRomParametersAsVersion1(parameters_file_folder, parameters_file_name) + f['assembling_strategy'] = self.general_rom_manager_parameters['assembling_strategy'].GetString() if self.general_rom_manager_parameters.Has('assembling_strategy') else 'global' + self._AddHromParametersToRomParameters(f) + if simulation_to_run=='GalerkinROM': + f['projection_strategy']="galerkin" + f['train_hrom']=False + f['run_hrom']=False + f["rom_settings"]['rom_bns_settings'] = self._SetGalerkinBnSParameters() + elif simulation_to_run=='trainHROMGalerkin': + f['train_hrom']=True + f['run_hrom']=False + f["rom_settings"]['rom_bns_settings'] = self._SetGalerkinBnSParameters() + elif simulation_to_run=='runHROMGalerkin': + f['projection_strategy']="galerkin" + f['train_hrom']=False + f['run_hrom']=True + f["rom_settings"]['rom_bns_settings'] = self._SetGalerkinBnSParameters() + elif simulation_to_run == 'lspg': + f['train_hrom'] = False + f['run_hrom'] = False + f['projection_strategy'] = "lspg" + f["rom_settings"]['rom_bns_settings'] = self._SetLSPGBnSParameters() + elif simulation_to_run == 'trainHROMLSPG': + f['train_hrom'] = True + f['run_hrom'] = False + f['projection_strategy'] = "lspg" + f["rom_settings"]['rom_bns_settings'] = self._SetLSPGBnSParameters() + elif simulation_to_run == 'runHROMLSPG': + f['train_hrom'] = False + f['run_hrom'] = True + f['projection_strategy'] = "lspg" + f["rom_settings"]['rom_bns_settings'] = self._SetLSPGBnSParameters() + elif simulation_to_run == 'TrainPG': + f['train_hrom'] = False + f['run_hrom'] = False + f['projection_strategy'] = "lspg" + f["rom_settings"]['rom_bns_settings'] = self._SetLSPGBnSParameters() + f["rom_settings"]['rom_bns_settings']['train_petrov_galerkin'] = True # Override the default + elif simulation_to_run=='PG': + f['train_hrom']=False + f['run_hrom']=False + f['projection_strategy']="petrov_galerkin" + f["rom_settings"]['rom_bns_settings'] = self._SetPetrovGalerkinBnSParameters() + elif simulation_to_run=='trainHROMPetrovGalerkin': + f['train_hrom']=True + f['run_hrom']=False + f['projection_strategy']="petrov_galerkin" + f["rom_settings"]['rom_bns_settings'] = self._SetPetrovGalerkinBnSParameters() + elif simulation_to_run=='runHROMPetrovGalerkin': + f['train_hrom']=False + f['run_hrom']=True + f['projection_strategy']="petrov_galerkin" + f["rom_settings"]['rom_bns_settings'] = self._SetPetrovGalerkinBnSParameters() + elif simulation_to_run=='GalerkinROM_ANN': + f['train_hrom']=False + f['run_hrom']=False + f['projection_strategy']="galerkin_ann" + f["rom_settings"]['rom_bns_settings'] = self._SetGalerkinBnSParameters() + elif simulation_to_run=='lspg_ANN': + f['train_hrom']=False + f['run_hrom']=False + f['projection_strategy']="lspg_ann" + f["rom_settings"]['rom_bns_settings'] = self._SetLSPGBnSParameters() + else: + raise Exception(f'Unknown flag "{simulation_to_run}" change for RomParameters.json') + rom_parameters.WriteRomParameters(parameters_file_folder, parameters_file_name, f) def _SetGalerkinBnSParameters(self): # Retrieve the default parameters as a JSON string and parse it into a dictionary diff --git a/applications/RomApplication/python_scripts/rom_parameters.py b/applications/RomApplication/python_scripts/rom_parameters.py new file mode 100644 index 000000000000..1bbfe5a7e010 --- /dev/null +++ b/applications/RomApplication/python_scripts/rom_parameters.py @@ -0,0 +1,243 @@ +"""Single owner of the RomParameters.json format. + +Two layouts exist: +- Version 1 (legacy): flat file mixing run-mode flags ('train_hrom', 'run_hrom', 'rom_manager'), solver settings, + inline data ('nodal_modes', 'elements_and_weights') and HROM training settings. Numpy data files are found by naming convention. +- Version 2: a manifest describing the trained ROM as typed blocks. It contains a 'deployment' block with the default + deployment options, a list of 'components' (one per solver), each with its 'projection' and 'manifold', an optional + 'coupling' block and a 'hyper_reduction' block. Numpy data files are referenced explicitly. + The 'manifold' type sets the scope: 'global' (a single 'decoder') or 'local' (a 'selector' and a list of 'clusters', + each with its own 'decoder'). The 'decoder' type sets how the reduced coordinates map to the full solution + ('linear', 'ann_enhanced', ...). Scope and decoder are independent, so every decoder type is usable in both scopes. + +Every read and write of RomParameters.json goes through this module. Files of either version can be read. +Files are still written as version 1 (see _OUTPUT_VERSION) until all the readers use the version 2 layout. +""" + +import copy +import json +from pathlib import Path + +ROM_PARAMETERS_VERSION = 2 +_OUTPUT_VERSION = 1 + +# Conventional names of the numpy files stored next to a version 1 RomParameters.json +_RIGHT_BASIS_FILE = "RightBasisMatrix.npy" +_NODE_IDS_FILE = "NodeIds.npy" +_LEFT_BASIS_FILE = "LeftBasisMatrix.npy" +_HROM_FILES = { + "element_ids" : "HROM_ElementIds.npy", + "element_weights" : "HROM_ElementWeights.npy", + "condition_ids" : "HROM_ConditionIds.npy", + "condition_weights" : "HROM_ConditionWeights.npy" +} + +# Values assumed by the readers when a version 1 key is missing +_V1_DEFAULTS = { + "rom_manager" : False, + "train_hrom" : False, + "run_hrom" : False, + "projection_strategy" : "galerkin", + "assembling_strategy" : "global", + "rom_format" : "json", + "rom_settings" : {"rom_bns_settings" : {}}, + "hrom_settings" : {}, + "nodal_modes" : {}, + "elements_and_weights" : {} +} + +_ANN_SUFFIX = "_ann" + + +def GetRomParametersFilePath(rom_basis_output_folder, rom_basis_output_name): + return Path(rom_basis_output_folder) / Path(rom_basis_output_name).with_suffix(".json") + + +def LoadRomParameters(rom_basis_output_folder, rom_basis_output_name): + """Returns the content of the RomParameters file in the version 2 layout, whatever the version stored.""" + with GetRomParametersFilePath(rom_basis_output_folder, rom_basis_output_name).open('r') as f: + data = json.load(f) + return data if IsVersion2(data) else UpgradeToVersion2(data) + + +def ReadRomParametersAsVersion1(rom_basis_output_folder, rom_basis_output_name): + """Returns the content of the RomParameters file in the version 1 layout, whatever the version stored. + Transitional: used by the readers that have not been moved to the version 2 layout yet.""" + with GetRomParametersFilePath(rom_basis_output_folder, rom_basis_output_name).open('r') as f: + data = json.load(f) + return DowngradeToVersion1(data) if IsVersion2(data) else data + + +def WriteRomParameters(rom_basis_output_folder, rom_basis_output_name, data): + """Writes the RomParameters file. 'data' can be given in either layout; it is stored in the current output version.""" + if _OUTPUT_VERSION == 1: + data = DowngradeToVersion1(data) if IsVersion2(data) else data + else: + data = data if IsVersion2(data) else UpgradeToVersion2(data) + + file_path = GetRomParametersFilePath(rom_basis_output_folder, rom_basis_output_name) + file_path.parent.mkdir(parents=True, exist_ok=True) + with file_path.open('w') as f: + json.dump(data, f, indent=4) + + +def IsVersion2(data): + return data.get("rom_parameters_version", 1) == 2 + + +def UpgradeToVersion2(v1_data): + """Converts a version 1 RomParameters dictionary into the version 2 layout. No information is lost.""" + v1 = copy.deepcopy(v1_data) + rom_settings = v1.pop("rom_settings", {}) + + # Projection: the ANN-enhanced strategies are the plain strategies with a nonlinear decoder + strategy = v1.pop("projection_strategy", _V1_DEFAULTS["projection_strategy"]) + decoder_type = "linear" + if strategy.endswith(_ANN_SUFFIX): + strategy = strategy[:-len(_ANN_SUFFIX)] + decoder_type = "ann_enhanced" + projection = { + "strategy" : strategy, + "assembling_strategy" : v1.pop("assembling_strategy", _V1_DEFAULTS["assembling_strategy"]), + "bns_settings" : rom_settings.pop("rom_bns_settings", {}) + } + + # Right basis + rom_format = v1.pop("rom_format", _V1_DEFAULTS["rom_format"]) + basis = {"format" : rom_format} + if "number_of_rom_dofs" in rom_settings: + basis["number_of_modes"] = rom_settings.pop("number_of_rom_dofs") + if rom_format == "numpy": + basis["file"] = _RIGHT_BASIS_FILE + basis["node_ids"] = _NODE_IDS_FILE + nodal_modes = v1.pop("nodal_modes", {}) + if nodal_modes or rom_format == "json": + basis["nodal_modes"] = nodal_modes + + # Left basis (Petrov-Galerkin) + left_basis = {} + if "petrov_galerkin_number_of_rom_dofs" in rom_settings: + left_basis["number_of_modes"] = rom_settings.pop("petrov_galerkin_number_of_rom_dofs") + if "petrov_galerkin_nodal_modes" in v1: + left_basis["nodal_modes"] = v1.pop("petrov_galerkin_nodal_modes") + if rom_format == "numpy" and left_basis.get("number_of_modes", 0) > 0: + left_basis["file"] = _LEFT_BASIS_FILE + if left_basis: + left_basis["format"] = rom_format + projection["left_basis"] = left_basis + + component = { + "name" : "", + "model_part_name" : "", + "nodal_unknowns" : rom_settings.pop("nodal_unknowns", []), + "projection" : projection, + "manifold" : {"type" : "global", "decoder" : {"type" : decoder_type, "basis" : basis}} + } + if rom_settings: + component["legacy_rom_settings"] = rom_settings + + # Hyper-reduction + hrom_settings = v1.pop("hrom_settings", {}) + hyper_reduction = {"type" : hrom_settings.get("element_selection_type", "empirical_cubature")} + hrom_format = hrom_settings.pop("hrom_format", None) + if hrom_format is not None: + hyper_reduction["format"] = hrom_format + if hrom_format == "numpy": + hyper_reduction.update(_HROM_FILES) + hyper_reduction["elements_and_weights"] = v1.pop("elements_and_weights", {}) + hyper_reduction["training_settings"] = hrom_settings + + v2 = { + "rom_parameters_version" : ROM_PARAMETERS_VERSION, + "deployment" : { + "use_hyper_reduction" : v1.pop("run_hrom", _V1_DEFAULTS["run_hrom"]), + "train_hyper_reduction" : v1.pop("train_hrom", _V1_DEFAULTS["train_hrom"]), # Deprecated: hyper-reduction training no longer happens inside the simulation + "rom_manager" : v1.pop("rom_manager", _V1_DEFAULTS["rom_manager"]) + }, + "components" : [component], + "hyper_reduction" : hyper_reduction + } + + # Coupled solvers sharing a single basis. The order of the solvers defines their column of HROM weights + if "coupled_solvers" in v1: + v2["coupling"] = {"basis_layout" : "monolithic", "solvers" : v1.pop("coupled_solvers")} + + if v1: + v2["legacy"] = v1 + return v2 + + +def DowngradeToVersion1(v2_data): + """Converts a version 2 RomParameters dictionary into the version 1 layout. + Only the version 2 content expressible in version 1 is supported (single component, global manifold, linear or ANN-enhanced decoder).""" + v2 = copy.deepcopy(v2_data) + components = v2["components"] + if len(components) != 1: + raise Exception(f"RomParameters with {len(components)} components cannot be written in the version 1 layout.") + component = components[0] + projection = component["projection"] + manifold = component["manifold"] + if manifold["type"] != "global": + raise Exception(f"'{manifold['type']}' manifold cannot be written in the version 1 layout, which only supports 'global' manifolds.") + decoder = manifold["decoder"] + if decoder["type"] not in ("linear", "ann_enhanced"): + raise Exception(f"'{decoder['type']}' decoder cannot be written in the version 1 layout, which only supports 'linear' and 'ann_enhanced' decoders.") + basis = decoder["basis"] + _CheckConventionalFileNames(basis, {"file" : _RIGHT_BASIS_FILE, "node_ids" : _NODE_IDS_FILE}) + + strategy = projection["strategy"] + if decoder["type"] == "ann_enhanced": + strategy += _ANN_SUFFIX + + rom_settings = {"rom_bns_settings" : projection.get("bns_settings", {})} + rom_settings["nodal_unknowns"] = component["nodal_unknowns"] + if "number_of_modes" in basis: + rom_settings["number_of_rom_dofs"] = basis["number_of_modes"] + left_basis = projection.get("left_basis", {}) + _CheckConventionalFileNames(left_basis, {"file" : _LEFT_BASIS_FILE}) + if "number_of_modes" in left_basis: + rom_settings["petrov_galerkin_number_of_rom_dofs"] = left_basis["number_of_modes"] + rom_settings.update(component.get("legacy_rom_settings", {})) + + hyper_reduction = v2.get("hyper_reduction", {}) + _CheckConventionalFileNames(hyper_reduction, _HROM_FILES) + hrom_settings = {} + if "format" in hyper_reduction: + hrom_settings["hrom_format"] = hyper_reduction["format"] + hrom_settings.update(hyper_reduction.get("training_settings", {})) + + deployment = v2.get("deployment", {}) + v1 = { + "rom_manager" : deployment.get("rom_manager", _V1_DEFAULTS["rom_manager"]), + "train_hrom" : deployment.get("train_hyper_reduction", _V1_DEFAULTS["train_hrom"]), + "run_hrom" : deployment.get("use_hyper_reduction", _V1_DEFAULTS["run_hrom"]), + "projection_strategy" : strategy, + "assembling_strategy" : projection.get("assembling_strategy", _V1_DEFAULTS["assembling_strategy"]), + "rom_format" : basis.get("format", _V1_DEFAULTS["rom_format"]), + "rom_settings" : rom_settings, + "hrom_settings" : hrom_settings, + "nodal_modes" : basis.get("nodal_modes", {}), + "elements_and_weights" : hyper_reduction.get("elements_and_weights", {}) + } + if "nodal_modes" in left_basis: + v1["petrov_galerkin_nodal_modes"] = left_basis["nodal_modes"] + if "coupling" in v2: + v1["coupled_solvers"] = v2["coupling"]["solvers"] + v1.update(v2.get("legacy", {})) + return v1 + + +def NormalizeVersion1(v1_data): + """Returns a copy of a version 1 dictionary with the defaults assumed by the readers filled in. + Two version 1 dictionaries describe the same ROM if their normalized forms are equal.""" + v1 = copy.deepcopy(v1_data) + for key, default in _V1_DEFAULTS.items(): + v1.setdefault(key, copy.deepcopy(default)) + v1["rom_settings"].setdefault("rom_bns_settings", {}) + return v1 + + +def _CheckConventionalFileNames(block, conventional_names): + for key, conventional_name in conventional_names.items(): + if key in block and block[key] != conventional_name: + raise Exception(f"'{key}': '{block[key]}' cannot be written in the version 1 layout, which requires '{conventional_name}'.") diff --git a/applications/RomApplication/tests/test_RomApplication.py b/applications/RomApplication/tests/test_RomApplication.py index d89827797302..9072de426a7d 100644 --- a/applications/RomApplication/tests/test_RomApplication.py +++ b/applications/RomApplication/tests/test_RomApplication.py @@ -28,6 +28,7 @@ from test_numpy_output_process import TestNumpyOutputProcess from test_coupled_fluid_thermal_rom import TestCoupledFluidThermalRom from test_projected_residuals_output_process import TestProjectedResidualsOutputProcess +from test_rom_parameters import TestRomParameters def AssembleTestSuites(): ''' Populates the test suites to run. @@ -71,6 +72,7 @@ def AssembleTestSuites(): smallSuite.addTests(KratosUnittest.TestLoader().loadTestsFromTestCases([TestNumpyOutputProcess])) smallSuite.addTests(KratosUnittest.TestLoader().loadTestsFromTestCases([TestCoupledFluidThermalRom])) smallSuite.addTests(KratosUnittest.TestLoader().loadTestsFromTestCases([TestProjectedResidualsOutputProcess])) + smallSuite.addTests(KratosUnittest.TestLoader().loadTestsFromTestCases([TestRomParameters])) diff --git a/applications/RomApplication/tests/test_rom_parameters.py b/applications/RomApplication/tests/test_rom_parameters.py new file mode 100644 index 000000000000..1ba7b0f078d2 --- /dev/null +++ b/applications/RomApplication/tests/test_rom_parameters.py @@ -0,0 +1,114 @@ +import copy +import json +from pathlib import Path + +import KratosMultiphysics +import KratosMultiphysics.KratosUnittest as KratosUnittest +import KratosMultiphysics.kratos_utilities as kratos_utilities +from KratosMultiphysics.RomApplication import rom_parameters + + +class TestRomParameters(KratosUnittest.TestCase): + + def setUp(self): + self.work_folder = Path("rom_parameters_test_output") + # Version 1 files currently committed with the tests + tests_folder = Path(__file__).parent + self.version_1_files = sorted(tests_folder.glob("**/rom_data/RomParameters*.json")) + with (tests_folder / "coupled_fluid_thermal_test_files/ROM/rom_data/RomParameters.json").open('r') as f: + self.coupled_version_1 = json.load(f) + + def tearDown(self): + with KratosUnittest.WorkFolderScope(".", __file__): + kratos_utilities.DeleteDirectoryIfExisting(str(self.work_folder)) + + def test_committed_files_round_trip(self): + self.assertGreater(len(self.version_1_files), 0) + for file_path in self.version_1_files: + with self.subTest(file=str(file_path)): + with file_path.open('r') as f: + version_1 = json.load(f) + version_2 = rom_parameters.UpgradeToVersion2(version_1) + self.assertTrue(rom_parameters.IsVersion2(version_2)) + self.assertEqual(len(version_2["components"]), 1) + self.assertEqual(rom_parameters.NormalizeVersion1(rom_parameters.DowngradeToVersion1(version_2)), rom_parameters.NormalizeVersion1(version_1)) + + def test_coupled_solvers(self): + # Content written by the RomManager for coupled solvers sharing a single basis + version_1 = copy.deepcopy(self.coupled_version_1) + version_1["coupled_solvers"] = ["fluid_solver", "thermal_solver"] + version_2 = rom_parameters.UpgradeToVersion2(version_1) + self.assertEqual(version_2["coupling"], {"basis_layout" : "monolithic", "solvers" : ["fluid_solver", "thermal_solver"]}) + self.assertNotIn("legacy", version_2) + self.assertEqual(rom_parameters.DowngradeToVersion1(version_2), version_1) + + def test_ann_enhanced(self): + for strategy in ["galerkin", "lspg"]: + with self.subTest(strategy=strategy): + version_1 = copy.deepcopy(self.coupled_version_1) + version_1["projection_strategy"] = f"{strategy}_ann" + version_2 = rom_parameters.UpgradeToVersion2(version_1) + component = version_2["components"][0] + self.assertEqual(component["projection"]["strategy"], strategy) + self.assertEqual(component["manifold"]["type"], "global") + self.assertEqual(component["manifold"]["decoder"]["type"], "ann_enhanced") + self.assertEqual(rom_parameters.DowngradeToVersion1(version_2), version_1) + + def test_numpy_files_are_referenced(self): + version_2 = rom_parameters.UpgradeToVersion2(self.coupled_version_1) + manifold = version_2["components"][0]["manifold"] + self.assertEqual(manifold["type"], "global") + self.assertEqual(manifold["decoder"]["type"], "linear") + basis = manifold["decoder"]["basis"] + self.assertEqual(basis["file"], "RightBasisMatrix.npy") + self.assertEqual(basis["node_ids"], "NodeIds.npy") + self.assertEqual(basis["number_of_modes"], self.coupled_version_1["rom_settings"]["number_of_rom_dofs"]) + self.assertEqual(version_2["hyper_reduction"]["element_weights"], "HROM_ElementWeights.npy") + + def test_read_and_write(self): + with KratosUnittest.WorkFolderScope(".", __file__): + # A version 2 file is read by the version 1 readers + version_2 = rom_parameters.UpgradeToVersion2(self.coupled_version_1) + self.work_folder.mkdir(parents=True, exist_ok=True) + with rom_parameters.GetRomParametersFilePath(self.work_folder, "RomParametersV2").open('w') as f: + json.dump(version_2, f) + self.assertEqual(rom_parameters.ReadRomParametersAsVersion1(self.work_folder, "RomParametersV2"), self.coupled_version_1) + self.assertEqual(rom_parameters.LoadRomParameters(self.work_folder, "RomParametersV2"), version_2) + + # Files are written in the version 1 layout for the moment + rom_parameters.WriteRomParameters(self.work_folder, "RomParametersV1", version_2) + with rom_parameters.GetRomParametersFilePath(self.work_folder, "RomParametersV1").open('r') as f: + self.assertEqual(json.load(f), self.coupled_version_1) + + def test_unsupported_downgrade(self): + version_2 = rom_parameters.UpgradeToVersion2(self.coupled_version_1) + version_2["components"].append(copy.deepcopy(version_2["components"][0])) + with self.assertRaisesRegex(Exception, "2 components"): + rom_parameters.DowngradeToVersion1(version_2) + + version_2 = rom_parameters.UpgradeToVersion2(self.coupled_version_1) + version_2["components"][0]["manifold"]["decoder"]["basis"]["file"] = "OtherBasis.npy" + with self.assertRaisesRegex(Exception, "RightBasisMatrix.npy"): + rom_parameters.DowngradeToVersion1(version_2) + + # Local manifolds only exist in the version 2 layout + version_2 = rom_parameters.UpgradeToVersion2(self.coupled_version_1) + decoder = version_2["components"][0]["manifold"]["decoder"] + version_2["components"][0]["manifold"] = { + "type" : "local", + "selector" : {"type" : "nearest_centroid", "centroids" : "centroids.npy"}, + "clusters" : [{"decoder" : decoder}, {"decoder" : decoder}] + } + with self.assertRaisesRegex(Exception, "'local' manifold"): + rom_parameters.DowngradeToVersion1(version_2) + + # So do the nonlinear decoders other than the ANN-enhanced one + version_2 = rom_parameters.UpgradeToVersion2(self.coupled_version_1) + version_2["components"][0]["manifold"]["decoder"]["type"] = "rbf" + with self.assertRaisesRegex(Exception, "'rbf' decoder"): + rom_parameters.DowngradeToVersion1(version_2) + + +if __name__ == "__main__": + KratosMultiphysics.Logger.GetDefaultOutput().SetSeverity(KratosMultiphysics.Logger.Severity.WARNING) + KratosUnittest.main()