Utils

physXAI.utils.logging

Classes

Logger

Source code in physXAI/utils/logging.py

class Logger:

    save_name_preprocessing: str = 'preprocessing_config.json'
    save_name_model_config: str = 'model_config.json'
    save_name_constructed: str = 'constructed_config.json'
    save_name_training_data_multi_step: str = 'training_data'
    save_name_training_data_multi_step_format: str = 'zip'
    save_name_training_data_json: str = 'training_data.json'
    base_path = 'stored_data'
    save_name_model: str = 'model'
    save_name_model_online_learning: str = 'model_ol'

    print_level: str = 'info'  # options: 'debug', 'info', 'warning', 'error'
    _print_levels = ['debug', 'info', 'warning', 'error']

    _logger = None
    _override = False

    @staticmethod
    def print(message: str, print_level: str = 'info'):
        if Logger.check_print_level(print_level):
            print(message)

    @staticmethod
    def check_print_level(print_level: str) -> bool:
        if str(print_level).lower() not in Logger._print_levels:
            raise ValueError(f"Invalid print level: {str(print_level).lower()}. Valid options are: {Logger._print_levels}")
        if Logger._print_levels.index(str(print_level).lower()) >= Logger._print_levels.index(Logger.print_level):
            return True
        else:
            return False

    @staticmethod
    def verbosity() -> Union[int, str]:
        if Logger._print_levels.index(Logger.print_level) >= Logger._print_levels.index('warning'):
            return 0
        else:
            return "auto"

    @staticmethod
    def verbosity_int() -> int:
        if Logger._print_levels.index(Logger.print_level) >= Logger._print_levels.index('warning'):
            return 0
        else:
            return 1

    @staticmethod
    def override_question(path: str):  # pragma: no cover
        if os.path.exists(path) and not Logger._override:
            try:
                user_input = input(f"Path {path} already exists. Do you want to override it (y/n)?").strip().lower()
                if user_input in ['y', 'yes', 'j', 'ja', 'true', '1']:
                    shutil.rmtree(path)
                else:
                    raise OSError(f"Path {path} already exists.")
            except OSError as e:
                raise e

    @staticmethod
    def already_exists_question(path: str):  # pragma: no cover
        if os.path.exists(path) and not Logger._override:
            try:
                user_input = input(f"Path {path} already exists. Do you want to proceed (y/n)?").strip().lower()
                if user_input in ['y', 'yes', 'j', 'ja', 'true', '1']:
                    return
                else:
                    raise OSError(f"Path {path} already exists.")
            except OSError as e:
                raise e

    @staticmethod
    def setup_logger(folder_name: str = None, override: bool = False, base_path: str = None, print_level: str = None):
        if base_path is None:
            base_path = Logger.base_path
        if folder_name is None:
            folder_name = datetime.now().strftime("%d.%m.%y %H:%M:%S")
            folder_name = os.path.join(base_path, folder_name)
        else:
            folder_name = os.path.join(base_path, folder_name)
        path = get_full_path(folder_name, raise_error=False)
        if not override and os.path.exists(path):
            Logger.already_exists_question(path)
        create_full_path(path)

        Logger._logger = path
        Logger._override = override

        if print_level is not None:
            if str(print_level).lower() not in Logger._print_levels:
                raise ValueError(f"Invalid print level: {str(print_level).lower()}. Valid options are: {Logger._print_levels}")
            Logger.print_level = str(print_level).lower()

    @staticmethod
    def log_setup(preprocessing=None, model=None, save_name_preprocessing=None, save_name_model=None,
                  save_name_constructed=None):
        if Logger._logger is None:
            Logger.setup_logger()

        if preprocessing is not None:
            try:
                preprocessing_dict = preprocessing.get_config()
            except AttributeError:  # pragma: no cover
                raise AttributeError('Error: Preprocessing object has no attribute "get_config()".')  # pragma: no cover
            if save_name_preprocessing is None:
                save_name_preprocessing = Logger.save_name_preprocessing
            path = os.path.join(Logger._logger, save_name_preprocessing)
            path = create_full_path(path)
            Logger.override_question(path)
            with open(path, "w") as f:
                json.dump(preprocessing_dict, f, indent=4)

            from physXAI.preprocessing.constructed import FeatureConstruction
            constructed_config = FeatureConstruction.get_config()
            if len(constructed_config) > 0:
                if save_name_constructed is None:
                    save_name_constructed = Logger.save_name_constructed
                path = os.path.join(Logger._logger, save_name_constructed)
                path = create_full_path(path)
                Logger.override_question(path)
                with open(path, "w") as f:
                    json.dump(constructed_config, f, indent=4)

            FeatureConstruction.reset()

        if model is not None:
            try:
                model_dict = model.get_config()
            except AttributeError:  # pragma: no cover
                raise AttributeError('Error: Model object has no attribute "get_config()".')  # pragma: no cover
            if save_name_model is None:
                save_name_model = Logger.save_name_model_config
            path = os.path.join(Logger._logger, save_name_model)
            path = create_full_path(path)
            Logger.override_question(path)
            with open(path, "w") as f:
                json.dump(model_dict, f, indent=4)

    @staticmethod
    def save_training_data(training_data, path: str = None):
        if Logger._logger is None:
            Logger.setup_logger()

        try:
            td_dict = training_data.get_config()
        except AttributeError:  # pragma: no cover
            raise AttributeError('Error: Training data object has no attribute "get_config()".')  # pragma: no cover

        if path is None:
            path = Logger.save_name_training_data_json
        else:
            if len(path.split('.json')) == 1:
                # join .json to path in case it is not yet included
                path = path + '.json'

        p = os.path.join(Logger._logger, path)
        p = create_full_path(p)
        Logger.override_question(p)
        with open(p, "w") as f:
            json.dump(td_dict, f, indent=4)

        from physXAI.preprocessing.training_data import TrainingDataMultiStep
        if isinstance(training_data, TrainingDataMultiStep):
            training_data = copy.copy(training_data)
            training_data.train_ds = None
            training_data.val_ds = None
            training_data.test_ds = None

        p = p.split('.json')[0]
        with open(p + '.pkl', "wb") as f:
             pickle.dump(training_data, f)

    @staticmethod
    def get_model_savepath():
        if Logger._logger is None:
            Logger.setup_logger()

        p = os.path.join(Logger._logger, Logger.save_name_model)

        return p

