diff --git a/.agents/skills/cli_executable_sweep/SKILL.md b/.agents/skills/cli_executable_sweep/SKILL.md index 5cf4ae5..0159d05 100644 --- a/.agents/skills/cli_executable_sweep/SKILL.md +++ b/.agents/skills/cli_executable_sweep/SKILL.md @@ -58,4 +58,3 @@ Executables often involve complex MPI environments or library path dependencies - **Absolute Paths**: For the `--app` flag, it is often safer to use an absolute path (`$(pwd)/my_binary`) or a clear relative path (`./my_binary`). - **Dry-Run first**: Use `--dry-run` to check the grid size before launching parallel processes. - **Save results**: Always use one of the `--save-*` flags to ensure the final data is persisted in a format you can easily parse later. -- **Remote Execution**: You can pass `--globus-compute-endpoint ` to execute the sequence remotely. However, for executables with complex file dependencies, prefer using the `programmatic_sweep_api` skill to handle data movement explicitly. diff --git a/.agents/skills/cli_python_sweep/SKILL.md b/.agents/skills/cli_python_sweep/SKILL.md index f10b742..8a3e375 100644 --- a/.agents/skills/cli_python_sweep/SKILL.md +++ b/.agents/skills/cli_python_sweep/SKILL.md @@ -51,4 +51,3 @@ ensemblesweep py --func module_name.function_name \ - `key=val1,val2`: Discrete list. - `key=start:stop:num`: Linear range (`np.linspace`). - **Imports**: Ensure you are in the same directory as the target module, as `ensemblesweep` automatically identifies the current working directory. -- **Remote Execution**: You can pass `--globus-compute-endpoint ` to execute the grid remotely. However, for complex environments/dependencies, prefer using the `programmatic_sweep_api` skill to handle data placement and imports correctly. diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..fcd9da6 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,224 @@ +# AGENTS.md + +Guidance for coding agents working in this repository. This file is intentionally more operational than the README: it captures architecture, current design decisions, and pitfalls that are easy to miss. + +## Project overview + +`ensemblesweep` is a small Python package for parameter sweeps backed by [libEnsemble](https://libensemble.readthedocs.io/). It supports: + +- Sweeping Python objective functions. +- Sweeping external executables. +- A simple `Sweep` API. +- A `concurrent.futures`-style `SweepExecutor` API. +- A Click-based CLI. + +The package source lives under `src/ensemblesweep/`. + +## Important files + +- `src/ensemblesweep/data.py` + - Defines `Data`. + - Builds the Cartesian product of input parameters. + - Converts parameter combinations into libEnsemble `H0` arrays via `to_h0()`. + +- `src/ensemblesweep/sweep.py` + - Defines `Sweep`. + - Owns libEnsemble setup and execution. + - Keeps the raw libEnsemble history in `self._H_total`. + - Tracks completed count in `self.evaluated`. + - Exposes `self.results` as a `SweepResults` view over all completed rows. + +- `src/ensemblesweep/results.py` + - Defines canonical result types: + - `SweepResult`: one completed evaluation. + - `SweepResults`: collection view over completed rows or a submitted batch. + - This replaces the older `ResultWrapper` concept. Do not reintroduce a parallel result wrapper unless there is a strong reason. + +- `src/ensemblesweep/executor.py` + - Defines `SweepExecutor`. + - Provides a `concurrent.futures`-style interface. + - `submit_sweep(...)` creates a `Sweep` and returns a single `Future`. + - `submit(sweep, n=None)` submits an existing `Sweep`, optionally for a partial batch. + +- `src/ensemblesweep/sim_funcs.py` + - libEnsemble simulation functions for Python callables and external executables. + +- `src/ensemblesweep/cli.py` + - Click CLI entry points: `ensemblesweep py ...` and `ensemblesweep exe ...`. + +- `README.md` + - User-facing examples and API description. + +- `.agents/skills/` + - Agent skills for using the project as a tool. Keep these in sync when user-facing CLI/API behavior changes. + +## Current API design decisions + +### libEnsemble remains the execution engine + +Do not replace libEnsemble with raw `concurrent.futures` for point-level execution. The desired architecture is: + +```text +User API -> Sweep/SweepExecutor -> libEnsemble -> objective function/executable +``` + +libEnsemble is responsible for evaluating individual parameter points concurrently. + +### `SweepExecutor` returns one future per libEnsemble run + +A libEnsemble run is blocking at the batch level. The executor API intentionally returns one `Future` for a whole sweep or partial batch, not one future per parameter point. + +Example: + +```python +with SweepExecutor(nworkers=4) as executor: + future = executor.submit_sweep(objective_function=f, input_data=data) + results = future.result() # SweepResults for this submitted batch +``` + +This means `as_completed(...)` is only useful across multiple submitted sweep/batch futures, not for individual points within one batch. + +### Do not add multiple parallel libEnsemble instances by default + +`SweepExecutor` currently uses an internal `ThreadPoolExecutor(max_workers=1)`. This is intentional. It avoids running multiple libEnsemble managers at the same time. + +If changing this, discuss the implications first: output directory conflicts, resource oversubscription, MPI/local comms behavior, and unclear semantics around `nworkers`. + +### Use `nworkers`, not `max_workers`, in public sweep APIs + +`nworkers` maps to libEnsemble workers. Avoid `max_workers` in public `SweepExecutor`/`Sweep` APIs because it is ambiguous with Python's executor worker count. + +### Results are canonicalized through `SweepResult` and `SweepResults` + +Use: + +```python +sweep.results # SweepResults over all completed rows +future.result() # SweepResults over only the submitted batch +``` + +A `SweepResult` has: + +- `index` +- `params` +- `outputs` +- `eval_time` +- `status` + +`SweepResults.to_numpy()` returns the raw completed libEnsemble rows for that view. `SweepResults.to_pandas()` returns a flattened DataFrame with `index`, params, outputs, `eval_time`, and `status`. + +Do not assume the string/repr form is the stable serialization format. Prefer `to_pandas()`, `to_numpy()`, or explicit iteration. + +## Execution model notes + +`Sweep.run(n=None)`: + +1. Determines how many total points remain. +2. Caps by `n` if supplied. +3. Sets libEnsemble `ExitCriteria(sim_max=target_sim_max)`. +4. Runs libEnsemble with local comms. +5. On the manager, stores `ensemble.H` back into `self._H_total` and updates `self.evaluated`. + +Default worker count is: + +```python +max(1, os.cpu_count() - 1) +``` + +unless `nworkers` is provided. + +## Common pitfalls + +### Do not validate Python objective functions with `python -c` + +libEnsemble worker processes may need to import the objective function. A function defined in `python -c` or other non-importable `__main__` contexts can fail with errors like: + +```text +AttributeError: module '__main__' has no attribute 'f' +``` + +Use a real script with an importable top-level function and an `if __name__ == "__main__":` guard. + +Good validation pattern: + +```python +from ensemblesweep import Data, SweepExecutor + + +def f(x, y): + return x * y + + +if __name__ == "__main__": + data = Data(x=[1, 2], y=[10, 20]) + with SweepExecutor(nworkers=2) as executor: + future = executor.submit_sweep(objective_function=f, input_data=data) + results = future.result() + print(len(results)) + print(results[0]) +``` + +### libEnsemble creates workflow/output directories + +Validation runs may create directories such as `sweep_/` and libEnsemble log/stat files. Clean up generated artifacts after tests when appropriate. Do not delete user-created data. + +### Batch-only executor results depend on completed-row ordering + +`SweepExecutor` currently captures `previously_evaluated` before `sweep.run(n)` and returns: + +```python +SweepResults(sweep, start_index=previously_evaluated) +``` + +This assumes the completed rows view is ordered compatibly with evaluation progress. If future changes allow cancellation, retries, out-of-order histories, or more complex libEnsemble allocation, consider selecting batch rows by explicit `sim_id`/row identity rather than by completed-row slice. + +### Objective output fields + +For Python objective functions: + +- If `objective_output` is omitted, output field is `"output"`. +- If `objective_output` is a string, it is normalized to a single-item list. +- Output dtype is currently float for each output field. + +For executables: + +- The executable receives `Data` parameters as positional command-line arguments in key order. +- Output is read from `objective_output` if supplied, otherwise from the executable stdout path used by `sim_funcs.py`. + +### `Data` dtype inference is simple + +`Data` infers dtype from the first value for each parameter. Strings use `U100`; unknown types fall back to `object`. Be careful when changing dtype behavior because libEnsemble structured arrays depend on these dtypes. + +### Keep CLI and README in sync with API changes + +If changing constructor arguments, result shapes, worker options, or supported execution modes, update: + +- `README.md` +- `src/ensemblesweep/cli.py` +- `.agents/skills/*/SKILL.md` when relevant +- `src/ensemblesweep/__init__.py` exports when adding/removing public classes + +## Development and validation commands + +This project uses Pixi. Useful commands: + +```bash +pixi run python -m compileall src/ensemblesweep +``` + +For linting, use the dev environment if needed: + +```bash +pixi run -e dev ruff check src +``` + +There may not be a full automated test suite yet. Prefer targeted real-script validations for libEnsemble behavior, especially for multiprocessing/importability-sensitive paths. + +## Style and change guidance + +- Make small, focused changes. +- Preserve libEnsemble as the core engine. +- Avoid broad rewrites of result semantics without updating all user-facing examples. +- Prefer existing dependencies and patterns. +- Do not run or leave long-lived servers/watchers. +- Do not commit changes unless explicitly asked. diff --git a/README.md b/README.md index ebd3a85..7fb8e2d 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # ensemblesweep -``ensemblesweep`` is a small library based on [libEnsemble](https://libensemble.readthedocs.io/en/latest/) +``ensemblesweep`` is a small library based on [libEnsemble](https://libensemble.readthedocs.io/en/latest/) for parallel parameter sweeps of objective functions or executables. Installation @@ -59,7 +59,7 @@ Executables - The ``Data`` parameters will be passed as arguments to the executable in the order they are defined. - The output of the executable will be read: - from the file specified by ``objective_output`` (if provided). - - or, if ``objective_output`` is not specified, the last line of the executable's stdout. + - or, if not provided, the last line of the executable's stdout. ```python @@ -101,6 +101,107 @@ Sweep an executable: ensemblesweep exe --app ./sim.x --var "m=1,2,3" --out-file out.stat ``` +Concurrent Futures Interface +---------------------------- + +A ``concurrent.futures``-style API. A libEnsemble run blocks, so ``submit_sweep`` returns one future for a whole batch. Individual parameter points are still evaluated concurrently. + +```python +from ensemblesweep import Data, SweepExecutor + + +data = Data(x=[1, 2, 3], y=[10, 20]) + + +def my_function(x, y): + return x * y + + +with SweepExecutor(nworkers=4) as executor: + + future = executor.submit_sweep( + objective_function=my_function, + input_data=data, + ) + + batch = future.result() + +for result in batch: + print(result.params, result.outputs, result.eval_time) +``` + +You can also submit an existing ``Sweep`` object, including a partial batch: + +```python +from ensemblesweep import Sweep, SweepExecutor + + +sweep = Sweep(objective_function=my_function, input_data=data) + +with SweepExecutor(nworkers=4) as executor: + future = executor.submit(sweep, n=10) + batch = future.result() +``` + +Results +------- + +``SweepResults`` is a collection of ``SweepResult`` objects. + +Each ``SweepResult`` has: +- ``index`` +- ``params`` (the input parameters) +- ``outputs`` (the objective function's return values) +- ``eval_time`` +- ``status`` + +Both the simple ``Sweep`` API and the ``concurrent.futures`` API return ``SweepResults``: + +```python +# Simple API +sweep = Sweep(objective_function=my_function, input_data=data) +sweep.run() +results = sweep.results # SweepResults (all completed) + +# Executor API +with SweepExecutor(nworkers=4) as executor: + future = executor.submit(sweep, n=10) + results = future.result() # SweepResults (this batch only) +``` + +Iterate over ``SweepResult`` objects: + +```python +for result in results: + print(result.index, result.params, result.outputs, result.eval_time) +``` + +Index into results: + +```python +print(results[0]) # single SweepResult +print(results[:3]) # list of SweepResult +``` + +Export: + +```python +results.to_numpy() # raw libEnsemble array +results.to_pandas() # DataFrame with index + params + outputs +``` + +The ``SweepResult`` dataclass shape: + +```python +SweepResult( + index=0, + params={"x": 1, "y": 10}, + outputs={"output": 10.0}, + eval_time=2.8e-06, + status="completed", +) +``` + Additional Features ------------------- @@ -127,17 +228,3 @@ if __name__ == "__main__": sweep.estimated_time(4) ``` - -- Get results as NumPy - -```python - -# print the results, numpy -print(sweep.results.to_numpy()) -``` - -- Get results as Pandas - -```python -print(sweep.results.to_pandas()) -``` diff --git a/src/ensemblesweep/__init__.py b/src/ensemblesweep/__init__.py index bc0b7fb..706811f 100644 --- a/src/ensemblesweep/__init__.py +++ b/src/ensemblesweep/__init__.py @@ -1,4 +1,12 @@ -from .sweep import Sweep from .data import Data +from .executor import SweepExecutor +from .results import SweepResult, SweepResults +from .sweep import Sweep -__all__ = ["Sweep", "Data"] +__all__ = [ + "Sweep", + "Data", + "SweepExecutor", + "SweepResult", + "SweepResults", +] diff --git a/src/ensemblesweep/cli.py b/src/ensemblesweep/cli.py index 0f410f3..ee273a9 100644 --- a/src/ensemblesweep/cli.py +++ b/src/ensemblesweep/cli.py @@ -5,13 +5,11 @@ import click import numpy as np -import pandas as pd from .data import Data from .sweep import Sweep from .utils import parse_var_string -# Ensure current directory is in path for module resolution sys.path.append(os.getcwd()) @@ -36,10 +34,6 @@ def common_options(f): click.option("--save-pandas", type=click.Path(), help="Save results to Parquet file"), click.option("--quiet", "-q", is_flag=True, help="Suppress libEnsemble logs"), click.option("--dry-run", is_flag=True, help="Show sample of combinations and exit"), - click.option( - "--globus-compute-endpoint", - help="Optional Globus Compute endpoint UUID for remote execution", - ), ] for option in reversed(options): f = option(f) @@ -118,8 +112,7 @@ def py(func, var, **kwargs): sweep = Sweep( objective_function=objective_function, input_data=data, - num_workers=kwargs.get("workers"), - globus_compute_endpoint=kwargs.get("globus_compute_endpoint"), + nworkers=kwargs.get("workers"), ) handle_sweep(data, sweep, **kwargs) @@ -142,8 +135,7 @@ def exe(app, out_file, var, **kwargs): objective_executable=app, objective_output=out_file, input_data=data, - num_workers=kwargs.get("workers"), - globus_compute_endpoint=kwargs.get("globus_compute_endpoint"), + nworkers=kwargs.get("workers"), ) handle_sweep(data, sweep, **kwargs) diff --git a/src/ensemblesweep/executor.py b/src/ensemblesweep/executor.py new file mode 100644 index 0000000..967cdb1 --- /dev/null +++ b/src/ensemblesweep/executor.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from concurrent.futures import Future, ThreadPoolExecutor + +from .results import SweepResults +from .sweep import Sweep + + +class SweepExecutor: + def __init__(self, nworkers=None): + self.nworkers = nworkers + self._executor = ThreadPoolExecutor(max_workers=1) + + def submit(self, sweep: Sweep, n=None) -> Future: + if self.nworkers is not None: + sweep.nworkers = self.nworkers + return self._executor.submit(_run_sweep, sweep, n) + + def submit_sweep( + self, + *, + objective_function=None, + input_data=None, + objective_executable=None, + objective_output=None, + n=None, + ) -> Future: + sweep = Sweep( + objective_function=objective_function, + input_data=input_data, + objective_executable=objective_executable, + objective_output=objective_output, + nworkers=self.nworkers, + ) + return self.submit(sweep, n=n) + + def shutdown(self, wait=True, cancel_futures=False): + self._executor.shutdown(wait=wait, cancel_futures=cancel_futures) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + self.shutdown() + + +def _run_sweep(sweep: Sweep, n=None) -> SweepResults: + previously_evaluated = int(sweep.evaluated) + sweep.run(n) + return SweepResults(sweep, start_index=previously_evaluated) diff --git a/src/ensemblesweep/results.py b/src/ensemblesweep/results.py new file mode 100644 index 0000000..f156258 --- /dev/null +++ b/src/ensemblesweep/results.py @@ -0,0 +1,123 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Iterator + +import numpy as np + + +@dataclass(frozen=True) +class SweepResult: + index: int + params: dict[str, Any] + outputs: dict[str, Any] + eval_time: float | None = None + status: str = "completed" + + +class SweepResults: + """Collection view over completed sweep results. + + Can represent all completed results on a sweep (``sweep.results``) + or a subset corresponding to a single submitted batch + (``future.result()``). + """ + + def __init__(self, sweep, start_index: int = 0, stop_index: int | None = None): + self.sweep = sweep + self.start_index = start_index + self.stop_index = stop_index + + @property + def _completed(self) -> np.ndarray: + completed = self.sweep._H_total[self.sweep._H_total["sim_ended"]] + start = self.start_index + stop = self.stop_index if self.stop_index is not None else len(completed) + return completed[start:stop] + + def __len__(self) -> int: + return len(self._completed) + + def __getitem__(self, index): + completed = self._completed + if isinstance(index, slice): + return [_row_to_result(row, self.sweep) for row in completed[index]] + return _row_to_result(completed[index], self.sweep) + + def __iter__(self) -> Iterator[SweepResult]: + for row in self._completed: + yield _row_to_result(row, self.sweep) + + def to_numpy(self) -> np.ndarray: + return self._completed + + def to_pandas(self): + try: + import pandas as pd + except ImportError: + raise ImportError("Pandas is required for to_pandas(). Install it with 'pip install pandas'.") + + return pd.DataFrame([self._result_to_row_dict(r) for r in self]) + + def __str__(self) -> str: + results = self._completed + if len(results) == 0: + return "[]" + return str([self._result_to_row_dict(r) for r in self]) + + def __repr__(self) -> str: + return f"SweepResults({[self._result_to_row_dict(r) for r in self]!r})" + + @staticmethod + def _result_to_row_dict(result: SweepResult) -> dict[str, Any]: + row = {"index": result.index} + row.update(result.params) + row.update(result.outputs) + row["eval_time"] = result.eval_time + row["status"] = result.status + return row + + +def _row_to_result(row: np.void, sweep) -> SweepResult: + input_keys = sweep.input_data._keys + internal_fields = set(_internal_fields()) + result_fields = [name for name in row.dtype.names if name not in input_keys and name not in internal_fields] + + index = int(row["sim_id"]) if "sim_id" in row.dtype.names else 0 + eval_time = _to_python_value(row["eval_time"]) if "eval_time" in row.dtype.names else None + status = "completed" if bool(row["sim_ended"]) else "pending" + + return SweepResult( + index=index, + params={key: _to_python_value(row[key]) for key in input_keys}, + outputs={key: _to_python_value(row[key]) for key in result_fields if key != "eval_time"}, + eval_time=eval_time, + status=status, + ) + + +def _to_python_value(value): + if isinstance(value, np.generic): + return value.item() + return value + + +def _internal_fields(): + return [ + "sim_id", + "sim_started", + "sim_started_time", + "sim_ended", + "sim_ended_time", + "sim_worker", + "sim_time", + "given", + "given_time", + "cancel_requested", + "kill_sent", + "gen_informed", + "gen_informed_time", + "gen_started_time", + "gen_ended_time", + "gen_worker", + ] diff --git a/src/ensemblesweep/sim_funcs.py b/src/ensemblesweep/sim_funcs.py index b98093d..641422c 100644 --- a/src/ensemblesweep/sim_funcs.py +++ b/src/ensemblesweep/sim_funcs.py @@ -11,17 +11,14 @@ def generic_function_simf(H, persis_info, sim_specs, *args): """ start_time = time.time() - # Get user function and expected parameters func = sim_specs["user"]["objective_function"] input_keys = sim_specs["user"]["input_keys"] - # Extract arguments from H, passing only the first row's data args = [H[k][0] for k in input_keys] calc_status = WORKER_DONE result = np.nan try: - # Evaluate function result = func(*args) except Exception as e: print(f"Error evaluating objective_function: {e}") @@ -29,7 +26,6 @@ def generic_function_simf(H, persis_info, sim_specs, *args): eval_time = time.time() - start_time - # Format output based on the result type output = np.zeros(1, dtype=sim_specs["out"]) out_fields = [n for n in output.dtype.names if n != "eval_time"] @@ -73,13 +69,11 @@ def generic_executable_simf(H, persis_info, sim_specs, libE_info): output_file = sim_specs["user"]["objective_output"] input_keys = sim_specs["user"]["input_keys"] - # Build arguments string args_list = [str(H[k][0]) for k in input_keys] args = " ".join(args_list) exctr = libE_info["executor"] - # Execute without redirecting stdout unless we don't have an output file if output_file: task = exctr.submit(app_name="executable", app_args=args) else: @@ -92,7 +86,6 @@ def generic_executable_simf(H, persis_info, sim_specs, libE_info): if task.state == "FINISHED": filepath = os.path.join(task.workdir, output_file) try: - # We assume output can be processed by np.loadtxt data = np.loadtxt(filepath) final_data = data[-1] if data.ndim > 0 else data.item() except Exception as e: diff --git a/src/ensemblesweep/sweep.py b/src/ensemblesweep/sweep.py index 49f99e1..2a91279 100644 --- a/src/ensemblesweep/sweep.py +++ b/src/ensemblesweep/sweep.py @@ -8,96 +8,34 @@ from libensemble.comms.logs import LogConfig from libensemble.specs import AllocSpecs, ExitCriteria, SimSpecs +from .results import SweepResults from .sim_funcs import generic_executable_simf, generic_function_simf logs = LogConfig.config logs.stat_filename = "stats.txt" -class ResultWrapper: - def __init__(self, sweep): - self.sweep = sweep - - def __len__(self): - return int(np.sum(self.sweep._H_total["sim_ended"])) - - def __getitem__(self, i): - results = self.sweep._H_total[self.sweep._H_total["sim_ended"]] - if len(results) == 0: - return [] - - sliced = results[i] - - # format out fields, exclude internal libEnsemble properties - ignore_fields = [ - "sim_id", - "sim_started", - "sim_started_time", - "sim_ended", - "sim_ended_time", - "sim_worker", - "sim_time", - "given", - "given_time", - "cancel_requested", - "kill_sent", - "gen_informed", - "gen_informed_time", - "gen_started_time", - "gen_ended_time", - "gen_worker", - ] - - if isinstance(sliced, np.void): - out = {} - for name in sliced.dtype.names: - if name not in ignore_fields: - out[name] = sliced[name] - return out - else: - out_list = [] - for item in sliced: - out = {} - for name in item.dtype.names: - if name not in ignore_fields: - out[name] = item[name] - out_list.append(out) - return out_list - - def to_numpy(self): - return self.sweep._H_total[self.sweep._H_total["sim_ended"]] - - def to_pandas(self): - try: - import pandas as pd - except ImportError: - raise ImportError("Pandas is required for to_pandas(). Install it with 'pip install pandas'.") - return pd.DataFrame(self[:]) - - def __str__(self): - # We can format it nicely - results = self.sweep._H_total[self.sweep._H_total["sim_ended"]] - if len(results) == 0: - return "[]" - return str(self[:]) - - def __repr__(self): - return repr(self[:]) - - class Sweep: - def __init__(self, objective_function=None, input_data=None, objective_executable=None, objective_output=None): + def __init__( + self, + objective_function=None, + input_data=None, + objective_executable=None, + objective_output=None, + nworkers=None, + ): self.objective_function = objective_function self.objective_executable = objective_executable self.objective_output = objective_output self.input_data = input_data + self.nworkers = nworkers if self.objective_function and self.objective_executable: raise ValueError("Provide either objective_function or objective_executable, not both.") self._H_total = input_data.to_h0() self.evaluated = 0 - self.results = ResultWrapper(self) + self.results = SweepResults(self) def run(self, n=None): total_points = len(self._H_total) @@ -111,10 +49,10 @@ def run(self, n=None): target_sim_max = self.evaluated + to_evaluate - cores = max(1, os.cpu_count() - 1) + nworkers = self.nworkers if self.nworkers is not None else max(1, os.cpu_count() - 1) libE_specs = { "comms": "local", - "nworkers": cores, + "nworkers": nworkers, "sim_dirs_make": True, "ensemble_dir_path": f"sweep_{int(time.time())}", "reuse_output_dir": True, @@ -158,7 +96,6 @@ def run(self, n=None): from libensemble.executors.mpi_executor import MPIExecutor exctr = MPIExecutor() - # Register using absolute path effectively exctr.register_app(full_path=self.objective_executable, app_name="executable") ensemble.run() @@ -181,7 +118,7 @@ def estimated_time(self, n=None): if n is None: n = len(self._H_total) - self.evaluated - cores = max(1, os.cpu_count() - 1) + nworkers = self.nworkers if self.nworkers is not None else max(1, os.cpu_count() - 1) - batches = math.ceil(n / cores) + batches = math.ceil(n / nworkers) return batches * avg_time