axfluxmdo.optimize¶

Requires the [opt] extra for the pymoo / OpenMDAO / scikit-learn backends; DesignProblem and DesignDataset work with the core install.

axfluxmdo.optimize.problem ¶

Backend-agnostic optimization problem specification.

This module is the single source of truth for parsing the SPEC-style optimization grammar against the stable AnalyticalResult.to_dict() keys:

variables: {"outer_radius": (0.05, 0.12), "pole_pairs": [8, 10, 12]} — 2-tuples are continuous bounds (or integer bounds when the motor field is an int), lists are discrete choices; dotted paths reach nested fields ("tolerances.runout_m").
objectives: "maximize_torque_density" / "minimize_mass" — the suffix resolves through the alias map to a to_dict() key.
constraints: "winding_temp_c < 140" — parsed to the pymoo/OpenMDAO g <= 0 convention. The model's built-in ConstraintRecords are enforced in addition by default, so optimizer output is always result.feasible.

Design vectors that violate geometric validation (frozen-dataclass __post_init__ ValueError, e.g. inner_radius >= outer_radius) are penalized with large finite objective/constraint values rather than raising — GA constraint-domination buries them and gradient drivers stay numeric.

Only numpy is required here; pymoo/OpenMDAO backends import this module, not the other way around.

UnknownKeyError ¶

Bases: ValueError

A user-facing name did not resolve to any result key or alias.

UserConstraint `dataclass` ¶

UserConstraint(key: str, op: str, bound: float, label: str)

violation ¶

violation(value: float) -> float

g <= 0 convention: positive means violated.

Source code in src/axfluxmdo/optimize/problem.py

def violation(self, value: float) -> float:
    """g <= 0 convention: positive means violated."""
    if not math.isfinite(value):
        return MARGIN_CLAMP
    if self.op in ("<", "<="):
        return value - self.bound
    return self.bound - value

EvalRecord `dataclass` ¶

EvalRecord(f_min: ndarray, g: ndarray, result: AnalyticalResult | None, motor: AxialFluxMotor | None)

One design evaluation in optimizer space.

DesignProblem ¶

DesignProblem(motor: AxialFluxMotor, op: OperatingPoint, *, variables: Mapping[str, object], objectives: Sequence[str], constraints: Sequence[str] = (), model: Model | None = None, enforce_model_constraints: bool = True)

Validated design variables / objectives / constraints around a seed motor.

Source code in src/axfluxmdo/optimize/problem.py

def __init__(
    self,
    motor: AxialFluxMotor,
    op: OperatingPoint,
    *,
    variables: Mapping[str, object],
    objectives: Sequence[str],
    constraints: Sequence[str] = (),
    model: Model | None = None,
    enforce_model_constraints: bool = True,
):
    self.motor = motor
    self.op = op
    self.model = model or AnalyticalModel()
    self.enforce_model_constraints = enforce_model_constraints

    # Baseline evaluation defines the available result keys and sanity-checks
    # that the seed design is evaluable.
    self.baseline_result = self.model.evaluate(motor, op)
    available = self.baseline_result.to_dict().keys()

    self.continuous: dict[str, tuple[float, float]] = {}
    self.integer: dict[str, tuple[int, int]] = {}
    self.choices: dict[str, list] = {}
    for name, spec in variables.items():
        baseline_value = self._baseline_value(name)  # validates the field path
        if isinstance(spec, tuple) and len(spec) == 2:
            lo, hi = spec
            if lo >= hi:
                raise ValueError(f"variable {name!r}: lower bound must be < upper bound")
            if isinstance(baseline_value, int) and not isinstance(baseline_value, bool):
                self.integer[name] = (int(lo), int(hi))
            else:
                self.continuous[name] = (float(lo), float(hi))
        elif isinstance(spec, (list, range, np.ndarray)):
            options = list(spec)
            if not options:
                raise ValueError(f"variable {name!r}: empty choice list")
            self.choices[name] = options
        else:
            raise ValueError(
                f"variable {name!r}: spec must be a (low, high) tuple or a list of "
                f"choices, got {spec!r}"
            )

    self.objectives = [parse_objective(s, available) for s in objectives]
    if not self.objectives:
        raise ValueError("at least one objective is required")
    self.user_constraints = [parse_constraint(s, available) for s in constraints]

    self.model_constraint_names = (
        [c.name for c in self.baseline_result.constraints] if enforce_model_constraints else []
    )

apply ¶

apply(x: Mapping[str, object]) -> AxialFluxMotor

