energytrackr.utils package¶

Submodules¶

energytrackr.utils.args_parser module¶

Command-line argument parser for the Energy Pipeline CLI.

energytrackr.utils.args_parser.parse_args() → Namespace[source]¶

Parse command-line arguments.

Returns:: Parsed command-line arguments.
Return type:: argparse.Namespace

energytrackr.utils.exceptions module¶

Custom exceptions for the application.

exception energytrackr.utils.exceptions.AttributeNotFoundError(mod_path: str, attr: str)[source]¶

Bases: ImportError

Raised when the attribute is not found in the specified module.

exception energytrackr.utils.exceptions.BokehValidationIssuesError(tab_title: str, issues: ValidationIssues)[source]¶

Bases: ValueError

Exception raised when Bokeh validation issues are detected in a tab.

exception energytrackr.utils.exceptions.CantFindFileError(path: str)[source]¶

Bases: FileNotFoundError

Exception raised when a file cannot be found.

Inherits from FileNotFoundError to provide a more specific error message.

exception energytrackr.utils.exceptions.CommitStatsMissingOrEmptyDataFrameError(column: str)[source]¶

Bases: ValueError

Exception raised when the DataFrame is missing or empty in CommitStats, or the specified column is not found.

exception energytrackr.utils.exceptions.ConfigurationSingletonAlreadySetError[source]¶

Bases: ValueError

Custom ValueError for errors related to the configuration singleton.

exception energytrackr.utils.exceptions.ConfigurationSingletonNotSetError[source]¶

Bases: Exception

Custom Exception for errors related to the configuration singleton.

exception energytrackr.utils.exceptions.InvalidDottedPathError(dotted: str)[source]¶

Bases: ValueError

Raised when the dotted path is not in the expected ‘mod:attr’ format.

exception energytrackr.utils.exceptions.InvalidLegendClickPolicyError(policy: str, valid: list)[source]¶

Bases: ValueError

Exception raised when an invalid legend click policy is provided.

exception energytrackr.utils.exceptions.JavaHomeNotFoundError(java_home: str)[source]¶

Bases: ValueError

Exception raised when the specified JAVA_HOME path does not exist or is not a directory.

exception energytrackr.utils.exceptions.MissingContextKeyError(key: str)[source]¶

Bases: Exception

Exception raised when a required key is missing in the context.

exception energytrackr.utils.exceptions.MissingEnergyFieldError[source]¶

Bases: ValueError

Exception raised when no energy fields are specified in the configuration.

exception energytrackr.utils.exceptions.MissingEnergyTrackrKeyError[source]¶

Bases: KeyError

Exception raised when the top-level ‘energytrackr’ key is missing in the YAML configuration.

exception energytrackr.utils.exceptions.ModuleDidNotResolveToClassError(module_path: str)[source]¶

Bases: TypeError

Exception raised when a module path does not resolve to a class.

exception energytrackr.utils.exceptions.MustImplementError(class_name: str, cls: type)[source]¶

Bases: TypeError

Exception raised when a class does not implement the Transform interface.

exception energytrackr.utils.exceptions.NotADataclassTypeError(config_cls: type)[source]¶

Bases: TypeError

Exception raised when a provided class is not a dataclass type.

exception energytrackr.utils.exceptions.NotAPlotObjTypeError(name: str)[source]¶

Bases: TypeError

Exception raised when an object is not a PlotObj.

exception energytrackr.utils.exceptions.PlotAlreadyRegisteredError(name: str)[source]¶

Bases: ValueError

Exception raised when a plot class is already registered.

exception energytrackr.utils.exceptions.PlotLabelsNotSetError(plot_key: str)[source]¶

Bases: ValueError

Exception raised when plot labels are not set.

exception energytrackr.utils.exceptions.PlotObjectDidNotInitializeFigureError(module: str)[source]¶

Bases: RuntimeError

Raised when the first PlotObj does not initialize ctx.fig.

exception energytrackr.utils.exceptions.PlotSourcesNotSetError(plot_key: str)[source]¶

Bases: ValueError

Exception raised when plot sources are not set.

exception energytrackr.utils.exceptions.SettingsReadOnlyError[source]¶

Bases: AttributeError

Exception raised when attempting to modify a read-only Settings object.

exception energytrackr.utils.exceptions.SourceDirectoryNotFoundError(path: Path)[source]¶

Bases: Exception

Exception raised when the source directory is not found.

exception energytrackr.utils.exceptions.TargetDirectoryNotFoundError(path: Path)[source]¶

Bases: Exception

Exception raised when the target directory is not found.

