Analysis

abses.utils.analysis.ResultAnalyzer

ResultAnalyzer(path)

Bases: _BaseAnalyzer

Analyzer for a single Hydra run result.

This class analyzes the output of a single Hydra experiment run, including reading configuration, loading data files, and extracting reporter information.

Attributes:

Name	Type	Description
data	DataFrame	Raw data loaded from CSV or datacollector output.
configs	Dict[str, Any]	Full configuration dictionary.
model_reporter	Dict[str, Any]	Model-level reporter configuration.
agent_reporter	Dict[str, Dict[str, Any]]	Agent-level reporter configuration.
final_reporter	Dict[str, Any]	Final reporter configuration.

Parameters:

Name	Type	Description	Default
path	PathLike	Path to the single run output directory.	required

Raises:

Type	Description
FileNotFoundError	If the path is not a valid directory.

Source code in abses/utils/analysis.py

def __init__(self, path: PathLike) -> None:
    """Initialize the result analyzer.

    Args:
        path: Path to the single run output directory.

    Raises:
        FileNotFoundError: If the path is not a valid directory.
    """
    # Initialize attributes
    self.configs: Dict[str, Any] = {}
    self.model_reporter: Dict[str, Any] = {}
    self.agent_reporter: Dict[str, Dict[str, Any]] = {}
    self.final_reporter: Dict[str, Any] = {}

    super().__init__(path=path)
    if not self.path.is_dir():
        raise FileNotFoundError(f"{path} is not a directory.")
    self._hydra = self.path / ".hydra"
    if self._hydra.is_dir():
        self.config = self._hydra / "config.yaml"
        self._load_hydra_cfg(self._hydra)
    else:
        # If no .hydra directory, try to find config.yaml in the path
        config_path = self.path / "config.yaml"
        if config_path.is_file():
            self.config = config_path
            self._load_hydra_cfg(self.path)
        else:
            raise FileNotFoundError(
                f"No configuration found in {path}. Expected .hydra/config.yaml or config.yaml"
            )
    self.read_data()

data property writable

data

Raw data loaded from CSV or datacollector output.

Returns:

Type	Description
DataFrame	DataFrame containing the raw data.

Raises:

Type	Description
AttributeError	If data has not been loaded yet.

read_data

read_data(suffix='csv')

Read and merge result csv files under the experiment folder.

This method will: - First, look for all files matching *_cities.csv (e.g. 1_cities.csv, 2_cities.csv ...) under self.path. - If found, read them all and vertically concatenate them into a single dataframe. - If none are found, fall back to reading a single cities.csv file.

Source code in abses/utils/analysis.py

def read_data(self, suffix: str = "csv") -> pd.DataFrame:
    """Read and merge result csv files under the experiment folder.

    This method will:
    - First, look for all files matching ``*_cities.csv`` (e.g. ``1_cities.csv``,
      ``2_cities.csv`` ...) under ``self.path``.
    - If found, read them all and vertically concatenate them into a single
      dataframe.
    - If none are found, fall back to reading a single ``cities.csv`` file.
    """
    # Prefer numbered runs like 1_cities.csv, 2_cities.csv, ...
    csv_files = sorted(self.path.glob(f"*.{suffix}"))

    if csv_files:
        data_frames = []
        for csv_file in csv_files:
            try:
                df = self.read_csv(path=csv_file)
                data_frames.append(df)
            except FileNotFoundError:
                logger.warning(f"Skip missing file: {csv_file}")
        if not data_frames:
            raise FileNotFoundError(
                f"No valid *.{suffix} files found under {self.path}."
            )
        self.data = pd.concat(data_frames, ignore_index=True)
        logger.info(
            "Loaded and merged result files: "
            f"{[f.name for f in csv_files]} from {self.path}."
        )
        return self.data
    else:
        # Backward compatibility: fall back to a single cities.csv
        logger.warning(f"No valid *.{suffix} files found under {self.path}.")
        self.data = pd.DataFrame()
        return self.data

read_csv

read_csv(path)

Read a CSV file into a DataFrame.

Parameters:

Name	Type	Description	Default
path	PathLike	Path to the CSV file.	required

Returns:

Type	Description
DataFrame	DataFrame containing the CSV data.

Raises:

Type	Description
FileNotFoundError	If the file does not exist or is invalid.

Source code in abses/utils/analysis.py

def read_csv(self, path: PathLike) -> pd.DataFrame:
    """Read a CSV file into a DataFrame.

    Args:
        path: Path to the CSV file.

    Returns:
        DataFrame containing the CSV data.

    Raises:
        FileNotFoundError: If the file does not exist or is invalid.
    """
    if isinstance(path, str):
        path = Path(path)
    if not path.is_file():
        raise FileNotFoundError(f"CSV file not found: {path}")
    if path.suffix != ".csv":
        raise FileNotFoundError(f"File is not a CSV: {path}")

    # Try reading with index_col=0, fallback to no index
    try:
        return pd.read_csv(path, index_col=0)
    except (ValueError, IndexError):
        return pd.read_csv(path)

