Skip to content

Commit

Permalink
make better files in aux, models_zoo.py + some small fix
Browse files Browse the repository at this point in the history
  • Loading branch information
LukyanovKirillML committed Nov 21, 2024
1 parent 7d06840 commit 87854b3
Show file tree
Hide file tree
Showing 8 changed files with 210 additions and 63 deletions.
36 changes: 28 additions & 8 deletions src/aux/custom_decorators.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,19 +2,28 @@
from functools import wraps
import logging
import functools
from typing import Callable

logging.basicConfig(level=logging.INFO)


def retry(max_tries=3, delay_seconds=1):
def retry(
max_tries: int = 3,
delay_seconds: int = 1
):
"""
Allows you to re-execute the program after the Nth amount of time
:param max_tries: number of restart attempts
:param delay_seconds: time interval between attempts
"""
def decorator_retry(func):
def decorator_retry(
func: Callable
) -> Callable:
@wraps(func)
def wrapper_retry(*args, **kwargs):
def wrapper_retry(
*args,
**kwargs
):
tries = 0
while tries < max_tries:
try:
Expand All @@ -30,13 +39,17 @@ def wrapper_retry(*args, **kwargs):
return decorator_retry


def memoize(func):
def memoize(
func: Callable
) -> Callable:
"""
Caching function
"""
cache = {}

def wrapper(*args):
def wrapper(
*args
):
if args in cache:
return cache[args]
else:
Expand All @@ -47,11 +60,16 @@ def wrapper(*args):
return wrapper


def timing_decorator(func):
def timing_decorator(
func: Callable
) -> Callable:
"""
Timing functions
"""
def wrapper(*args, **kwargs):
def wrapper(
*args,
**kwargs
):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
Expand All @@ -61,7 +79,9 @@ def wrapper(*args, **kwargs):
return wrapper


def log_execution(func):
def log_execution(
func: Callable
) -> Callable:
"""
Function call logging
"""
Expand Down
85 changes: 63 additions & 22 deletions src/aux/data_info.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@
import importlib.util
import json
import logging
from typing import List, Tuple, Union

# from pydantic.utils import deep_update
# from pydantic.v1.utils import deep_update

Expand All @@ -24,7 +26,8 @@ class DataInfo:
"""

@staticmethod
def refresh_all_data_info():
def refresh_all_data_info(
) -> None:
"""
Calling all files to update with information about saved objects
"""
Expand All @@ -35,7 +38,8 @@ def refresh_all_data_info():
DataInfo.refresh_explanations_dir_structure()

@staticmethod
def refresh_data_dir_structure():
def refresh_data_dir_structure(
) -> None:
"""
Calling a file update with information about saved raw datasets
"""
Expand All @@ -51,7 +55,8 @@ def refresh_data_dir_structure():
prev_path = path

@staticmethod
def refresh_models_dir_structure():
def refresh_models_dir_structure(
) -> None:
"""
Calling a file update with information about saved models
"""
Expand All @@ -62,7 +67,8 @@ def refresh_models_dir_structure():
f.write(str(Path(*path)) + '\n')

@staticmethod
def refresh_explanations_dir_structure():
def refresh_explanations_dir_structure(
) -> None:
"""
Calling a file update with information about saved explanations
"""
Expand All @@ -73,7 +79,8 @@ def refresh_explanations_dir_structure():
f.write(str(Path(*path)) + '\n')

@staticmethod
def refresh_data_var_dir_structure():
def refresh_data_var_dir_structure(
) -> None:
"""
Calling a file update with information about saved prepared datasets
"""
Expand All @@ -84,7 +91,9 @@ def refresh_data_var_dir_structure():
f.write(str(Path(*path)) + '\n')

@staticmethod
def take_keys_etc_by_prefix(prefix):
def take_keys_etc_by_prefix(
prefix: Tuple
) -> [List, List, dict, int]:
"""
:param prefix: what data and in what order were used to form the path when saving the object
Expand Down Expand Up @@ -114,7 +123,11 @@ def take_keys_etc_by_prefix(prefix):
return keys_list, full_keys_list, dir_structure, empty_dir_shift

@staticmethod
def values_list_by_path_and_keys(path, full_keys_list, dir_structure):
def values_list_by_path_and_keys(
path: Union[str, Path],
full_keys_list: List,
dir_structure: dict
) -> List:
"""
:param path: path of the saved object
Expand All @@ -133,7 +146,10 @@ def values_list_by_path_and_keys(path, full_keys_list, dir_structure):
return parts_val

@staticmethod
def values_list_and_technical_files_by_path_and_prefix(path, prefix):
def values_list_and_technical_files_by_path_and_prefix(
path: Union[str, Path],
prefix: Tuple
) -> [List, dict]:
"""
:param path: path of the saved object
Expand Down Expand Up @@ -169,12 +185,16 @@ def values_list_and_technical_files_by_path_and_prefix(path, prefix):
else:
file_name = file_info_dict["file_name"]
file_name += file_info_dict["format"]
description_info.update({key: {parts_val[-1]: os.path.join(os.path.join(*path[:parts_parse]), file_name)}})
description_info.update(
{key: {parts_val[-1]: os.path.join(os.path.join(*path[:parts_parse]), file_name)}})
parts_parse += 1
return parts_val, description_info