exception energytrackr.utils.exceptions.UnexpectedArgumentError(invalid_args: list[str], cls: str)[source]¶

Bases: TypeError

Exception raised when MACDLine receives unexpected keyword arguments.

exception energytrackr.utils.exceptions.UnknownCommandError(command: str)[source]¶

Bases: Exception

Exception raised for unknown commands.

command¶

The command that caused the exception.

Type:: str

energytrackr.utils.git_utils module¶

Git utility functions for cloning repositories and gathering commits.

energytrackr.utils.git_utils.clone_or_open_repo(repo_path: str, repo_url: str, clone_options: list[str] | None = None) → Repo[source]¶

Clone a Git repository from a URL or open an existing repository from a local path.

Parameters:

repo_path (str) – The local path where the repository should be cloned or opened.
repo_url (str) – The URL of the remote Git repository.
clone_options (list, optional) – Additional options to pass to the clone command.

Returns:

An instance of the Git repository at the specified path.

Return type:

git.Repo

energytrackr.utils.git_utils.gather_commits(repo: Repo) → list[Commit][source]¶

Gather the commits that should be processed according to the execution plan.

For branches, we just take one commit per branch. For tags, we take the num_commits newest commits for each tag. For commits, we take the specified number of commits from the specified branch, starting from the oldest commit. If oldest_commit is specified, we start from there. If newest_commit is specified, we stop at that commit.

Parameters:: repo (git.Repo) – The git repository to gather commits from.
Returns:: The list of commits to process.
Return type:: list[git.Commit]

energytrackr.utils.git_utils.generate_commit_link(remote_url: str, commit_hash: str) → str[source]¶

Generate a URL to view a specific commit on a remote Git repository.

Given a remote repository URL (either SSH or HTTPS) and a commit hash, this function constructs a direct link to the commit on platforms that use GitHub-style URLs.

Parameters:

remote_url (str) – The remote repository URL (SSH or HTTPS).
commit_hash (str) – The commit hash to link to.

Returns:

The URL to view the commit, or “N/A” if the URL format is unsupported.

Return type:

str

Examples

>>> generate_commit_link("git@github.com:user/repo.git", "abc123")
'https://github.com/user/repo/commit/abc123'
>>> generate_commit_link("https://github.com/user/repo.git", "abc123")
'https://github.com/user/repo/commit/abc123'

energytrackr.utils.git_utils.get_commit_details_from_git(commit_hash: str, repo: Repo) → dict[str, Any][source]¶

Retrieve details about a specific git commit from a repository.

Parameters:

commit_hash (str) – The hash of the commit to retrieve details for.
repo (Repo) – An instance of a GitPython Repo object representing the repository.

Returns:

A dictionary containing the following keys:

”commit_summary” (str): The summary message of the commit.
”commit_link” (str): A URL to view the commit in a remote repository, or “N/A” if unavailable.
”commit_date” (str): The date of the commit in “YYYY-MM-DD” format.
”files_modified” (list[str]): A list of file paths modified in the commit.

Return type:

dict[str, Any]

If an error occurs, returns a dictionary with “N/A” or empty values for all fields.

energytrackr.utils.logger module¶

Logger with context support for logging messages in a structured way.

This module provides ContextLogger, a subclass of logging.Logger, which can buffer log messages (including their formatting arguments) into a context dictionary instead of emitting them immediately. You can then call log_context_buffer() to replay those buffered messages with their original formatting.

class energytrackr.utils.logger.ContextLogger(name, level=0)[source]¶

Bases: Logger

Custom logger class that supports logging with additional context.

ContextLogger extends the standard logging.Logger to allow buffering of log calls (including format strings and arguments) into a supplied context dict under the “log_buffer” key. Later, you can flush that buffer with log_context_buffer().

If no context is provided, or if the context does not contain “worker_process”: True, messages are emitted immediately as normal.

None¶

Type:: inherits everything from logging.Logger

info, debug, warning, error, critical, exception, log: Same signature as base Logger, but support an extra context kwarg.

critical(msg: object, *args: Any, **kwargs: Any) → None[source]¶

Log a message with severity ‘CRITICAL’, optionally buffering it.

Parameters:

msg – The log message or format string.
*args – Arguments for the format string.
**kwargs – Keyword arguments. May include context: dict.

debug(msg: object, *args: Any, **kwargs: Any) → None[source]¶

Log a message with severity ‘DEBUG’, optionally buffering it.

Parameters:

msg – The log message or format string.
*args – Arguments for the format string.
**kwargs – Keyword arguments. May include context: dict.

error(msg: object, *args: Any, **kwargs: Any) → None[source]¶

Log a message with severity ‘ERROR’, optionally buffering it.

