Benchmark Orchestration API

Benchmark orchestration module.

Coordinates the full benchmark execution pipeline: 1. Build benchmarks (if needed) 2. Execute benchmarks 3. Aggregate results 4. Combine methods 5. Calculate carbon 6. Calculate GreenScore

File and path logistics are delegated to ResultsCollector.

pgsi_analyzer.benchmark.orchestrator.get_benchmark_path(algorithm: str, method: str, registry: Dict[str, Dict[str, str]] | None = None) → Path[source]: Backward-compatible path resolver used by tests and orchestrator internals.

pgsi_analyzer.benchmark.orchestrator.resolve_algorithms(algorithms: List[str], registry: Dict[str, Dict[str, str]] | None = None) → List[str][source]

Resolve algorithm list, expanding ‘all’ to all available algorithms.

Parameters:: algorithms – List of algorithm names or [‘all’]
Returns:: Sorted list of algorithm names

pgsi_analyzer.benchmark.orchestrator.resolve_methods(methods: List[str], algorithm: str | None = None, registry: Dict[str, Dict[str, str]] | None = None) → List[str][source]

Resolve method list, expanding ‘all’ to all available methods.

Parameters:

methods – List of method names or [‘all’]
algorithm – Optional algorithm name for validation

Returns:

List of method names in deterministic order

pgsi_analyzer.benchmark.orchestrator.run_benchmark_suite(algorithms: List[str], methods: List[str], runs: int = 50, algorithm_runs: Dict[str, int] | None = None, output_dir: Path = None, carbon_intensity: float = 0.000475, alpha: float = 0.4, beta: float = 0.4, gamma: float = 0.2, tool_paths: ToolPaths | None = None, path_sources: dict | None = None, provider: FileSystemProvider | None = None, benchmarks_dir: Path | None = None) → Path[source]

Execute full benchmark suite and generate GreenScore CSV.

This is the main orchestration function that: 1. Resolves algorithm and method lists 2. Builds benchmarks that require compilation 3. Executes all benchmark × method combinations 4. Aggregates raw CSVs 5. Combines results across methods 6. Calculates carbon footprint 7. Calculates GreenScore 8. Writes final GreenScore.csv

Parameters:

algorithms – List of algorithm names or [‘all’]
methods – List of method names or [‘all’]
runs – Number of runs per benchmark (default: 50)
algorithm_runs – Optional per-algorithm run overrides (e.g. {“hanoi”: 10})
output_dir – Base directory for all outputs (defaults to ./results)
carbon_intensity – Carbon intensity factor in gCO₂e/J (default: 0.000475)
alpha – Energy weight for GreenScore (default: 0.4)
beta – Carbon weight for GreenScore (default: 0.4)
gamma – Time weight for GreenScore (default: 0.2)
tool_paths – Optional ToolPaths configuration for Python/PyPy/C compiler
path_sources – Optional dict of path source metadata for audit report
provider – Optional FileSystemProvider for workspace/path I/O (default: real provider)

Returns:

Path to final GreenScore.csv file

Raises:

ValueError – If algorithms or methods are invalid
ConfigurationError – If build fails
AnalysisError – If data processing fails

Benchmark build system for Cython and ctypes benchmarks.

Handles compilation of Cython extensions and C shared libraries before benchmark execution. Compilation is done separately from measurement to ensure only execution time/energy is measured.

pgsi_analyzer.benchmark.builder.requires_build(method: str) → bool[source]

Check if a method requires compilation before execution.

Parameters:: method – Execution method name
Returns:: True if method requires build step

pgsi_analyzer.benchmark.builder.build_cython(benchmark_path: Path, build_dir: Path | None = None, tool_paths: ToolPaths | None = None) → Path[source]

Build Cython extension module.

Runs: python setup.py build_ext –inplace

Parameters:

benchmark_path – Path to Cython benchmark directory (contains setup.py)
build_dir – Optional directory for build artifacts (defaults to benchmark_path)
tool_paths – Optional ToolPaths configuration for Python executable