@staticmethod
def fill_prefix_storage(prefix, file_with_paths):
def fill_prefix_storage(
prefix: Tuple,
file_with_paths: Union[str, Path]
) -> [PrefixStorage, dict]:
"""
Fill prefix storage by file with paths
Expand All @@ -183,7 +203,7 @@ def fill_prefix_storage(prefix, file_with_paths):
:param file_with_paths: file with paths of saved objects
:return: fill prefix storage and dict with description_info about objects use hash
"""
keys_list, full_keys_list, dir_structure, empty_dir_shift =\
keys_list, full_keys_list, dir_structure, empty_dir_shift = \
DataInfo.take_keys_etc_by_prefix(prefix=prefix)
ps = PrefixStorage(keys_list)
with open(file_with_paths, 'r', encoding='utf-8') as f:
Expand All @@ -202,7 +222,10 @@ def fill_prefix_storage(prefix, file_with_paths):
return ps, description_info

@staticmethod
def deep_update(d, u):
def deep_update(
d: dict,
u: dict
) -> dict:
for k, v in u.items():
if isinstance(v, collections.abc.Mapping):
d[k] = DataInfo.deep_update(d.get(k, {}), v)
Expand All @@ -211,14 +234,19 @@ def deep_update(d, u):
return d

@staticmethod
def description_info_with_paths_to_description_info_with_files_values(description_info, root_path):
def description_info_with_paths_to_description_info_with_files_values(
description_info: dict,
root_path: Union[str, Path]
) -> dict:
for description_info_key, description_info_val in description_info.items():
for obj_name, obj_file_path in description_info_val.items():
with open(os.path.join(root_path, obj_file_path)) as f:
description_info[description_info_key][obj_name] = f.read()
return description_info

@staticmethod
def explainers_parse():
def explainers_parse(
) -> [PrefixStorage, dict]:
"""
Parses the path to explainers from a technical file with the paths of all saved explainers.
"""
Expand All @@ -232,7 +260,8 @@ def explainers_parse():
return ps, description_info

@staticmethod
def models_parse():
def models_parse(
) -> [PrefixStorage, dict]:
"""
Parses the path to models from a technical file with the paths of all saved models.
"""
Expand All @@ -246,7 +275,8 @@ def models_parse():
return ps, description_info

@staticmethod
def data_parse():
def data_parse(
) -> [PrefixStorage, dict]:
"""
Parses the path to raw datasets from a technical file with the paths of all saved raw datasets.
"""
Expand All @@ -257,7 +287,8 @@ def data_parse():
return ps

@staticmethod
def data_var_parse():
def data_var_parse(
) -> [PrefixStorage, dict]:
"""
Parses the path to prepared datasets from a technical file with the paths of all saved prepared datasets.
"""
Expand All @@ -268,7 +299,9 @@ def data_var_parse():
return ps

@staticmethod
def clean_prepared_data(dry_run=False):
def clean_prepared_data(
dry_run: bool = False
) -> None:
"""
Remove all prepared data for all datasets.
"""
Expand All @@ -279,7 +312,9 @@ def clean_prepared_data(dry_run=False):
shutil.rmtree(path)

@staticmethod
def all_obj_ver_by_obj_path(obj_dir_path):
def all_obj_ver_by_obj_path(
obj_dir_path: Union[str, Path]
) -> set:
"""
:param obj_dir_path: path to the saved object
Expand All @@ -294,7 +329,9 @@ def all_obj_ver_by_obj_path(obj_dir_path):
return set(vers_ind)

@staticmethod
def del_all_empty_folders(dir_path):
def del_all_empty_folders(
dir_path: Union[str, Path]
) -> None:
"""
Deletes all empty folders and files with meta information in the selected directory
Expand All @@ -316,7 +353,8 @@ def del_all_empty_folders(dir_path):

class UserCodeInfo:
@staticmethod
def user_models_list_ref():
def user_models_list_ref(
) -> dict:
"""
:return: dict with information about user models objects in directory <project root>/user_model_list
Contains information about objects class name, objects names and import paths
Expand Down Expand Up @@ -371,7 +409,10 @@ def models_init():
return user_models_obj_dict_info

@staticmethod
def take_user_model_obj(user_file_path, obj_name: str):
def take_user_model_obj(
user_file_path: Union[str, Path],
obj_name: str
) -> object:
"""
:param user_file_path: path to the user file with user model
:param obj_name: user object name
Expand Down
Loading

0 comments on commit 87854b3

Please sign in to comment.