energytrackr.pipeline package

Subpackages

Submodules

energytrackr.pipeline.pipeline module

Pipeline orchestrator for running stages on commits in a repository.

class energytrackr.pipeline.pipeline.Pipeline(stages: dict[str, list[PipelineStage]], repo_path: str)[source]

Bases: object

Orchestrates the provided stages for each commit in sequence.

run(batches: list[list[Commit]]) → None[source]

Executes the pipeline over a list of batches, where each batch contains a list of commits.

Parameters:

batches (list[list[git.Commit]]) – A list of batches, where each batch is a list of git.Commit objects.

energytrackr.pipeline.pipeline.clean_cache_dir(repo_path: str) → None[source]

Remove all entries in the cache directory (siblings of the cloned repo) to free disk space.

Given that repo_path points to:

<project_dir>/.cache/.cache_<project_name>

this will delete everything under <project_dir>/.cache/ except the live repo folder.

Parameters:

repo_path (str) – Absolute path to the cloned repository.

energytrackr.pipeline.pipeline.compile_stages() → dict[str, list[PipelineStage]][source]

Compile the pipeline stages based on the execution plan.

Returns:

The compiled list of pipeline stages.

Return type:

list[PipelineStage]

energytrackr.pipeline.pipeline.create_batches(commits: list[Commit], batch_size: int, num_runs: int, num_repeats: int, randomize_tasks: bool) → list[list[Any]][source]

Divide commits into batches and expand according to runs/repeats.

Parameters:

  • commits (list[Commit]) – List of git.Commit objects to process.

  • batch_size (int) – Number of commits per batch.

  • num_runs (int) – Number of runs per commit.

  • num_repeats (int) – Number of repeats for each run.

  • randomize_tasks (bool) – Whether to randomize the order of tasks in each batch.

Returns:

List of batches, where each batch is a list of commits.

Return type:

list[list[Any]]

energytrackr.pipeline.pipeline.log_context_buffer(context: dict[str, Any]) → None[source]

Flushes buffered log calls from a context dict to the main logger.

Reads the list of buffered entries under context[“log_buffer”], and replays each one (including its original format string and args).

Parameters:

context – A dict which should contain: - “log_buffer”: List of tuples (level, msg, args, kwargs) - “commit”: Optional str commit identifier for header logging

energytrackr.pipeline.pipeline.measure(config_path: str) → None[source]

Executes the measurement process for a given repository based on the provided configuration.

This function performs the following steps: 1. Loads the pipeline configuration from the specified path. 2. Sets up the repository directory and clones or opens the repository. 3. Optionally runs system-level setup commands. 4. Collects all commits from the repository. 5. Runs the pre-stages once on the complete commit list for initial filtering. 6. Divides the filtered commits into batches. 7. Executes the remaining pipeline stages on these batches. 8. Restores the repository’s HEAD to the latest commit on the specified branch.

Parameters:

config_path (str) – The file path to the configuration file.

Raises:

Exceptions raised during the execution of repository operations or pipeline processing. –

energytrackr.pipeline.pipeline.restore_head(repo: Repo, branch: str) → None[source]

Restore the repository’s HEAD to the latest commit on the specified branch.

Parameters:

  • repo (git.Repo) – The Git repository object.

  • branch (str) – The branch to restore HEAD to.

energytrackr.pipeline.pipeline.run_pre_stages(commits: list[Commit], repo_path: str) → bool[source]

Run pre-stages on the full commit list. Returns True if pipeline should continue.

Parameters:

  • commits (list[git.Commit]) – List of git.Commit objects to process.

  • repo_path (str) – Path to the repository.

Returns:

True if the pipeline should continue, False if it should abort.

Return type:

bool

energytrackr.pipeline.pipeline.run_pre_test_stages_for_commit(commit_hexsha: str, repo_path: str) → dict[str, Any][source]

Process the pre-test stages for a single commit in a separate process.

Instead of receiving a git.Commit object (which might not be picklable), we pass the commit’s hexsha. Each process reopens the repository using repo_path and retrieves the commit object.

Parameters:

  • commit_hexsha (str) – The hexsha of the commit to process.

  • repo_path (str) – Path to the repository.

Returns:

The context after processing the stages.

Return type:

dict

energytrackr.pipeline.pipeline.run_setup_commands(commands: list[str]) → None[source]

Run system-level setup commands if provided.

Parameters:

commands (list[str]) – List of shell commands to run.

energytrackr.pipeline.pipeline.setup_project_dirs(config: PipelineConfig, config_dir: str) → str[source]

Set up project, cache directories and return their paths.

Parameters:

config (Config) – The configuration object containing repository information.

Returns:

Paths for project directory, cache directory, and repository path.

Return type:

tuple[str, str, str]

energytrackr.pipeline.stage_interface module

Interface for pipeline stages.

class energytrackr.pipeline.stage_interface.PipelineStage[source]

Bases: ABC

Abstract base for pipeline stages.

Each stage receives a shared ‘context’ dict and can read/write data.

abstractmethod run(context: dict[str, Any]) → None[source]

Execute the logic for this stage, possibly modifying context.

If the stage fails critically, set context[“abort_pipeline”] = True.

Parameters:

context (dict[str, Any]) – Shared context for the pipeline.

Module contents

Energy Pipeline.