Utils Subpackage

This subpackage contains the utility modules used by the YugiQuery package.

Dirs Module

class yugiquery.utils.dirs.Dirs

Bases: object

Singleton class to manage directory paths for the application.

Variables:

APP (Path) – The path to the application (YugiQuery) directory.
ASSETS (Path) – The path to the assets directory.
DATA (Path) – The path to the data directory.
NOTEBOOKS (Path) – The path to the notebooks directory.
REPORTS (Path) – The path to the reports directory.
UTILS (Path:) – The path to the utils subpackage directory.
WORK (Path) – The path to the working directory.

property ASSETS: SimpleNamespace

Returns a SimpleNamespace object containing the paths to the package and user assets directories.

Returns:: A SimpleNamespace object containing the paths to the package and user assets directories.
Return type:: SimpleNamespace

property DATA: Path

Return the path to the data directory.

Returns:: The path to the data directory.
Return type:: Path

property NOTEBOOKS: SimpleNamespace

Return the path to the notebooks directory.

Returns:: The path to the notebooks directory.
Return type:: Path

property REPORTS: Path

Return the path to the reports directory.

Returns:: The path to the reports directory.
Return type:: Path

static __new__(cls, *args, **kwargs): Create a new instance of the Dirs class if one does not already exist.

find_notebooks(notebooks: str | List[str] | List[Path] = 'all') → List[Path]

Finds the paths of the specified notebooks.

Parameters:: notebooks (str | List[str] | List[Path], optional) – A list of notebook names or paths. If “all”, finds all notebooks in the NOTEBOOKS directory. If “user”, finds all notebooks in the user directory. Defaults to “all”.

get_asset(*parts: str) → Path

Return the path to an asset file.

Parameters:: *parts (str) – The path parts to the asset file or directory.
Returns:: The path to the asset file or directory.
Return type:: Path
Raises:: FileNotFoundError – If the asset file is not found

get_notebook(*parts: str) → Path

Return the path to a notebook file. Names are case insensitive. If more than one occurrence is found, the first path will be returned. Notebooks in the user directory will take precedence over the pkg director.

Parameters:: *parts (str) – The path parts to the notebook file.
Returns:: The path to the notebook file.
Return type:: Path
Raises:: FileNotFoundError – If the notebook file is not found.

property is_notebook: bool

Check if the current environment is a Jupyter notebook.

Returns:: True if the current environment is a Jupyter notebook, False otherwise.
Return type:: bool

make() → None: Ensure that the DATA and REPORTS directories exist.

print() → None: Prints the directory paths managed by this class.

property secrets_file: Path | None

first ASSETS, then WORK. Returns none if the file is not found.

Returns:: The path to the secrets file, or None if not found.
Return type:: Path | None
Type:: Return the path to the secrets file following the hierarchy

Git Module

yugiquery.utils.git.commit(files: str | List[str | Path], message: str = '', repo: Repo | None = None) → str

Commits the specified files to the git repository after staging them.

Parameters:

files (str | List[str | Path]) – A list of file paths to be committed.
message (str, optional) – The commit message. If not provided, a default message will be used.
repo (git.Repo | None, optional) – The git repository object. If none provided, the current repository will be used.

Raises:

git.GitCommandError – If an error occurs while committing the changes.
Exception – For any other unexpected errors.

Returns:

The commit result.

Return type:

str

yugiquery.utils.git.ensure_repo() → Repo

Ensures the execution happens inside a git repository. Initializes a repository if one is not found.

Raises:

git.InvalidGitRepositoryError – If the dirs.SCRIPT is not in a git repository.
Exception – For any other unexpected errors.
RuntimeError – If a Git repository cannot be initialized.

Returns:

The git repository object.

Return type:

git.Repo

yugiquery.utils.git.get_repo() → Repo

Gets the current git repository if there is one.

Raises:

git.InvalidGitRepositoryError – If the dirs.SCRIPT is not in a git repository.
git.GitCommandError – If an error occurs while committing the changes.
Exception – For any other unexpected errors.

