Coverage for mlair/run_modules/run_environment.py: 77%

1"""Implementation of run environment.""" 

2 

3__author__ = "Lukas Leufen" 

4__date__ = '2019-11-25' 

5 

6import json 

7import logging 

8import os 

9import shutil 

10import time 

11import sys 

12 

13from mlair.helpers.datastore import DataStoreByScope as DataStoreObject 

14from mlair.helpers.datastore import NameNotFoundInDataStore 

15from mlair.helpers import Logger, to_list, TimeTracking 

16from mlair.plotting.tracker_plot import TrackPlot 

17 

18 

19class RunEnvironment(object): 

20 """ 

21 Basic run class to measure execution time. 

22 

23 Either call this class by 'with' statement or delete the class instance after finishing the measurement. The 

24 duration result is logged. 

25 

26 .. code-block:: python 

27 

28 >>> with RunEnvironment(): 

29 <your code> 

30 INFO: RunEnvironment started 

31 ... 

32 INFO: RunEnvironment finished after 00:00:04 (hh:mm:ss) 

33 

34 If you want to embed your custom module in a RunEnvironment, you can easily call it inside the with statement. If 

35 you want to exchange between different modules in addition, create your module as inheritance of the RunEnvironment 

36 and call it after you initialised the RunEnvironment itself. 

37 

38 .. code-block:: python 

39 

40 class CustomClass(RunEnvironment): 

41 

42 def __init__(self): 

43 super().__init__() 

44 ... 

45 ... 

46 

47 

48 >>> with RunEnvironment(): 

49 CustomClass() 

50 INFO: RunEnvironment started 

51 INFO: CustomClass started 

52 INFO: CustomClass finished after 00:00:04 (hh:mm:ss) 

53 INFO: RunEnvironment finished after 00:00:04 (hh:mm:ss) 

54 

55 All data that is stored in the data store will be available for all other modules that inherit from RunEnvironment 

56 as long the RunEnvironemnt base class is running. If the base class is deleted either by hand or on exit of the with 

57 statement, this storage is cleared. 

58 

59 .. code-block:: python 

60 

61 class CustomClassA(RunEnvironment): 

62 

63 def __init__(self): 

64 super().__init__() 

65 self.data_store.set("testVar", 12) 

66 

67 

68 class CustomClassB(RunEnvironment): 

69 

70 def __init__(self): 

71 super().__init__() 

72 self.test_var = self.data_store.get("testVar") 

73 logging.info(f"testVar = {self.test_var}") 

74 

75 

76 >>> with RunEnvironment(): 

77 CustomClassA() 

78 CustomClassB() 

79 INFO: RunEnvironment started 

80 INFO: CustomClassA started 

81 INFO: CustomClassA finished after 00:00:01 (hh:mm:ss) 

82 INFO: CustomClassB started 

83 INFO: testVar = 12 

84 INFO: CustomClassB finished after 00:00:02 (hh:mm:ss) 

85 INFO: RunEnvironment finished after 00:00:03 (hh:mm:ss) 

86 

87 """ 

88 

89 # set data store and logger (both are mutable!) 

90 del_by_exit = False 

91 data_store = None 

92 logger = None 

93 tracker_list = [] 

94 

95 def __init__(self, name=None, log_level_stream=None): 

96 """Start time tracking automatically and logs as info.""" 

97 if RunEnvironment.data_store is None: 

98 RunEnvironment.data_store = DataStoreObject() 

99 if RunEnvironment.logger is None: 

100 RunEnvironment.logger = Logger(level_stream=log_level_stream) 

101 self._name = name if name is not None else self.__class__.__name__ 

102 self.time = TimeTracking(name=name) 

103 logging.info(f"{self._name} started") 

104 # atexit.register(self.__del__) 

105 self.data_store.tracker.append({}) 

106 self.tracker_list.extend([{self._name: self.data_store.tracker[-1]}]) 

107 

108 def __del__(self): 