Build the motor variant for a design vector; may raise ValueError.

Source code in src/axfluxmdo/optimize/problem.py

def apply(self, x: Mapping[str, object]) -> AxialFluxMotor:
    """Build the motor variant for a design vector; may raise ValueError."""
    motor = self.motor
    for name, value in x.items():
        if name in self.integer or name in self.choices:
            baseline = self._baseline_value(name)
            if isinstance(baseline, int) and not isinstance(value, int):
                value = int(round(float(value)))  # noqa: PLW2901
        motor = replace_field(motor, name, value)
    return motor

objective_values ¶

objective_values(result: AnalyticalResult) -> np.ndarray

Human-readable (un-negated) objective values.

Source code in src/axfluxmdo/optimize/problem.py

def objective_values(self, result: AnalyticalResult) -> np.ndarray:
    """Human-readable (un-negated) objective values."""
    d = result.to_dict()
    return np.array([d[obj.name] for obj in self.objectives])

resolve_key ¶

resolve_key(name: str, available: Collection[str]) -> str

Resolve a user-facing name (alias or canonical) to a to_dict() key.

Source code in src/axfluxmdo/optimize/problem.py

def resolve_key(name: str, available: Collection[str]) -> str:
    """Resolve a user-facing name (alias or canonical) to a to_dict() key."""
    candidate = ALIASES.get(name, name)
    if candidate in available:
        return candidate
    options = sorted(set(available) | set(ALIASES))
    raise UnknownKeyError(f"unknown result key {name!r}; available keys and aliases: {options}")

parse_objective ¶

parse_objective(spec: str, available: Collection[str]) -> Objective

Parse 'maximize_' / 'minimize_' (key may be an alias).

Source code in src/axfluxmdo/optimize/problem.py

def parse_objective(spec: str, available: Collection[str]) -> Objective:
    """Parse 'maximize_<key>' / 'minimize_<key>' (key may be an alias)."""
    if spec.startswith("maximize_"):
        sense, raw = +1, spec[len("maximize_") :]
    elif spec.startswith("minimize_"):
        sense, raw = -1, spec[len("minimize_") :]
    else:
        raise ValueError(
            f"objective {spec!r} must start with 'maximize_' or 'minimize_' "
            f"(e.g. 'maximize_torque_density')"
        )
    key = resolve_key(raw, available)
    if key == "feasible":
        raise ValueError("'feasible' is not a valid objective; constraints handle feasibility")
    return Objective(name=key, sense=sense, label=spec)

parse_constraint ¶

parse_constraint(spec: str, available: Collection[str]) -> UserConstraint

Parse ' < bound' style constraint strings (<=, >=, <, > supported).

Source code in src/axfluxmdo/optimize/problem.py

def parse_constraint(spec: str, available: Collection[str]) -> UserConstraint:
    """Parse '<key> < bound' style constraint strings (<=, >=, <, > supported)."""
    match = _CONSTRAINT_RE.match(spec)
    if match is None:
        raise ValueError(
            f"cannot parse constraint {spec!r}; expected '<key> < bound' with one of <, <=, >, >="
        )
    raw_key, op, bound = match.groups()
    return UserConstraint(
        key=resolve_key(raw_key, available), op=op, bound=float(bound), label=spec
    )

axfluxmdo.optimize.pymoo_runner ¶

Pareto-front optimization via pymoo's mixed-variable GA.

pymoo is imported lazily so the base package works without the [opt] extra. Mixed continuous/integer/choice design variables use MixedVariableGA with rank-and-crowding survival (the multi-objective configuration recommended by pymoo 0.6.x for mixed search spaces) — never NSGA2-with-rounding, which breaks Choice semantics and duplicate elimination.

ParetoStudy `dataclass` ¶

ParetoStudy(variables: list[str], objectives: list[Objective], X: list[dict], F: ndarray, results: list[AnalyticalResult], motors: list[AxialFluxMotor], seed: int, pop_size: int, n_gen: int, problem: DesignProblem)

Feasible non-dominated designs from one optimize_pareto run.

F holds human-readable objective values (maximize objectives are NOT negated here; the sign convention lives only inside the optimizer). Every point satisfies the user constraints and result.feasible.

to_records ¶

to_records() -> list[dict]

One flat dict per point: design variables merged with result.to_dict().

Source code in src/axfluxmdo/optimize/pymoo_runner.py

def to_records(self) -> list[dict]:
    """One flat dict per point: design variables merged with result.to_dict()."""
    return [x | r.to_dict() for x, r in zip(self.X, self.results, strict=True)]