Returns:

Path to benchmark directory (build artifacts are in-place)

Raises:

ConfigurationError – If setup.py not found or build fails

pgsi_analyzer.benchmark.builder.build_ctypes(benchmark_path: Path, build_dir: Path | None = None, tool_paths: ToolPaths | None = None) → Path[source]

Build C shared library for ctypes benchmark.

Compiles .c files to .so (Linux) or .dll (Windows).

Parameters:

benchmark_path – Path to ctypes benchmark directory (contains .c files)
build_dir – Optional directory for build artifacts (defaults to benchmark_path)
tool_paths – Optional ToolPaths configuration for C compiler

Returns:

Path to benchmark directory (shared library is in-place)

Raises:

ConfigurationError – If compilation fails
PlatformError – If C compiler not available

pgsi_analyzer.benchmark.builder.build_benchmark(algorithm: str, method: str, benchmark_path: Path, tool_paths: ToolPaths | None = None) → Path[source]

Build a benchmark if it requires compilation.

Parameters:

algorithm – Algorithm name (for error messages)
method – Execution method
benchmark_path – Path to benchmark directory
tool_paths – Optional ToolPaths configuration

Returns:

Path to benchmark directory (ready for execution)

Raises:

ConfigurationError – If build fails
PlatformError – If build tools not available

Benchmark execution module.

Handles subprocess execution of benchmarks with proper isolation and environment setup. Ensures compilation time is excluded from measurements. Writes audit log (.audit.log) in output_dir for each run. Supports AuditLogger for path identity verification and audit_report.json.

pgsi_analyzer.benchmark.executor.get_runtime_executable(interpreter_path: str, env: Dict[str, str], cwd: str | None = None) → str | None[source]: Run the interpreter with -c “import sys; print(sys.executable)” and return the path the runtime reports. Used for path identity verification.

class pgsi_analyzer.benchmark.executor.AuditLogger[source]

Bases: object

Captures cmd list and env for every subprocess.run, and path identity (resolved vs runtime-reported) per method for audit_report.json.

log_execution(method: str, algorithm: str, cmd: List[str], env_slice: Dict[str, str], resolved_interpreter: str, runtime_reported_path: str | None = None) → None[source]

set_path_identity(method: str, resolved: str, runtime_reported: str | None) → None[source]: Set or update path identity for a method (e.g. after path identity check).

to_report_dict(path_sources: Dict[str, Dict[str, Any]]) → Dict[str, Any][source]: Build the structure for audit_report.json: path_integrity per method, requested/resolved/runtime_reported, source (env/cli/system_default), and severity HIGH if mismatch.

pgsi_analyzer.benchmark.executor.find_python_executable(method: str, tool_paths: ToolPaths | None = None) → str[source]

Find the correct Python executable for a method.

Parameters:

method – Execution method (‘cpython’, ‘pypy’, etc.)
tool_paths – Optional ToolPaths configuration. If None, uses defaults.

Returns:

Path to Python executable (as string for subprocess)

Raises:

PlatformError – If required runtime not found

pgsi_analyzer.benchmark.executor.prepare_py_compile(benchmark_path: Path, tool_paths: ToolPaths | None = None) → Path[source]

Pre-compile Python file to .pyc for py_compile method.

Parameters:

benchmark_path – Path to main.py file or to the py_compile directory
tool_paths – Optional ToolPaths configuration for Python executable

Returns:

Path to compiled .pyc file (or main.py if compilation fails)

pgsi_analyzer.benchmark.executor.execute_benchmark(algorithm: str, method: str, benchmark_path: Path, runs: int = 50, output_dir: Path = None, env: Dict[str, str] | None = None, tool_paths: ToolPaths | None = None, audit_logger: AuditLogger | None = None) → Dict[str, Path][source]

Execute a benchmark and collect measurement CSVs.

This function runs the benchmark in a subprocess, allowing the measurement decorators to capture energy and time data.

