utils

aws

download_s3(bucket_name, object_name, output_path)

Download a file from an S3 bucket and save it to the local file system.

Parameters:

Name	Type	Description	Default
bucket_name	str	The name of the S3 bucket.	required
object_name	str	The key or path of the object to be downloaded from the bucket.	required
output_path	str	The local file path to save the downloaded object.	required

Source code in aimet_ml/utils/aws.py

def download_s3(bucket_name: str, object_name: str, output_path: str):
    """
    Download a file from an S3 bucket and save it to the local file system.

    Args:
        bucket_name (str): The name of the S3 bucket.
        object_name (str): The key or path of the object to be downloaded from the bucket.
        output_path (str): The local file path to save the downloaded object.
    """
    s3 = boto3.client("s3")
    s3.download_file(bucket_name, object_name, output_path)

upload_dir_s3(bucket_name, bucket_dir_path, src_dir_path)

Upload a local directory to an S3 bucket, preserving the directory structure.

Parameters:

Name	Type	Description	Default
bucket_name	str	The name of the S3 bucket.	required
bucket_dir_path	str	The path within the bucket where the local directory will be uploaded.	required
src_dir_path	str	The local directory path to be uploaded.	required

Source code in aimet_ml/utils/aws.py

def upload_dir_s3(bucket_name: str, bucket_dir_path: str, src_dir_path: str):
    """
    Upload a local directory to an S3 bucket, preserving the directory structure.

    Args:
        bucket_name (str): The name of the S3 bucket.
        bucket_dir_path (str): The path within the bucket where the local directory will be uploaded.
        src_dir_path (str): The local directory path to be uploaded.
    """
    s3 = boto3.client("s3")
    src_dir_name = os.path.basename(src_dir_path)
    for r, _, f in tqdm(os.walk(src_dir_path)):
        for n in f:
            file_path = os.path.join(r, n)
            relpath = os.path.relpath(r, src_dir_path)
            object_name = os.path.join(bucket_dir_path, src_dir_name, "" if relpath == "." else relpath, n)
            s3.upload_file(file_path, bucket_name, object_name)

upload_files_s3(bucket_name, bucket_dir_path, src_file_paths)

Upload multiple local files to an S3 bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	The name of the S3 bucket.	required
bucket_dir_path	str	The path within the bucket where the files will be uploaded.	required
src_file_paths	list	A list of local file paths to be uploaded to the bucket.	required

Source code in aimet_ml/utils/aws.py

def upload_files_s3(bucket_name: str, bucket_dir_path: str, src_file_paths: list):
    """
    Upload multiple local files to an S3 bucket.

    Args:
        bucket_name (str): The name of the S3 bucket.
        bucket_dir_path (str): The path within the bucket where the files will be uploaded.
        src_file_paths (list): A list of local file paths to be uploaded to the bucket.
    """
    s3 = boto3.client("s3")
    for file_path in tqdm(src_file_paths):
        object_name = os.path.join(bucket_dir_path, os.path.basename(file_path))
        s3.upload_file(file_path, bucket_name, object_name)

git

get_commit_id(short=True)

Get the Git commit ID of the current repository.

Parameters:

Name	Type	Description	Default
short	bool	Whether to get a short or full Git commit ID. Defaults to True.	True

Returns:

Name	Type	Description
str	str	The Git commit ID as a string.

Source code in aimet_ml/utils/git.py

def get_commit_id(short: bool = True) -> str:
    """
    Get the Git commit ID of the current repository.

    Args:
        short (bool, optional): Whether to get a short or full Git commit ID. Defaults to True.

    Returns:
        str: The Git commit ID as a string.
    """
    if short:
        git_command = "git rev-parse --short HEAD"
    else:
        git_command = "git rev-parse HEAD"

    commit_id = os.popen(git_command).read().strip()

    return commit_id

hamd_7

score_to_severity(score)

Convert a score to a severity category.

Parameters:

Name	Type	Description	Default
score	int	The input score.	required

Returns:

Name	Type	Description
str	str	The corresponding severity category ("normal", "mild", "moderate", or "severe").

Source code in aimet_ml/utils/hamd_7.py

def score_to_severity(score: int) -> str:
    """
    Convert a score to a severity category.

    Args:
        score (int): The input score.

    Returns:
        str: The corresponding severity category ("normal", "mild", "moderate", or "severe").
    """
    if score < 5:
        return "normal"
    if score < 13:
        return "mild"
    if score < 21:
        return "moderate"
    return "severe"

io_utils

read_json(file_path)

Read and parse a JSON file.

Parameters:

Name	Type	Description	Default
file_path	str	The path to the JSON file to be read.	required

Returns:

Name	Type	Description
dict	dict	A dictionary containing the parsed JSON data.

Source code in aimet_ml/utils/io_utils.py

def read_json(file_path: str) -> dict:
    """
    Read and parse a JSON file.

    Args:
        file_path (str): The path to the JSON file to be read.

    Returns:
        dict: A dictionary containing the parsed JSON data.
    """
    with open(file_path, "r") as f:
        data = json.load(f)
    return data

read_pickle(file_path)

Read and unpickle a binary pickle file.

Parameters:

Name	Type	Description	Default
file_path	str	The path to the pickle file to be read.	required

Returns:

Name	Type	Description
Any		The unpickled object.

Source code in aimet_ml/utils/io_utils.py

def read_pickle(file_path: str):
    """
    Read and unpickle a binary pickle file.

    Args:
        file_path (str): The path to the pickle file to be read.

    Returns:
        Any: The unpickled object.
    """
    with open(file_path, "rb") as f:
        data = pickle.load(f)
    return data

read_yaml(file_path)

Read and parse a YAML file.

Parameters:

Name	Type	Description	Default
file_path	str	The path to the YAML file to be read.	required

Returns:

Name	Type	Description
dict	dict	A dictionary containing the parsed YAML data.

Source code in aimet_ml/utils/io_utils.py

def read_yaml(file_path: str) -> dict:
    """
    Read and parse a YAML file.

    Args:
        file_path (str): The path to the YAML file to be read.

    Returns:
        dict: A dictionary containing the parsed YAML data.
    """
    with open(file_path, "r") as f:
        result = yaml.safe_load(f)
    return result

write_json(file_path, data, indent=None)

Write data to a JSON file.

Parameters:

Name	Type	Description	Default
file_path	str	The path to the JSON file to be written.	required
data	dict	The data to be written to the JSON file.	required
indent	int	The number of spaces to use for indentation.	None

Source code in aimet_ml/utils/io_utils.py

def write_json(file_path: str, data: dict, indent: Optional[int] = None):
    """
    Write data to a JSON file.

    Args:
        file_path (str): The path to the JSON file to be written.
        data (dict): The data to be written to the JSON file.
        indent (int, optional): The number of spaces to use for indentation.
    """
    with open(file_path, "w") as f:
        json.dump(data, f, indent=indent)

write_pickle(file_path, data)

Write data to a binary pickle file.

Parameters:

Name	Type	Description	Default
file_path	str	The path to the pickle file to be written.	required
data		The data to be pickled and written to the file.	required

Source code in aimet_ml/utils/io_utils.py

def write_pickle(file_path: str, data):
    """
    Write data to a binary pickle file.

    Args:
        file_path (str): The path to the pickle file to be written.
        data: The data to be pickled and written to the file.
    """
    with open(file_path, "wb") as f:
        pickle.dump(data, f)

write_yaml(file_path, data, default_flow_style=False)

Write data to a YAML file.

Parameters:

Name	Type	Description	Default
file_path	str	The path to the YAML file to be written.	required
data	dict	The data to be written to the YAML file.	required
default_flow_style	bool	Whether to use the default flow style for YAML.	False

Source code in aimet_ml/utils/io_utils.py

