Coverage for smartmdao / optimization.py: 100%
100 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-07-05 17:31 +0200
1import logging
2from dataclasses import dataclass, field
3from typing import Any, Callable, Dict, List, Literal, Optional, Protocol, Tuple, Type, Union
5from .core import Pipeline
7logger = logging.getLogger(__name__)
9class PipelineEvaluator:
10 """
11 A generic, stateful bridge to interface the Pipeline with external optimizers
12 (SciPy, OpenTURNS, PyOptSparse, etc.).
14 It caches the last evaluation to prevent redundant pipeline runs when optimizers
15 request objectives and constraints independently for the same state.
16 """
17 def __init__(self,
18 pipeline: Pipeline,
19 design_vars: List[str],
20 constants: Dict[str, Any] = None):
21 """
22 :param pipeline: The instantiated smartmdao.
23 :param design_vars: Ordered list of variable names corresponding to the optimizer's input array `x`.
24 :param constants: Optional dictionary of variables that remain fixed during optimization.
25 """
26 self.pipeline = pipeline
27 self.design_vars = design_vars
28 self.constants = constants or {}
30 self.last_x = None
31 self.last_results = None
32 self.eval_count = 0
34 def evaluate(self, x) -> Dict[str, Any]:
35 """Runs the pipeline if the design variables have changed."""
36 # Convert to tuple for hashable comparison
37 x_tuple = tuple(x)
39 if self.last_x != x_tuple:
40 self.eval_count += 1
42 # 1. Map the numeric array 'x' back to named variables
43 inputs = dict(zip(self.design_vars, x))
45 # 2. Inject constants
46 inputs.update(self.constants)
48 # 3. Execute
49 self.last_results = self.pipeline.run(**inputs)
50 self.last_x = x_tuple
52 return self.last_results
54 def get_objective(self, output_name: str) -> Callable:
55 """
56 Factory method that returns a callable objective function for the optimizer.
57 """
58 def _objective(x):
59 return self.evaluate(x)[output_name]
60 return _objective
62 def get_constraint(self, output_name: str, multiplier: float = 1.0) -> Callable:
63 """
64 Factory method that returns a callable constraint function.
65 :param multiplier: Useful for flipping constraint signs.
66 (e.g., SciPy expects f(x) >= 0. If pipeline outputs f(x) <= 0, use multiplier=-1.0)
67 """
68 def _constraint(x):
69 return multiplier * self.evaluate(x)[output_name]
70 return _constraint
73# ==============================================================================
74# Backend-agnostic optimization: define the problem once, swap the solver.
75# ==============================================================================
77@dataclass
78class ConstraintSpec:
79 """
80 One inequality or equality constraint on a named pipeline output.
81 Both built-in backends follow the same convention: 'ineq' means
82 h(x) >= 0 and 'eq' means h(x) == 0 - use `multiplier` to flip the sign
83 of a discipline that was naturally written the other way around.
84 """
85 name: str
86 kind: Literal["ineq", "eq"] = "ineq"
87 multiplier: float = 1.0
90@dataclass
91class OptimizationProblem:
92 """
93 A backend-agnostic description of an optimization problem, built on top
94 of an existing PipelineEvaluator. Any OptimizerBackend can consume this
95 without knowing anything about SmartMDAO's Pipeline internals.
96 """
97 evaluator: PipelineEvaluator
98 initial_guess: List[float]
99 bounds: Optional[List[Tuple[float, float]]] = None
100 objective: str = "objective"
101 constraints: List[ConstraintSpec] = field(default_factory=list)
104@dataclass
105class OptimizationResult:
106 """
107 Normalized optimization outcome, common across every backend.
108 `raw` keeps the underlying backend-specific result object (e.g. a
109 scipy.optimize.OptimizeResult) for anyone who needs backend-specific detail.
110 """
111 x: List[float]
112 objective_value: float
113 success: bool
114 message: str
115 state: Dict[str, Any]
116 raw: Any = None
119class OptimizerBackend(Protocol):
120 """Interface every optimizer backend implements."""
121 def solve(self, problem: OptimizationProblem, **options: Any) -> OptimizationResult:
122 ...
125_BACKENDS: Dict[str, Type[OptimizerBackend]] = {}
128def register_backend(name: str):
129 """
130 Registers an OptimizerBackend under a short string key, so it becomes
131 selectable via optimize(problem, backend=name) - the same "name it once,
132 use it everywhere" pattern as @pipeline.step. Custom backends (PyOptSparse,
133 an in-house solver, ...) can register themselves the exact same way.
134 """
135 def decorator(cls: Type[OptimizerBackend]) -> Type[OptimizerBackend]:
136 _BACKENDS[name] = cls
137 return cls
138 return decorator
141def optimize(
142 problem: OptimizationProblem,
143 backend: Union[str, OptimizerBackend] = "scipy",
144 **options: Any,
145) -> OptimizationResult:
146 """
147 Runs `problem` through the given backend and returns a normalized result.
148 `backend` can be a registered name ('scipy', 'openturns', ...) or any
149 object implementing OptimizerBackend directly (bring your own optimizer).
150 """
151 if isinstance(backend, str):
152 try:
153 backend_cls = _BACKENDS[backend]
154 except KeyError:
155 raise ValueError(
156 f"Unknown optimizer backend '{backend}'. Registered backends: {sorted(_BACKENDS)}."
157 ) from None
158 backend = backend_cls()
160 logger.info(f"Running optimization with backend '{type(backend).__name__}'.")
161 return backend.solve(problem, **options)
164@register_backend("scipy")
165class ScipyBackend:
166 """
167 Bridges to scipy.optimize.minimize. Defaults to SLSQP, since it natively
168 supports bounds plus inequality/equality constraints - the most common case.
169 Extra keyword arguments are forwarded as-is to `minimize()` (e.g. `tol=1e-6`
170 or `options={"disp": True}`).
171 """
172 def solve(self, problem: OptimizationProblem, method: str = "SLSQP", **options: Any) -> OptimizationResult:
173 from scipy.optimize import minimize
175 constraints = [
176 {
177 "type": c.kind,
178 "fun": problem.evaluator.get_constraint(c.name, multiplier=c.multiplier),
179 }
180 for c in problem.constraints
181 ]
183 result = minimize(
184 problem.evaluator.get_objective(problem.objective),
185 problem.initial_guess,
186 method=method,
187 bounds=problem.bounds,
188 constraints=constraints,
189 **options,
190 )
192 x_opt = [float(v) for v in result.x]
193 return OptimizationResult(
194 x=x_opt,
195 objective_value=float(result.fun),
196 success=bool(result.success),
197 message=str(result.message),
198 state=problem.evaluator.evaluate(x_opt),
199 raw=result,
200 )
203@register_backend("openturns")
204class OpenTURNSBackend:
205 """
206 Bridges to OpenTURNS' OptimizationAlgorithm classes. Defaults to Cobyla,
207 a derivative-free algorithm that handles inequality/equality constraints
208 without requiring gradients. `method` is looked up as a class name on the
209 `openturns` module (e.g. "Cobyla", "SLSQP", "AbdoRackwitz", ...), so any
210 algorithm OpenTURNS ships is selectable without adding a branch here.
211 """
212 def solve(
213 self,
214 problem: OptimizationProblem,
215 method: str = "Cobyla",
216 max_iterations: int = 1000,
217 **options: Any,
218 ) -> OptimizationResult:
219 import openturns as ot
221 dim = len(problem.initial_guess)
222 objective_fn = ot.PythonFunction(
223 dim, 1, lambda x: [problem.evaluator.get_objective(problem.objective)(x)]
224 )
225 ot_problem = ot.OptimizationProblem(objective_fn)
227 if problem.bounds:
228 lower = [b[0] for b in problem.bounds]
229 upper = [b[1] for b in problem.bounds]
230 ot_problem.setBounds(ot.Interval(lower, upper))
232 ineq = [c for c in problem.constraints if c.kind == "ineq"]
233 eq = [c for c in problem.constraints if c.kind == "eq"]
235 if ineq:
236 ot_problem.setInequalityConstraint(
237 ot.PythonFunction(
238 dim, len(ineq),
239 lambda x, _cs=ineq: [problem.evaluator.get_constraint(c.name, c.multiplier)(x) for c in _cs],
240 )
241 )
242 if eq:
243 ot_problem.setEqualityConstraint(
244 ot.PythonFunction(
245 dim, len(eq),
246 lambda x, _cs=eq: [problem.evaluator.get_constraint(c.name, c.multiplier)(x) for c in _cs],
247 )
248 )
250 algo = getattr(ot, method)(ot_problem)
251 algo.setStartingPoint(problem.initial_guess)
252 algo.setMaximumIterationNumber(max_iterations)
253 for key, value in options.items():
254 getattr(algo, f"set{key}")(value)
255 algo.run()
257 ot_result = algo.getResult()
258 x_opt = list(ot_result.getOptimalPoint())
260 return OptimizationResult(
261 x=x_opt,
262 objective_value=float(ot_result.getOptimalValue()[0]),
263 success=True, # OpenTURNS raises on failure rather than reporting a boolean.
264 message=f"Completed after {ot_result.getIterationNumber()} iteration(s).",
265 state=problem.evaluator.evaluate(x_opt),
266 raw=ot_result,
267 )