109 """ 

110 Finalise class. 

111 

112 Only stop time tracking, if not already called by exit method to prevent duplicated logging (__exit__ is always 

113 executed before __del__) it this class was used in a with statement. If instance is called as base class and 

114 not as inheritance from this class, log file is copied and data store is cleared. 

115 """ 

116 if not self.del_by_exit: 

117 if hasattr(self, "time"): 

118 self.time.stop() 

119 try: 

120 logging.info(f"{self._name} finished after {self.time}") 

121 except NameError: 

122 pass 

123 self.del_by_exit = True 

124 # copy log file and clear data store only if called as base class and not as super class 

125 if self.__class__.__name__ == "RunEnvironment": 

126 try: 

127 self.__plot_tracking() 

128 self.__save_tracking() 

129 self.__move_log_file() 

130 except (FileNotFoundError, NameError): 

131 pass 

132 self.data_store.clear_data_store() 

133 

134 def __enter__(self): 

135 """Enter run environment.""" 

136 return self 

137 

138 def __exit__(self, exc_type, exc_val, exc_tb): 

139 """Exit run environment.""" 

140 if exc_type: 140 ↛ 141line 140 didn't jump to line 141, because the condition on line 140 was never true

141 logging.error(exc_val, exc_info=(exc_type, exc_val, exc_tb)) 

142 self.__del__() 

143 

144 def __move_log_file(self): 

145 try: 

146 new_file = self.__find_file_pattern("logging_%03i.log") 

147 logging.info(f"Move log file to {new_file}") 

148 shutil.move(self.logger.log_file, new_file) 

149 try: 

150 os.rmdir(os.path.dirname(self.logger.log_file)) 

151 except (OSError, FileNotFoundError): 

152 pass 

153 except (NameNotFoundInDataStore, FileNotFoundError): 

154 pass 

155 

156 def __save_tracking(self): 

157 tracker = self.data_store.tracker 

158 new_file = self.__find_file_pattern("tracking_%03i.json") 

159 logging.info(f"Copy tracker file to {new_file}") 

160 with open(new_file, "w") as f: 

161 json.dump(tracker, f) 

162 

163 def __plot_tracking(self): 

164 try: 

165 plot_folder, plot_name = os.path.split(self.__find_file_pattern("tracking_%03i.pdf")) 

166 TrackPlot(self.tracker_list, sparse_conn_mode=True, plot_folder=plot_folder, plot_name=plot_name) 

167 except Exception as e: 

168 logging.info(f"Could not plot tracking plot due to:\n{sys.exc_info()[0]}\n" 

169 f"{sys.exc_info()[1]}\n{sys.exc_info()[2]}") 

170 

171 def __find_file_pattern(self, name): 

172 counter = 0 

173 filename_pattern = os.path.join(self.data_store.get_default("logging_path", os.path.realpath(".")), name) 

174 new_file = filename_pattern % counter 

175 while os.path.exists(new_file): 

176 counter += 1 

177 new_file = filename_pattern % counter 

178 return new_file 

179 

180 @classmethod 

181 def update_datastore(cls, new_data_store: DataStoreObject, excluded_params=None, apply_full_replacement=False): 

182 if apply_full_replacement is True: 

183 RunEnvironment.data_store = new_data_store 

184 else: 

185 assert type(RunEnvironment.data_store) == type(new_data_store) 

186 scopes = new_data_store.list_all_scopes() 

187 excluded_params = to_list(excluded_params) 

188 for s in scopes: 

189 # name, scope and value 

190 entries = new_data_store.search_scope(s, current_scope_only=True, return_all=True) 

191 for e in entries: 

192 if e[0] not in excluded_params: 

193 # name, value, scope 

194 RunEnvironment.data_store.set(e[0], e[2], e[1]) 

195 

196 @staticmethod 

197 def do_stuff(length=2): 

198 """Just a placeholder method for testing without any sense.""" 

199 time.sleep(length)