def write_yaml(file_path: str, data: dict, default_flow_style: bool = False):
    """
    Write data to a YAML file.

    Args:
        file_path (str): The path to the YAML file to be written.
        data (dict): The data to be written to the YAML file.
        default_flow_style (bool, optional): Whether to use the default flow style for YAML.
    """
    with open(file_path, "w") as f:
        yaml.dump(data, f, default_flow_style=default_flow_style)

plots

add_bar_label(bar_chart, with_percent=False, percent_digits=2)

Add labels to a bar chart with optional percentage values.

Parameters:

Name	Type	Description	Default
bar_chart	Axes	The bar chart object.	required
with_percent	bool	Whether to include percentage values. Defaults to False.	False
percent_digits	int	Number of decimal digits for percentage values. Defaults to 2.	2

Source code in aimet_ml/utils/plots.py

def add_bar_label(bar_chart: plt.Axes, with_percent: bool = False, percent_digits: int = 2) -> None:
    """
    Add labels to a bar chart with optional percentage values.

    Args:
        bar_chart (plt.Axes): The bar chart object.
        with_percent (bool, optional): Whether to include percentage values. Defaults to False.
        percent_digits (int, optional): Number of decimal digits for percentage values. Defaults to 2.
    """
    containers = bar_chart.containers[0]
    labels = None
    if with_percent:
        datavalues = containers.datavalues
        total = sum(datavalues)
        labels = [f"{v:,} ({v/total*100:.{percent_digits}f}%)" for v in datavalues]
    bar_chart.bar_label(containers, labels)

plt2arr(fig, draw=True)

Convert a Matplotlib figure to a NumPy array.

Parameters:

Name	Type	Description	Default
fig	Figure	The Matplotlib figure to be converted.	required
draw	bool	Whether to draw the figure. Defaults to True.	True

Returns:

Type	Description
ndarray	np.ndarray: The converted figure as a NumPy array.

Source code in aimet_ml/utils/plots.py

def plt2arr(fig: Figure, draw: bool = True) -> np.ndarray:
    """
    Convert a Matplotlib figure to a NumPy array.

    Args:
        fig (Figure): The Matplotlib figure to be converted.
        draw (bool, optional): Whether to draw the figure. Defaults to True.

    Returns:
        np.ndarray: The converted figure as a NumPy array.
    """
    if draw:
        fig.canvas.draw()
    rgba_buf = fig.canvas.buffer_rgba()
    (w, h) = fig.canvas.get_width_height()
    rgba_arr = np.frombuffer(rgba_buf, dtype=np.uint8).reshape((h, w, 4))
    return rgba_arr

set_font(font_path)

Set the font for Matplotlib using the provided font file.

Parameters:

Name	Type	Description	Default
font_path	str	Path to the font file.	required

Source code in aimet_ml/utils/plots.py

def set_font(font_path: str) -> None:
    """
    Set the font for Matplotlib using the provided font file.

    Args:
        font_path (str): Path to the font file.
    """
    import matplotlib
    import matplotlib.font_manager

    font_prop = matplotlib.font_manager.FontProperties(fname=font_path)
    matplotlib.font_manager.fontManager.addfont(font_path)
    matplotlib.rc("font", family=font_prop.get_name())

set_thai_font()

Set the Thai font for Matplotlib using a predefined font path.

Source code in aimet_ml/utils/plots.py

def set_thai_font() -> None:
    """Set the Thai font for Matplotlib using a predefined font path."""
    font_path = str(PWD.parent / "resources" / "fonts" / "thsarabunnew-webfont.ttf")
    set_font(font_path)

wandb_utils

list_artifact_names(api, artifact_type, with_versions=True, with_aliases=True, per_page=100)

List available artifact names for a specific artifact type.

Parameters:

