[RomApp] RomParameters V2.0

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):

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.")
 

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")
 

applications/RomApplication/python_scripts/rom_analysis.py

@@ -1,11 +1,13 @@
 import importlib
+import json
 
 import KratosMultiphysics
 import KratosMultiphysics.RomApplication as KratosROM
 from KratosMultiphysics.RomApplication import python_solvers_wrapper_rom
 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"])

applications/RomApplication/python_scripts/rom_manager.py

@@ -1,14 +1,14 @@
 import numpy as np
 import importlib
 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.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