Attributes

save_name_preprocessing: str = 'preprocessing_config.json' class-attribute instance-attribute

save_name_model_config: str = 'model_config.json' class-attribute instance-attribute

save_name_constructed: str = 'constructed_config.json' class-attribute instance-attribute

save_name_training_data_multi_step: str = 'training_data' class-attribute instance-attribute

save_name_training_data_multi_step_format: str = 'zip' class-attribute instance-attribute

save_name_training_data_json: str = 'training_data.json' class-attribute instance-attribute

base_path = 'stored_data' class-attribute instance-attribute

save_name_model: str = 'model' class-attribute instance-attribute

save_name_model_online_learning: str = 'model_ol' class-attribute instance-attribute

print_level: str = 'info' class-attribute instance-attribute

Functions

print(message: str, print_level: str = 'info') staticmethod

Source code in physXAI/utils/logging.py

@staticmethod
def print(message: str, print_level: str = 'info'):
    if Logger.check_print_level(print_level):
        print(message)

check_print_level(print_level: str) -> bool staticmethod

Source code in physXAI/utils/logging.py

@staticmethod
def check_print_level(print_level: str) -> bool:
    if str(print_level).lower() not in Logger._print_levels:
        raise ValueError(f"Invalid print level: {str(print_level).lower()}. Valid options are: {Logger._print_levels}")
    if Logger._print_levels.index(str(print_level).lower()) >= Logger._print_levels.index(Logger.print_level):
        return True
    else:
        return False

verbosity() -> Union[int, str] staticmethod

Source code in physXAI/utils/logging.py

@staticmethod
def verbosity() -> Union[int, str]:
    if Logger._print_levels.index(Logger.print_level) >= Logger._print_levels.index('warning'):
        return 0
    else:
        return "auto"

verbosity_int() -> int staticmethod

Source code in physXAI/utils/logging.py

@staticmethod
def verbosity_int() -> int:
    if Logger._print_levels.index(Logger.print_level) >= Logger._print_levels.index('warning'):
        return 0
    else:
        return 1

override_question(path: str) staticmethod

Source code in physXAI/utils/logging.py

@staticmethod
def override_question(path: str):  # pragma: no cover
    if os.path.exists(path) and not Logger._override:
        try:
            user_input = input(f"Path {path} already exists. Do you want to override it (y/n)?").strip().lower()
            if user_input in ['y', 'yes', 'j', 'ja', 'true', '1']:
                shutil.rmtree(path)
            else:
                raise OSError(f"Path {path} already exists.")
        except OSError as e:
            raise e

