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(),
)

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.
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:")

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,
    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,
)

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

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)