sweep.SweepManager

sweep.SweepManager(
    config,
    registry,
    cli,
    job_set,
    session_id,
    _owns_registry=False,
    _owns_cli=False,
    _catalog=None,
    _experiment_id=None,
    _last_run_ids=dict(),
    _batch_remote_target=None,
    _batch_no_wait=False,
    _batch_poll_interval=10,
    _batch_timeout=None,
    _batch_auto_ingest=True,
    _collision_policy='fail',
)

Convenience orchestrator for parameter sweeps.

Encapsulates the common workflow of expanding, running, and collecting sweep results. For more control, use the underlying components directly.

Attributes

Name	Type	Description
config	JobConfig	The job configuration.
registry	RunRegistry	DuckDB registry for tracking.
cli	JoshCLI	Josh CLI for execution.
job_set	JobSet	Expanded jobs.
session_id	str	Session ID in registry.

Examples

>>> # Simple usage with defaults
>>> with SweepManager.from_config(config, registry=":memory:") as manager:
...     results = manager.run()
...     manager.load_results()
...     df = manager.query("averageHeight", group_by="maxGrowth")

>>> # Builder pattern for more control
>>> manager = (
...     SweepManager.builder(config)
...     .with_registry("experiment.duckdb", experiment_name="my_sweep")
...     .with_cli(jar_path=Path("josh.jar"))
...     .build()
... )

Methods

Name	Description
builder	Create a builder for complex configuration.
cleanup	Clean up temporary files.
close	Close registry connection if owned.
from_config	Create from a JobConfig.
from_yaml	Create from a YAML config file.
ingest	Recover and ingest results from MinIO (or local) by label.
load_results	Load CSV results into registry.
query	Query results with optional grouping and filtering.
run	Execute all jobs in the sweep.

builder

sweep.SweepManager.builder(config)

Create a builder for complex configuration.

Parameters

Name	Type	Description	Default
config	JobConfig	The job configuration to expand and run.	required

Returns

Name	Type	Description
	SweepManagerBuilder	A SweepManagerBuilder for fluent configuration.

Examples

>>> manager = (
...     SweepManager.builder(config)
...     .with_registry("experiment.duckdb")
...     .with_cli(jar_path=Path("josh.jar"))
...     .build()
... )

cleanup

sweep.SweepManager.cleanup()

Clean up temporary files.

Removes temporary config files created during job expansion. Should be called when you’re done with the SweepManager.

Examples

>>> manager = SweepManager.from_yaml(Path("sweep.yaml"))
>>> try:
...     manager.run()
...     manager.load_results()
... finally:
...     manager.cleanup()

close

sweep.SweepManager.close()

Close registry connection if owned.

Only closes the registry if it was created by the builder, not if an existing registry was passed in.

Examples

>>> manager = SweepManager.from_yaml(Path("sweep.yaml"))
>>> try:
...     manager.run()
... finally:
...     manager.close()

from_config

sweep.SweepManager.from_config(config, **kwargs)

Create from a JobConfig.

Parameters

Name	Type	Description	Default
config	JobConfig	The job configuration.	required
**kwargs	Any	Passed to with_defaults() (registry, experiment_name, jar_path).	{}

Returns

Name	Type	Description
	SweepManager	Configured SweepManager ready for use.

Examples

>>> config = JobConfig(
...     template_path=Path("template.jshc.j2"),
...     source_path=Path("simulation.josh"),
...     simulation="Main",
...     replicates=3,
...     sweep=SweepConfig(config_parameters=[...]),
... )
>>> manager = SweepManager.from_config(config, registry=":memory:")

from_yaml

sweep.SweepManager.from_yaml(path, **kwargs)

Create from a YAML config file.

Parameters

Name	Type	Description	Default
path	Path	Path to YAML file containing JobConfig.	required
**kwargs	Any	Passed to with_defaults() (registry, experiment_name, jar_path).	{}

Returns

Name	Type	Description
	SweepManager	Configured SweepManager ready for use.

Examples

>>> manager = SweepManager.from_yaml(Path("sweep.yaml"), registry=":memory:")

ingest

sweep.SweepManager.ingest(
    export_type='patch',
    download=False,
    output_dir=None,
    minio_bucket=None,
    quiet=False,
)

Recover and ingest results from MinIO (or local) by label.

Uses ingest_results() to look up the run by label, discover export paths, and load CSVs into the registry. Unlike load_results() this does not require a prior run() call – it works from the registry alone.

Parameters

Name	Type	Description	Default
export_type	str	Type of export to load (“patch”, “meta”, “entity”).	'patch'
download	bool	If True, download CSVs locally instead of S3 direct read.	False
output_dir	Path | None	Download destination (only used with download=True).	None
minio_bucket	str | None	Override MinIO bucket name.	None
quiet	bool	Suppress progress output.	False

Returns

Name	Type	Description
	int	Total number of rows loaded.

