Experiment

abses.core.experiment.Experiment

Experiment(model_cls, cfg, seed=None, **kwargs)

Experiment class.

Source code in abses/core/experiment.py

def __init__(
    self,
    model_cls: Type[MainModelProtocol],
    cfg: Configurations,
    seed: Optional[int] = None,
    **kwargs,
):
    self._job_id = 0
    self._extra_kwargs = kwargs
    self._overrides: Dict[str, Any] = {}
    self._base_seed = seed
    self._manager = ExperimentManager(model_cls)
    self.cfg = cfg

    # Setup experiment-level logger (separate from model run loggers)
    # This ensures experiment-level messages don't mix with model run logs
    # Pass DictConfig directly, don't convert to dict (log_parser needs DictConfig)
    self._logger: Optional[logging.Logger] = None
    if isinstance(cfg, DictConfig):
        # Create a copy to avoid modifying original
        cfg_dict = OmegaConf.to_container(cfg, resolve=True)
        if isinstance(cfg_dict, dict):
            cfg_dict["outpath"] = str(self.outpath)  # Convert Path to string
            cfg_copy = OmegaConf.create(cfg_dict)
            self._logger = setup_exp_logger(cfg_copy)
    elif isinstance(cfg, dict):
        # Create a copy to avoid modifying original input
        cfg_copy = cfg.copy()
        cfg_copy["outpath"] = str(self.outpath)  # Convert Path to string
        self._logger = setup_exp_logger(cfg_copy)

model_cls property

model_cls

Model class.

name property

name

Experiment name from configuration.

Returns:

Type	Description
str	Experiment name from exp.name config, or 'experiment' if not set.

logger property

logger

Experiment-level logger for recording experiment logs.

Use this logger to write messages to the experiment log file (e.g., fire_spread.log) rather than model run logs.

Example

exp.logger.info("Experiment started") exp.logger.debug("Processing parameters...")

Returns:

Type	Description
Logger	The experiment-level logger instance.

cfg property writable

cfg

Configuration

hydra_config property

hydra_config

Hydra runtime configuration object (HydraConf).

folder property

folder

Output dir path.

outpath property

outpath

Output dir path.

overrides property writable

overrides

Overrides

job_id property

job_id

Job id. Each job means a combination of the configuration. If the experiment is running in Hydra, it will return the hydra's job id.

new classmethod

new(model_cls, cfg, **kwargs)

Create a new experiment for the singleton class Experiment. This method will delete all currently available exp results and settings. Then, it initialize a new instance of experiment.

Parameters:

Name	Type	Description	Default
model_cls	Type[MainModelProtocol]	Using which model class to initialize the experiment.	required

Raises:

Type	Description
TypeError	If the model class model_cls is not a valid ABSESpy model.

Returns:

Type	Description
'Experiment'	An experiment.

Source code in abses/core/experiment.py

@classmethod
def new(
    cls, model_cls: Type[MainModelProtocol], cfg: Configurations, **kwargs
) -> "Experiment":
    """Create a new experiment for the singleton class `Experiment`.
    This method will delete all currently available exp results and settings.
    Then, it initialize a new instance of experiment.

    Parameters:
        model_cls:
            Using which model class to initialize the experiment.

    Raises:
        TypeError:
            If the model class `model_cls` is not a valid `ABSESpy` model.

    Returns:
        An experiment.
    """
    ExperimentManager(model_cls).clean()
    return cls(model_cls, cfg, **kwargs)

is_hydra_job staticmethod

is_hydra_job()

Returns True if the experiment is running in Hydra.

Source code in abses/core/experiment.py

@staticmethod
def is_hydra_job() -> bool:
    """Returns True if the experiment is running in Hydra."""
    return GlobalHydra().is_initialized()

summary

summary()

Summary of the experiment.

Source code in abses/core/experiment.py

def summary(self) -> pd.DataFrame:
    """Summary of the experiment."""
    return self._manager.get_datasets(seed=bool(self._base_seed))

batch_run

batch_run(
    repeats=1,
    parallels=None,
    display_progress=True,
    overrides=None,
)

Run the experiment multiple times.

Source code in abses/core/experiment.py

def batch_run(
    self,
    repeats: int = 1,
    parallels: Optional[int] = None,
    display_progress: bool = True,
    overrides: Optional[Dict[str, str | Iterable[Number]]] = None,
) -> None:
    """Run the experiment multiple times."""
    self.logger.info(
        f"Running experiment with {repeats} repeats and {parallels} parallels."
    )
    cfg = deepcopy(self._cfg)

    if not overrides:
        # 如果没有覆写，直接运行
        self._batch_run_repeats(cfg, repeats, parallels, display_progress)
        return

    # 获取所有配置组合
    all_configs = list(self._overriding(cfg, overrides))
    # 使用一个总进度条
    for config, overrides_ in tqdm(
        all_configs,
        disable=not display_progress,
        desc=f"{len(all_configs)} jobs (repeats {repeats} times each).",
        position=0,
    ):
        self.overrides = overrides_
        # 内层任务只显示简单信息，不显示进度条
        self._batch_run_repeats(
            config,
            repeats,
            parallels,
            display_progress=False,  # 关闭内层进度条
        )
        self._job_id += 1
    self.overrides = {}

add_hooks

add_hooks(hooks)

Add hooks to the experiment.

Source code in abses/core/experiment.py

def add_hooks(
    self,
    hooks: List[HookFunc] | Dict[str, HookFunc] | HookFunc,
) -> None:
    """Add hooks to the experiment."""
    if hasattr(hooks, "__call__"):
        hooks = [hooks]
    if isinstance(hooks, (list, tuple)):
        for hook in hooks:
            self._manager.add_a_hook(hook_func=hook)
    elif isinstance(hooks, dict):
        for hook_name, hook_func in hooks.items():
            self._manager.add_a_hook(hook_func=hook_func, hook_name=hook_name)
    else:
        raise TypeError(f"Invalid hooks type: {type(hooks)}.")