already_exists_question(path: str) staticmethod

Source code in physXAI/utils/logging.py

@staticmethod
def already_exists_question(path: str):  # pragma: no cover
    if os.path.exists(path) and not Logger._override:
        try:
            user_input = input(f"Path {path} already exists. Do you want to proceed (y/n)?").strip().lower()
            if user_input in ['y', 'yes', 'j', 'ja', 'true', '1']:
                return
            else:
                raise OSError(f"Path {path} already exists.")
        except OSError as e:
            raise e

setup_logger(folder_name: str = None, override: bool = False, base_path: str = None, print_level: str = None) staticmethod

Source code in physXAI/utils/logging.py

@staticmethod
def setup_logger(folder_name: str = None, override: bool = False, base_path: str = None, print_level: str = None):
    if base_path is None:
        base_path = Logger.base_path
    if folder_name is None:
        folder_name = datetime.now().strftime("%d.%m.%y %H:%M:%S")
        folder_name = os.path.join(base_path, folder_name)
    else:
        folder_name = os.path.join(base_path, folder_name)
    path = get_full_path(folder_name, raise_error=False)
    if not override and os.path.exists(path):
        Logger.already_exists_question(path)
    create_full_path(path)

    Logger._logger = path
    Logger._override = override

    if print_level is not None:
        if str(print_level).lower() not in Logger._print_levels:
            raise ValueError(f"Invalid print level: {str(print_level).lower()}. Valid options are: {Logger._print_levels}")
        Logger.print_level = str(print_level).lower()

log_setup(preprocessing=None, model=None, save_name_preprocessing=None, save_name_model=None, save_name_constructed=None) staticmethod

Source code in physXAI/utils/logging.py

@staticmethod
def log_setup(preprocessing=None, model=None, save_name_preprocessing=None, save_name_model=None,
              save_name_constructed=None):
    if Logger._logger is None:
        Logger.setup_logger()

    if preprocessing is not None:
        try:
            preprocessing_dict = preprocessing.get_config()
        except AttributeError:  # pragma: no cover
            raise AttributeError('Error: Preprocessing object has no attribute "get_config()".')  # pragma: no cover
        if save_name_preprocessing is None:
            save_name_preprocessing = Logger.save_name_preprocessing
        path = os.path.join(Logger._logger, save_name_preprocessing)
        path = create_full_path(path)
        Logger.override_question(path)
        with open(path, "w") as f:
            json.dump(preprocessing_dict, f, indent=4)

        from physXAI.preprocessing.constructed import FeatureConstruction
        constructed_config = FeatureConstruction.get_config()
        if len(constructed_config) > 0:
            if save_name_constructed is None:
                save_name_constructed = Logger.save_name_constructed
            path = os.path.join(Logger._logger, save_name_constructed)
            path = create_full_path(path)
            Logger.override_question(path)
            with open(path, "w") as f:
                json.dump(constructed_config, f, indent=4)

        FeatureConstruction.reset()

    if model is not None:
        try:
            model_dict = model.get_config()
        except AttributeError:  # pragma: no cover
            raise AttributeError('Error: Model object has no attribute "get_config()".')  # pragma: no cover
        if save_name_model is None:
            save_name_model = Logger.save_name_model_config
        path = os.path.join(Logger._logger, save_name_model)
        path = create_full_path(path)
        Logger.override_question(path)
        with open(path, "w") as f:
            json.dump(model_dict, f, indent=4)

save_training_data(training_data, path: str = None) staticmethod

Source code in physXAI/utils/logging.py

@staticmethod
def save_training_data(training_data, path: str = None):
    if Logger._logger is None:
        Logger.setup_logger()

    try:
        td_dict = training_data.get_config()
    except AttributeError:  # pragma: no cover
        raise AttributeError('Error: Training data object has no attribute "get_config()".')  # pragma: no cover

    if path is None:
        path = Logger.save_name_training_data_json
    else:
        if len(path.split('.json')) == 1:
            # join .json to path in case it is not yet included
            path = path + '.json'

    p = os.path.join(Logger._logger, path)
    p = create_full_path(p)
    Logger.override_question(p)
    with open(p, "w") as f:
        json.dump(td_dict, f, indent=4)

    from physXAI.preprocessing.training_data import TrainingDataMultiStep
    if isinstance(training_data, TrainingDataMultiStep):
        training_data = copy.copy(training_data)
        training_data.train_ds = None
        training_data.val_ds = None
        training_data.test_ds = None

    p = p.split('.json')[0]
    with open(p + '.pkl', "wb") as f:
         pickle.dump(training_data, f)

