Coverage for smart_pipeline / optimization.py: 100%

1import logging 

2from typing import List, Dict, Any, Callable 

3 

4from .core import Pipeline 

5 

6logger = logging.getLogger(__name__) 

7 

8class PipelineEvaluator: 

9 """ 

10 A generic, stateful bridge to interface the Pipeline with external optimizers  

11 (SciPy, OpenTURNS, PyOptSparse, etc.). 

12  

13 It caches the last evaluation to prevent redundant pipeline runs when optimizers 

14 request objectives and constraints independently for the same state. 

15 """ 

16 def __init__(self, 

17 pipeline: Pipeline, 

18 design_vars: List[str], 

19 constants: Dict[str, Any] = None): 

20 """ 

21 :param pipeline: The instantiated smart_pipeline. 

22 :param design_vars: Ordered list of variable names corresponding to the optimizer's input array `x`. 

23 :param constants: Optional dictionary of variables that remain fixed during optimization. 

24 """ 

25 self.pipeline = pipeline 

26 self.design_vars = design_vars 

27 self.constants = constants or {} 

28 

29 self.last_x = None 

30 self.last_results = None 

31 self.eval_count = 0 

32 

33 def evaluate(self, x) -> Dict[str, Any]: 

34 """Runs the pipeline if the design variables have changed.""" 

35 # Convert to tuple for hashable comparison 

36 x_tuple = tuple(x) 

37 

38 if self.last_x != x_tuple: 

39 self.eval_count += 1 

40 

41 # 1. Map the numeric array 'x' back to named variables 

42 inputs = dict(zip(self.design_vars, x)) 

43 

44 # 2. Inject constants 

45 inputs.update(self.constants) 

46 

47 # 3. Execute 

48 self.last_results = self.pipeline.run(**inputs) 

49 self.last_x = x_tuple 

50 

51 return self.last_results 

52 

53 def get_objective(self, output_name: str) -> Callable: 

54 """ 

55 Factory method that returns a callable objective function for the optimizer. 

56 """ 

57 def _objective(x): 

58 return self.evaluate(x)[output_name] 

59 return _objective 

60 

61 def get_constraint(self, output_name: str, multiplier: float = 1.0) -> Callable: 

62 """ 

63 Factory method that returns a callable constraint function. 

64 :param multiplier: Useful for flipping constraint signs.  

65 (e.g., SciPy expects f(x) >= 0. If pipeline outputs f(x) <= 0, use multiplier=-1.0) 

66 """ 

67 def _constraint(x): 

68 return multiplier * self.evaluate(x)[output_name] 

69 return _constraint