Name	Type	Description	Default
api	Api	The WandB API client.	required
artifact_type	str	The type of artifact for which names are listed.	required
with_versions	bool	Include version suffixes. Defaults to True.	True
with_aliases	bool	Include artifact aliases. Defaults to True.	True
per_page	int	Number of items to retrieve per page. Defaults to 100.	100

Returns:

Name	Type	Description
list	list	A sorted list of available artifact names with optional suffixes (versions or aliases).

Source code in aimet_ml/utils/wandb_utils.py

def list_artifact_names(
    api: wandb.Api, artifact_type: str, with_versions: bool = True, with_aliases: bool = True, per_page: int = 100
) -> list:
    """
    List available artifact names for a specific artifact type.

    Args:
        api (wandb.Api): The WandB API client.
        artifact_type (str): The type of artifact for which names are listed.
        with_versions (bool, optional): Include version suffixes. Defaults to True.
        with_aliases (bool, optional): Include artifact aliases. Defaults to True.
        per_page (int, optional): Number of items to retrieve per page. Defaults to 100.

    Returns:
        list: A sorted list of available artifact names with optional suffixes (versions or aliases).
    """
    available_artifact_names = set()

    for collection in api.artifact_type(artifact_type).collections():
        suffixes = set()

        if with_aliases:
            suffixes.update(collection.aliases)

        if with_versions:
            suffixes.update([v.version for v in collection.versions(per_page=per_page)])

        if suffixes:
            available_artifact_names.update([f"{collection.name}:{suffix}" for suffix in suffixes])
        else:
            available_artifact_names.update([collection.name])

    return sorted(available_artifact_names)

load_artifact(api, artifact_type, artifact_name, artifact_alias, per_page=100)

Load a WandB artifact by name and alias.

Parameters:

Name	Type	Description	Default
api	Api	The WandB API client.	required
artifact_type	str	The type of artifact to load.	required
artifact_name	str	The base name of the artifact.	required
artifact_alias	str	The alias of the artifact.	required
per_page	int	Number of items to retrieve per page. Defaults to 100.	100

Returns:

Type	Description
Union[Artifact, None]	wandb.Artifact: The loaded WandB artifact or None if it doesn't exist.

Source code in aimet_ml/utils/wandb_utils.py

def load_artifact(
    api: wandb.Api, artifact_type: str, artifact_name: str, artifact_alias: str, per_page: int = 100
) -> Union[wandb.Artifact, None]:
    """
    Load a WandB artifact by name and alias.

    Args:
        api (wandb.Api): The WandB API client.
        artifact_type (str): The type of artifact to load.
        artifact_name (str): The base name of the artifact.
        artifact_alias (str): The alias of the artifact.
        per_page (int, optional): Number of items to retrieve per page. Defaults to 100.

    Returns:
        wandb.Artifact: The loaded WandB artifact or None if it doesn't exist.
    """
    available_artifact_types = [t.name for t in api.artifact_types()]
    if artifact_type not in available_artifact_types:
        return None

    available_artifact_names = list_artifact_names(api, artifact_type, per_page=per_page)

    artifact_name_with_alias = f"{artifact_name}:{artifact_alias}"

    if artifact_name_with_alias not in available_artifact_names:
        return None

    return wandb.use_artifact(f"{api.settings['entity']}/{api.settings['project']}/{artifact_name_with_alias}")

table_to_dataframe(table)

Convert a WandB table to a Pandas DataFrame.

Parameters:

Name	Type	Description	Default
table	Table	The WandB table to be converted.	required

Returns:

Type	Description
DataFrame	pd.DataFrame: A Pandas DataFrame containing the data from the WandB table.

Source code in aimet_ml/utils/wandb_utils.py

def table_to_dataframe(table: wandb.Table) -> pd.DataFrame:
    """
    Convert a WandB table to a Pandas DataFrame.

    Args:
        table (wandb.Table): The WandB table to be converted.

    Returns:
        pd.DataFrame: A Pandas DataFrame containing the data from the WandB table.
    """
    return pd.DataFrame(data=table.data, columns=table.columns)