File indexing completed on 2026-11-09 09:29:52
0001 """Evaluate DTLZ2 through Python callables, CLI arguments, or a design JSON file."""
0002
0003 from __future__ import annotations
0004
0005 import argparse
0006 import json
0007 import math
0008 import sys
0009 import time
0010 from pathlib import Path
0011 from typing import TYPE_CHECKING, Any, Dict, List
0012
0013 if TYPE_CHECKING:
0014 from aid2e.utilities.workflows import JobContext
0015
0016 def dtlz2_both_objectives(x: List[float]) -> Dict[str, float]:
0017 """Compute both DTLZ2 objectives in one function."""
0018 g = sum((value - 0.5) ** 2 for value in x[1:])
0019 f1 = (1 + g) * math.cos(x[0] * math.pi / 2)
0020 f2 = (1 + g) * math.sin(x[0] * math.pi / 2)
0021 return {"f1": float(f1), "f2": float(f2)}
0022
0023 def dtlz2_f1_only(x: List[float]) -> float:
0024 """Compute only f1 objective of DTLZ2."""
0025 g = sum((value - 0.5) ** 2 for value in x[1:])
0026 f1 = (1 + g) * math.cos(x[0] * math.pi / 2)
0027 return float(f1)
0028
0029 def dtlz2_f2_only(x: List[float]) -> float:
0030 """Compute only f2 objective of DTLZ2."""
0031 g = sum((value - 0.5) ** 2 for value in x[1:])
0032 f2 = (1 + g) * math.sin(x[0] * math.pi / 2)
0033 return float(f2)
0034
0035 def design_vector(design_point: Dict[str, Any], prefix: str = "DTLZ2_variables.") -> List[float]:
0036 """Return all DTLZ2 variables in numeric order (x1, ..., x10)."""
0037 names = sorted(
0038 (name for name in design_point if name.startswith(f"{prefix}x")),
0039 key=lambda name: int(name.rsplit("x", 1)[1]),
0040 )
0041 if len(names) < 2 or names != [f"{prefix}x{i}" for i in range(1, len(names) + 1)]:
0042 raise ValueError("DTLZ2 requires consecutive variables x1 through xN, with N >= 2")
0043 return [float(design_point[name]) for name in names]
0044
0045
0046 def objective_payload(
0047 *,
0048 design_point: Dict[str, Any],
0049 **kwargs,
0050 ) -> Dict[str, float]:
0051 """Compute DTLZ2 objectives for the config-driven objective plan."""
0052 return dtlz2_both_objectives(design_vector(design_point))
0053
0054
0055 def prepare_objective_inputs(*, design_point: Dict[str, Any], **kwargs) -> Dict[str, List[float]]:
0056 """Pass ordered design values to the dependent objective step."""
0057 return {"values": design_vector(design_point)}
0058
0059
0060 def evaluate_prepared_objective(*, inputs: Dict[str, Any], **kwargs) -> Dict[str, float]:
0061 """Compute f2 from the prepare step's output."""
0062 return {"f2": dtlz2_f2_only(inputs["prepare"]["values"])}
0063
0064
0065 def evaluate_both_objectives_wrapper(context: JobContext) -> Dict[str, float]:
0066 """Wrapper to evaluate both objectives from JobContext."""
0067 design_point = context.design_point
0068 x = design_vector(design_point, prefix="")
0069 objectives = dtlz2_both_objectives(x)
0070 context.add_log(f"Design point: {x}")
0071 context.add_log(f"Objectives: {objectives}")
0072 context.xcom_push("objectives", objectives)
0073
0074 required_keys = {"f1", "f2"}
0075 missing = required_keys - objectives.keys()
0076 if missing:
0077 raise ValueError(f"evaluate_both_objectives_wrapper: Missing objectives {missing} in result dict. Got: {objectives}")
0078 return objectives
0079
0080 def evaluate_qualified_both_objectives(context: JobContext) -> Dict[str, float]:
0081 """Evaluate DTLZ2 objectives from canonical qualified design keys."""
0082 design_point = context.design_point
0083 x = design_vector(design_point)
0084 objectives = dtlz2_both_objectives(x)
0085 context.add_log(f"Qualified design point: {x}")
0086 context.add_log(f"Objectives: {objectives}")
0087 context.xcom_push("objectives", objectives)
0088 return objectives
0089
0090 def evaluate_f1_wrapper(context: JobContext) -> float:
0091 """Wrapper to evaluate f1 from JobContext."""
0092 design_point = context.design_point
0093 x = design_vector(design_point, prefix="")
0094 f1 = dtlz2_f1_only(x)
0095 context.add_log(f"Design point: {x}")
0096 context.add_log(f"f1 = {f1}")
0097 context.xcom_push("f1", f1)
0098 return f1
0099
0100 def evaluate_f2_wrapper(context: JobContext) -> float:
0101 """Wrapper to evaluate f2 from JobContext."""
0102 design_point = context.design_point
0103 x = design_vector(design_point, prefix="")
0104 f2 = dtlz2_f2_only(x)
0105 context.add_log(f"Design point: {x}")
0106 context.add_log(f"f2 = {f2}")
0107 context.xcom_push("f2", f2)
0108 return f2
0109
0110
0111 def main() -> int:
0112 parser = argparse.ArgumentParser(description=__doc__)
0113 inputs = parser.add_mutually_exclusive_group(required=True)
0114 inputs.add_argument("--design", "--design_params_file", type=Path)
0115 inputs.add_argument("--x1", type=float)
0116 for index in range(2, 11):
0117 parser.add_argument(f"--x{index}", type=float)
0118 parser.add_argument("--output", "--output_file", type=Path, required=True)
0119 parser.add_argument("--label", type=str, default="slurm-dtlz2")
0120 parser.add_argument("--sleep-sec", type=float, default=0.0)
0121 parser.add_argument("--repeat", type=int, default=2)
0122 args = parser.parse_args()
0123
0124 values = [getattr(args, f"x{index}") for index in range(1, 11)]
0125 if args.design:
0126 if any(value is not None for value in values):
0127 parser.error("Use either --design or --x1 through --x10, not both")
0128 design = json.loads(args.design.read_text(encoding="utf-8"))
0129 values = design_vector(design)
0130 print(f"[{args.label}] design file: {args.design}", flush=True)
0131 elif any(value is None for value in values):
0132 parser.error("Direct argument mode requires --x1 through --x10")
0133
0134 for step in range(args.repeat):
0135 print(f"[{args.label}] progress {step + 1}/{args.repeat}", flush=True)
0136 print(f"[{args.label}] stderr heartbeat {step + 1}", file=sys.stderr, flush=True)
0137 if args.sleep_sec > 0:
0138 time.sleep(args.sleep_sec)
0139
0140 payload = dtlz2_both_objectives(values)
0141 args.output.parent.mkdir(parents=True, exist_ok=True)
0142 args.output.write_text(json.dumps(payload, indent=2), encoding="utf-8")
0143 print(json.dumps(payload, indent=2), flush=True)
0144 return 0
0145
0146
0147 if __name__ == "__main__":
0148 raise SystemExit(main())