Parameters:

msg – The log message or format string.
*args – Arguments for the format string.
**kwargs – Keyword arguments. May include context: dict.

exception(msg: object, *args: Any, **kwargs: Any) → None[source]¶

Log a message with severity ‘ERROR’ including exception info.

Parameters:

msg – The log message or format string.
*args – Arguments for the format string.
**kwargs – Keyword arguments. May include context: dict. If exc_info isn’t provided, it is set to True to include trace.

info(msg: object, *args: Any, **kwargs: Any) → None[source]¶

Log a message with severity ‘INFO’, optionally buffering it.

Parameters:

msg – The log message or format string.
*args – Arguments for the format string.
**kwargs – Keyword arguments. May include context: dict.

log(level: int, msg: object, *args: Any, **kwargs: Any) → None[source]¶

Log a message with a specified level, optionally buffering it.

Parameters:

level – Numeric log level.
msg – The log message or format string.
*args – Arguments for the format string.
**kwargs – Keyword arguments. May include context: dict.

warning(msg: object, *args: Any, **kwargs: Any) → None[source]¶

Log a message with severity ‘WARNING’, optionally buffering it.

Parameters:

msg – The log message or format string.
*args – Arguments for the format string.
**kwargs – Keyword arguments. May include context: dict.

energytrackr.utils.sort module¶

Sorts a CSV file containing commit hashes and energy values based on the commit history of a Git repository.

energytrackr.utils.sort.get_commit_history(repo_path: str) → list[str][source]¶

Retrieve the commit history from a Git repository in chronological order.

Parameters:: repo_path (str) – The file system path to the Git repository.
Returns:: A list of commit hashes in chronological order (oldest to newest).
Return type:: list[str]

energytrackr.utils.sort.read_csv(file_path: str) → list[tuple[str, str]][source]¶

Reads a CSV file containing commit hashes and energy values, preserving duplicate entries.

Parameters:: file_path (str) – The path to the CSV file to read.
Returns:: A list of tuples, each containing a commit hash and its corresponding energy value.
Return type:: list[tuple[str, str]]

energytrackr.utils.sort.reorder_commits(csv_file: str, repo_path: str, output_file: str) → None[source]¶

Reorders the rows of a CSV file containing commit hashes.

Reorders the rows of a CSV file containing commit hashes to match the chronological order of commits in a Git repository, preserving duplicate entries, and writes the result to a new CSV file.

Parameters:

csv_file (str) – Path to the input CSV file where the first column contains commit hashes.
repo_path (str) – Path to the local Git repository to retrieve commit history.
output_file (str) – Path to the output CSV file where the reordered data will be written.

Notes

Rows with commit hashes not found in the repository history are placed at the end of the output file.
Assumes the existence of helper functions: get_commit_history, read_csv, and write_csv.

energytrackr.utils.sort.write_csv(file_path: str, sorted_data: list[tuple[str, str]]) → None[source]¶

Write the sorted commit data to a CSV file, preserving duplicates.

Parameters:

file_path (str) – The path to the output CSV file.
sorted_data (list[tuple[str, str]]) – A list of tuples, each containing a commit identifier and its associated energy value.

energytrackr.utils.utils module¶

Utility functions.

energytrackr.utils.utils.get_local_env(env: Environment, template_path: str) → Environment[source]¶

Returns a Jinja2 Environment instance configured for the directory containing the specified template.

If the provided environment’s loader is already set to the template’s parent directory, it is returned as-is. Otherwise, a new Environment is created with a FileSystemLoader pointing to the template’s parent directory.

Parameters:

env (Environment) – The current Jinja2 Environment instance.
template_path (str) – The file path to the template.

Returns:

A Jinja2 Environment configured for the template’s directory.

Return type:

Environment

energytrackr.utils.utils.nice_number(x: float) → float[source]¶

Rounds a given number to a “nice” number, which is a value that is easy to interpret.

The function determines the order of magnitude of the input number and selects a “nice” fraction based on predefined thresholds.

Parameters:: x (float) – The input number to be rounded.
Returns:: A “nice” number that is close to the input value.
Return type:: float

energytrackr.utils.utils.run_command(arg: str, cwd: str | None = None, context: dict[str, Any] | None = None) → CompletedProcess[str][source]¶

Executes a shell command, streaming and capturing its output in real time.

Parameters:

arg (str) – The command to run.
cwd (str | None) – The working directory for the command.
context (dict[str, Any] | None) – The context dictionary for logging.

Returns:

The completed process object containing the command’s output and return code.

Return type:

subprocess.CompletedProcess[str]

Module contents¶

Utily functions for the project.