File indexing completed on 2026-08-12 08:24:55
0001 import numpy as np
0002 from typing import Dict, List
0003 from aid2e.utilities.workflows import JobContext
0004
0005 def dtlz2_both_objectives(x: List[float]) -> Dict[str, float]:
0006 """Compute both DTLZ2 objectives in one function."""
0007 x = np.array(x)
0008 g = np.sum((x[1:] - 0.5) ** 2)
0009 f1 = (1 + g) * np.cos(x[0] * np.pi / 2) * np.cos(x[1] * np.pi / 2)
0010 f2 = (1 + g) * np.cos(x[0] * np.pi / 2) * np.sin(x[1] * np.pi / 2)
0011 return {"f1": float(f1), "f2": float(f2)}
0012
0013 def dtlz2_f1_only(x: List[float]) -> float:
0014 """Compute only f1 objective of DTLZ2."""
0015 x = np.array(x)
0016 g = np.sum((x[1:] - 0.5) ** 2)
0017 f1 = (1 + g) * np.cos(x[0] * np.pi / 2) * np.cos(x[1] * np.pi / 2)
0018 return float(f1)
0019
0020 def dtlz2_f2_only(x: List[float]) -> float:
0021 """Compute only f2 objective of DTLZ2."""
0022 x = np.array(x)
0023 g = np.sum((x[1:] - 0.5) ** 2)
0024 f2 = (1 + g) * np.cos(x[0] * np.pi / 2) * np.sin(x[1] * np.pi / 2)
0025 return float(f2)
0026
0027 def evaluate_both_objectives_wrapper(context: JobContext) -> Dict[str, float]:
0028 """Wrapper to evaluate both objectives from JobContext."""
0029 design_point = context.design_point
0030 x = [design_point['x1'], design_point['x2'], design_point['x3']]
0031 objectives = dtlz2_both_objectives(x)
0032 context.add_log(f"Design point: {x}")
0033 context.add_log(f"Objectives: {objectives}")
0034 context.xcom_push("objectives", objectives)
0035
0036 required_keys = {"f1", "f2"}
0037 missing = required_keys - objectives.keys()
0038 if missing:
0039 raise ValueError(f"evaluate_both_objectives_wrapper: Missing objectives {missing} in result dict. Got: {objectives}")
0040 return objectives
0041
0042 def evaluate_f1_wrapper(context: JobContext) -> float:
0043 """Wrapper to evaluate f1 from JobContext."""
0044 design_point = context.design_point
0045 x = [design_point['x1'], design_point['x2'], design_point['x3']]
0046 f1 = dtlz2_f1_only(x)
0047 context.add_log(f"Design point: {x}")
0048 context.add_log(f"f1 = {f1}")
0049 context.xcom_push("f1", f1)
0050 return f1
0051
0052 def evaluate_f2_wrapper(context: JobContext) -> float:
0053 """Wrapper to evaluate f2 from JobContext."""
0054 design_point = context.design_point
0055 x = [design_point['x1'], design_point['x2'], design_point['x3']]
0056 f2 = dtlz2_f2_only(x)
0057 context.add_log(f"Design point: {x}")
0058 context.add_log(f"f2 = {f2}")
0059 context.xcom_push("f2", f2)
0060 return f2
