Flume Overview
The purpose of this file is to outline in detail the use of Flume and a workflow for the framework. For a new user, this file will broadly outline the following:
Nomenclature utilized for the framework
Construction of an Analysis class, which internally uses State objects to define variables and outputs
Defining instances of Analysis classes and the role of
sub_analysesto connect outputs to inputsAssembly of a System with specific instances of Analysis classes To address these points, this notebook will provide sections of code with accompanying descriptions regarding the use of the various base classes and how to set up an optimization problem.
Framework Nomenclature
Before including any code, it is important to clarify the nomenclature that is used within the framework. All Analysis classes contain at least one variable and output, and they may optionally include parameters. In the context of Flume, these are defined as follows.
Parameter: Numerical, boolean, string, or other data type that are nominally set during the construction of an Analysis object. Parameters represent one type of inputs to an analysis procedure, but they are not candidates for design variables, as they are set during object construction and are (generally) not changed after this point. Parameters are stored within a dictionary, an attribute of the Analysis class entitled
parameters, where the keys correspond to the parameters’ names and the values take on the actual values of the parameters.Variable: Numerical quantities (either floats or NumPy arrays) that are candidates for design variables during optimization. This is the second type of inputs to an analysis procedure and are numerical quantities used to compute outputs of an Analysis class. Variables are stored within a dictionary, an attribute of the Analysis class entitled
variables, where the keys correspond to the local variable name and the values are instances of State objects. By defining each variable as a State, it contains information beyond its numerical value, including the data type, shape, description, derivative, and source. If sub-analyses exist for an Analysis class, then variables may be sourced from another object if its local variable name is the same as an output of a connected sub-analysis (see insert section for additional details).Output: Outputs are similar in nature to variables, but they are the quantities determined from performing the analysis procedure with a given set of parameters and variables. Output values are also wrapped within State objects, so information about an output is accessed in the same way as variables by accessing the
outputsattribute of an Analysis class. As mentioned above, outputs of a sub-analysis can be connected to variables for another Analysis class if they share the same local names.
These three terms are used throughout the framework but will be further explained in context below.
Defining an Analysis Class
Now, consider the code below, which defines the Rosenbrock Analysis, which is responsible for evaluating
where \(a\) and \(b\) are parameter values and \(x\) and \(y\) are current variable values. The entire code is shown first, and then each method is explained in greater detail.
from flume.base_classes.analysis import Analysis
from flume.base_classes.state import State
class Rosenbrock(Analysis):
def __init__(self, obj_name: str, sub_analyses=[], **kwargs):
# Set the default parameters
self.default_parameters = {"a": 1.0, "b": 100.0}
# Perform the base class object initialization
super().__init__(obj_name=obj_name, sub_analyses=sub_analyses, **kwargs)
# Set the default State for the variables
xvar = State(value=0.0, desc="x state value", source=self)
yvar = State(value=0.0, desc="y state value", source=self)
# Construct variables dictionary
self.variables = {"x": xvar, "y": yvar}
return
def _analyze(self):
# Extract the variable values
x = self.variables["x"].value
y = self.variables["y"].value
# Extract the parameter values
a = self.parameters["a"]
b = self.parameters["b"]
# Compute the value of the Rosenbrock function
f = (a - x) ** 2 + b * (y - x**2) ** 2
# Update the analyzed attribute
self.analyzed = True
# Store the outputs
self.outputs = {}
self.outputs["f"] = State(
value=f, desc="Rosenbrock function value", source=self
)
return
def _analyze_adjoint(self):
# Extract the derivatives of the outputs
fb = self.outputs["f"].deriv
# Extract the variable values
x = self.variables["x"].value
y = self.variables["y"].value
# Extract the variable derivatives
xb = self.variables["x"].deriv
yb = self.variables["y"].deriv
# Extract the parameter values
a = self.parameters["a"]
b = self.parameters["b"]
# Compute xb
xb += (2 * (a - x) * -1 + 2.0 * b * (y - x**2) * -2 * x) * fb
# Compute yb
yb += (2 * b * (y - x**2)) * fb
# Update the analyzed adjoint attribute
self.adjoint_analyzed = True
# Assign the derivative values
self.variables["x"].set_deriv_value(deriv_val=xb)
self.variables["y"].set_deriv_value(deriv_val=yb)
return
__init__ Method
The __init__ method is responsible for constructing the object, similar to a regular Python class. Regardless of the purpose of the Analysis class, this method must provide an object name obj_name, a list of sub-analyses sub_analyses, and any keyword arguments **kwargs. When constructing a specific instance of this class, the sub_analyses list may be empty or contain other Analysis objects, but the definition of the __init__ method does not have to contain this
information.
Next, the default parameter values are defined for the current class with the following line
self.default_parameters = {"a": 1.0, "b": 100.0}
Default parameters must be provided here, as they are one of the required inputs to compute the outputs. If the user wants to override the default parameter values, they are passed in as keyword arguments during object construction. In the event that an Analysis class does not have any parameters, the default_parameters attribute should be defined as an empty dictionary to denote this. The base class object initialization is then performed by calling
super().__init__(obj_name=obj_name, sub_analyses=sub_analyses, **kwargs)
This performs various backend assignment procedures and is necessary to properly interface with the framework. Similar to the parameters, default variable States are defined with
xvar = State(value=0.0, desc="x state value", source=self)
yvar = State(value=0.0, desc="y state value", source=self)
Here, the variables for \(x\) and \(y\) define the default value, description, and source information. When defining the source information, it is always self. Even if the variable is intended to be sourced from another object, this information is updated on the backend if a connection is identified between an outputs and variables. Finally, the variable States are assigned into the variables dictionary
self.variables = {"x": xvar, "y": yvar}
This concludes the required parts of the constructor method, but additional modifications are possible based on a user’s individual needs for their Analysis classes.
_analyze Method
The primary intent of this method is to compute the outputs of interest using the input values for the parameters and variables. To do so, the variable values and parameters are extracted with
# Extract the variable values
x = self.variables["x"].value
y = self.variables["y"].value
# Extract the parameter values
a = self.parameters["a"]
b = self.parameters["b"]
Here, it should be noted that variables must include the additional .value to access the value contained within the associated State object. Parameters do not require this because they are not wrapped with the State class. The next section of this method is dedicated to evaluating discipline specific computations that determine the outputs. For the Rosenbrock function, this is done by simply evaluating
f = (a - x) ** 2 + b * (y - x**2) ** 2
After the outputs are computed, the user should update the analyzed attribute to denote that computations have concluded for the current Analysis.
self.analyzed = True
This is used internally to avoid recomputing data if an Analysis appears multiple times within a System before variable values are updated. Then, the outputs are stored within the outputs dictionary, where the values are again wrapped in State objects.
# Store the outputs
self.outputs = {}
self.outputs["f"] = State(
value=f, desc="Rosenbrock function value", source=self
)
Nothing is returned from this method, as the output values are stored within the class instance and are accessed by interacting with this object.
_analyze_adjoint Method
Comparable to the previous method, _analyze_adjoint is responsible for propagating the derivatives from the outputs back to the variables. Within Flume, the variables for an individual discipline are treated as design variables for the adjoint method. As a result, if the outputs are explicit functions of the input variables, then the user is responsible for computing the partial derivatives of the outputs with respect to the variables and accounting for the impact of the adjoint variables.
However, if an Analysis internally performs some linear or nonlinear solve to compute additional state variables, the full form of the adjoint method should be used to compute the contributions to the variables’ derivatives. A full discussion of the adjoint method is not in the scope of this overview, but its role within Flume can be understood through this example. To begin the method, variable and output derivatives are extracted along with the variable values. Similar to values,
derivatives are accessed using the .deriv extension for State objects.
# Extract the derivatives of the outputs
fb = self.outputs["f"].deriv
# Extract the variable values
x = self.variables["x"].value
y = self.variables["y"].value
# Extract the variable derivatives
xb = self.variables["x"].deriv
yb = self.variables["y"].deriv
# Extract the parameter values
a = self.parameters["a"]
b = self.parameters["b"]
Next, since this discipline computes \(f\) as an explicit function of \(x\) and \(y\), the adjoint method simplifies to multiplying the adjoint variables \(\bar{f}\) or fb by the partial derivatives of the output with respect to each variable. Note that the addition assignment += operator is used to update, not overwrite, the derivatives for the variables. This is critical to ensure that derivatives already computed for \(\bar{x}\) (xb) and \(\bar{y}\)
(yb) do not get discarded.
# Compute contributions to xb
xb += (2 * (a - x) * -1 + 2.0 * b * (y - x**2) * -2 * x) * fb
# Compute contributions to yb
yb += (2 * b * (y - x**2)) * fb
After computing the derivatives, the user should update the adjoint_analyzed attribute to denote that computations for the derivatives have concluded, similar to the behavior for the _analyze method.
# Update the analyzed adjoint attribute
self.adjoint_analyzed = True
Finally, the derivative values are assigned to the State objects stored within the variables dictionary, which ensures that these quantities are stored before concluding the current _analyze_adjoint method.
# Assign the derivative values
self.variables["x"].set_deriv_value(deriv_val=xb)
self.variables["y"].set_deriv_value(deriv_val=yb)
Again, nothing is returned from this method, as the derivatives are updated within the State objects for the specific class instance.
Setting up a System
Using the procedure outlined above to define various Analysis classes, emphasis will now be placed on combining Analysis instances together to construct a System. Here, the classes associated with the constrained Rosenbrock example will be utilized, which are located here. First, the entire script to define a System is shown, and detailed descriptions will be included below.
# Construct the design variables object
rosenbrock_dvs = RosenbrockDVs(obj_name="dvs", sub_analyses=[])
# Construct the analysis object for the Rosenbrock function
a = 1.0
b = 100.0
rosenbrock = Rosenbrock(
obj_name="rosenbrock", sub_analyses=[rosenbrock_dvs], a=a, b=b
)
# Construct the analysis object for the constraint on the design variables
rosenbrock_con = RosenbrockConstraint(
obj_name="con", sub_analyses=[rosenbrock_dvs]
)
# Construct the system
flume_sys = System(
sys_name="rosen_sys_con",
top_level_analysis_list=[rosenbrock, rosenbrock_con],
log_name="flume.log",
log_prefix="examples/rosenbrock_constrained",
)
# Graph the system network to visualize connections
graph = flume_sys.graph_network(
filename="ConstrainedRosenbrock", output_directory="examples/rosenbrock_constrained"
)
Creating Analsis Instances
Before assembling the System, it is necessary to create instances of the Analysis classes that are intended to be used within the System. The first object is the RosenBrockDVs class:
# Construct the design variables object
rosenbrock_dvs = RosenbrockDVs(obj_name="dvs", sub_analyses=[])
This simply maps design variables x_dv and y_dv to x and y. This creates a single set of outputs for x and y that can then be distributed to the other parts of the System. A unique name is given to the object, and no sub-analyses are provided, as this object defines the source for x_dv and y_dv.
Click here to view the RosenbrockDVs class
from flume.base_classes.analysis import Analysis
from flume.base_classes.state import State
class RosenbrockDVs(Analysis):
def __init__(self, obj_name: str, sub_analyses=[], **kwargs):
"""
Analysis class that defines the design variables for the Rosenbrock function. This is needed to
define single values for x and y that can be treated as design variables, which are then
distributed throughout the System (i.e. to the objective and constriant functions).
Parameters
----------
obj_name : str
Name for the analysis object
sub_analyses : list
A list of sub-analyses for the object
Keyword Arguments
-----------------
No input parameters for this object, so no **kwargs.
"""
# Set the default parameters
self.default_parameters = {}
# Perform the base class object initialization
super().__init__(obj_name=obj_name, sub_analyses=sub_analyses, **kwargs)
# Set the default states for the variables
x_dv_var = State(
value=0.0, desc="Value to use for the x design variable", source=self
)
y_dv_var = State(
value=0.0, desc="Value to use for the y design variable", source=self
)
self.variables = {"x_dv": x_dv_var, "y_dv": y_dv_var}
return
def _analyze(self):
"""
Maps x_dv, y_dv -> x, y, where x, y are then distributed throughout the System.
"""
# Extract the variables
x_dv = self.variables["x_dv"].value
y_dv = self.variables["y_dv"].value
# Update the analyzed attribute
self.analyzed = True
# Store the outputs in the outputs dictionary
self.outputs = {}
self.outputs["x"] = State(value=x_dv, desc="Value for x", source=self)
self.outputs["y"] = State(value=y_dv, desc="Value for y", source=self)
return
def _analyze_adjoint(self):
"""
Propagates the derivatives for x and y back to the design variables x_dv and y_dv
"""
# Extract the output derivatives
xb = self.outputs["x"].deriv
yb = self.outputs["y"].deriv
# Extract the variable derivatives
x_dvb = self.variables["x_dv"].deriv
y_dvb = self.variables["y_dv"].deriv
# Update the derivative values
x_dvb += xb
y_dvb += yb
# Update the analyzed adjoint attribute
self.adjoint_analyzed = True
# Set the derivative values
self.variables["x_dv"].set_deriv_value(deriv_val=x_dvb)
self.variables["y_dv"].set_deriv_value(deriv_val=y_dvb)
return
Next, instances of the Rosenbrock and RosenbrockConstraint Analysis classes are created
# Construct the analysis object for the Rosenbrock function
a = 1.0
b = 100.0
rosenbrock = Rosenbrock(
obj_name="rosenbrock", sub_analyses=[rosenbrock_dvs], a=a, b=b
)
# Construct the analysis object for the constraint on the design variables
rosenbrock_con = RosenbrockConstraint(
obj_name="con", sub_analyses=[rosenbrock_dvs]
)
Here, the values for a and b are passed as keyword arguments to the Rosenbrock class to set the parameter values. Also, each class takes the rosenbrock_dvs instance as a sub-analysis. This creates a connection between the outputs for x and y from rosenbrock_dvs to the input variables for both rosenbrock and rosenbrock_con. If this is not done, then the values for x and y will not be distributed throughout the model correctly.
Click here to view the RosenbrockConstraint class
from flume.base_classes.analysis import Analysis
from flume.base_classes.state import State
class RosenbrockConstraint(Analysis):
def __init__(self, obj_name: str, sub_analyses=[RosenbrockDVs], **kwargs):
"""
Analysis class that computes the value of the constraint function, which is used to constrain
the design space to a circular region.
Parameters
----------
obj_name : str
Name for the analysis object
sub_analyses : list
A list of sub-analyses for the object, which nominally contains an instance of
the RosenbrockDVs class
Keyword Arguments
-----------------
No input parameters for this object, so no **kwargs.
"""
# Set the default parameters
self.default_parameters = {}
# Perform the base class object initialization
super().__init__(obj_name=obj_name, sub_analyses=sub_analyses, **kwargs)
# Set the default State for the variables
xvar = State(value=0.0, desc="x state value", source=self)
yvar = State(value=0.0, desc="y state value", source=self)
# Construct variables dictionary
self.variables = {"x": xvar, "y": yvar}
return
def _analyze(self):
"""
Computes the distance from the origin for the current values of x and y.
"""
# Extract the variable values
x = self.variables["x"].value
y = self.variables["y"].value
# Compute the value of the constraint
g = x**2 + y**2
# Update the analyzed attribute
self.analyzed = True
# Store the outputs in the outputs dictionary
self.outputs = {}
self.outputs["g"] = State(
value=g,
desc="Distance from the origin, which is used to constrain the design space to a circle",
source=self,
)
return
def _analyze_adjoint(self):
"""
Performs the adjoint analysis for the constraint function to propagate the derivatives back to x and y.
"""
# Extract the variable values
x = self.variables["x"].value
y = self.variables["y"].value
# Extract the variable derivatives
xb = self.variables["x"].deriv
yb = self.variables["y"].deriv
# Extract the output derivatives
gb = self.outputs["g"].deriv
# Add the contributions to xb and yb
xb += gb * 2 * x
yb += gb * 2 * y
# Update the analyzed adjoint attribute
self.adjoint_analyzed = True
# Assign the derivative values
self.variables["x"].set_deriv_value(deriv_val=xb)
self.variables["y"].set_deriv_value(deriv_val=yb)
return
Finally, the System is constructed with the following
# Construct the system
flume_sys = System(
sys_name="rosen_sys_con",
top_level_analysis_list=[rosenbrock, rosenbrock_con],
log_name="flume.log",
log_prefix="examples/rosenbrock_constrained",
)
A name is provided for the system, and the top-level analysis list is specified by providing the rosenbrock and rosenbrock_con objects as a list. This top-level analysis list corresponds to the Analysis objects whose outputs will be used as the objective or constraint functions for optimization. Additionally, a name for the log file and output directory are provided for file management purposes. To ensure that outputs and variables are connected correctly, the structure of the System
can be visualized by calling the following function.
# Graph the system network to visualize connections
graph = flume_sys.graph_network(
filename="ConstrainedRosenbrock", output_directory="examples/rosenbrock_constrained"
)
This constructs a document using graphviz that provides a visual representation of the System network. If it is found that outputs and variables are not connected within the diagram, this likely indicates that a sub_analyses list is missing an entry to define the flow of information.
Formulating an Optimization Problem
With the System assembled, the next step is to declare the design variables, objective function, and constraints that will be utilized for the optimization problem. In all cases, global names are utilized to identify these States, which is denoted with object_name.local_state_name. For the design variables
# Declare the design variables for the system
flume_sys.declare_design_vars(
global_var_name={
"dvs.x_dv": {"lb": -1.5, "ub": 1.5},
"dvs.y_dv": {"lb": -1.5, "ub": 1.5},
}
)
The user can optionally provide applicable scalar lower and upper bound values. If the variable is an array, then the bounds are still provided as scalars, and each variable in the array is then given the same bounds. Next, the objective function is specified
# Declare the objective
flume_sys.declare_objective(global_obj_name="rosenbrock.f", obj_scale=1.0)
The second argument obj_scale is optional, but it defines a scaling factor that, when multiplied by the objective function’s value, should scale the value to \(\mathcal{O}(1)\). This product is the quantity that the optimizer will see. Finally, constraints are specified with
# Declare the constraint
flume_sys.declare_constraints(
global_con_name={"con.g": {"direction": "leq", "rhs": 1.0}}
)
The syntax here corresponds to providing a dictionary of dictionaries, where each outer key corresponds to the constraint affecting the optimization problem. The inner dictionaries provide information regarding the constraint type and direction, that is less than or equal to leq, greater than or equal to geq, or an equality constraint both. Additionally, the right-hand side value rhs for the constraint is provided, which is also used to scale the constraints internally. For this
example, the declared constraint is such that the output g of the object named con should be less than or equal to \(1.0\).
Optimizing with the SciPy Interface
After constructing the System and specifying the optimization problem formulation, numerical optimization can be performed by using one of Flume’s optimizer interfaces. To date, the supported interfaces include ParOpt and SciPy. In this example, the SciPy interface will be utilized, which is done with the following lines
# Construct the Scipy interface
interface = FlumeScipyInterface(flume_sys=flume_sys)
# Set a random starting point
init_global_vars = {
"dvs.x_dv": np.random.uniform(low=-1.0, high=1.0),
"dvs.y_dv": np.random.uniform(low=1.0, high=1.0)
}
initial_point = interface.set_initial_point(
initial_global_vars=init_global_vars
)
# Optimize the problem with SciPy minimize
x, res = interface.optimize_system(x0=initial_point, method="SLSQP")
First, an instance of the FlumeScipyInterface is constructed by providing the class with the instance of the System. Then, an initial starting point, random in this case, is determined and mapped into a NumPy array with the set_initial_point method. Finally, the System is optimized by calling optimize_system with the initial point and desired method. The default method is SLSQP, and the user can optionally provide a maximum number of iterations and options for the optimization
method that align with the SciPy documentation. The output of this method is the optimal set of design variables x and the OptimizationResult instance from SciPy res.
Executing the Optimization Script
Below the setup of the System and optimization portions of the script are repeated and can be executed to explore Flume in action. This will print the optimal values for the design variables, which should be \(x^*\approx 0.786, y^* \approx 0.618\) for the constrained Rosenbrock example. It will also output the OptimizeResult instance returned by SciPy.
[ ]:
import sys
import os
!pip install -r requirements.txt --quiet
import numpy as np
# Add the root directory of the repository
sys.path.append(os.path.abspath("../.."))
from flume.base_classes.system import System
from flume.interfaces.scipy_interface import FlumeScipyInterface
from examples.rosenbrock.rosenbrock_problem_classes import RosenbrockDVs, Rosenbrock, RosenbrockConstraint
# Construct the design variables object
rosenbrock_dvs = RosenbrockDVs(obj_name="dvs", sub_analyses=[])
# Construct the analysis object for the Rosenbrock function
a = 1.0
b = 100.0
rosenbrock = Rosenbrock(
obj_name="rosenbrock", sub_analyses=[rosenbrock_dvs], a=a, b=b
)
# Construct the analysis object for the constraint on the design variables
rosenbrock_con = RosenbrockConstraint(
obj_name="con", sub_analyses=[rosenbrock_dvs]
)
# Construct the system
flume_sys = System(
sys_name="rosen_sys_con",
top_level_analysis_list=[rosenbrock, rosenbrock_con],
log_name="flume.log",
log_prefix=".",
)
# Declare the design variables for the system
flume_sys.declare_design_vars(
global_var_name={
"dvs.x_dv": {"lb": -1.5, "ub": 1.5},
"dvs.y_dv": {"lb": -1.5, "ub": 1.5},
}
)
# Declare the objective
flume_sys.declare_objective(global_obj_name="rosenbrock.f")
# Declare the constraint
flume_sys.declare_constraints(
global_con_name={"con.g": {"direction": "leq", "rhs": 1.0}}
)
# Construct the Scipy interface
interface = FlumeScipyInterface(flume_sys=flume_sys)
# Set a random starting point
init_global_vars = {
"dvs.x_dv": np.random.uniform(low=-1.0, high=1.0),
"dvs.y_dv": np.random.uniform(low=1.0, high=1.0)
}
initial_point = interface.set_initial_point(
initial_global_vars=init_global_vars
)
# Optimize the problem with SciPy minimize
x, res = interface.optimize_system(x0=initial_point, method="SLSQP")
print("Optimal design variables: ", x)
print(res)
WARNING: There was an error checking the latest version of pip.
Optimal design variables: [0.78641515 0.61769831]
message: Optimization terminated successfully
success: True
status: 0
fun: 0.0456748087195002
x: [ 7.864e-01 6.177e-01]
nit: 26
jac: [-1.911e-01 -1.501e-01]
nfev: 35
njev: 26
Method SLSQP does not use Hessian information (hess).
Unknown solver options: verbose
Other Resources to Review
This concludes the overview of Flume and the descriptions provided in context of the constrained Rosenbrock example. For new users, exploring additional examples within the examples gallery will provide further insight into Flume in the context of different problems. A few other useful methods for Analysis classes are also discussed in the methods overview, although these are intended to be helper methods and do not constitute critical features for using the framework.