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..1a3af7fc2197 100644 --- a/applications/RomApplication/python_scripts/calculate_rom_basis_output_process.py +++ b/applications/RomApplication/python_scripts/calculate_rom_basis_output_process.py @@ -5,6 +5,7 @@ # Importing the Kratos Library import KratosMultiphysics +from KratosMultiphysics.kratos_utilities import IssueDeprecationWarning from KratosMultiphysics.RomApplication.randomized_singular_value_decomposition import RandomizedSingularValueDecomposition def Factory(settings, model): @@ -18,6 +19,8 @@ class CalculateRomBasisOutputProcess(KratosMultiphysics.OutputProcess): def __init__(self, model, settings): KratosMultiphysics.OutputProcess.__init__(self) + IssueDeprecationWarning("CalculateRomBasisOutputProcess", "This process is deprecated and will be removed. Please use the RomManager to compute the ROM basis.") + # Validate input settings against defaults settings.ValidateAndAssignDefaults(self.GetDefaultParameters()) diff --git a/applications/RomApplication/python_scripts/rom_analysis.py b/applications/RomApplication/python_scripts/rom_analysis.py index fdb0915b4f06..7d17c61d0920 100644 --- a/applications/RomApplication/python_scripts/rom_analysis.py +++ b/applications/RomApplication/python_scripts/rom_analysis.py @@ -12,7 +12,7 @@ from os import remove from pathlib import Path -def CreateRomAnalysisInstance(cls, global_model, parameters, nn_rom_interface=None): +def CreateRomAnalysisInstance(cls, global_model, parameters, nn_rom_interface=None, rom_basis_output_folder=None, rom_basis_output_name=None): class RomAnalysis(cls): def __init__(self,global_model, parameters): @@ -34,6 +34,11 @@ def _CreateSolver(self): self.rom_basis_output_name = parameter_set["Parameters"]["rom_basis_output_name"].GetString() if parameter_set["Parameters"].Has("rom_basis_output_folder"): self.rom_basis_output_folder = parameter_set["Parameters"]["rom_basis_output_folder"].GetString() + # Explicitly provided values (e.g. from the RomManager) take precedence + if rom_basis_output_name is not None: + self.rom_basis_output_name = rom_basis_output_name + if rom_basis_output_folder is not None: + self.rom_basis_output_folder = rom_basis_output_folder self.rom_basis_output_name = Path(self.rom_basis_output_name) self.rom_basis_output_folder = Path(self.rom_basis_output_folder) diff --git a/applications/RomApplication/python_scripts/rom_manager.py b/applications/RomApplication/python_scripts/rom_manager.py index b3f3655f0ecd..e3f3502fe179 100644 --- a/applications/RomApplication/python_scripts/rom_manager.py +++ b/applications/RomApplication/python_scripts/rom_manager.py @@ -1,12 +1,12 @@ import numpy as np import importlib +import shutil import json from pathlib import Path import types import KratosMultiphysics from KratosMultiphysics.RomApplication.rom_database import RomDatabase from KratosMultiphysics.RomApplication.rom_testing_utilities import SetUpSimulationInstance -from KratosMultiphysics.RomApplication.calculate_rom_basis_output_process import CalculateRomBasisOutputProcess from KratosMultiphysics.RomApplication.randomized_singular_value_decomposition import RandomizedSingularValueDecomposition from KratosMultiphysics.RomApplication.empirical_cubature_method import EmpiricalCubatureMethod from KratosMultiphysics.RomApplication.rom_nn_interface import NN_ROM_Interface @@ -17,7 +17,6 @@ class RomManager(object): def __init__(self,project_parameters_name="ProjectParameters.json", general_rom_manager_parameters=None, CustomizeSimulation=None, UpdateProjectParameters=None,UpdateMaterialParametersFile=None, mu_names=None): #FIXME: - # - Use a method (upcoming) for smothly retrieving solutions. In here we are using the RomBasisOutput process in order to store the solutions # - There is some redundancy between the methods that launch the simulations. Can we create a single method? # - Not yet paralellised with COMPSs self.project_parameters_name = project_parameters_name @@ -381,7 +380,7 @@ def _LaunchFOM(self, mu_train, gid_and_vtk_name='FOM_Fit'): nonconverged_fom_in_database, _ = self.data_base.check_if_in_database("NonconvergedFOM", mu) if not fom_in_database or (NonConvergedSolutionsGathering and not nonconverged_fom_in_database): parameters_copy = self.UpdateProjectParameters(parameters.Clone(), mu) - parameters_copy = self._AddBasisCreationToProjectParameters(parameters_copy) #TODO stop using the RomBasisOutputProcess to store the snapshots. Use instead the upcoming build-in function + parameters_copy = self._AddNumpyOutputToProjectParameters(parameters_copy) parameters_copy = self._StoreResultsByName(parameters_copy,gid_and_vtk_name,mu,Id) materials_file_name = self._GetMaterialsFileName(parameters_copy) self.UpdateMaterialParametersFile(materials_file_name, mu) @@ -393,10 +392,7 @@ def _LaunchFOM(self, mu_train, gid_and_vtk_name='FOM_Fit'): simulation.Run() if not fom_in_database: self.data_base.add_to_database("QoI_FOM", mu, simulation.GetFinalData()) - for process in simulation._GetListOfOutputProcesses(): - if isinstance(process, CalculateRomBasisOutputProcess): - BasisOutputProcess = process - SnapshotsMatrix = BasisOutputProcess._GetSnapshotsMatrix() #TODO add a CustomMethod() as a standard method in the Analysis Stage to retrive some solution + SnapshotsMatrix = self._GetSnapshotsMatrixFromNumpyOutput() self.data_base.add_to_database("FOM", mu, SnapshotsMatrix) if NonConvergedSolutionsGathering: self.data_base.add_to_database("NonconvergedFOM", mu, simulation.GetNonconvergedSolutions()) @@ -406,18 +402,16 @@ def _LaunchFOM(self, mu_train, gid_and_vtk_name='FOM_Fit'): def _LaunchComputeSolutionBasis(self, mu_train): in_database, hash_basis = self.data_base.check_if_in_database("RightBasis", mu_train) if not in_database: - BasisOutputProcess = self.InitializeDummySimulationForBasisOutputProcess() if self.general_rom_manager_parameters["ROM"]["use_non_converged_sols"].GetBool(): - u,sigma = BasisOutputProcess._ComputeSVD(self.data_base.get_snapshots_matrix_from_database(mu_train, table_name='NonconvergedFOM')) #TODO this might be too large for single opeartion, add partitioned svd + u,sigma = self._ComputeSVD(self.data_base.get_snapshots_matrix_from_database(mu_train, table_name='NonconvergedFOM')) #TODO this might be too large for single opeartion, add partitioned svd else: - u,sigma = BasisOutputProcess._ComputeSVD(self.data_base.get_snapshots_matrix_from_database(mu_train, table_name='FOM')) - BasisOutputProcess._PrintRomBasis(u, sigma) #Calling the RomOutput Process for creating the RomParameter.json + u,sigma = self._ComputeSVD(self.data_base.get_snapshots_matrix_from_database(mu_train, table_name='FOM')) + self._PrintRomBasis(u, sigma) self.data_base.add_to_database("RightBasis", mu_train, u ) self.data_base.add_to_database("SingularValues_Solution", mu_train, sigma ) else: - BasisOutputProcess = self.InitializeDummySimulationForBasisOutputProcess() _ , hash_sigma = self.data_base.check_if_in_database("SingularValues_Solution", mu_train) - BasisOutputProcess._PrintRomBasis(self.data_base.get_single_numpy_from_database(hash_basis), self.data_base.get_single_numpy_from_database(hash_sigma) ) #this updates the RomParameters.json + self._PrintRomBasis(self.data_base.get_single_numpy_from_database(hash_basis), self.data_base.get_single_numpy_from_database(hash_sigma) ) #this updates the RomParameters.json self.GenerateDatabaseSummary() def _LoadSolutionBasis(self, mu_train): @@ -433,9 +427,8 @@ def _LoadSolutionBasis(self, mu_train): err_msg = f'ROM basis not found for indicated training snapshots. Please run the Train method() to create it.' raise Exception(err_msg) else: - BasisOutputProcess = self.InitializeDummySimulationForBasisOutputProcess() _ , hash_sigma = self.data_base.check_if_in_database("SingularValues_Solution", mu_train) - BasisOutputProcess._PrintRomBasis(self.data_base.get_single_numpy_from_database(hash_basis), self.data_base.get_single_numpy_from_database(hash_sigma) ) #this updates the RomParameters.json + self._PrintRomBasis(self.data_base.get_single_numpy_from_database(hash_basis), self.data_base.get_single_numpy_from_database(hash_sigma) ) #this updates the RomParameters.json self.GenerateDatabaseSummary() @@ -446,25 +439,21 @@ def _LaunchROM(self, mu_train, gid_and_vtk_name='ROM_Fit', nn_rom_interface = No """ with open(self.project_parameters_name,'r') as parameter_file: parameters = KratosMultiphysics.Parameters(parameter_file.read()) - BasisOutputProcess = None for Id, mu in enumerate(mu_train): in_database, _ = self.data_base.check_if_in_database("ROM", mu) if not in_database: parameters_copy = self.UpdateProjectParameters(parameters.Clone(), mu) - parameters_copy = self._AddBasisCreationToProjectParameters(parameters_copy) #TODO stop using the RomBasisOutputProcess to store the snapshots. Use instead the upcoming build-in function + parameters_copy = self._AddNumpyOutputToProjectParameters(parameters_copy) parameters_copy = self._StoreResultsByName(parameters_copy,gid_and_vtk_name,mu,Id) materials_file_name = self._GetMaterialsFileName(parameters_copy) self.UpdateMaterialParametersFile(materials_file_name, mu) model = KratosMultiphysics.Model() - analysis_stage_class = type(SetUpSimulationInstance(model, parameters_copy, nn_rom_interface=nn_rom_interface)) + analysis_stage_class = type(self._SetUpRomSimulationInstance(model, parameters_copy, nn_rom_interface=nn_rom_interface)) simulation = self.CustomizeSimulation(analysis_stage_class,model,parameters_copy, mu) simulation.Run() self.data_base.add_to_database("QoI_ROM", mu, simulation.GetFinalData()) - for process in simulation._GetListOfOutputProcesses(): - if isinstance(process, CalculateRomBasisOutputProcess): - BasisOutputProcess = process - SnapshotsMatrix = BasisOutputProcess._GetSnapshotsMatrix() #TODO add a CustomMethod() as a standard method in the Analysis Stage to retrive some solution + SnapshotsMatrix = self._GetSnapshotsMatrixFromNumpyOutput() self.data_base.add_to_database("ROM", mu, SnapshotsMatrix ) self.GenerateDatabaseSummary() @@ -484,12 +473,11 @@ def _LaunchTrainPG(self, mu_train): in_database, _ = self.data_base.check_if_in_database("PetrovGalerkinSnapshots", mu) if not in_database: parameters_copy = self.UpdateProjectParameters(parameters.Clone(), mu) - parameters_copy = self._AddBasisCreationToProjectParameters(parameters_copy) parameters_copy = self._StoreNoResults(parameters_copy) materials_file_name = self._GetMaterialsFileName(parameters_copy) self.UpdateMaterialParametersFile(materials_file_name, mu) model = KratosMultiphysics.Model() - analysis_stage_class = type(SetUpSimulationInstance(model, parameters_copy)) + analysis_stage_class = type(self._SetUpRomSimulationInstance(model, parameters_copy)) simulation = self.CustomizeSimulation(analysis_stage_class,model,parameters_copy, mu) simulation.Run() PetrovGalerkinTrainingUtility = simulation.GetPetrovGalerkinTrainUtility() @@ -527,7 +515,6 @@ def _LaunchTrainHROM(self, mu_train, nn_rom_interface = None): in_database, _ = self.data_base.check_if_in_database("ResidualsProjected", mu) if not in_database: parameters_copy = self.UpdateProjectParameters(parameters.Clone(), mu) - parameters_copy = self._AddBasisCreationToProjectParameters(parameters_copy) #TODO Remove the basis creation process. Here it generates the RomParameters.json, find a workaround parameters_copy = self._AddResidualsProjectedOutputProcessToProjectParameters(parameters_copy) #This deals with the creation of residuals parameters_copy = self._StoreNoResults(parameters_copy) # Remove the residuals of previous runs, so that only the ones of this mu are collected @@ -537,7 +524,7 @@ def _LaunchTrainHROM(self, mu_train, nn_rom_interface = None): materials_file_name = self._GetMaterialsFileName(parameters_copy) self.UpdateMaterialParametersFile(materials_file_name, mu) model = KratosMultiphysics.Model() - analysis_stage_class = type(SetUpSimulationInstance(model, parameters_copy, nn_rom_interface)) + analysis_stage_class = type(self._SetUpRomSimulationInstance(model, parameters_copy, nn_rom_interface=nn_rom_interface)) simulation = self.CustomizeSimulation(analysis_stage_class,model,parameters_copy, mu) simulation.Run() ResidualProjected = self._GetResidualProjected() # this method fetches the residuals projected from the corresponding folder. @@ -623,24 +610,20 @@ def _LaunchHROM(self, mu_train, nn_rom_interface=None,gid_and_vtk_name ='HROM_Fi """ with open(self.project_parameters_name,'r') as parameter_file: parameters = KratosMultiphysics.Parameters(parameter_file.read()) - BasisOutputProcess = None for Id, mu in enumerate(mu_train): in_database, _ = self.data_base.check_if_in_database("HROM", mu) if not in_database: parameters_copy = self.UpdateProjectParameters(parameters.Clone(), mu) - parameters_copy = self._AddBasisCreationToProjectParameters(parameters_copy) + parameters_copy = self._AddNumpyOutputToProjectParameters(parameters_copy) parameters_copy = self._StoreResultsByName(parameters_copy,gid_and_vtk_name,mu,Id) materials_file_name = self._GetMaterialsFileName(parameters_copy) self.UpdateMaterialParametersFile(materials_file_name, mu) model = KratosMultiphysics.Model() - analysis_stage_class = type(SetUpSimulationInstance(model, parameters_copy,nn_rom_interface)) + analysis_stage_class = type(self._SetUpRomSimulationInstance(model, parameters_copy, nn_rom_interface=nn_rom_interface)) simulation = self.CustomizeSimulation(analysis_stage_class,model,parameters_copy, mu) simulation.Run() self.data_base.add_to_database("QoI_HROM", mu, simulation.GetFinalData()) - for process in simulation._GetListOfOutputProcesses(): - if isinstance(process, CalculateRomBasisOutputProcess): - BasisOutputProcess = process - SnapshotsMatrix = BasisOutputProcess._GetSnapshotsMatrix() #TODO add a CustomMethod() as a standard method in the Analysis Stage to retrive some solution + SnapshotsMatrix = self._GetSnapshotsMatrixFromNumpyOutput() self.data_base.add_to_database("HROM", mu, SnapshotsMatrix) self.GenerateDatabaseSummary() @@ -676,7 +659,7 @@ def _LaunchRunROM(self, mu_run, nn_rom_interface=None): materials_file_name = self._GetMaterialsFileName(parameters_copy) self.UpdateMaterialParametersFile(materials_file_name, mu) model = KratosMultiphysics.Model() - analysis_stage_class = type(SetUpSimulationInstance(model, parameters_copy, nn_rom_interface=nn_rom_interface)) + analysis_stage_class = type(self._SetUpRomSimulationInstance(model, parameters_copy, nn_rom_interface=nn_rom_interface)) simulation = self.CustomizeSimulation(analysis_stage_class,model,parameters_copy, mu) simulation.Run() self.QoI_Run_ROM.append(simulation.GetFinalData()) @@ -702,7 +685,7 @@ def _LaunchRunHROM(self, mu_run, use_full_model_part, nn_rom_interface): materials_file_name = self._GetMaterialsFileName(parameters_copy) self.UpdateMaterialParametersFile(materials_file_name, mu) model = KratosMultiphysics.Model() - analysis_stage_class = type(SetUpSimulationInstance(model, parameters_copy,nn_rom_interface)) + analysis_stage_class = type(self._SetUpRomSimulationInstance(model, parameters_copy, nn_rom_interface=nn_rom_interface)) simulation = self.CustomizeSimulation(analysis_stage_class,model,parameters_copy, mu) simulation.Run() self.QoI_Run_HROM.append(simulation.GetFinalData()) @@ -720,28 +703,23 @@ def _LaunchTestNeuralNetworkReconstruction(self,mu_train, mu_validation): rom_nn_trainer = RomNeuralNetworkTrainer(self.general_rom_manager_parameters, mu_train, mu_validation, self.data_base) rom_nn_trainer.EvaluateNetwork() - def InitializeDummySimulationForBasisOutputProcess(self): + def InitializeDummySimulationForSnapshotsModelPart(self): with open(self.project_parameters_name,'r') as parameter_file: parameters = KratosMultiphysics.Parameters(parameter_file.read()) - parameters = self._AddBasisCreationToProjectParameters(parameters) parameters = self._StoreNoResults(parameters) model = KratosMultiphysics.Model() analysis_stage_class = self._GetAnalysisStageClass(parameters) simulation = self.CustomizeSimulation(analysis_stage_class,model,parameters) simulation.Initialize() - for process in simulation._GetListOfOutputProcesses(): - if isinstance(process, CalculateRomBasisOutputProcess): - BasisOutputProcess = process - return BasisOutputProcess + return model[self.general_rom_manager_parameters["ROM"]["model_part_name"].GetString()] def InitializeDummySimulationForHromTrainingUtility(self, nn_rom_interface=None): with open(self.project_parameters_name,'r') as parameter_file: parameters = KratosMultiphysics.Parameters(parameter_file.read()) - parameters = self._AddBasisCreationToProjectParameters(parameters) parameters = self._StoreNoResults(parameters) model = KratosMultiphysics.Model() - analysis_stage_class = type(SetUpSimulationInstance(model, parameters,nn_rom_interface=nn_rom_interface)) + analysis_stage_class = type(self._SetUpRomSimulationInstance(model, parameters, nn_rom_interface=nn_rom_interface)) simulation = self.CustomizeSimulation(analysis_stage_class,model,parameters) simulation.Initialize() return simulation.GetHROM_utility() @@ -750,10 +728,9 @@ def InitializeDummySimulationForHromTrainingUtility(self, nn_rom_interface=None) def InitializeDummySimulationForPetrovGalerkinTrainingUtility(self): with open(self.project_parameters_name,'r') as parameter_file: parameters = KratosMultiphysics.Parameters(parameter_file.read()) - parameters = self._AddBasisCreationToProjectParameters(parameters) parameters = self._StoreNoResults(parameters) model = KratosMultiphysics.Model() - analysis_stage_class = type(SetUpSimulationInstance(model, parameters)) + analysis_stage_class = type(self._SetUpRomSimulationInstance(model, parameters)) simulation = self.CustomizeSimulation(analysis_stage_class,model,parameters) simulation.Initialize() return simulation.GetPetrovGalerkinTrainUtility() @@ -948,6 +925,17 @@ def _UpdateDefaultsWithRomParams(self, defaults, rom_params): defaults[key] = rom_params[key].GetDouble() return defaults + def _SetUpRomSimulationInstance(self, model, parameters, nn_rom_interface=None): + rom_params = self.general_rom_manager_parameters["ROM"] + return SetUpSimulationInstance(model, parameters, nn_rom_interface=nn_rom_interface, + rom_basis_output_folder=rom_params["rom_basis_output_folder"].GetString(), + rom_basis_output_name=rom_params["rom_basis_output_name"].GetString()) + + + def _GetSnapshotsOutputPath(self): + return Path(self.general_rom_manager_parameters["ROM"]["rom_basis_output_folder"].GetString()) / "numpy_snapshots" + + def _GetMaterialsFileName(self, parameters): """Returns the materials file name, or None if it is not defined at the solver_settings level (e.g. coupled solvers).""" solver_settings = parameters["solver_settings"] @@ -1050,16 +1038,108 @@ def _SetTrainHROMFlag(self, train_hrom): json.dump(f, parameter_file, indent=4) parameter_file.truncate() - #TODO get rid of this process in favor or Numpy output process - def _AddBasisCreationToProjectParameters(self, parameters): - #FIXME make sure no other rom_output already existed. If so, erase the prior and keep only the one in self.general_rom_manager_parameters["ROM"] - parameters["output_processes"].AddEmptyArray("rom_output") - rom_basis_parameters = self._SetUpRomBasisParameters() - parameters["output_processes"]["rom_output"].Append(rom_basis_parameters) + def _AddNumpyOutputToProjectParameters(self, parameters): + rom_params = self.general_rom_manager_parameters["ROM"] + numpy_output_parameters = KratosMultiphysics.Parameters("""{ + "python_module" : "numpy_output_process", + "kratos_module" : "KratosMultiphysics.RomApplication", + "process_name" : "NumpyOutputProcess", + "Parameters" : {} + }""") + numpy_output_parameters["Parameters"].AddString("model_part_name", rom_params["model_part_name"].GetString()) + numpy_output_parameters["Parameters"].AddString("output_control_type", rom_params["snapshots_control_type"].GetString()) + numpy_output_parameters["Parameters"].AddDouble("output_interval", rom_params["snapshots_interval"].GetDouble()) + numpy_output_parameters["Parameters"].AddValue("nodal_results", rom_params["nodal_unknowns"]) + numpy_output_parameters["Parameters"].AddString("output_path", str(self._GetSnapshotsOutputPath())) + parameters["output_processes"].AddEmptyArray("rom_manager_numpy_output") + parameters["output_processes"]["rom_manager_numpy_output"].Append(numpy_output_parameters) + + # Remove leftovers from previous runs so that only the upcoming simulation snapshots are fetched + shutil.rmtree(self._GetSnapshotsOutputPath(), ignore_errors=True) return parameters + def _GetSnapshotsMatrixFromNumpyOutput(self): + snapshots_output_path = self._GetSnapshotsOutputPath() + # Files are named 'solution_' or 'solution_