Examples

>>> manager.ingest()  # reads directly from S3
>>> manager.ingest(download=True, output_dir=Path("./local"))

load_results

sweep.SweepManager.load_results(export_type='patch', quiet=False)

Load CSV results into registry.

Parameters

Name	Type	Description	Default
export_type	str	Type of export to load (“patch”, “meta”, “entity”).	'patch'
quiet	bool	Suppress progress output.	False

Returns

Name	Type	Description
	int	Total number of rows loaded.

Examples

>>> rows = manager.load_results()
>>> print(f"Loaded {rows} rows")

query

sweep.SweepManager.query(variable, *, group_by=None, step=None, **filters)

Query results with optional grouping and filtering.

Parameters

Name	Type	Description	Default
variable	str	Variable name to query (e.g., “averageHeight”).	required
group_by	str | None	Optional parameter name to group by.	None
step	int | None	Optional timestep filter.	None
**filters	Any	Additional parameter filters.	{}

Returns

Name	Type	Description
	pd.DataFrame	DataFrame with query results.

Examples

>>> # Get parameter comparison
>>> df = manager.query("averageHeight", group_by="maxGrowth")

>>> # Get comparison at specific step
>>> df = manager.query("averageHeight", group_by="maxGrowth", step=50)

run

sweep.SweepManager.run(
    remote=False,
    api_key=None,
    endpoint=None,
    batch_remote=_UNSET,
    target=_UNSET,
    batch_no_wait=_UNSET,
    poll_interval=_UNSET,
    batch_timeout=_UNSET,
    auto_ingest=_UNSET,
    stop_on_failure=True,
    dry_run=False,
    quiet=False,
    on_complete=None,
    objective=None,
    jfr=None,
    enable_profiler=False,
    stream_output=False,
    bottle=None,
    bottle_dir=None,
    bottle_omit_jshd=False,
    force=False,
    collision_policy=_UNSET,
)

Execute all jobs in the sweep.

Automatically detects adaptive vs batch strategy and dispatches to the appropriate runner. Runs are automatically recorded in the registry.

For adaptive strategies (OptunaStrategy), uses run_adaptive_sweep() which generates jobs on-demand based on Optuna’s suggestions.

For non-adaptive strategies (CartesianStrategy), uses run_sweep() which executes all pre-expanded jobs.

Parameters

Name	Type	Description	Default
remote	bool	If True, use run_remote() for remote execution.	False
api_key	str | None	API key for authentication (optional for local servers).	None
endpoint	str | None	Custom endpoint URL (optional).	None
stop_on_failure	bool	If True (default), stop on first job failure. For adaptive strategies, raises SweepExecutionError with full error details. For batch strategies, returns partial results.	True
dry_run	bool	If True, print plan without executing (batch only).	False
quiet	bool	If True, suppress progress output.	False
on_complete	Callable[[ExpandedJob, Any], None] | None	Optional additional callback invoked after each job. Signature: callback(job, result) -> None. Called after registry recording. Use for progress reporting, logging, etc.	None
objective	Any | None	Objective function for adaptive strategies. If not provided, uses the objective from the strategy configuration.	None
jfr	Any | None	Optional JFR profiling configuration. When provided, each job gets its own recording file with the run_hash in the filename.	None
enable_profiler	bool	Enable Josh evaluation profiler (–enable-profiler).	False
stream_output	bool	If True, stream JAR stdout/stderr to the terminal in real time while still capturing them in CLIResult.	False
force	bool	Removed. Passing force=True now raises ValueError. The old behavior aliased collision_policy="overwrite", which couldn’t keep the registry in sync with re-run outputs. To add replicates use collision_policy="pool"; to redo a config, registry.drop_run(...) it then re-run.	False
collision_policy	str | Any	Override the builder-configured collision policy for this call. One of :data:COLLISION_POLICIES: "fail" (default) triggers the static check; "pool" reconciles by count, dispatching only the number of replicates still needed at fresh indices; "skip" is idempotent (no-op if any prior replicates exist).	_UNSET

Returns

Name	Type	Description
	SweepResult	SweepResult for batch strategies, AdaptiveSweepResult for adaptive.
	SweepResult	Both are compatible (same base interface).

Examples

>>> # Local execution (auto-detects strategy)
>>> results = manager.run()

>>> # Remote execution (Josh Cloud)
>>> results = manager.run(remote=True, api_key="your-api-key")

>>> # Remote execution (local server, no API key)
>>> results = manager.run(remote=True, endpoint="http://localhost:8080")

>>> # Dry run to see what would be executed (batch only)
>>> results = manager.run(dry_run=True)

>>> # Adaptive sweep with custom objective
>>> def my_objective(registry, run_hash, job):
...     return registry.query("SELECT AVG(value) FROM cell_data WHERE run_hash=?", [run_hash]).fetchone()[0]
>>> results = manager.run(objective=my_objective)