to_arrays ¶

to_arrays(*fields_: str) -> dict[str, np.ndarray]

Extract named fields (aliases, to_dict keys, or design variables) as arrays.

Source code in src/axfluxmdo/optimize/pymoo_runner.py

def to_arrays(self, *fields_: str) -> dict[str, np.ndarray]:
    """Extract named fields (aliases, to_dict keys, or design variables) as arrays."""
    records = self.to_records()
    available = records[0].keys()
    out = {}
    for name in fields_:
        key = name if name in available else resolve_key(name, available)
        out[name] = np.array([rec[key] for rec in records])
    return out

best ¶

best(objective: str) -> int

Index of the best point for one objective (alias-aware).

'Best' respects the study's sense for that key if it is an objective (max for maximize_*), otherwise defaults to maximum.

Source code in src/axfluxmdo/optimize/pymoo_runner.py

def best(self, objective: str) -> int:
    """Index of the best point for one objective (alias-aware).

    'Best' respects the study's sense for that key if it is an objective
    (max for maximize_*), otherwise defaults to maximum.
    """
    records = self.to_records()
    key = objective if objective in records[0] else resolve_key(objective, records[0].keys())
    values = np.array([rec[key] for rec in records])
    sense = next((obj.sense for obj in self.objectives if obj.name == key), +1)
    return int(np.argmax(sense * values))

optimize_pareto ¶

optimize_pareto(motor: AxialFluxMotor, op: OperatingPoint, *, variables: Mapping[str, object], objectives: Sequence[str], constraints: Sequence[str] = (), model: Model | None = None, algorithm=None, pop_size: int = 60, n_gen: int = 40, seed: int = 1, verbose: bool = False, enforce_model_constraints: bool = True) -> ParetoStudy

Multi-objective Pareto optimization of the motor design (SPEC flagship API).

Note: unlike the SPEC sketch, the operating point is an explicit argument — every objective (torque, efficiency, winding temperature) depends on it.

Source code in src/axfluxmdo/optimize/pymoo_runner.py

def optimize_pareto(
    motor: AxialFluxMotor,
    op: OperatingPoint,
    *,
    variables: Mapping[str, object],
    objectives: Sequence[str],
    constraints: Sequence[str] = (),
    model: Model | None = None,
    algorithm=None,
    pop_size: int = 60,
    n_gen: int = 40,
    seed: int = 1,
    verbose: bool = False,
    enforce_model_constraints: bool = True,
) -> ParetoStudy:
    """Multi-objective Pareto optimization of the motor design (SPEC flagship API).

    Note: unlike the SPEC sketch, the operating point is an explicit argument —
    every objective (torque, efficiency, winding temperature) depends on it.
    """
    _require_pymoo()
    from pymoo.core.mixed import MixedVariableGA
    from pymoo.core.problem import ElementwiseProblem
    from pymoo.core.variable import Choice, Integer, Real
    from pymoo.optimize import minimize

    try:
        from pymoo.operators.survival.rank_and_crowding import RankAndCrowding
    except ImportError:  # pragma: no cover - older pymoo layout
        from pymoo.algorithms.moo.nsga2 import RankAndCrowdingSurvival as RankAndCrowding

    design = DesignProblem(
        motor,
        op,
        variables=variables,
        objectives=objectives,
        constraints=constraints,
        model=model,
        enforce_model_constraints=enforce_model_constraints,
    )

    pymoo_vars: dict[str, object] = {}
    for name, (lo, hi) in design.continuous.items():
        pymoo_vars[name] = Real(bounds=(lo, hi))
    for name, (lo, hi) in design.integer.items():
        pymoo_vars[name] = Integer(bounds=(lo, hi))
    for name, options in design.choices.items():
        pymoo_vars[name] = Choice(options=options)

    class _MotorProblem(ElementwiseProblem):
        def __init__(self):
            super().__init__(vars=pymoo_vars, n_obj=design.n_obj, n_ieq_constr=design.n_constr)

        def _evaluate(self, x, out, *args, **kwargs):
            record = design.evaluate(x)
            out["F"] = record.f_min
            out["G"] = record.g

    if algorithm is None:
        algorithm = MixedVariableGA(pop_size=pop_size, survival=RankAndCrowding())

    res = minimize(_MotorProblem(), algorithm, ("n_gen", n_gen), seed=seed, verbose=verbose)

    if res.X is None:
        raise RuntimeError(
            "optimize_pareto found no feasible designs; relax the constraints, widen "
            "the variable bounds, or increase pop_size/n_gen"
        )
    xs = [dict(x) for x in np.atleast_1d(res.X)]

    # Re-evaluate the front for full results; keep strictly feasible points only.
    points = []
    for x in xs:
        record = design.evaluate(x)
        if record.result is not None and record.result.feasible and np.all(record.g <= 0):
            points.append((x, record))
    if not points:
        raise RuntimeError(
            "optimize_pareto found no feasible designs after re-evaluation; relax the "
            "constraints or increase the budget"
        )

    return ParetoStudy(
        variables=design.variable_names,
        objectives=design.objectives,
        X=[x for x, _ in points],
        F=np.array([design.objective_values(rec.result) for _, rec in points]),
        results=[rec.result for _, rec in points],
        motors=[rec.motor for _, rec in points],
        seed=seed,
        pop_size=pop_size,
        n_gen=n_gen,
        problem=design,
    )