get_model_savepath() staticmethod

Source code in physXAI/utils/logging.py

@staticmethod
def get_model_savepath():
    if Logger._logger is None:
        Logger.setup_logger()

    p = os.path.join(Logger._logger, Logger.save_name_model)

    return p

Functions

get_parent_working_directory() -> str

Finds the root directory of the Git repository that contains the current working directory.

This function is useful for locating project-relative paths when the script might be run from different subdirectories within a Git project.

Returns:

Name	Type	Description
str	str	The absolute path to the root of the Git working tree if found. Returns an empty string if not in a Git repository or if an error occurs.

Source code in physXAI/utils/logging.py

def get_parent_working_directory() -> str:
    """
    Finds the root directory of the Git repository that contains the current working directory.

    This function is useful for locating project-relative paths when the script
    might be run from different subdirectories within a Git project.

    Returns:
        str: The absolute path to the root of the Git working tree if found.
             Returns an empty string if not in a Git repository or if an error occurs.
    """

    try:
        repo = git.Repo(search_parent_directories=True)
        git_root = repo.working_tree_dir
        return git_root
    except git.InvalidGitRepositoryError:  # pragma: no cover
        Logger.print(f"Error: Cannot find git root directory.", 'error')  # pragma: no cover
        return ''  # pragma: no cover
    except Exception as e:  # pragma: no cover
        Logger.print(f"Error: An unexpected error occurred when searching for parent directory: {e}", 'error')  # pragma: no cover
        return ''  # pragma: no cover

get_full_path(path: str, raise_error=True) -> str

Resolves a given path to an absolute path. If the path is relative, it first checks relative to the current working directory. If not found, it attempts to resolve it relative to the Git project's root directory.

Parameters:

Name	Type	Description	Default
path	str	The path string to resolve (can be absolute or relative).	required
raise_error	bool	If True (default), raises a FileNotFoundError if the path cannot be resolved. If False, returns the constructed path even if it doesn't exist.	True

Returns:

Name	Type	Description
str	str	The resolved absolute path. If raise_error is False and the path is not found, it returns the last attempted path construction.

Raises:

Type	Description
FileNotFoundError	If raise_error is True and the path cannot be found.

Source code in physXAI/utils/logging.py

def get_full_path(path: str, raise_error=True) -> str:
    """
    Resolves a given path to an absolute path.
    If the path is relative, it first checks relative to the current working directory.
    If not found, it attempts to resolve it relative to the Git project's root directory.

    Args:
        path (str): The path string to resolve (can be absolute or relative).
        raise_error (bool, optional): If True (default), raises a FileNotFoundError
                                      if the path cannot be resolved. If False,
                                      returns the constructed path even if it doesn't exist.

    Returns:
        str: The resolved absolute path. If `raise_error` is False and the path
             is not found, it returns the last attempted path construction.

    Raises:
        FileNotFoundError: If `raise_error` is True and the path cannot be found.
    """

    if os.path.exists(path):
        return path
    parent = get_parent_working_directory()
    path = os.path.join(parent, path)
    if os.path.exists(path):
        return path
    elif raise_error:
        raise FileNotFoundError(f'Path "{path}" does not exist.')
    else:
        return path

create_full_path(path: str) -> str

Ensures that the directory structure for a given file path exists, creating it if necessary. Returns the absolute version of the input path.

Parameters:

Name	Type	Description	Default
path	str	The file path for which the directory structure should be created. This can be a path to a file or just a directory.	required

Returns:

Name	Type	Description
str	str	The absolute path, with its directory structure ensured to exist.

Raises:

Type	Description
OSError	If os.makedirs fails for reasons other than the directory already existing.

Source code in physXAI/utils/logging.py

def create_full_path(path: str) -> str:
    """
    Ensures that the directory structure for a given file path exists, creating
    it if necessary. Returns the absolute version of the input path.

    Args:
        path (str): The file path for which the directory structure should be created.
                    This can be a path to a file or just a directory.

    Returns:
        str: The absolute path, with its directory structure ensured to exist.

    Raises:
        OSError: If `os.makedirs` fails for reasons other than the directory already existing.
    """

    directory = os.path.dirname(path)
    file = os.path.basename(path)

    directory = get_full_path(directory, raise_error=False)
    path = os.path.join(directory, file)

    if not os.path.exists(directory):
        try:
            os.makedirs(directory, exist_ok=True)
        except OSError as e:
            raise e

    return path