-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathdata_path_manager.py
66 lines (56 loc) · 3.02 KB
/
data_path_manager.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
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
import os
PROJECT_ROOT_NAME = 'data_extraction'
PROJECT_ROOT_PATH = '/home/jvn/GameTheorist/Computing/PythonProjects/data_extraction'
# See PathConfig check_project_path, we need a more consistent way of getting project root path.
# maybe a file that we are always sure is in project first level will do the trick
DATA_BASE_DIR = 'Data'
RAW_DATA_BASE_DIR_NAME = 'Raw_Data'
PROCESSED_DATA_DIR_NAME = 'Processed_Data'
FINAL_DATA_DIR_NAME = 'Final_Data'
class PathConfig:
def __init__(self, project_root_name=PROJECT_ROOT_NAME,
project_root_path_abs=PROJECT_ROOT_PATH,
data_base_dir_name=DATA_BASE_DIR,
data_types=None,
dir_name_of_data_type_dict=None,
data_base_root_path_rel='',
structure='STANDARD_TREE'):
self.project_root_name = project_root_name
self.project_root_path_abs = project_root_path_abs
self.data_base_dir_name = data_base_dir_name
if data_types is None:
self.data_types = ['Raw', 'Processed', 'Final']
else:
self.data_types = data_types
if dir_name_of_data_type_dict is None:
self.dir_name_of_data_type_dict = {x: x + '_Data' for x in self.data_types}
else:
self.dir_name_of_data_type_dict = dir_name_of_data_type_dict
self.structure = structure
self.data_base_root_path_rel = data_base_root_path_rel
if self.structure == 'STANDARD_TREE':
self.data_base_dir_path_rel = os.path.join(self.data_base_root_path_rel, self.data_base_dir_name)
self.path_rel_of_data_type_dict = {
x: os.path.join(self.data_base_dir_path_rel, self.dir_name_of_data_type_dict[x])
for x in self.data_types}
def check_project_path(self, print_warnings_regardless=False):
if (os.getcwd() != self.project_root_path_abs) or print_warnings_regardless:
print('Project Warning: project root path is different from current directory')
print(f'Current Working Directory Is {os.getcwd()}')
print(f'Project Root Path Is {self.project_root_path_abs}')
if (os.path.dirname(__file__) != self.project_root_path_abs) or print_warnings_regardless:
print('Project Warning: project root path is different from directory of current file')
print(f'Directory of Current file Is {os.path.dirname(__file__)}')
print(f'Project Root Path Is {self.project_root_path_abs}')
print(f'Path of Current file Is {__file__}')
else:
print('Paths related to Project Root seems to be correct')
def add_dtype_with_update(self, **dtypes):
for key, value in dtypes.items():
self.data_types.append(key)
self.dir_name_of_data_type_dict[key] = value
self.path_rel_of_data_type_dict[key] = os.path.join(
self.data_base_dir_path_rel, self.dir_name_of_data_type_dict[key])
DEFAULT_PATH_CONFIG = PathConfig()
if __name__ == '__main__':
d = DEFAULT_PATH_CONFIG