axfluxmdo.optimize.sensitivity ¶

One-at-a-time design sensitivities for tornado charts.

compute_sensitivities ¶

compute_sensitivities(motor: AxialFluxMotor, op: OperatingPoint, variables: Sequence[str] | Mapping[str, object], *, output: str = 'torque_density_nm_kg', rel_step: float = 0.05, model: Model | None = None) -> SensitivityResult

One-at-a-time low/high perturbations of each variable (tornado-chart input).

variables may be a sequence of field names (perturbed by ±rel_step of the baseline value) or an optimize-style dict: tuple bounds clamp the perturbations, choice lists step one option below/above the baseline's nearest choice. Integer fields are rounded and de-duplicated. A perturbation that produces invalid geometry is skipped with a warning (the baseline output stands in for that side).

Source code in src/axfluxmdo/optimize/sensitivity.py

def compute_sensitivities(
    motor: AxialFluxMotor,
    op: OperatingPoint,
    variables: Sequence[str] | Mapping[str, object],
    *,
    output: str = "torque_density_nm_kg",
    rel_step: float = 0.05,
    model: Model | None = None,
) -> SensitivityResult:
    """One-at-a-time low/high perturbations of each variable (tornado-chart input).

    ``variables`` may be a sequence of field names (perturbed by ±rel_step of
    the baseline value) or an optimize-style dict: tuple bounds clamp the
    perturbations, choice lists step one option below/above the baseline's
    nearest choice. Integer fields are rounded and de-duplicated. A
    perturbation that produces invalid geometry is skipped with a warning
    (the baseline output stands in for that side).
    """
    model = model or AnalyticalModel()
    baseline_result = model.evaluate(motor, op)
    key = resolve_key(output, baseline_result.to_dict().keys())
    baseline = baseline_result.to_dict()[key]

    specs: dict[str, object | None]
    if isinstance(variables, Mapping):
        specs = dict(variables)
    else:
        specs = {name: None for name in variables}

    entries = []
    for name, spec in specs.items():
        base_value = _get_field(motor, name)
        low_in, high_in = _perturbations(base_value, spec, rel_step)
        if low_in == high_in:
            warnings.warn(f"variable {name!r}: no usable perturbation; skipped", stacklevel=2)
            continue
        low_out = _evaluate_side(model, motor, op, name, low_in, key, baseline)
        high_out = _evaluate_side(model, motor, op, name, high_in, key, baseline)
        entries.append(
            SensitivityEntry(
                variable=name,
                low_input=float(low_in),
                high_input=float(high_in),
                low_output=low_out,
                high_output=high_out,
            )
        )

    entries.sort(key=lambda e: abs(e.swing), reverse=True)
    return SensitivityResult(output=key, baseline=baseline, entries=entries)

axfluxmdo.optimize.dataset ¶

Dataset of evaluated designs (numpy-only; usable without the [opt] extra).

Each record pairs a design vector x (dict over the declared variables) with the flat outputs dict of its evaluation (result.to_dict(), plus any extra keys such as an expensive-FEA objective).

Feature encoding for surrogates is ordinal-as-float: continuous and integer variables are cast to float; Choice variables use the option VALUE when numeric (e.g. pole_pairs — an ordered physical quantity) and the option INDEX otherwise. One-hot encoding for genuinely unordered categoricals is documented future work; the GP's per-dimension ARD length scales absorb the differing column scales.

Persistence is JSON Lines with a versioned header (axfluxmdo-dataset-v1) — dependency-free, append-friendly, git-diffable; these datasets are O(10^2..10^3) rows, so columnar formats buy nothing.