# About
ChemFit is a Python package for concurrent force-field parameter optimization. It can be used with ASE calculators and external executables.
+# Documentation
+
+Please check the **documentation** for details [here](https://chemfit.readthedocs.io).
+
+
# Installation
From PyPi:
@@ -21,18 +26,14 @@ git clone git@github.com:MSallermann/chemfit.git
pip install chemfit
```
-# Documentation
-
-Please check the **documentation** for details [here](https://chemfit.readthedocs.io/en/latest/).
-
# Citation
If you find ChemFit useful and happen to use it in any academic context, please use this reference to cite it:
```
-@misc{sallermann2026chemfitconcurrentframeworkmodel,
- title={ChemFit: A concurrent framework for model parametrization},
- author={Moritz Sallermann and Amrita Goswami and Hannes Jónsson and Elvar Ö. Jónsson and Jorge R. Espinosa},
+@misc{sallermann2026chemfitframeworkautomatedhighdimensional,
+ title={ChemFit: A framework for automated high-dimensional model parameter optimization},
+ author={Moritz Sallermann and Amrita Goswami and Rosana Collepardo-Guevara and Alberto Ocana and Hannes Jónsson and Elvar Ö. Jónsson and Jorge R. Espinosa},
year={2026},
eprint={2603.11769},
archivePrefix={arXiv},
diff --git a/docs/src/images/chemfit_logo.svg b/docs/src/images/chemfit_logo.svg
index 4e463ba..5183b77 100644
--- a/docs/src/images/chemfit_logo.svg
+++ b/docs/src/images/chemfit_logo.svg
@@ -8,7 +8,7 @@
version="1.1"
id="svg1"
xml:space="preserve"
- sodipodi:docname="chemfit_logo.svg"
+ sodipodi:docname="chemfit_logo_portable.svg"
inkscape:version="1.1.2 (0a00cf5339, 2022-02-04)"
inkscape:export-filename="chemfit_logo.pdf"
inkscape:export-xdpi="96"
@@ -28,14 +28,14 @@
inkscape:deskcolor="#d1d1d1"
inkscape:document-units="mm"
inkscape:zoom="2"
- inkscape:cx="-9.75"
- inkscape:cy="151.25"
+ inkscape:cx="59.75"
+ inkscape:cy="142.75"
inkscape:window-width="1920"
inkscape:window-height="1080"
inkscape:window-x="0"
inkscape:window-y="0"
inkscape:window-maximized="1"
- inkscape:current-layer="layer2"
+ inkscape:current-layer="layer1"
inkscape:pageshadow="0"
showgrid="false" />hemFit dict[str, Any]:
- a = parameters["a"]
- b = parameters["b"]
- y_hat = [a * xi + b for xi in self.x]
- residuals = [yh - yt for yh, yt in zip(y_hat, self.y)]
- return {"y_hat": y_hat, "residuals": residuals}
-
-A simple loss function
-----------------------
-
-.. code-block:: python
-
- from typing import Any
-
- def mse_loss(quantities: dict[str, Any]) -> float:
- r = quantities["residuals"]
- return sum(ri * ri for ri in r) / len(r)
-
-Wiring it together as an objective
-----------------------------------
-
-.. code-block:: python
-
- from chemfit.abstract_objective_function import QuantityComputerObjectiveFunction
-
- x = [0.0, 1.0, 2.0, 3.0]
- y = [1.0, 3.1, 4.9, 7.2]
-
- qc = LinearModelComputer(x, y)
- objective = QuantityComputerObjectiveFunction(
- loss_function=mse_loss,
- quantity_computer=qc,
- )
-
- loss = objective({"a": 2.0, "b": 1.0})
- print(loss) # -> a float
-
- # Introspection
- meta = objective.get_meta_data()
- # meta["last_loss"] is the last computed loss
- # meta["computer"]["last"] contains the most recent quantities dict
-
-Loss as an ``ObjectiveFunctor`` (optional)
-------------------------------------------
-
-If your loss needs its own state/metadata, implement it as an :py:class:`~chemfit.abstract_objective_function.ObjectiveFunctor`
-over quantities:
-
-.. code-block:: python
-
- from typing import Any
- from chemfit.abstract_objective_function import ObjectiveFunctor, SupportsGetMetaData
-
- class RobustL1Loss(ObjectiveFunctor):
- def __init__(self):
- self._last: float | None = None
-
- def __call__(self, quantities: dict[str, Any]) -> float:
- r = quantities["residuals"]
- self._last = sum(abs(ri) for ri in r) / len(r)
- return self._last
-
- def get_meta_data(self) -> dict[str, Any]:
- return {"last_loss": self._last}
-
- robust = RobustL1Loss()
- objective = QuantityComputerObjectiveFunction(robust, qc)
- _ = objective({"a": 2.0, "b": 1.0})
- # objective.get_meta_data()["loss_function"] now includes RobustL1Loss metadata.
-
-
-Design Notes & Best Practices
-=============================
-
-- **Keep losses pure** when possible: accept only ``quantities`` and return
- a ``float``. This simplifies testing and reuse.
-- **Use ``QuantityComputer`` to cache** expensive intermediate results. The base
- class already stores the last computed dictionary in metadata.
-- **Validate inputs early** (e.g., check required keys in ``parameters`` and
- ``quantities``) to fail fast during development.
-- **Log via metadata**: expose anything useful for debugging (timings,
- convergence flags, shapes) through ``get_meta_data()``.
-- **Composability**: multiple objectives can wrap the same ``QuantityComputer``
- with different losses, enabling multi-criteria exploration.
diff --git a/docs/src/usage/concepts.rst b/docs/src/usage/concepts.rst
index 59fdca8..6ec0e16 100644
--- a/docs/src/usage/concepts.rst
+++ b/docs/src/usage/concepts.rst
@@ -54,6 +54,7 @@ independent contributions.
-------------------------
.. _concepts_parallel_eval:
+
Independent terms and parallelism
==================================
diff --git a/docs/src/usage/fitter.rst b/docs/src/usage/fitter.rst
index 2204e81..bbcc0df 100644
--- a/docs/src/usage/fitter.rst
+++ b/docs/src/usage/fitter.rst
@@ -29,6 +29,9 @@ ChemFit currently supports two optimization backends:
1. :py:meth:`~chemfit.fitter.Fitter.fit_scipy`
2. :py:meth:`~chemfit.fitter.Fitter.fit_nevergrad`
+Custom optimizers can be connected through the user-driven
+``init``/``ask``/``tell`` interface.
+
Both operate on the same parameter-dictionary interface.
----------------------------------
@@ -251,6 +254,108 @@ ask/tell interface.
optimizer_str="NgIohTuned",
)
+----------------------------------
+User-supplied ask/tell interface
+----------------------------------
+
+An optimizer can be integrated without a dedicated backend. The user owns the
+loop while ChemFit evaluates parameter dictionaries, maintains contexts and
+dispatches its callbacks.
+
+.. code-block:: python
+
+ optimizer = MyOptimizer(initial_params, bounds)
+ fitter.init()
+
+ for _ in range(100):
+ params = optimizer.ask()
+ loss = fitter.ask(params)
+ optimizer.tell(params, loss)
+ fitter.tell() # dispatch registered progress callbacks
+
+ opt_params = fitter.finish(optimizer.recommendation())
+
+``fitter.ask`` also accepts a list of candidates and evaluates it in parallel
+when ``fitter.init`` is configured with ``num_workers`` and an executor. If no
+recommendation is passed to ``finish``, ChemFit returns the best parameters it
+actually evaluated.
+
+A complete custom-loop example
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The optimizer only needs to propose parameter dictionaries and accept their
+losses. For example, this small optimizer searches a predefined set of points:
+
+.. code-block:: python
+
+ class CandidateSearch:
+ def __init__(self, candidates):
+ self.candidates = iter(candidates)
+ self.observations = []
+
+ def ask(self):
+ return next(self.candidates)
+
+ def tell(self, params, loss):
+ self.observations.append((loss, params))
+
+ def recommendation(self):
+ return min(self.observations, key=lambda item: item[0])[1]
+
+
+ def objective(params):
+ return (params["x"] - 2.0) ** 2
+
+
+ fitter = Fitter(objective, initial_params={"x": 0.0})
+ optimizer = CandidateSearch([{"x": 0.0}, {"x": 2.0}, {"x": 4.0}])
+
+ fitter.register_callback(print_progress, n_steps=1)
+ fitter.init()
+
+ for _ in range(3):
+ params = optimizer.ask()
+ loss = fitter.ask(params)
+ optimizer.tell(params, loss)
+ fitter.tell()
+
+ opt_params = fitter.finish(optimizer.recommendation())
+
+Calling ``fitter.ask`` applies the same objective wrapping, invalid-value
+handling and context bookkeeping as the built-in SciPy and Nevergrad fitting
+methods. Calling ``fitter.tell`` marks the end of an optimizer step and invokes
+callbacks registered for that step. ``fitter.finish`` runs the usual post-fit
+checks.
+
+User-driven parallel batches
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Candidates can also be evaluated in batches while the user retains control of
+the optimizer loop:
+
+.. code-block:: python
+
+ from concurrent.futures import ThreadPoolExecutor
+
+ with ThreadPoolExecutor(4) as executor:
+ fitter.init(num_workers=4, executor=executor)
+
+ for _ in range(25):
+ candidates = [optimizer.ask() for _ in range(4)]
+ losses = fitter.ask(candidates)
+
+ for params, loss in zip(candidates, losses):
+ optimizer.tell(params, loss)
+
+ fitter.tell()
+
+ opt_params = fitter.finish(optimizer.recommendation())
+
+One :class:`~chemfit.fitter.FitterEvaluateContext` is maintained per worker.
+The losses returned by ``fitter.ask`` have the same order as the candidate
+list. If no executor is supplied, ``fitter.init(num_workers=...)`` creates and
+later shuts down a thread pool automatically.
+
----------------------------------
Parallel Nevergrad execution
----------------------------------
@@ -264,6 +369,10 @@ Parallel evaluation is supported via ``num_workers``:
num_workers=4,
)
+The evaluation budget is exact. When it is not divisible by ``num_workers``,
+the final batch contains only the remaining candidates. For example,
+``budget=10`` with four workers evaluates batches of four, four and two.
+
Each worker uses its own
:py:class:`~chemfit.fitter.FitterEvaluateContext`.
diff --git a/docs/src/usage/text_file_based_computer.rst b/docs/src/usage/text_file_based_computer.rst
index f25982f..8f69610 100644
--- a/docs/src/usage/text_file_based_computer.rst
+++ b/docs/src/usage/text_file_based_computer.rst
@@ -290,6 +290,32 @@ To inspect failures, you can also enable dump files:
write_dump_file_after_crash=True,
)
+Parsing output from a failed command
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Some external programs return a non-zero exit status after writing usable
+output files. Set ``try_parsing_after_exception=True`` to let the computer
+continue waiting for its configured output files and run the output parsers
+after :py:func:`subprocess.run` raises
+:py:class:`subprocess.CalledProcessError`:
+
+.. code-block:: python
+
+ computer = FileBasedQuantityComputer(
+ ...,
+ try_parsing_after_exception=True,
+ )
+
+The default is ``False``, so a non-zero exit status normally fails the
+evaluation without parsing. When enabled, the evaluation succeeds only if all
+configured output files appear within ``wait_timeout`` and every parser
+succeeds. The subprocess failure is logged as a warning, and dump-file settings
+still apply.
+
+Enable this only when a non-zero exit status is known to leave complete,
+trustworthy output. It can otherwise turn a failed calculation into an
+apparently successful result based on partial files.
+
During execution, useful information is stored in the context, including:
- the working directory
@@ -307,6 +333,9 @@ The constructor exposes additional options:
- ``poll_interval`` - how often file existence is checked
- ``subprocess_run_args`` - arguments passed to ``subprocess.run``
- ``delete_temp_workdirs`` - whether to remove directories after success
+- ``write_dump_file_after_crash`` - whether to write subprocess diagnostics
+- ``keep_temp_workdir_after_crash`` - whether to retain failed work directories
+- ``try_parsing_after_exception`` - whether to parse output after a non-zero exit
Subclassing
diff --git a/logo/chemfit_logo_portable.svg b/logo/chemfit_logo_portable.svg
new file mode 100644
index 0000000..5183b77
--- /dev/null
+++ b/logo/chemfit_logo_portable.svg
@@ -0,0 +1,222 @@
+
+
+
+
diff --git a/pyproject.toml b/pyproject.toml
index f27c6d6..90973c1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,7 +1,7 @@
[project]
name = "chemfit"
description = "A package to support fitting the parameters potentials described by an ASE calculator."
-version = "3.1.0"
+version = "3.1.1"
dependencies = [
"numpy < 2.4.0", # Nevergrad is not compatible with the newest numpy deprecations (see https://github.com/facebookresearch/nevergrad/issues/1707)
"ase",
diff --git a/src/chemfit/file_based_computer.py b/src/chemfit/file_based_computer.py
index 886d69f..333a3be 100644
--- a/src/chemfit/file_based_computer.py
+++ b/src/chemfit/file_based_computer.py
@@ -23,6 +23,12 @@
logger = logging.getLogger(__name__)
+def _subprocess_output_to_text(output: bytes | str) -> str:
+ if isinstance(output, bytes):
+ return output.decode("utf-8", errors="replace")
+ return output
+
+
@runtime_checkable
class OutputParser(Protocol):
"""Protocol for parsing output files into a quantity dictionary."""
@@ -75,6 +81,7 @@ def __init__(
delete_temp_workdirs: bool = True,
write_dump_file_after_crash: bool = True,
keep_temp_workdir_after_crash: bool = True,
+ try_parsing_after_exception: bool = False,
):
"""
Initialize a file-based quantity computer.
@@ -122,7 +129,9 @@ def __init__(
subprocess output when command execution fails.
keep_temp_workdir_after_crash: Whether to keep the temporary
working directory for inspection after a failed evaluation.
-
+ try_parsing_after_exception: Whether to continue waiting for and parsing
+ output files when ``subprocess.run`` raises
+ ``subprocess.CalledProcessError``. Defaults to False.
Raises:
Exception: If any path in `output_files` is absolute rather
@@ -136,6 +145,7 @@ def __init__(
self.base_working_directory = Path(base_working_directory)
self.write_dump_file_after_crash = write_dump_file_after_crash
self.keep_temp_workdir_after_crash = keep_temp_workdir_after_crash
+ self.try_parsing_after_exception = try_parsing_after_exception
# We need to make sure none of the output files is absolute.
# The reason for this is that, to facilitate multiple concurrent evaluations,
@@ -464,7 +474,8 @@ def _compute( # noqa: PLR0912, PLR0915
)
if e.stderr is not None:
- msg += f" stderr (if captured) = {e.stderr.decode('utf-8')}\n"
+ stderr = _subprocess_output_to_text(e.stderr)
+ msg += f" stderr (if captured) = {stderr}\n"
# Try to write a dump file
if self.write_dump_file_after_crash:
@@ -475,16 +486,25 @@ def _compute( # noqa: PLR0912, PLR0915
with dump_path.open("w") as f:
if e.stderr is not None:
f.write("Stderr:\n")
- f.write(e.stderr.decode("utf-8"))
+ f.write(_subprocess_output_to_text(e.stderr))
if e.stdout is not None:
f.write("Stdout:\n")
- f.write(e.stdout.decode("utf-8"))
+ f.write(_subprocess_output_to_text(e.stdout))
f.write("ctx.temp:\n")
f.write(f"{ctx.temp}")
msg += f"\nWrote dump file to `{dump_path}`."
except Exception as exc_dump:
msg += f"\nCould not write dump file to `{dump_path}`, because of {exc_dump}."
- raise Exception(msg) from e
+
+ msg += f"\n`{self.try_parsing_after_exception = }`."
+
+ if self.try_parsing_after_exception:
+ msg += "\nWill attempt to parse output files."
+ logger.warning(msg)
+ else:
+ stop.set()
+ watcher.join(timeout=1)
+ raise Exception(msg) from e
# Block here until file appears (or timeout)
# The main reason to implement this extra check is to eventually support remote execution, e.g. on clusters
diff --git a/src/chemfit/fitter.py b/src/chemfit/fitter.py
index b44c1f7..6a4e592 100644
--- a/src/chemfit/fitter.py
+++ b/src/chemfit/fitter.py
@@ -301,7 +301,119 @@ def _hook_post_fit(self, opt_params: dict[str, Any]):
f" parameter = {kp}, lower = {lower}, value = {vp}, upper = {upper}"
)
- def fit_nevergrad( # noqa: PLR0912, PLR0915
+ def init(
+ self,
+ num_workers: int = 1,
+ contexts: list[FitterEvaluateContext] | None = None,
+ executor: ExecutorLike | None = None,
+ ) -> None:
+ """
+ Initialize a user-driven optimization session.
+
+ After initialization the user owns the optimization loop: obtain
+ candidates from an optimizer, pass them to :meth:`ask`, feed the
+ returned losses back to the optimizer, and call :meth:`tell` once per
+ optimizer step. Call :meth:`finish` with the optimizer's final
+ recommendation when the loop is complete.
+ """
+
+ if num_workers < 1:
+ msg = "num_workers must be at least 1"
+ raise ValueError(msg)
+ if contexts is not None and len(contexts) != num_workers:
+ msg = "contexts must contain one context per worker"
+ raise ValueError(msg)
+
+ self._user_owns_executor = num_workers != 1 and executor is None
+ if self._user_owns_executor:
+ executor = ThreadPoolExecutor(num_workers)
+
+ self._hook_pre_fit()
+ self._user_executor = executor
+ self._user_num_workers = num_workers
+ self._user_step = 0
+ self.contexts = (
+ [FitterEvaluateContext() for _ in range(num_workers)]
+ if contexts is None
+ else contexts
+ )
+
+ def ask(
+ self,
+ parameters: dict[str, Any] | list[dict[str, Any]],
+ context_index: int = 0,
+ ) -> float | list[float]:
+ """
+ Evaluate one candidate or a parallel batch proposed by the user.
+
+ A dictionary produces one loss. A list produces a list of losses in
+ input order and may contain at most ``num_workers`` candidates.
+ """
+
+ if not hasattr(self, "_user_num_workers"):
+ msg = "call fitter.init() before fitter.ask()"
+ raise RuntimeError(msg)
+
+ if isinstance(parameters, dict):
+ return self.objective_function(parameters, self.contexts[context_index])
+
+ if len(parameters) > self._user_num_workers:
+ msg = "a batch cannot contain more candidates than workers"
+ raise ValueError(msg)
+ if len(parameters) == 0:
+ return []
+ if len(parameters) == 1:
+ return [self.objective_function(parameters[0], self.contexts[0])]
+
+ if self._user_executor is None:
+ msg = "parallel evaluation requires an executor"
+ raise RuntimeError(msg)
+ return map_with_context(
+ self._user_executor,
+ self.objective_function,
+ parameters,
+ ctxs=self.contexts[: len(parameters)],
+ )
+
+ def tell(self, step: int | None = None) -> None:
+ """
+ Notify ChemFit that the user completed an optimizer step.
+
+ This dispatches registered fitter callbacks. If ``step`` is omitted,
+ an internal zero-based step counter is used and advanced automatically.
+ """
+
+ if not hasattr(self, "_user_step"):
+ msg = "call fitter.init() before fitter.tell()"
+ raise RuntimeError(msg)
+ current_step = self._user_step if step is None else step
+ callback, n_steps = self._unify_callbacks()
+ if callback is not None and current_step % n_steps == 0:
+ callback(current_step, self.contexts)
+ self._user_step = current_step + 1
+
+ def finish(self, opt_params: dict[str, Any] | None = None) -> dict[str, Any]:
+ """
+ Finalize a user-driven session and return its chosen parameters.
+
+ When no recommendation is supplied, the best candidate evaluated by
+ ChemFit is used.
+ """
+
+ if opt_params is None:
+ evaluated = [ctx for ctx in self.contexts if ctx.opt_loss is not None]
+ if not evaluated:
+ msg = "cannot finish before evaluating a candidate"
+ raise RuntimeError(msg)
+ best_context = min(evaluated, key=lambda ctx: cast("float", ctx.opt_loss))
+ assert best_context.opt_params is not None
+ opt_params = dict(best_context.opt_params)
+ self._hook_post_fit(opt_params)
+ if self._user_owns_executor:
+ cast("ThreadPoolExecutor", self._user_executor).shutdown()
+ return opt_params
+
+ def fit_nevergrad(
self,
budget: int,
optimizer_str: str = "NgIohTuned",
@@ -348,7 +460,7 @@ def fit_nevergrad( # noqa: PLR0912, PLR0915
Raises:
KeyError: If ``optimizer_str`` is not found in the nevergrad
optimizer registry.
- AssertionError: If ``contexts`` is provided and its length does
+ ValueError: If ``contexts`` is provided and its length does
not equal ``num_workers``.
Side Effects:
@@ -359,11 +471,6 @@ def fit_nevergrad( # noqa: PLR0912, PLR0915
"""
- if num_workers != 1 and executor is None:
- executor = ThreadPoolExecutor(num_workers)
-
- self._hook_pre_fit()
-
flat_bounds = flatten_dict(self.bounds)
flat_initial_params = flatten_dict(self.initial_parameters)
@@ -378,28 +485,18 @@ def fit_nevergrad( # noqa: PLR0912, PLR0915
try:
OptimizerCls = ng.optimizers.registry[optimizer_str]
except KeyError as e:
- e.add_note(f"Available solvers: {list(ng.optimizers.registry.keys())}")
- raise e
+ available_solvers = list(ng.optimizers.registry.keys())
+ msg = (
+ f"Unknown nevergrad optimizer {optimizer_str!r}. "
+ f"Available solvers: {available_solvers}"
+ )
+ raise KeyError(msg) from e
optimizer = OptimizerCls(
parametrization=instru, budget=budget, num_workers=num_workers
)
- def f_ng(parameters: dict[str, Any], ctx: FitterEvaluateContext) -> float:
- params = unflatten_dict(parameters, dict_factory=dict)
- return self.objective_function(params, ctx)
-
- callback, n_steps = self._unify_callbacks()
-
- # We need one context per worker
- if contexts is None:
- self.contexts = [FitterEvaluateContext() for _ in range(num_workers)]
- else:
- assert len(contexts) == num_workers
- self.contexts = contexts
-
- # After the if statements we know that we have a list of FitterEvaluateContexts and not None
- self.contexts = cast("list[FitterEvaluateContext]", self.contexts)
+ self.init(num_workers=num_workers, contexts=contexts, executor=executor)
# This applies restart parameters, **if** they are within the bounds
if initial_observations is not None:
@@ -430,31 +527,31 @@ def f_ng(parameters: dict[str, Any], ctx: FitterEvaluateContext) -> float:
)
optimizer.tell(asked_params, post_processed_loss_value)
- for step in range(budget // num_workers):
+ for step, batch_start in enumerate(range(0, budget, num_workers)):
+ batch_size = min(num_workers, budget - batch_start)
+
# On the first evaluation we ensure that the optimizer suggests the initial params
if step == 0:
optimizer.suggest(flat_initial_params)
- # Ask for num_workers parameters to evaluate in parallel
- asked_params = [optimizer.ask() for _ in range(num_workers)]
+ # The final batch may be smaller than num_workers when the budget
+ # is not evenly divisible by the worker count.
+ asked_params = [optimizer.ask() for _ in range(batch_size)]
flat_params = [p.value[0][0] for p in asked_params]
-
- if num_workers == 1:
- losses = [f_ng(flat_params[0], self.contexts[0])]
- else:
- assert executor is not None
- assert self.contexts is not None
- losses = map_with_context(
- executor, f_ng, flat_params, ctxs=self.contexts
- )
+ nested_params = [
+ unflatten_dict(params, dict_factory=dict[str, Any])
+ for params in flat_params
+ ]
+ asked_losses = self.ask(nested_params)
+ assert isinstance(asked_losses, list)
+ losses = asked_losses
[
optimizer.tell(params, loss)
for params, loss in zip(asked_params, losses, strict=True)
]
- if callback is not None and step % n_steps == 0:
- callback(step, self.contexts)
+ self.tell(step)
recommendation = optimizer.provide_recommendation()
@@ -464,9 +561,7 @@ def f_ng(parameters: dict[str, Any], ctx: FitterEvaluateContext) -> float:
opt_params = unflatten_dict(flat_opt_params, dict_factory=dict[str, Any])
- self._hook_post_fit(opt_params)
-
- return opt_params
+ return self.finish(opt_params)
def fit_scipy(
self,
@@ -504,8 +599,6 @@ def fit_scipy(
"""
- self._hook_pre_fit()
-
# Scipy expects a function with n real-valued parameters f(x)
# but our objective function takes a dictionary of parameters.
# Moreover, the dictionary might not be flat but nested.
@@ -526,10 +619,7 @@ def fit_scipy(
bounds = np.array([flat_bounds.get(k, (None, None)) for k in self._keys])
# Since we know that scipy.optimize works synchronously, we create a single context, which we'll keep alive.
- if ctx is None:
- self.contexts = [FitterEvaluateContext()]
- else:
- self.contexts = [ctx]
+ self.init(contexts=None if ctx is None else [ctx])
# The local objective function first creates a flat dictionary from the `x` array
# by zipping it with the captured flattened keys and then unflattens the dictionary
@@ -537,11 +627,9 @@ def fit_scipy(
def f_scipy(x: npt.NDArray) -> float:
p = unflatten_dict(dict(zip(self._keys, x)), dict_factory=dict[str, Any])
cast("dict[str, Any]", p)
- assert self.contexts is not None
- return self.objective_function(p, ctx=self.contexts[0])
-
- # First concatenate the list of callbacks into a single function
- callback, n_steps = self._unify_callbacks()
+ loss = self.ask(p)
+ assert isinstance(loss, float)
+ return loss
def callback_scipy(intermediate_result: OptimizeResult):
if "nit" in intermediate_result:
@@ -549,8 +637,7 @@ def callback_scipy(intermediate_result: OptimizeResult):
else:
step = self.contexts[0].n_evals
- if callback is not None and step % n_steps == 0:
- callback(step, self.contexts)
+ self.tell(step)
res = minimize(
f_scipy, x0, method=method, bounds=bounds, **kwargs, callback=callback_scipy
@@ -563,6 +650,4 @@ def callback_scipy(intermediate_result: OptimizeResult):
opt_params = unflatten_dict(opt_params)
- self._hook_post_fit(opt_params)
-
- return opt_params
+ return self.finish(opt_params)
diff --git a/tests/test_file_based_computer.py b/tests/test_file_based_computer.py
index 6ab7629..7d53321 100644
--- a/tests/test_file_based_computer.py
+++ b/tests/test_file_based_computer.py
@@ -1,8 +1,13 @@
-from collections.abc import Iterable
+from __future__ import annotations
+
+import logging
+import subprocess
+import sys
from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any, NoReturn
import numpy as np
+import pytest
from chemfit.abstract_objective_function import (
EvaluateContext,
@@ -10,6 +15,9 @@
from chemfit.file_based_computer import FileBasedQuantityComputer
from chemfit.fitter import Fitter
+if TYPE_CHECKING:
+ from collections.abc import Iterable
+
class MyOutputParser:
def __call__(self, output_files: list[Path]) -> dict[str, Any]:
@@ -53,7 +61,12 @@ def callable_cmd(
script_file: Path,
output_file: Path,
) -> list[str]:
- return f"python {script_file} {parameters['prefactor']} {workdir / output_file}".split()
+ return [
+ sys.executable,
+ str(script_file),
+ str(parameters["prefactor"]),
+ str(workdir / output_file),
+ ]
output_parser = MyOutputParser()
@@ -80,9 +93,91 @@ def callable_cmd(
assert np.isclose(opt_params["prefactor"], 2.0)
-if __name__ == "__main__":
- import logging
+def test_try_parsing_after_subprocess_exception(
+ tmp_path: Path,
+ monkeypatch: pytest.MonkeyPatch,
+ caplog: pytest.LogCaptureFixture,
+):
+ output_file = Path("result.txt")
+
+ def failing_run(
+ cmd: list[str], *, check: bool, cwd: Path | str, **_kwargs: Any
+ ) -> NoReturn:
+ assert check
+ (Path(cwd) / output_file).write_text("42", encoding="utf-8")
+ raise subprocess.CalledProcessError(
+ returncode=1,
+ cmd=cmd,
+ output="partial output",
+ stderr="expected failure",
+ )
+
+ def parse_output(output_files: list[Path]) -> dict[str, int]:
+ return {"result": int(output_files[0].read_text(encoding="utf-8"))}
+
+ monkeypatch.setattr(subprocess, "run", failing_run)
+ computer = FileBasedQuantityComputer(
+ output_files=[output_file],
+ output_parsers=parse_output,
+ base_working_directory=tmp_path,
+ subprocess_run_args={},
+ try_parsing_after_exception=True,
+ ).with_cmd(lambda _parameters, _workdir: ["failing-command"])
+ ctx = EvaluateContext()
+ with caplog.at_level(logging.WARNING):
+ result = computer({}, ctx)
+
+ assert result == {"result": 42}
+ assert not ctx.temp.workdir.exists()
+ assert "Will attempt to parse output files." in caplog.text
+ dump_files = list(tmp_path.glob("*.dump"))
+ assert len(dump_files) == 1
+ assert "expected failure" in dump_files[0].read_text(encoding="utf-8")
+
+
+def test_subprocess_exception_does_not_parse_by_default(
+ tmp_path: Path,
+ monkeypatch: pytest.MonkeyPatch,
+):
+ output_file = Path("result.txt")
+ parser_called = False
+
+ def failing_run(
+ cmd: list[str], *, check: bool, cwd: Path | str, **_kwargs: Any
+ ) -> NoReturn:
+ assert check
+ (Path(cwd) / output_file).write_text("42", encoding="utf-8")
+ raise subprocess.CalledProcessError(returncode=1, cmd=cmd)
+
+ def parse_output(output_files: list[Path]) -> dict[str, int]:
+ nonlocal parser_called
+ assert output_files
+ parser_called = True
+ return {"result": 42}
+
+ monkeypatch.setattr(subprocess, "run", failing_run)
+ computer = FileBasedQuantityComputer(
+ output_files=[output_file],
+ output_parsers=parse_output,
+ base_working_directory=tmp_path,
+ subprocess_run_args={},
+ delete_temp_workdirs=True,
+ keep_temp_workdir_after_crash=False,
+ write_dump_file_after_crash=False,
+ ).with_cmd(lambda _parameters, _workdir: ["failing-command"])
+ ctx = EvaluateContext()
+
+ with pytest.raises(Exception, match="Exception in `_compute`") as exc_info:
+ computer({}, ctx)
+
+ subprocess_exception = exc_info.value.__cause__.__cause__
+ assert isinstance(subprocess_exception, subprocess.CalledProcessError)
+ assert not parser_called
+ assert not ctx.temp.workdir.exists()
+
+
+if __name__ == "__main__":
logging.basicConfig(filename="test_file_based.log")
test_squares_file_based()
diff --git a/tests/test_fitter.py b/tests/test_fitter.py
index 487b288..b7a0f75 100644
--- a/tests/test_fitter.py
+++ b/tests/test_fitter.py
@@ -331,5 +331,56 @@ def objective(params: dict[str, float]) -> float:
assert 0.0 <= opt_params["x"] <= 5.0
+def test_nevergrad_evaluates_partial_final_batch():
+ n_calls = 0
+
+ def objective(params: dict[str, float]) -> float:
+ nonlocal n_calls
+ n_calls += 1
+ return params["x"] ** 2
+
+ fitter = Fitter(objective, initial_params={"x": 1.0})
+ fitter.fit_nevergrad(budget=3, num_workers=2)
+
+ assert n_calls == 3
+
+
+def test_user_supplied_ask_tell_interface():
+ candidates = iter([{"x": 0.0}, {"x": 2.0}, {"x": 4.0}])
+ observations = []
+
+ fitter = Fitter(lambda params: (params["x"] - 2.0) ** 2, {"x": 0.0})
+ fitter.init()
+ for params in candidates:
+ loss = fitter.ask(params)
+ observations.append((params, loss))
+ fitter.tell()
+ result = fitter.finish()
+
+ assert result == {"x": 2.0}
+ assert observations == [
+ ({"x": 0.0}, 4.0),
+ ({"x": 2.0}, 0.0),
+ ({"x": 4.0}, 4.0),
+ ]
+ assert fitter.contexts[0].n_evals == 3
+
+
+def test_user_supplied_ask_tell_recommendation_and_partial_batch():
+ fitter = Fitter(lambda params: params["x"] ** 2, {"x": 0.0})
+
+ with ThreadPoolExecutor(2) as executor:
+ fitter.init(num_workers=2, executor=executor)
+ losses = fitter.ask([{"x": 0.0}, {"x": 1.0}])
+ fitter.tell()
+ final_loss = fitter.ask([{"x": 2.0}])
+ fitter.tell()
+ result = fitter.finish({"x": 0.5})
+
+ assert losses == [0.0, 1.0]
+ assert final_loss == [4.0]
+ assert result == {"x": 0.5}
+
+
if __name__ == "__main__":
test_with_square_func()
+ 