Skip to content

Analysis Registry

analysis_registry

Analysis Registry - Configuration Management.

Provides registry for loading and managing analysis configurations. Each analysis (Use Case) has a JSON configuration defining its parameters, filters, and visualization settings.

Classes:

Name Description
AnalysisRegistry

Registry for analysis configurations with JSON support

Classes

AnalysisRegistry

AnalysisRegistry(analyses_dir: Optional[Path] = None)

Registry for analysis configurations.

Loads analysis definitions from JSON files and provides access to analysis metadata, parameters, and settings.

Attributes:

Name Type Description
analyses_dir Path

Directory containing analysis JSON files

_analyses Dict[str, Dict[str, Any]]

Loaded analysis configurations

Methods:

Name Description
get_analysis

Get analysis configuration by ID

get_all_analyses

Get all registered analyses

get_analyses_by_use_case

Get all analyses for a specific Use Case

get_analysis_ids

Get list of all analysis IDs

get_use_cases

Get list of unique Use Cases

analysis_exists

Check if analysis exists

get_analysis_plot_type

Get plot type for analysis

reload

Reload all analysis configurations

get_stats

Get registry statistics

Initialize analysis registry.

Parameters:

Name Type Description Default
analyses_dir Optional[Path]

Directory containing analysis JSON files. Defaults to 'config/analyses/'.

None
Source code in src/infrastructure/config/analysis_registry.py
def __init__(self, analyses_dir: Optional[Path] = None):
    """
    Initialize analysis registry.

    Parameters
    ----------
    analyses_dir : Optional[Path], default=None
        Directory containing analysis JSON files.
        Defaults to 'config/analyses/'.
    """
    if analyses_dir is None:
        analyses_dir = Path("config/analyses")

    self.analyses_dir = analyses_dir
    self._analyses: Dict[str, Dict[str, Any]] = {}
    self._load_analyses()
Functions
get_analysis
get_analysis(analysis_id: str) -> Optional[Dict[str, Any]]

Get analysis configuration by ID.

Parameters:

Name Type Description Default
analysis_id str

Analysis identifier (e.g., 'UC1_1')

required

Returns:

Type Description
Optional[Dict[str, Any]]

Analysis configuration or None if not found

Source code in src/infrastructure/config/analysis_registry.py
def get_analysis(self, analysis_id: str) -> Optional[Dict[str, Any]]:
    """
    Get analysis configuration by ID.

    Parameters
    ----------
    analysis_id : str
        Analysis identifier (e.g., 'UC1_1')

    Returns
    -------
    Optional[Dict[str, Any]]
        Analysis configuration or None if not found
    """
    analysis = self._analyses.get(analysis_id)

    if analysis is None:
        logger.warning(f"Analysis not found: {analysis_id}")

    return analysis
get_all_analyses
get_all_analyses() -> Dict[str, Dict[str, Any]]

Get all registered analyses.

Returns:

Type Description
Dict[str, Dict[str, Any]]

Dictionary mapping analysis IDs to configurations.

Source code in src/infrastructure/config/analysis_registry.py
def get_all_analyses(self) -> Dict[str, Dict[str, Any]]:
    """
    Get all registered analyses.

    Returns
    -------
    Dict[str, Dict[str, Any]]
        Dictionary mapping analysis IDs to configurations.
    """
    return self._analyses.copy()
get_analyses_by_use_case
get_analyses_by_use_case(use_case: str) -> List[Dict[str, Any]]

Get all analyses for a specific Use Case.

Parameters:

Name Type Description Default
use_case str

Use Case identifier (e.g., 'UC1')

required

Returns:

Type Description
List[Dict[str, Any]]

List of analysis configurations

Source code in src/infrastructure/config/analysis_registry.py
def get_analyses_by_use_case(self, use_case: str) -> List[Dict[str, Any]]:
    """
    Get all analyses for a specific Use Case.

    Parameters
    ----------
    use_case : str
        Use Case identifier (e.g., 'UC1')

    Returns
    -------
    List[Dict[str, Any]]
        List of analysis configurations
    """
    return [
        analysis
        for analysis_id, analysis in self._analyses.items()
        if analysis_id.startswith(use_case)
    ]
get_analysis_ids
get_analysis_ids() -> List[str]

Get list of all analysis IDs.

Returns:

Type Description
List[str]

List of analysis identifiers.

Source code in src/infrastructure/config/analysis_registry.py
def get_analysis_ids(self) -> List[str]:
    """
    Get list of all analysis IDs.

    Returns
    -------
    List[str]
        List of analysis identifiers.
    """
    return list(self._analyses.keys())
get_use_cases
get_use_cases() -> List[str]

Get list of unique Use Cases.

Returns:

Type Description
List[str]

List of Use Case identifiers

Source code in src/infrastructure/config/analysis_registry.py
def get_use_cases(self) -> List[str]:
    """
    Get list of unique Use Cases.

    Returns
    -------
    List[str]
        List of Use Case identifiers
    """
    use_cases = set()

    for analysis_id in self._analyses.keys():
        # Extract UC prefix (e.g., 'UC1' from 'UC1_1')
        if "_" in analysis_id:
            uc = analysis_id.split("_")[0]
            use_cases.add(uc)

    return sorted(list(use_cases))
analysis_exists
analysis_exists(analysis_id: str) -> bool

Check if analysis exists.

Parameters:

Name Type Description Default
analysis_id str

Analysis identifier.

required

Returns:

Type Description
bool

True if analysis is registered.

Source code in src/infrastructure/config/analysis_registry.py
def analysis_exists(self, analysis_id: str) -> bool:
    """
    Check if analysis exists.

    Parameters
    ----------
    analysis_id : str
        Analysis identifier.

    Returns
    -------
    bool
        True if analysis is registered.
    """
    return analysis_id in self._analyses
get_analysis_plot_type
get_analysis_plot_type(analysis_id: str) -> Optional[str]

Get plot type for analysis.

Parameters:

Name Type Description Default
analysis_id str

Analysis identifier

required

Returns:

Type Description
Optional[str]

Plot type (e.g., 'heatmap', 'bar_chart') or None

Source code in src/infrastructure/config/analysis_registry.py
def get_analysis_plot_type(self, analysis_id: str) -> Optional[str]:
    """
    Get plot type for analysis.

    Parameters
    ----------
    analysis_id : str
        Analysis identifier

    Returns
    -------
    Optional[str]
        Plot type (e.g., 'heatmap', 'bar_chart') or None
    """
    analysis = self.get_analysis(analysis_id)

    if analysis is None:
        return None

    return analysis.get("plot_type")
reload
reload() -> None

Reload all analysis configurations from files.

Source code in src/infrastructure/config/analysis_registry.py
def reload(self) -> None:
    """Reload all analysis configurations from files."""
    logger.info("Reloading analysis configurations")
    self._analyses.clear()
    self._load_analyses()
get_stats
get_stats() -> Dict[str, Any]

Get registry statistics.

Returns:

Type Description
Dict[str, Any]

Statistics including total analyses, use cases, etc.

Source code in src/infrastructure/config/analysis_registry.py
def get_stats(self) -> Dict[str, Any]:
    """
    Get registry statistics.

    Returns
    -------
    Dict[str, Any]
        Statistics including total analyses, use cases, etc.
    """
    use_cases = self.get_use_cases()
    use_case_counts = {
        uc: len(self.get_analyses_by_use_case(uc)) for uc in use_cases
    }

    return {
        "total_analyses": len(self._analyses),
        "total_use_cases": len(use_cases),
        "use_cases": use_cases,
        "analyses_per_use_case": use_case_counts,
    }

Functions