get_data cached

get_data(**kwargs)

Get processed data with optional transformations.

This method can be overridden or extended to support different aggregation levels or data transformations.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Additional arguments for data processing.	{}

Returns:

Type	Description
DataFrame	Processed DataFrame.

Source code in abses/utils/analysis.py

@lru_cache(maxsize=1)
def get_data(self, **kwargs: Any) -> pd.DataFrame:
    """Get processed data with optional transformations.

    This method can be overridden or extended to support different
    aggregation levels or data transformations.

    Args:
        **kwargs: Additional arguments for data processing.

    Returns:
        Processed DataFrame.
    """
    return self.data.copy()

select

select(key)

Select a value from the configuration.

Parameters:

Name	Type	Description	Default
key	str	Configuration key path.	required

Returns:

Type	Description
Any	The value at the specified key path.

Source code in abses/utils/analysis.py

def select(self, key: str) -> Any:
    """Select a value from the configuration.

    Args:
        key: Configuration key path.

    Returns:
        The value at the specified key path.
    """
    return OmegaConf.select(self.config, key=key)

abses.utils.analysis.ExpAnalyzer

ExpAnalyzer(path, enable_logger=True)

Bases: _BaseAnalyzer

Analyzer for a group of Hydra multirun experiment results.

This class analyzes multiple experiment runs from a Hydra multirun, including parsing configuration overrides, aggregating data, and comparing differences between runs.

Attributes:

Name	Type	Description
overrides	Dict[str, List[str]]	Dictionary of configuration overrides from multirun.yaml.
results	Generator[ResultAnalyzer, None, None]	Generator yielding ResultAnalyzer for each run.

Parameters:

Name	Type	Description	Default
path	PathLike	Path to the multirun output directory.	required
enable_logger	bool	Whether to enable logging (default: True).	True

Source code in abses/utils/analysis.py

def __init__(self, path: PathLike, enable_logger: bool = True) -> None:
    """Initialize the experiment analyzer.

    Args:
        path: Path to the multirun output directory.
        enable_logger: Whether to enable logging (default: True).
    """
    super().__init__(path=path)
    multirun_config = self.path / "multirun.yaml"
    if multirun_config.is_file():
        self.config = multirun_config
    else:
        # Try alternative location
        multirun_config = self.path.parent / "multirun.yaml"
        if multirun_config.is_file():
            self.config = multirun_config
        else:
            if enable_logger:
                logger.warning(
                    f"multirun.yaml not found in {self.path}. "
                    f"Some features may not work correctly."
                )
            # Create an empty config
            self._config = OmegaConf.create({})

overrides property

overrides

Configuration overrides from multirun.yaml.

Parses the hydra.overrides.task section to extract parameter overrides and their values.

Returns:

Type	Description
Dict[str, List[str]]	Dictionary mapping parameter names to lists of values.

results property

results

Generator yielding ResultAnalyzer for each run.

Yields:

Type	Description
ResultAnalyzer	ResultAnalyzer instance for each subdirectory in the multirun output.

diff_runs cached property

diff_runs

DataFrame showing configuration differences between runs.

Returns:

Type	Description
DataFrame	DataFrame with columns for each override parameter and rows
DataFrame	for each run, showing the actual values used.

Raises:

Type	Description
NotImplementedError	If unexpected configuration values are found.

agg_data cached property

agg_data

Aggregated data from all runs.

This property aggregates data from all runs and adds configuration override columns to identify each run.

Returns:

Type	Description
DataFrame	DataFrame containing aggregated data from all runs.

Note

This is a cached property. To refresh, delete the attribute or use a new instance.

apply

apply(func, *args, **kwargs)

Apply a function to each run's ResultAnalyzer.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	Function to apply. Should accept ResultAnalyzer as first argument.	required
*args	Any	Additional positional arguments for the function.	()
**kwargs	Any	Additional keyword arguments for the function.	{}

Returns:

Type	Description
Series	Series with results from applying the function to each run.

Source code in abses/utils/analysis.py

def apply(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> pd.Series:
    """Apply a function to each run's ResultAnalyzer.

    Args:
        func: Function to apply. Should accept ResultAnalyzer as first argument.
        *args: Additional positional arguments for the function.
        **kwargs: Additional keyword arguments for the function.

    Returns:
        Series with results from applying the function to each run.
    """
    results = []
    for run in self._results_list:
        try:
            result = func(run, *args, **kwargs)
            results.append(result)
        except Exception as e:
            logger.warning(f"Failed to apply {func.__name__} to {run.path}: {e}")
            results.append(None)

    return pd.Series(results, name=func.__name__)