sweep.SweepManager

sweep.SweepManager(
    config,
    registry,
    cli,
    job_set,
    session_id,
    _owns_registry=False,
    _owns_cli=False,
)

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(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=False,
    dry_run=False,
    quiet=False,
    on_complete=None,
)

Execute all jobs in the sweep.

Runs are automatically recorded in the registry.

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, stop on first failure.	False
dry_run	bool	If True, print plan without executing.	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

Returns

Name	Type	Description
	SweepResult	SweepResult with all job outcomes.

Examples

>>> # Local execution
>>> 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
>>> results = manager.run(dry_run=True)