code/utils.py

import logging
import os
from typing import Optional


def is_float(value: str):
    """Check if the input value is float.

    :param value: value
    :return: True / False based on whether the value is float or not
    """
    try:
        float(value)
    except (TypeError, ValueError):
        return False
    else:
        return True


def is_int(value: str):
    """Check if the input value is int.

    :param value: value
    :return: True / False based on whether the value is int or not
    """
    try:
        float_value = float(value)
        int_value = int(value)
    except (TypeError, ValueError):
        return False
    else:
        return float_value == int_value


def is_list(value: str):
    """Check if the input value is list.

    Currently, we support list in the following format -

    1. "a,b,c,d"
    2. "1,2,3,4"

    :param value: value
    :return: True / False based on whether the value is list or not
    """
    return "," in value


def is_boolean(value: str):
    """Check if the input value is boolean.

    :param value: value
    :return: True / False based on whether the value is boolean or not
    """
    return value.lower() in ["true", "false"]


def parse_boolean(value: str):
    """Parse the boolean value.

    :param value: value
    :return: Parsed boolean values
    """
    return True if value.lower() == "true" else False


def parse_list(value: str):
    """Parse the list value.

    Currently, we support list in the following format -

    1. "a,b,c,d"
    2. "1,2,3,4"

    :param value: value
    :return: Parsed list
    """
    values = value.split(",")
    clean_values = [v.strip(" \"'") for v in values]
    return [infer_type_and_cast_value(v) for v in clean_values]


def infer_type_and_cast_value(value: Optional[str]):
    """Infer the type of value and casts it accordingly.

    :param value: value
    :return: casted value
    """
    if value is None:
        return value
    elif is_int(value):
        return int(value)
    elif is_float(value):
        return float(value)
    elif is_boolean(value):
        return parse_boolean(value)
    elif is_list(value):
        return parse_list(value)
    else:
        return value


def __setup_fault_handler(file_path: str = None):
    """Set up fault handler.

    :param file_path: path to the error file
    :return:
    """
    try:
        import faulthandler

        if not faulthandler.is_enabled():
            if file_path is not None:
                faulthandler.enable(os.open(file_path, os.O_APPEND), all_threads=True)
            else:
                faulthandler.enable()
    except ImportError:
        logging.warn("No faulthandler found")


def get_error_logger():
    """Return the logger from logging for id ERROR_LOGGER_ID ."""
    return logging.getLogger("error")


def setup_trusted_log(error_volume: str, error_file_path: str):
    """Set up trusted logs for the script.

    :param error_volume: volume where the errors should be written
    :param error_file_path: path to the error_file
    :return: trusted logger
    """
    trusted_log_formatter = logging.Formatter(
        "[%(asctime)s %(levelname)s %(thread)d %(filename)s:%(lineno)d] %(message)s",
        datefmt="%m/%d/%Y %H:%M:%S",
    )
    os.makedirs(error_volume, exist_ok=True)
    trusted_log_handler = logging.FileHandler(error_file_path)
    __setup_fault_handler(file_path=error_file_path)
    trusted_log_handler.setFormatter(trusted_log_formatter)
    trusted_log_handler.setLevel(logging.INFO)

    error_logger = get_error_logger()
    error_logger.addHandler(trusted_log_handler)
    error_logger.propagate = False


def write_trusted_log_info(private_info_message):
    """Write private info message to the trusted log channel.

    :param private_info_message: private trusted log message
    :return:
    """
    trusted_logger = get_error_logger()
    trusted_logger.info(private_info_message)


def write_failure_reason(failure_reason_text, file_path):
    """Write failure reason to failure file.

    :param failure_reason_text: reason for failure
    :param file_path: path to the failure file
    :return:
    """
    if not os.path.exists(os.path.dirname(file_path)):
        os.makedirs(os.path.dirname(file_path))
    with open(file_path, "w") as f:
        f.write(failure_reason_text)


def write_trusted_log_exception(
    error_message, caused_by, failure_file_path, failure_prefix="Algorithm Error"
):
    """Write private exception message to the trusted error channel.

    :param error_message: error_message
    :param caused_by: cause for the error
    :param failure_file_path: failure file path. Usually /opt/ml/output/failure
    :param failure_prefix: prefix to attach to the error message
    :return:
    """
    message = "{}: {}".format(failure_prefix, error_message)
    error_detail = "Caused by: {}".format(caused_by)
    message += "\n\n{}".format(error_detail)
    err_logger = get_error_logger()
    err_logger.exception(message)
    write_failure_reason(message, failure_file_path)
    return message