Parameters:

algorithm – Algorithm name (for CSV naming)
method – Execution method
benchmark_path – Path to benchmark directory or main.py
runs – Number of runs (passed to benchmark script)
output_dir – Directory where CSVs will be written (defaults to benchmark_path)
env – Optional environment variables for subprocess
tool_paths – Optional ToolPaths configuration for Python/PyPy executables

Returns:

‘energy_csv’: Path to energy CSV file
’time_csv’: Path to time CSV file
’system_info’: Path to system info JSON

Return type:

Dictionary with keys

Raises:

MeasurementError – If execution fails

File-system provider for the benchmark pipeline.

Separates pipeline coordination from filesystem I/O: workspace preparation, path resolution, and aggregated path collection. The orchestrator uses this provider so that I/O can be mocked in tests without touching disk.

class pgsi_analyzer.benchmark.provider.FileSystemProvider[source]

Bases: object

Handles filesystem operations for the benchmark pipeline: creating workspaces, copying raw CSVs by pattern, resolving output paths, and collecting aggregated paths.

prepare_aggregation_workspace(output_dir: Path, method: str, raw_dirs: List[Path], kind: str) → Path[source]

Create a workspace directory and copy raw CSVs from raw_dirs into it.

Parameters:

output_dir – Base output directory (e.g. results/).
method – Execution method name (e.g. cpython, pypy).
raw_dirs – List of directories that contain raw CSV files.
kind – “energy” or “time” (used for temp dir naming and pattern).

Returns:

Path to the created directory containing the copied CSV files.

get_output_path(output_dir: Path, method: str | None = None, file_type: str | None = None) → Path[source]

Return the canonical path for an output file (aggregated, combined, or final).

Creates method subdirectories when file_type is energy_aggregated or time_aggregated.

Parameters:

output_dir – Base output directory.
method – Execution method (required for energy_aggregated, time_aggregated).
file_type – One of energy_aggregated, time_aggregated, energy_combined, time_combined, carbon_footprint, GreenScore.

Returns:

Path where the file should be written.

collect_aggregated_paths(output_dir: Path, methods: List[str]) → Dict[str, Dict[str, Path]][source]

Return paths to energy_aggregated.csv and time_aggregated.csv for each method.

Parameters:

output_dir – Base output directory.
methods – List of method names (e.g. cpython, pypy).

Returns:

{method: path}, “time”: {method: path}}

Return type:

{“energy”

Results collection and filesystem layout for the benchmark pipeline.

Groups raw CSV paths by method (collect_paths). Filesystem I/O (workspace preparation, output path resolution) is delegated to FileSystemProvider; ResultsCollector retains backward compatibility by delegating to a default provider when prepare_aggregation_workspace or get_output_path are called.

class pgsi_analyzer.benchmark.results_collector.ResultsCollector(provider: FileSystemProvider = None)[source]

Bases: object

Handles grouping of raw CSV paths by method. Workspace and path resolution delegate to FileSystemProvider (default instance) for backward compatibility.

collect_paths(execution_results: Dict[str, Dict[str, Dict[str, Any]]]) → Dict[str, Dict[str, List[Path]]][source]

Group energy and time CSV paths by execution method.

execution_results is expected to be a nested mapping in this shape: {algorithm: {method: {"energy_csv": Path, "time_csv": Path, ...}}}.

Returns:: Grouped CSV parent directories: {"energy": {method: [Path, ...]}, "time": {method: [Path, ...]}}.
Return type:: Dict[str, Dict[str, List[Path]]]
Raises:: AuditError – If a method is not in the registry whitelist (VALID_METHODS).

prepare_aggregation_workspace(output_dir: Path, method: str, raw_dirs: List[Path], kind: str) → Path[source]: Delegate to FileSystemProvider. See provider.prepare_aggregation_workspace.

get_output_path(output_dir: Path, method: str = None, file_type: str = None) → Path[source]: Delegate to FileSystemProvider. See provider.get_output_path.