Returns:

The git repository object.

Return type:

git.Repo

yugiquery.utils.git.pull(passphrase: str = '', repo: Repo | None = None) → str

Pulls changes from the remote git repository.

Parameters:

passphrase (str, optional) – The passphrase to unlock your Git credential store. Defaults to empty.
repo (git.Repo | None, optional) – The git repository object. If none provided, the current repository will be used.

Raises:

git.GitCommandError – If an error occurs while committing the changes.
Exception – For any other unexpected errors.

Returns:

None

yugiquery.utils.git.push(passphrase: str = '', repo: Repo | None = None) → str

Pushes commits to the remote git repository.

Parameters:

passphrase (str, optional) – The passphrase to unlock your Git credential store. Defaults to empty.
repo (git.Repo | None, optional) – The git repository object. If none provided, the current repository will be used.

Raises:

git.GitCommandError – If an error occurs while committing the changes.
Exception – For any other unexpected errors.

Returns:

None

yugiquery.utils.git.restore(files: str | List[str], repo: Repo | None = None) → str

Restores the specified files on the git repository.

Parameters:

files (str | List[str]) – A list of file paths to be restored.
repo (git.Repo | None, optional) – The git repository object. If none provided, the current repository will be used.

Raises:

git.GitCommandError – If an error occurs while committing the changes.
Exception – For any other unexpected errors.

Returns:

The restore result.

Return type:

str

yugiquery.utils.git.squash_commits(start_commit: Commit, repo: Repo | None = None, message: str = '') → str

Squashes all commits from the start_commit to HEAD into a single commit.

Parameters:

start_commit (git.Commit) – The commit object to start squashing from.
repo (git.Repo | None, optional) – The git repository object. If none provided, the current repository will be used.
message (str, optional) – The commit message. If not provided, a default message will generated from commits.

yugiquery.utils.git.unlock(passphrase: str = '') → CompletedProcess

Unlock the git credential store.

Parameters:

passphrase (str, optional) – The passphrase to unlock your Git credential store. Defaults to empty.

Returns:

The result of the unlock operation.

Return type:

subprocess.CompletedProcess

Raises:

FileNotFoundError – If the git unlock script is not found.
OSError – If there is an error starting the git unlock script.
subprocess.CalledProcessError – If the git unlock script returns a non-zero exit code.

Helpers Module

yugiquery.utils.helpers.auto_or_bool(value: str) → bool | Literal['auto']: Convert a string to a boolean (True or False) or “auto”.

yugiquery.utils.helpers.ensure_tuple_columns(df: DataFrame) → DataFrame

Ensures that if any value in a column is a tuple, all values in that column are tuples, but leaves NaN values as NaN.

Parameters:: df (pd.DataFrame) – The DataFrame to process.
Returns:: The processed DataFrame with consistent tuple columns.
Return type:: pd.DataFrame

yugiquery.utils.helpers.escape_chars(string: str, chars: List[str] = ['_', '.', '-', '+', '#', '@', '=']) → str

Escapes specified characters in a given string by adding a backslash before each occurrence.

Parameters:

string (str) – The input string to be processed.
chars (list, optional) – A list of characters to be escaped. Default is [“_”, “.”, “-”, “+”, “#”, “@”, “=”].

Returns:

The input string with the specified characters escaped.

Return type:

str

yugiquery.utils.helpers.filename_ts_fmt(ts: Arrow) → str

Formats a timestamp for use in filenames, converting it to UTC and using the format YYYYMMDDTHHmmZ. :param ts: The timestamp to format. :type ts: arrow.Arrow

Returns:: The formatted timestamp string.
Return type:: str

yugiquery.utils.helpers.get_ts_granularity(seconds: int) → List[Literal['auto', 'second', 'minute', 'hour', 'day', 'week', 'month', 'quarter', 'year']]

Humanizes a time interval given in seconds.

Parameters:: seconds (int) – The time interval in seconds.
Returns:: A list of human-readable granularities for the time interval.
Return type:: List[arrow.arrow._GRANULARITY]

yugiquery.utils.helpers.load_json(json_file: str | Path) → dict

Load data from a JSON file.

Parameters:: json_file (str | Path) – The file path to the JSON file.
Returns:: A dictionary containing the data from the JSON file. If the file does not exist, an empty dictionary is returned.
Return type:: dict

yugiquery.utils.helpers.load_secrets(requested_secrets: List[str] = [], secrets_file: str | Path | None = None, required: bool = False) → Dict[str, str | None]

Load secrets from environment variables and/or a .env file.

The secrets can be specified by name using the requested_secrets argument, which should be a list of strings. If requested_secrets is not specified, all available secrets will be returned.

The secrets_file argument is the path to a .env file containing additional secrets to load. If secrets_file is specified and the file exists, the function will load the secrets from the file and merge them with the secrets loaded from the environment variables giving priority to secrets obtained from the environment.

The required argument is a boolean or list of booleans indicating whether each requested secret is required to be present. If required is True, a KeyError will be raised if the secret is not found. If required is False or not specified, missing secrets will be skipped.

Parameters:

requested_secrets (List[str], optional) – A list of names of the secrets to retrieve. If empty or not specified, all available secrets will be returned. Defaults to [].
secrets_file (str | Path | None, optional) – The path to a .env file containing additional secrets to load. Defaults to None.
required (bool or List[bool], optional) – A boolean or list of booleans indicating whether each requested secret is required to be present. If True, a KeyError will be raised if the secret is not found. If False or not specified, missing secrets will be skipped. Defaults to False.

Returns:

A dictionary containing the requested secrets as key-value pairs.

Return type:

Dict[str, str | None]

Raises:

KeyError – If a required secret is not found in the environment variables or .env file.

yugiquery.utils.helpers.lock(file_name: str) → None

Acquire a file lock and handle stale locks using the same lock file.

Parameters:: file_name (str) – The name of the lock file to create
Raises:: RuntimeError – If another instance is already running.

yugiquery.utils.helpers.make_filename(report: str, timestamp: Arrow, previous_timestamp: Arrow | None = None) → str

Generates a standardized filename based on the provided parameters.

Parameters:

report (str) – The name or identifier of the report.
timestamp (arrow.Arrow) – The timestamp to be included in the filename.
previous_timestamp (arrow.Arrow | None) – The previous timestamp, if applicable. Defaults to None.

Returns:

The generated filename.

Return type:

str

yugiquery.utils.helpers.md5(name: str) → str

Generate the MD5 hash of a string.

Parameters:: name (str) – The string to hash.
Returns:: The MD5 hash of the string.
Return type:: str

yugiquery.utils.helpers.unlock(file_name: str) → None

Release a file lock and remove the lock file.

Parameters:: file_name (str) – The name of the lock file to remove

Image Module

Notebook Module

yugiquery.utils.notebook.buttons() → HTML | None

Generates HTML buttons for updating the index and rarities/regions dictionaries. If there is no buttons.html file in the ASSETS directory, prints error message and an returns None.

Parameters:: None
Returns:: The generated HTML buttons, or None if an error occurs.
Return type:: HTML | None

Convert a Jupyter notebook to HTML using nbconvert and save the output to disk.

Parameters:

input_path (str | Path | None, optional) – The path to the Jupyter notebook file to convert. If None, gets the notebook path with get_notebook_path. Defaults to None.
output_path (str | Path | None, optional) – The path to save the converted HTML file. If None, saves the file to the REPORTS directory. Defaults to None.
template (str, optional) – The name of the nbconvert template to use. Defaults to “lab”.
theme (str | None, optional) – The name of the nbconvert theme to use. Defaults to None. If template is “lab” and “auto” theme is installed, defaults to the “auto” theme.
no_input (bool, optional) – If True, excludes input cells from the output. Defaults to True.

Raises:

ValueError – If no notebook path is provided and cannot be found with get_notebook_path.

Returns:

None

yugiquery.utils.notebook.footer(timestamp: Arrow | None = None) → HTML | None

Generates an HTML footer with a timestamp. If there is no footer.html file in the ASSETS directory, prints an error message and returns None.

Parameters:: timestamp (arrow.Arrow | None, optional) – The timestamp to use. If None, uses the current time. Defaults to None.
Returns:: The generated HTML footer, or None if an error occurs.
Return type:: HTML | None

yugiquery.utils.notebook.get_notebook_path() → Path | None

Gets the path of the current notebook opened in JupyterLab. If the path cannot be obtained, returns None.

Parameters:: None
Returns:: The path of the current notebook. Returns None if the path cannot be obtained.
Return type:: Path | None

yugiquery.utils.notebook.header(name: str | None = None, timestamp: Arrow | None = None) → HTML | None

Generates an HTML header with a timestamp and the name of the notebook (if provided). If there is no header.html file in the ASSETS directory, prints an error message and returns None.

Parameters:

name (str | None, optional) – The name of the notebook. If None, uses the current notebook stem from get_notebook_path(); falls back to “Unnamed” when unavailable. Defaults to None.
timestamp (arrow.Arrow | None, optional) – The timestamp to use. If None, uses the current time. Defaults to None.

Returns:

The generated HTML header, or None if an error occurs.

Return type:

HTML | None

yugiquery.utils.notebook.make_jekyll_page(title: str | None = None, placeholders: dict[str, str] | None = None, output_path: str | Path | None = None) → Path

Generate a Jekyll markdown page from the assets/html/index.md template.

Parameters:

title (str | None, optional) – Report title used for @title@ replacement and default output filename. If None or empty, uses the current notebook stem from get_notebook_path(); falls back to “index” when unavailable. Defaults to None.
placeholders (dict[str, str] | None, optional) – Additional placeholder values. Keys should be provided without @ symbols, e.g. {"title": "Cards"}.
output_path (str | Path | None, optional) – Output path. If None, defaults to reports/<title>.md. Relative paths are resolved relative to the reports directory.

Returns:

The path to the generated Jekyll page.

Return type:

Path

yugiquery.utils.notebook.save_notebook() → None

Save the current notebook opened in JupyterLab to disk.

Parameters:: None
Returns:: None

Plot Module

yugiquery.utils.plot.DEFAULT_FONTSIZES: dict[str, int] = {'label': 14, 'legend': 12, 'suptitle': 20, 'tick': 12, 'title': 20}: Default font sizes for plot elements (labels, titles, ticks, legends, suptitles). Keys are element names, values are font sizes in points.

class yugiquery.utils.plot.MulticolorPatchHandler

Bases: object

Custom legend handler to display a multicolored rectangle with a single uniform hatch across the entire box.

__init__(colors: List[str], hatch: str | None = None, edgecolor: str = 'black', **kwargs)

yugiquery.utils.plot.arrows(arrows: Series, figsize: Tuple[int, int] = (6, 6), **kwargs) → Figure

Create a polar plot to visualize the frequency of each arrow direction in a pandas Series.

Parameters:

arrows (pandas.Series) – A pandas Series containing arrow symbols as string data type.
figsize (Tuple[int, int], optional) – The width and height of the figure. Defaults to (6, 6).
**kwargs – Additional keyword arguments to be passed to the bar() method.

Returns:

The generated figure.

Return type:

matplotlib.figure.Figure

Raises:

ValueError – If input Series is empty or contains invalid arrow symbols.

yugiquery.utils.plot.box(df: DataFrame, mean: bool = True, group_string: str = '%Y', x: str | None = None, y: str | None = None, **kwargs) → Figure

Plots a box plot of a given DataFrame using seaborn, with the year of the timestamp column on the x-axis and the remaining column on the y-axis.

Parameters:

df (pandas.DataFrame) – The input DataFrame containing the timestamps and another numeric column.
group_string (str, optional) – The string format to group the timestamps by. Defaults to “%Y”, representing year.
mean (bool, optional) – If True, plots a line representing the mean of each box. Defaults to True.
x (str | None, optional) – The column name to use for the x-axis. Defaults to None.
y (str | None, optional) – The column name to use for the y-axis. Defaults to None.
**kwargs – Additional keyword arguments to pass to seaborn.boxplot().

Returns:

The generated figure.

Return type:

matplotlib.figure.Figure

Raises:

ValueError – If the DataFrame is empty or has no suitable x/y columns.

yugiquery.utils.plot.colors_dict = {"5D's": '#CCCCCC', '?': 'y', '???': '#ff7f0e', 'ARC-V': '#1D9E74', 'Character Card': '#FDE68A', 'Counter': '#C0C0C0', 'DARK': '#745ea5', 'DIVINE': '#7e6537', 'Duel Marker': '#C0C0C0', 'Duel Monsters': '#FF8B53', 'EARTH': '#060d0a', 'Effect Monster': '#FF8B53', 'Equip Card': '#1D9E74', 'FIRE': '#fd1b1b', 'First Series': '#FDE68A', 'Fusion Monster': '#A086B7', 'GO RUSH!!': '#BC5A84', 'GX': '#A086B7', 'LIGHT': '#9d8047', 'Level': '#f1a41f', 'Link Monster': '#00008B', 'Maximum Monsters': '#FF8B53', 'Monster Card': '#FF8B53', 'Monster Token': '#C0C0C0', 'Non-game card': '#C0C0C0', 'Normal Monster': '#FDE68A', 'Pendulum Monster': '#1D9E74', 'Ritual Monster': '#9DB5CC', 'SEVENS': '#FF8B53', 'Skill Card': '#3571B4', 'Spell Card': '#1D9E74', 'Synchro Monster': '#CCCCCC', 'Ticket Card': '#FDE68A', 'Trap Card': '#BC5A84', 'VRAINS': '#00008B', 'WATER': '#03a9e6', 'WIND': '#77bb58', 'Xyz Monster': '#000000', 'ZEXAL': '#000000'}: Dictionary mapping card or type names to color hex codes, loaded from the colors.json asset. Used for consistent color assignment in plots.

yugiquery.utils.plot.deck_composition(deck_df: DataFrame, grid_cols: int = 3, ring_radius: float = 0.3, pctdistances: List[float] = [0.85, 0.75], scale: float = 1, pie_kwargs: dict = {}, bar_kwargs: dict = {}, **kwargs) → Figure

Create a grid of pie charts displaying the composition of each deck in a DataFrame.

Parameters:

deck_df (pd.DataFrame) – The DataFrame containing the deck data.
grid_cols (int, optional) – The number of columns in the grid. Defaults to 3.
ring_radius (float, optional) – The radius of the ring in the pie chart. Defaults to 0.3.
pctdistances (List[float], optional) – The distances of the percentage labels from the center of the chart rings. Defaults to [0.85, 0.75].
scale (float, optional) – The scale factor for the plot size. Defaults to 1.
pie_kwargs (dict, optional) – Additional keyword arguments to pass to the pie chart plotting function. Defaults to None.
bar_kwargs (dict, optional) – Additional keyword arguments to pass to the side bar plotting function. Defaults to None.
**kwargs – Additional keyword arguments such as font sizes.

Returns:

The generated figure.

Return type:

matplotlib.figure.Figure

yugiquery.utils.plot.deck_distribution(deck_df: DataFrame, column: str, scale: float = 1.0, grid_cols: int = 2, colors: Dict[str, str] | List[str] | None = None, hatches: Dict[str, str] | List[str] | str = '', edgecolors: Dict[str, str] | List[str] | str = 'white', align=False, **kwargs) → Figure

Create a grid of horizontal bar charts displaying the distribution of a specified column in each deck.

Parameters:

deck_df (pd.DataFrame) – The DataFrame containing the deck data.
column (str) – The column to be plotted.
scale (float, optional) – The scaling factor for the plot. Defaults to 1.0.
grid_cols (int, optional) – The number of columns in the grid. Defaults to 2.
colors (Dict[str, str] | List[str] | None, optional) – A dictionary of colors for each section, or a list of colors to be used in the plot. If not provided, colors_dict is used. Defaults to None.
hatches (Dict[str, str] | List[str] | str, optional) – A dictionary of hatches for each section, or a list of hatches to be used in the plot. If passed, must be the same length as the number of sections in deck_df or a single string for the entire plot. Defaults to “”.
edgecolors (Dict[str, str] | List[str] | str, optional) – The colors of the edges of the bars and hatches. If passed, must be the same length as the number of sections in deck_df or a single string for the entire plot. Defaults to “white”.
**kwargs – Extra keyword arguments such as font sizes. Passed to the bar plotting function.

Returns:

The generated figure.

Return type:

matplotlib.figure.Figure

yugiquery.utils.plot.deck_stem(deck_df: DataFrame, columns: str | List[str], scale: float = 1.0, grid_cols: int = 2, colors: Dict[str, str] | List[str] | None = None, markers: Dict[str, str] | List[str] = ['s', 'o', '+'], hollow: bool = False, marker_size: int = 10, align_zero: bool = False, **kwargs) → Figure

Create a grid of stem plots displaying the distribution of a specified column in each deck.

Parameters:

deck_df (pd.DataFrame) – The DataFrame containing the deck data.
columns (str | List[str]) – The column(s) to be plotted. If a single string is passed, it will be converted to a list with one element. Must be one or two columns.
scale (float, optional) – The scaling factor for the plot size. Defaults to 1.0.
grid_cols (int, optional) – The number of columns in the grid. Defaults to 2.
colors (Dict[str, str] | List[str] | None, optional) – A dictionary of colors for each section, or a list of colors to be used in the plot. If not provided, colors_dict is used. Defaults to None.
markers (Dict[str, str], optional) – A dictionary mapping section names to marker styles. Defaults to {“Section1”: “s”, “Section2”: “o”, “Section3”: “+”}.
hollow (bool, optional) – Whether to make the markers hollow. Defaults to False.
marker_size (int, optional) – The initial size of the markers. Defaults to 10.
align_zero (bool, optional) – Whether to align the zero point of the y-axis across all subplots. Defaults to False.
**kwargs – Additional keyword arguments such as font sizes. Passed to the stem plotting function.

Returns:

The generated figure.

Return type:

matplotlib.figure.Figure

yugiquery.utils.plot.get_color(name: str | Iterable[str]) → str | list[str]

Retrieve the color hex code for a given name or a list of names from the colors_dict.

Parameters:: name (str | Iterable[str]) – The name(s) of the color(s) to retrieve.
Returns:: The hex code(s) of the color(s) associated with the given name(s).
Return type:: str | list[str]

yugiquery.utils.plot.pyramid(series: Series, use_area: bool = False, figsize: Tuple[int, int] = (8, 8), grid: bool = False, colors: List[str] | None = None, alpha: float = 1, **kwargs) → Figure

Create a pyramid (horizontal stacked bar or area) plot from a pandas Series.

Parameters:

series (pd.Series) – Data to plot, with index as labels and values as bar/area sizes.
use_area (bool, optional) – If True, use area to represent values (triangular segments); if False, use bar width. Defaults to False.
figsize (Tuple[int, int], optional) – Figure size. Defaults to (8, 8).
grid (bool, optional) – Whether to show a grid on the x-axis. Defaults to False.
colors (List[str] | None, optional) – List of colors for each segment. Defaults to matplotlib default cycle.
alpha (float, optional) – Transparency of the bars/areas. Defaults to 1.
**kwargs – Additional keyword arguments (currently unused).

Returns:

The generated pyramid plot figure.

Return type:

matplotlib.figure.Figure

yugiquery.utils.plot.rate(df: DataFrame, figsize: Tuple[int, int] | None = None, title: str = '', colors: List[str] | None = None, cumsum: bool = True, bg: DataFrame | None = None, vlines: Series | None = None, fill: bool = False, limit_year: bool = False, subplots: bool = False, hspace: float = 0.05, **kwargs) → Figure

Create a visualization of rate changes over time for multiple variables in a DataFrame.

Parameters:

df (pd.DataFrame) – The DataFrame containing the data to plot. Index must be datetime-like.
figsize (Tuple[int, int] | None, optional) – Size of the figure. Defaults to None.
title (str, optional) – Title of the plot. Defaults to an empty string.
colors (List[str] | None, optional) – List of colors for the plot lines. Defaults to None.
cumsum (bool, optional) – Whether to plot cumulative sum of data. Defaults to True.
bg (pd.DataFrame | None, optional) – Data for background shading. Defaults to None.
vlines (pd.Series | None, optional) – Series for vertical lines. Defaults to None.
fill (bool, optional) – Whether to fill the area under the cumulative sum curve. Defaults to False.
limit_year (bool, optional) – Whether to limit the x-axis to the next full year. Defaults to False.
subplots (bool, optional) – Whether to create a grid of subplots for each column in the DataFrame. Defaults to False.
hspace (float, optional) – Height space between subplots. Defaults to 0.05.
**kwargs – Additional keyword arguments to pass to the plotting functions. Not implemented.

Returns:

The generated figure.

Return type:

matplotlib.figure.Figure

Raises:

ValueError – If input DataFrame is empty or index is not datetime-like.

Progress Handler Module

class yugiquery.utils.progress_handler.ProgressHandler

Bases: object

A progress handler class to comunicate progress of a process execution.

Parameters:

queue (multiprocessing.Queue | None, optional) – The multiprocessing queue to communicate progress status. If None, a new queue is created. Defaults to None.
progress_bar (tqdm | None, optional) – The tqdm progress bar implementation. Defaults to None.
pbar_kwargs (Dict[str, Any], optional) – Keyword arguments to customize the progress bar. Defaults to None.

queue

The multiprocessing queue to communicate progress status. If None, a new queue is created. Defaults to None.

Type:: multiprocessing.Queue | None, optional

progress_bar

The tqdm progress bar implementation. Defaults to None.

Type:: tqdm | None, optional

pbar_kwargs

Keyword arguments to customize the progress bar. Defaults to None.

Type:: Dict[str, Any], optional

__init__(queue: Queue | None = None, progress_bar: type[tqdm_asyncio] | None = None, pbar_kwargs: Dict[str, Any] | None = {})

Initializes the ProgressHandler class.

Parameters:

queue (multiprocessing.Queue | None, optional) – The multiprocessing queue to communicate progress status. If None, a new queue is created. Defaults to None.
progress_bar (Type[tqdm] | None, optional) – The tqdm progress bar class. Defaults to None.
pbar_kwargs (Dict[str, Any], optional) – Keyword arguments to customize the progress bar. Defaults to None.

async await_result(process: Process) → tuple[int | None, list]

Waits asynchronously for a multiprocessing process to finish and retrieves results from a queue.

Parameters:: process (multiprocessing.Process) – The multiprocessing process to await.
Returns:: A tuple containing the API status (int or None) and a list of errors.
Return type:: tuple[int | None, list]

Notes

The method periodically checks if the process is alive and waits until it finishes.
After completion, it extracts ‘API_status’ and ‘error’ messages from the queue.
The queue is closed after processing.

pbar(**kwargs) → tqdm_asyncio | None

Creates and returns a progress bar instance with merged kwargs.

Parameters:: **kwargs – Additional keyword arguments for the progress bar (merged with stored pbar_kwargs).
Returns:: The initialized progress bar instance or None if progress_bar is None.
Return type:: tqdm | None

send(**kwargs) → None

Puts the keyword arguments into the queue

Parameters:: **kwargs – Keyword arguments to put into the queue.
Returns:: None