Skip to content

Utils

physXAI.utils.logging

Classes

Logger

Source code in physXAI/utils/logging.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
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'

    _logger = None
    _override = False

    @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):
        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

    @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)

            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)

        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)

        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
Functions
override_question(path: str) staticmethod
Source code in physXAI/utils/logging.py 
116
117
118
119
120
121
122
123
124
125
126
@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 
128
129
130
131
132
133
134
135
136
137
138
@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) staticmethod
Source code in physXAI/utils/logging.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
@staticmethod
def setup_logger(folder_name: str = None, override: bool = False, base_path: 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
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 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@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)

        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)

    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 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
@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)

    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 
232
233
234
235
236
237
238
239
@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 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
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
        print(f"Error: Cannot find git root directory.")  # pragma: no cover
        return ''  # pragma: no cover
    except Exception as e:  # pragma: no cover
        print(f"Error: An unexpected error occurred when searching for parent directory: {e}")  # 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 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
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 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
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