[OptApp] Refactor the Optimization Problem (#12589)

* refactoring and adding Master COntrol and Reponse Routing to Opt Problem

* adding the has/get/remove

* remove AddProcessType

* unneeded

* hinting in GetComponents

---------

Co-authored-by: IAntonau <[email protected]>
Co-authored-by: IAntonau <[email protected]>

applications/OptimizationApplication/python_scripts/algorithms/NLOPT_algorithms.py

@@ -59,16 +59,20 @@ def __init__(self, model:Kratos.Model, parameters: Kratos.Parameters, optimizati
 
         # controls
         self.master_control = MasterControl()
+        self._optimization_problem.AddComponent(self.master_control)
         for control_name in parameters["controls"].GetStringArray():
             control = optimization_problem.GetControl(control_name)
             self.master_control.AddControl(control)
         self.__control_field = None
 
         # objective & constraints
         self.__objective = StandardizedNLOPTObjective(parameters["objective"], self.master_control, self._optimization_problem)
+        self._optimization_problem.AddComponent(self.__objective)
         self.__constraints = []
         for constraint_settings in parameters["constraints"]:
-            self.__constraints.append(StandardizedNLOPTConstraint(constraint_settings, self.master_control, self._optimization_problem))
+            constraint = StandardizedNLOPTConstraint(constraint_settings, self.master_control, self._optimization_problem)
+            self._optimization_problem.AddComponent(constraint)
+            self.__constraints.append(constraint)
 
         # nlopt settings
         NLOPT_settings = parameters["NLOPT_settings"]

applications/OptimizationApplication/python_scripts/algorithms/algorithm_gradient_projection.py

@@ -14,6 +14,7 @@
 from KratosMultiphysics.OptimizationApplication.utilities.logger_utilities import OptimizationAlgorithmTimeLogger
 from KratosMultiphysics.OptimizationApplication.utilities.list_collective_expression_utilities import CollectiveListCollectiveProduct
 from KratosMultiphysics.OptimizationApplication.utilities.list_collective_expression_utilities import CollectiveListVectorProduct
+from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem_utilities import OutputGradientFields
 
 def Factory(model: Kratos.Model, parameters: Kratos.Parameters, optimization_problem: OptimizationProblem):
     return AlgorithmGradientProjection(model, parameters, optimization_problem)
@@ -48,6 +49,7 @@ def __init__(self, model:Kratos.Model, parameters: Kratos.Parameters, optimizati
         parameters.ValidateAndAssignDefaults(self.GetDefaultParameters())
 
         self.master_control = MasterControl() # Need to fill it with controls
+        self._optimization_problem.AddComponent(self.master_control)
 
         for control_name in parameters["controls"].GetStringArray():
             control = optimization_problem.GetControl(control_name)
@@ -65,9 +67,11 @@ def __init__(self, model:Kratos.Model, parameters: Kratos.Parameters, optimizati
         self.__line_search_method = CreateLineSearch(settings["line_search"], self._optimization_problem)
 
         self.__objective = StandardizedObjective(parameters["objective"], self.master_control, self._optimization_problem)
+        self._optimization_problem.AddComponent(self.__objective)
         self.__constraints_list: 'list[StandardizedConstraint]' = []
         for constraint_param in parameters["constraints"].values():
             constraint = StandardizedConstraint(constraint_param, self.master_control, self._optimization_problem)
+            self._optimization_problem.AddComponent(constraint)
             self.__constraints_list.append(constraint)
         self.__control_field = None
         self.__obj_val = None
@@ -163,9 +167,9 @@ def GetCurrentControlField(self):
     @time_decorator()
     def Output(self) -> KratosOA.CollectiveExpression:
         self.algorithm_data.GetBufferedData()["control_field"] = self.__control_field.Clone()
-        self.__objective.OutputGradientFields(self._optimization_problem, True)
+        OutputGradientFields(self.__objective, self._optimization_problem, True)
         for constraint in self.__constraints_list:
-            constraint.OutputGradientFields(self._optimization_problem, constraint.GetStandardizedValue() > 0.0)
+            OutputGradientFields(constraint, self._optimization_problem, True)
         for process in self._optimization_problem.GetListOfProcesses("output_processes"):
             if process.IsOutputStep():
                 process.PrintOutput()

applications/OptimizationApplication/python_scripts/algorithms/algorithm_relaxed_gradient_projection.py

@@ -14,6 +14,7 @@
 from KratosMultiphysics.OptimizationApplication.utilities.logger_utilities import OptimizationAlgorithmTimeLogger
 from KratosMultiphysics.OptimizationApplication.utilities.list_collective_expression_utilities import CollectiveListCollectiveProduct
 from KratosMultiphysics.OptimizationApplication.utilities.list_collective_expression_utilities import CollectiveListVectorProduct
+from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem_utilities import OutputGradientFields
 
 import math
 
@@ -54,6 +55,7 @@ def __init__(self, model:Kratos.Model, parameters: Kratos.Parameters, optimizati
         parameters.ValidateAndAssignDefaults(self.GetDefaultParameters())
 
         self.master_control = MasterControl() # Need to fill it with controls
+        self._optimization_problem.AddComponent(self.master_control)
 
         for control_name in parameters["controls"].GetStringArray():
             control = optimization_problem.GetControl(control_name)
@@ -73,9 +75,11 @@ def __init__(self, model:Kratos.Model, parameters: Kratos.Parameters, optimizati
         self.__line_search_method = CreateLineSearch(settings["line_search"], self._optimization_problem)
 
         self.__objective = StandardizedObjective(parameters["objective"], self.master_control, self._optimization_problem)
+        self._optimization_problem.AddComponent(self.__objective)
         self.__constraints_list: 'list[StandardizedRGPConstraint]' = []
         for constraint_param in parameters["constraints"].values():
             constraint = StandardizedRGPConstraint(constraint_param, self.master_control, self._optimization_problem, self.GetMinimumBufferSize())
+            self._optimization_problem.AddComponent(constraint)
             self.__constraints_list.append(constraint)
         self.__control_field = None
         self.__obj_val = None
@@ -208,9 +212,9 @@ def GetCurrentControlField(self):
     @time_decorator()
     def Output(self) -> KratosOA.CollectiveExpression:
         self.algorithm_data.GetBufferedData()["control_field"] = self.__control_field.Clone()
-        self.__objective.OutputGradientFields(self._optimization_problem, True)
+        OutputGradientFields(self.__objective, self._optimization_problem, True)
         for constraint in self.__constraints_list:
-            constraint.OutputGradientFields(self._optimization_problem, constraint.GetStandardizedValue() > 0.0)
+            OutputGradientFields(constraint, self._optimization_problem, True)
         for process in self._optimization_problem.GetListOfProcesses("output_processes"):
             if process.IsOutputStep():
                 process.PrintOutput()

applications/OptimizationApplication/python_scripts/algorithms/algorithm_steepest_descent.py

@@ -42,6 +42,7 @@ def __init__(self, model:Kratos.Model, parameters: Kratos.Parameters, optimizati
         parameters.ValidateAndAssignDefaults(self.GetDefaultParameters())
 
         self.master_control = MasterControl() # Need to fill it with controls
+        self._optimization_problem.AddComponent(self.master_control)
 
         for control_name in parameters["controls"].GetStringArray():
             control = optimization_problem.GetControl(control_name)
@@ -59,6 +60,7 @@ def __init__(self, model:Kratos.Model, parameters: Kratos.Parameters, optimizati
         self.__line_search_method = CreateLineSearch(settings["line_search"], self._optimization_problem)
 
         self.__objective = StandardizedObjective(parameters["objective"], self.master_control, self._optimization_problem)
+        self._optimization_problem.AddComponent(self.__objective)
         self.__control_field = None
         self.__obj_val = None
 

applications/OptimizationApplication/python_scripts/controls/master_control.py

@@ -192,3 +192,6 @@ def Initialize(self) -> None:
 
     def Finalize(self) -> None:
         CallOnAll(self.__list_of_controls, Control.Finalize)
+
+    def GetName(self) -> str:
+        return "master_control"

applications/OptimizationApplication/python_scripts/execution_policies/execution_policy_decorator.py

@@ -3,7 +3,7 @@
 from KratosMultiphysics.OptimizationApplication.execution_policies.execution_policy import ExecutionPolicy
 from KratosMultiphysics.OptimizationApplication.utilities.logger_utilities import FileLogger
 from KratosMultiphysics.OptimizationApplication.utilities.logger_utilities import TimeLogger
-from KratosMultiphysics.OptimizationApplication.utilities.helper_utilities import OptimizationComponentFactory
+from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem_utilities import OptimizationComponentFactory
 from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem import OptimizationProblem
 from KratosMultiphysics.OptimizationApplication.utilities.helper_utilities import CallOnAll
 

applications/OptimizationApplication/python_scripts/optimization_analysis.py

@@ -4,8 +4,7 @@
 from KratosMultiphysics.OptimizationApplication.model_part_controllers.model_part_controller import ModelPartController
 from KratosMultiphysics.OptimizationApplication.execution_policies.execution_policy_decorator import ExecutionPolicyDecorator
 from KratosMultiphysics.OptimizationApplication.responses.response_function import ResponseFunction
-from KratosMultiphysics.OptimizationApplication.controls.control import Control
-from KratosMultiphysics.OptimizationApplication.utilities.helper_utilities import OptimizationComponentFactory
+from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem_utilities import OptimizationComponentFactory
 from KratosMultiphysics.OptimizationApplication.utilities.helper_utilities import CallOnAll
 from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem import OptimizationProblem
 from KratosMultiphysics.OptimizationApplication.utilities.logger_utilities import OptimizationAnalysisTimeLogger

applications/OptimizationApplication/python_scripts/processes/optimization_problem_ascii_output_process.py

@@ -7,9 +7,8 @@
 from KratosMultiphysics.OptimizationApplication.execution_policies.execution_policy import ExecutionPolicy
 from KratosMultiphysics.OptimizationApplication.utilities.component_data_view import ComponentDataView
 from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem import OptimizationProblem
-from KratosMultiphysics.OptimizationApplication.utilities.buffered_dict import BufferedDict
-from KratosMultiphysics.OptimizationApplication.utilities.helper_utilities import GetAllComponentFullNamesWithData
-from KratosMultiphysics.OptimizationApplication.utilities.helper_utilities import GetComponentHavingDataByFullName
+from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem_utilities import GetAllComponentFullNamesWithData
+from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem_utilities import GetComponentHavingDataByFullName
 
 def Factory(_: Kratos.Model, parameters: Kratos.Parameters, optimization_problem: OptimizationProblem) -> ExecutionPolicy:
     if not parameters.Has("settings"):
@@ -96,23 +95,23 @@ def __init__(self, parameters: Kratos.Parameters, optimization_problem: Optimiza
             raise RuntimeError("The \"bool_values\" should have only two strings corresponding to False and True values in the mentioned order.")
 
         self.list_of_components: 'list[Union[str, ResponseFunction, Control, ExecutionPolicy]]' = []
+        list_of_component_names = parameters["list_of_output_components"].GetStringArray()
+        if len(list_of_component_names) == 1 and list_of_component_names[0] == "all":
+            list_of_component_names = GetAllComponentFullNamesWithData(optimization_problem)
+
+        for component_name in list_of_component_names:
+            self.list_of_components.append(GetComponentHavingDataByFullName(component_name, optimization_problem))
+
         self.list_of_headers: 'list[tuple[Any, dict[str, Header]]]' = []
         self.initialized_headers = False
-        self.list_of_component_names = parameters["list_of_output_components"].GetStringArray()
 
     def IsOutputStep(self) -> bool:
         return True
 
     def PrintOutput(self) -> None:
         if not self.initialized_headers:
-            if len(self.list_of_component_names) == 1 and self.list_of_component_names[0] == "all":
-                self.list_of_component_names = GetAllComponentFullNamesWithData(self.optimization_problem)
-
-            for component_name in self.list_of_component_names:
-                self.list_of_components.append(GetComponentHavingDataByFullName(component_name, self.optimization_problem))
-
             # now get the buffered data headers
-            self.list_of_headers = self._GetHeaders(lambda x: x.GetBufferedData() if x.HasDataBuffer() else BufferedDict())
+            self.list_of_headers = self._GetHeaders(lambda x: x.GetBufferedData())
             # write the ehader information
             self._WriteHeaders()
             self.initialized_headers = True
@@ -169,12 +168,10 @@ def _WriteHeaders(self):
                     componend_data_view = ComponentDataView(component, self.optimization_problem)
                     buffered_dict = componend_data_view.GetUnBufferedData()
                     component_name = componend_data_view.GetComponentName()
-                    # check if there are values to be written under the component name, if not skip the component.
-                    if len(header_info_dict):
-                        msg_header = f"{msg_header}# \t" + component_name + ":\n"
-                        for k, header in header_info_dict.items():
-                            component_name_header = header.GetHeaderName().strip()[len(component_name)+1:]
-                            msg_header = f"{msg_header}# \t\t" + component_name_header + ": " + header.GetValueStr(buffered_dict[k]).strip() + "\n"
+                    msg_header = f"{msg_header}# \t" + component_name + ":\n"
+                    for k, header in header_info_dict.items():
+                        component_name_header = header.GetHeaderName().strip()[len(component_name)+1:]
+                        msg_header = f"{msg_header}# \t\t" + component_name_header + ": " + header.GetValueStr(buffered_dict[k]).strip() + "\n"
 
                 msg_header = f"{msg_header}# ------------ End of initial values ------------\n"
                 msg_header = f"{msg_header}# -----------------------------------------------\n"
@@ -207,6 +204,5 @@ def _GetHeaders(self, dict_getter_method) ->  'list[tuple[Any, dict[str, Header]
                     if header_name in [header.GetHeaderName().strip() for header in header_info_dict.values()]:
                         Kratos.Logger.PrintWarning(self.__class__.__name__, "Second value with same header name = \"" + header_name + "\" found.")
                     header_info_dict[k] = Header(header_name, v, self.format_info)
-            if len(header_info_dict):
-                list_of_headers.append([component, header_info_dict])
+            list_of_headers.append([component, header_info_dict])
         return list_of_headers

applications/OptimizationApplication/python_scripts/processes/optimization_problem_graph_output_process.py

@@ -6,7 +6,7 @@
 import KratosMultiphysics as Kratos
 from KratosMultiphysics.OptimizationApplication.utilities.component_data_view import ComponentDataView
 from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem import OptimizationProblem
-from KratosMultiphysics.OptimizationApplication.utilities.helper_utilities import GetComponentHavingDataByFullName
+from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem_utilities import GetComponentHavingDataByFullName
 
 def Factory(_: Kratos.Model, parameters: Kratos.Parameters, optimization_problem: OptimizationProblem) -> Kratos.OutputProcess:
     if not parameters.Has("settings"):

applications/OptimizationApplication/python_scripts/processes/optimization_problem_vtu_output_process.py

@@ -10,8 +10,8 @@
 from KratosMultiphysics.OptimizationApplication.execution_policies.execution_policy import ExecutionPolicy
 from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem import OptimizationProblem
 from KratosMultiphysics.OptimizationApplication.utilities.component_data_view import ComponentDataView
-from KratosMultiphysics.OptimizationApplication.utilities.helper_utilities import GetAllComponentFullNamesWithData
-from KratosMultiphysics.OptimizationApplication.utilities.helper_utilities import GetComponentHavingDataByFullName
+from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem_utilities import GetAllComponentFullNamesWithData
+from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem_utilities import GetComponentHavingDataByFullName
 from KratosMultiphysics.OptimizationApplication.utilities.union_utilities import ContainerExpressionTypes
 
 def Factory(_: Kratos.Model, parameters: Kratos.Parameters, optimization_problem: OptimizationProblem) -> Any:

applications/OptimizationApplication/python_scripts/responses/response_routine.py

@@ -4,8 +4,6 @@
 from KratosMultiphysics.OptimizationApplication.responses.response_function import ResponseFunction
 from KratosMultiphysics.OptimizationApplication.controls.master_control import MasterControl
 from KratosMultiphysics.OptimizationApplication.utilities.union_utilities import SupportedSensitivityFieldVariableTypes
-from KratosMultiphysics.OptimizationApplication.utilities.optimization_problem import OptimizationProblem
-from KratosMultiphysics.OptimizationApplication.utilities.component_data_view import ComponentDataView
 
 class ResponseRoutine:
     """A class which adds optimization-specific utilities to simplify routines
@@ -128,6 +126,9 @@ def CalculateGradient(self) -> KratosOA.CollectiveExpression:
         # calculate and return the control space gradients from respective controls
         self.__mapped_gradients = self.__master_control.MapGradient(self.__required_physical_gradients)
         return self.__mapped_gradients
+
+    def GetMappedGradients(self):
+        return self.__mapped_gradients
 
     def GetRequiredPhysicalGradients(self) -> 'dict[SupportedSensitivityFieldVariableTypes, KratosOA.CollectiveExpression]':
         """Returns required physical gradients by this response
@@ -139,29 +140,8 @@ def GetRequiredPhysicalGradients(self) -> 'dict[SupportedSensitivityFieldVariabl
             dict[SupportedSensitivityFieldVariableTypes, KratosOA.CollectiveExpression]: Required physical gradients.
         """
         return self.__required_physical_gradients
+
+    def GetName(self):
+        return self.GetReponse().GetName()
 
-    def OutputGradientFields(self, optimization_problem: OptimizationProblem, is_gradients_computed: bool) -> None:
-        unbuffered_data = ComponentDataView(self.GetReponse(), optimization_problem).GetUnBufferedData()
-        if is_gradients_computed:
-            # save the physical gradients for post processing in unbuffered data container.
-            for physical_var, physical_gradient in self.GetRequiredPhysicalGradients().items():
-                variable_name = f"d{self.GetResponseName()}_d{physical_var.Name()}"
-                for physical_gradient_expression in physical_gradient.GetContainerExpressions():
-                    unbuffered_data.SetValue(variable_name, physical_gradient_expression.Clone(), overwrite=True)
-
-            # save the filtered gradients for post processing in unbuffered data container.
-            for gradient_container_expression, control in zip(self.__mapped_gradients.GetContainerExpressions(), self.GetMasterControl().GetListOfControls()):
-                variable_name = f"d{self.GetResponseName()}_d{control.GetName()}"
-                unbuffered_data.SetValue(variable_name, gradient_container_expression.Clone(), overwrite=True)
-        else:
-            # save the physical gradients for post processing in unbuffered data container.
-            for physical_var, physical_gradient in self.GetRequiredPhysicalGradients().items():
-                variable_name = f"d{self.GetResponseName()}_d{physical_var.Name()}"
-                for physical_gradient_expression in physical_gradient.GetContainerExpressions():
-                    unbuffered_data.SetValue(variable_name, physical_gradient_expression.Clone() * 0.0, overwrite=True)
-
-            # save the filtered gradients for post processing in unbuffered data container.
-            for control in self.GetMasterControl().GetListOfControls():
-                variable_name = f"d{self.GetResponseName()}_d{control.GetName()}"
-                unbuffered_data.SetValue(variable_name, control.GetEmptyField(), overwrite=True)