Optimize¶

def to_config(self, point: dict[str, float]) -> SimulationConfig:
    """Build the [`SimulationConfig`][femforge.config.SimulationConfig] for one parameter-space ``point`` (knob name -> value), leaving the base untouched.

    The point's keys must be exactly the knob names; a missing or unknown key raises.
    """
    names: set[str] = {knob.name for knob in self.knobs}
    missing: set[str] = names - point.keys()
    unknown: set[str] = point.keys() - names
    if missing or unknown:
        raise ValueError(f"point keys {sorted(point)} do not match knobs {sorted(names)} (missing {sorted(missing)}, unknown {sorted(unknown)})")

    config: SimulationConfig = self.base
    for knob in self.knobs:
        config = knob.apply(config, point[knob.name])

    return config

def space(self) -> dict[str, ParamSpec]:
    """The free variables keyed by name, each with its declared bounds — the schema a search walks (or a future optimizer samples)."""
    return {knob.name: knob.spec() for knob in self.knobs}

def spec(self) -> ParamSpec:
    """The declared range of this knob, for [`Parameterization.space`][]."""
    return ParamSpec(low=self.low, high=self.high)

def apply(self, config: SimulationConfig, value: float) -> SimulationConfig:
    """Return a copy of ``config`` with this knob set to ``value``; implementations must not mutate ``config`` (the base config is shared across every grid point). The base raises."""
    raise NotImplementedError

def apply(self, config: SimulationConfig, value: float) -> SimulationConfig:
    """A copy of ``config`` with ``global_mesh_size`` set to ``value``."""
    mesh = config.mesh.model_copy(update={"global_mesh_size": value})
    return config.model_copy(update={"mesh": mesh})

def apply(self, config: SimulationConfig, value: float) -> SimulationConfig:
    """A copy of ``config`` with the ``mesh_size`` of the region named ``tag`` set to ``value``."""
    if not any(tag.name == self.tag for tag in config.domain.tags):
        raise ValueError(f"no region tagged {self.tag!r} in the base config")
    # Replace only the matching tag; other tags are shared unchanged — apply must not mutate the base.
    tags = [tag.model_copy(update={"mesh_size": value}) if tag.name == self.tag else tag for tag in config.domain.tags]

    domain = config.domain.model_copy(update={"tags": tags})
    return config.model_copy(update={"domain": domain})

def record(self, *, point: dict[str, float], run_id: str, result: Result, objective: float) -> StudyRecord:
    """Append and return a [`StudyRecord`][] for one evaluated point.

    Args:
        point: The parameter-space coordinate, by knob name.
        run_id: The run bundle's content-addressed id.
        result: The run's typed result.
        objective: The run's objective scalar.
    """
    entry: StudyRecord = StudyRecord(point=point, run_id=run_id, result=result, objective=objective)
    self.records.append(entry)
    return entry

def best(self, *, maximize: bool = True) -> StudyRecord:
    """The record with the extremal objective: argmax by default, argmin when ``maximize=False``."""
    if not self.records:
        raise ValueError("study has no records")
    chooser = max if maximize else min
    return chooser(self.records, key=lambda entry: entry.objective)

def to_dataframe(self) -> pd.DataFrame:
    """Flatten the records to a pandas DataFrame, the units-stripping analysis boundary.

    One row per record: the knob columns, then ``objective``, ``run_id``, ``status`` and the result quantities as bare SI floats.
    """
    import pandas as pd

    rows: list[dict[str, object]] = []
    for entry in self.records:
        row: dict[str, object] = dict(entry.point)
        row["run_id"] = entry.run_id
        row["objective"] = entry.objective
        row["status"] = entry.result.status

        force = entry.result.force
        row["force"] = force.to(U.N).magnitude.magnitude() if force is not None else None
        row["force_x"] = force.to(U.N).magnitude.x if force is not None else None
        row["force_y"] = force.to(U.N).magnitude.y if force is not None else None
        row["force_z"] = force.to(U.N).magnitude.z if force is not None else None

        torque = entry.result.torque
        row["torque"] = torque.to(U.N * U.m).magnitude.magnitude() if torque is not None else None

        flux_density_max = entry.result.flux_density_max
        row["flux_density_max"] = flux_density_max.to(U.T).magnitude if flux_density_max is not None else None

        row["residual"] = entry.result.residual
        row["wall_time"] = entry.result.wall_time
        rows.append(row)
    return pd.DataFrame(rows)

def run(self, driver: DriverLike) -> Study:
    """Evaluate every grid point and return the populated [`Study`][femforge.optimize.study.Study].

    Each record keeps the ``run_id`` the driver stamped, recomputed with [`run.cache_key`][femforge.run.cache_key] when a stand-in stamps none.
    """
    study: Study = Study()
    names: list[str] = list(self.grid.keys())

    for combo in itertools.product(*(self.grid[name] for name in names)):
        point: dict[str, float] = dict(zip(names, combo))
        config: SimulationConfig = self.parameterization.to_config(point)
        result: Result = driver.evaluate(config)

        run_id: str = result.run_id if result.run_id is not None else bundle_name(cache_key(config))
        study.record(
            point=point,
            run_id=run_id,
            result=result,
            objective=self.objective(result),
        )

    return study

def evaluate(self, config: SimulationConfig) -> Result:
    """Solve ``config`` and return its result."""
    ...