src.serena.utilities.thread_manager
1from dataclasses import dataclass 2from typing import List, Dict 3from enum import Enum 4import threading 5from datetime import datetime, time 6import time 7 8from serena.utilities.ensemble_structures import Sara2SecondaryStructure, Sara2StructureList 9from serena.utilities.ensemble_variation import EVShuttle, EV, EVToken, EnsembleVariation 10 11 12 13 14 15@dataclass 16class SwitchInput(): 17 sequence:str 18 fmn_struct:str 19 fmn_struct_free_energy:float 20 span:int 21 units:int 22 run_name:str 23 24 25class Temperature_Shuttle(): 26 pass 27 28class Temperature_Token(): 29 pass 30 31class Temperature_ThreadProcessor(): 32 33 def __init__(self, temp_list: List[int], switch_input:SwitchInput) -> None: 34 self._temp_list: List[int] = temp_list 35 num_temps:int = len(temp_list) 36 self._num_temps: int = num_temps 37 self._temp_token: Temperature_Token = Temperature_Token(num_temps) 38 self._SwitchAnalysis: EnsembleVariation = EnsembleVariation() 39 self._switch_input: SwitchInput = switch_input 40 41 @property 42 def temp_list(self): 43 return self._temp_list 44 45 @temp_list.setter 46 def temp_list(self, temp_list:List[int]): 47 self._temp_list = temp_list 48 49 @property 50 def switch_input(self): 51 return self._switch_input 52 53 @switch_input.setter 54 def switch_input(self, new_input:SwitchInput): 55 self._switch_input = new_input 56 57 @property 58 def num_temps(self): 59 return self._num_temps 60 61 @num_temps.setter 62 def num_temps(self, new_num:int): 63 self._num_temps = new_num 64 65 @property 66 def temp_token(self): 67 return self._temp_token 68 69 @temp_token.setter 70 def temp_token(self, new_token:Temperature_Token): 71 self._temp_token = new_token 72 73 @property 74 def SwitchAnalysis(self): 75 return self._SwitchAnalysis 76 77 @SwitchAnalysis.setter 78 def SwitchAnalysis(self, new_lmv:EnsembleVariation): 79 self._SwitchAnalysis = new_lmv 80 81 def run_SwitchAnalysis(self): 82 self.start_calculations() 83 self.wait_for_finish() 84 #the test should be done now 85 #check foor index that is -1 and if so then use prev value 86 num_groups:int = len(self.temp_token.temps_results) 87 #for index in range(1, num_groups): 88 # if self.temp_token.temps_results[index].ev_normalized == -1: 89 # previous_EV = self.group_token.group_results[index-1] 90 # self.group_token.group_results[index] = previous_EV 91 # self.group_token.group_dict[index] = previous_EV 92 return self.temp_token 93 94 def start_calculations(self): 95 for thread_index in range(self.num_temps): 96 sequence:str = self.switch_input.sequence 97 fmn_struct:str = self.switch_input.fmn_struct 98 fmn_struct_free_energy:float = self.switch_input.fmn_struct_free_energy 99 span:int = self.switch_input.span 100 units:int = self.switch_input.units 101 run_name:str = self.switch_input.run_name 102 new_shuttle: Temperature_Shuttle = Temperature_Shuttle(sequence=self.switch_input.sequence, 103 fmn_struct=self.switch_input.fmn_struct, 104 fmn_struct_free_energy=self.switch_input.fmn_struct_free_energy, 105 span=self.switch_input.span, 106 units=self.switch_input.units, 107 group_index=thread_index, 108 token=self.temp_token) 109 mew_thread = threading.Thread(target=self.SwitchAnalysis.process_ensemble_variation, args=[new_shuttle]) 110 mew_thread.start() 111 112 113 def wait_for_finish(self): 114 115 stop:bool = False 116 while stop == False: 117 print(f'Checking LMV status at {datetime.now()}') 118 current_status: List[bool] = self.group_token.group_done_status 119 is_done:bool = self.group_token.is_done 120 121 message: str = '' 122 for index in range(self.num_groups): 123 goup_value:str = self.group_token.group_values[index] 124 done_status: bool = self.group_token.group_done_status[index] 125 message = message + f'Group_{index+1}: kcal_group={goup_value}, status={done_status}\n' 126 print(message) 127 128 if is_done == True: 129 stop = True 130 print(f'Its done at {datetime.now()}') 131 else: 132 dwell_time:int = 5 133 print(f'dwelling for {dwell_time} seconds until next check') 134 time.sleep(dwell_time) 135 136 137class EV_ThreadProcessor(): 138 139 def __init__(self, stuctures: List[Sara2StructureList], comp_structure: Sara2SecondaryStructure = Sara2SecondaryStructure(), comp_struct_list_option: List[Sara2SecondaryStructure] = []) -> None: 140 self._sara2_groups: List[Sara2StructureList] = stuctures 141 num_groups:int = len(stuctures) 142 self._num_groups: int = num_groups 143 self._group_token: EVToken = EVToken(num_groups) 144 self._EV: EnsembleVariation = EnsembleVariation() 145 self._comparison_structure: Sara2SecondaryStructure = comp_structure 146 self._comp_struct_list_option:List[Sara2SecondaryStructure] = comp_struct_list_option 147 @property 148 def sara2_groups(self): 149 return self._sara2_groups 150 151 @sara2_groups.setter 152 def sara2_groups(self, new_list:List[Sara2StructureList]): 153 self._sara2_groups = new_list 154 155 @property 156 def comparison_structure(self): 157 return self._comparison_structure 158 159 @comparison_structure.setter 160 def comparison_structure(self, new_struct:Sara2SecondaryStructure): 161 self._comparison_structure = new_struct 162 163 @property 164 def comp_struct_list_option(self): 165 return self._comp_struct_list_option 166 167 @comp_struct_list_option.setter 168 def comp_struct_list_option(self, new_list:List[Sara2SecondaryStructure]): 169 self._comp_struct_list_option = new_list 170 171 @property 172 def num_groups(self): 173 return self._num_groups 174 175 @num_groups.setter 176 def num_groups(self, new_num:int): 177 self._num_groups = new_num 178 179 @property 180 def group_token(self): 181 return self._group_token 182 183 @group_token.setter 184 def group_token(self, new_token:EVToken): 185 self._group_token = new_token 186 187 @property 188 def EV(self): 189 return self._EV 190 191 #@EV.setter 192 #def EV(self, new_ev:EnsembleVariation): 193 # self._EV = new_ev 194 195 def run_EV(self): 196 self.start_calculations() 197 self.wait_for_finish() 198 #the test should be done now 199 #check foor index that is -1 and if so then use prev value 200 num_groups:int = len(self.group_token.group_results) 201 for index in range(1, num_groups): 202 if self.group_token.group_results[index].ev_normalized == -1: 203 previous_EV = self.group_token.group_results[index-1] 204 self.group_token.group_results[index] = previous_EV 205 self.group_token.group_dict[index] = previous_EV 206 return self.group_token 207 208 def start_calculations(self): 209 comp_structure: Sara2SecondaryStructure = Sara2SecondaryStructure() 210 for thread_index in range(self.num_groups): 211 if len(self.comp_struct_list_option) == self.num_groups: 212 comp_structure = self.comp_struct_list_option[thread_index] 213 else: 214 comp_structure = self.comparison_structure 215 sara2_structs: Sara2StructureList = self.sara2_groups[thread_index] 216 new_shuttle: EVShuttle = EVShuttle(structs_list=sara2_structs, mfe=comp_structure, group_index=thread_index,token=self.group_token) 217 mew_thread = threading.Thread(target=self.EV.thread_ev, args=[new_shuttle]) 218 mew_thread.start() 219 220 221 def wait_for_finish(self): 222 223 stop:bool = False 224 while stop == False: 225 print(f'Checking LMV status at {datetime.now()}') 226 current_status: List[bool] = self.group_token.group_done_status 227 is_done:bool = self.group_token.is_done 228 229 message: str = '' 230 for index in range(self.num_groups): 231 goup_value:str = self.group_token.group_values[index] 232 done_status: bool = self.group_token.group_done_status[index] 233 message = message + f'Group_{index+1}: kcal_group={goup_value}, status={done_status}\n' 234 print(message) 235 236 if is_done == True: 237 stop = True 238 print(f'Its done at {datetime.now()}') 239 else: 240 dwell_time:int = 5 241 print(f'dwelling for {dwell_time} seconds until next check') 242 time.sleep(dwell_time)
@dataclass
class
SwitchInput:
17@dataclass 18class SwitchInput(): 19 sequence:str 20 fmn_struct:str 21 fmn_struct_free_energy:float 22 span:int 23 units:int 24 run_name:str
class
Temperature_Shuttle:
class
Temperature_Token:
class
Temperature_ThreadProcessor:
33class Temperature_ThreadProcessor(): 34 35 def __init__(self, temp_list: List[int], switch_input:SwitchInput) -> None: 36 self._temp_list: List[int] = temp_list 37 num_temps:int = len(temp_list) 38 self._num_temps: int = num_temps 39 self._temp_token: Temperature_Token = Temperature_Token(num_temps) 40 self._SwitchAnalysis: EnsembleVariation = EnsembleVariation() 41 self._switch_input: SwitchInput = switch_input 42 43 @property 44 def temp_list(self): 45 return self._temp_list 46 47 @temp_list.setter 48 def temp_list(self, temp_list:List[int]): 49 self._temp_list = temp_list 50 51 @property 52 def switch_input(self): 53 return self._switch_input 54 55 @switch_input.setter 56 def switch_input(self, new_input:SwitchInput): 57 self._switch_input = new_input 58 59 @property 60 def num_temps(self): 61 return self._num_temps 62 63 @num_temps.setter 64 def num_temps(self, new_num:int): 65 self._num_temps = new_num 66 67 @property 68 def temp_token(self): 69 return self._temp_token 70 71 @temp_token.setter 72 def temp_token(self, new_token:Temperature_Token): 73 self._temp_token = new_token 74 75 @property 76 def SwitchAnalysis(self): 77 return self._SwitchAnalysis 78 79 @SwitchAnalysis.setter 80 def SwitchAnalysis(self, new_lmv:EnsembleVariation): 81 self._SwitchAnalysis = new_lmv 82 83 def run_SwitchAnalysis(self): 84 self.start_calculations() 85 self.wait_for_finish() 86 #the test should be done now 87 #check foor index that is -1 and if so then use prev value 88 num_groups:int = len(self.temp_token.temps_results) 89 #for index in range(1, num_groups): 90 # if self.temp_token.temps_results[index].ev_normalized == -1: 91 # previous_EV = self.group_token.group_results[index-1] 92 # self.group_token.group_results[index] = previous_EV 93 # self.group_token.group_dict[index] = previous_EV 94 return self.temp_token 95 96 def start_calculations(self): 97 for thread_index in range(self.num_temps): 98 sequence:str = self.switch_input.sequence 99 fmn_struct:str = self.switch_input.fmn_struct 100 fmn_struct_free_energy:float = self.switch_input.fmn_struct_free_energy 101 span:int = self.switch_input.span 102 units:int = self.switch_input.units 103 run_name:str = self.switch_input.run_name 104 new_shuttle: Temperature_Shuttle = Temperature_Shuttle(sequence=self.switch_input.sequence, 105 fmn_struct=self.switch_input.fmn_struct, 106 fmn_struct_free_energy=self.switch_input.fmn_struct_free_energy, 107 span=self.switch_input.span, 108 units=self.switch_input.units, 109 group_index=thread_index, 110 token=self.temp_token) 111 mew_thread = threading.Thread(target=self.SwitchAnalysis.process_ensemble_variation, args=[new_shuttle]) 112 mew_thread.start() 113 114 115 def wait_for_finish(self): 116 117 stop:bool = False 118 while stop == False: 119 print(f'Checking LMV status at {datetime.now()}') 120 current_status: List[bool] = self.group_token.group_done_status 121 is_done:bool = self.group_token.is_done 122 123 message: str = '' 124 for index in range(self.num_groups): 125 goup_value:str = self.group_token.group_values[index] 126 done_status: bool = self.group_token.group_done_status[index] 127 message = message + f'Group_{index+1}: kcal_group={goup_value}, status={done_status}\n' 128 print(message) 129 130 if is_done == True: 131 stop = True 132 print(f'Its done at {datetime.now()}') 133 else: 134 dwell_time:int = 5 135 print(f'dwelling for {dwell_time} seconds until next check') 136 time.sleep(dwell_time)
Temperature_ThreadProcessor( temp_list: List[int], switch_input: src.serena.utilities.thread_manager.SwitchInput)
35 def __init__(self, temp_list: List[int], switch_input:SwitchInput) -> None: 36 self._temp_list: List[int] = temp_list 37 num_temps:int = len(temp_list) 38 self._num_temps: int = num_temps 39 self._temp_token: Temperature_Token = Temperature_Token(num_temps) 40 self._SwitchAnalysis: EnsembleVariation = EnsembleVariation() 41 self._switch_input: SwitchInput = switch_input
def
run_SwitchAnalysis(self):
83 def run_SwitchAnalysis(self): 84 self.start_calculations() 85 self.wait_for_finish() 86 #the test should be done now 87 #check foor index that is -1 and if so then use prev value 88 num_groups:int = len(self.temp_token.temps_results) 89 #for index in range(1, num_groups): 90 # if self.temp_token.temps_results[index].ev_normalized == -1: 91 # previous_EV = self.group_token.group_results[index-1] 92 # self.group_token.group_results[index] = previous_EV 93 # self.group_token.group_dict[index] = previous_EV 94 return self.temp_token
def
start_calculations(self):
96 def start_calculations(self): 97 for thread_index in range(self.num_temps): 98 sequence:str = self.switch_input.sequence 99 fmn_struct:str = self.switch_input.fmn_struct 100 fmn_struct_free_energy:float = self.switch_input.fmn_struct_free_energy 101 span:int = self.switch_input.span 102 units:int = self.switch_input.units 103 run_name:str = self.switch_input.run_name 104 new_shuttle: Temperature_Shuttle = Temperature_Shuttle(sequence=self.switch_input.sequence, 105 fmn_struct=self.switch_input.fmn_struct, 106 fmn_struct_free_energy=self.switch_input.fmn_struct_free_energy, 107 span=self.switch_input.span, 108 units=self.switch_input.units, 109 group_index=thread_index, 110 token=self.temp_token) 111 mew_thread = threading.Thread(target=self.SwitchAnalysis.process_ensemble_variation, args=[new_shuttle]) 112 mew_thread.start()
def
wait_for_finish(self):
115 def wait_for_finish(self): 116 117 stop:bool = False 118 while stop == False: 119 print(f'Checking LMV status at {datetime.now()}') 120 current_status: List[bool] = self.group_token.group_done_status 121 is_done:bool = self.group_token.is_done 122 123 message: str = '' 124 for index in range(self.num_groups): 125 goup_value:str = self.group_token.group_values[index] 126 done_status: bool = self.group_token.group_done_status[index] 127 message = message + f'Group_{index+1}: kcal_group={goup_value}, status={done_status}\n' 128 print(message) 129 130 if is_done == True: 131 stop = True 132 print(f'Its done at {datetime.now()}') 133 else: 134 dwell_time:int = 5 135 print(f'dwelling for {dwell_time} seconds until next check') 136 time.sleep(dwell_time)
class
EV_ThreadProcessor:
139class EV_ThreadProcessor(): 140 141 def __init__(self, stuctures: List[Sara2StructureList], comp_structure: Sara2SecondaryStructure = Sara2SecondaryStructure(), comp_struct_list_option: List[Sara2SecondaryStructure] = []) -> None: 142 self._sara2_groups: List[Sara2StructureList] = stuctures 143 num_groups:int = len(stuctures) 144 self._num_groups: int = num_groups 145 self._group_token: EVToken = EVToken(num_groups) 146 self._EV: EnsembleVariation = EnsembleVariation() 147 self._comparison_structure: Sara2SecondaryStructure = comp_structure 148 self._comp_struct_list_option:List[Sara2SecondaryStructure] = comp_struct_list_option 149 @property 150 def sara2_groups(self): 151 return self._sara2_groups 152 153 @sara2_groups.setter 154 def sara2_groups(self, new_list:List[Sara2StructureList]): 155 self._sara2_groups = new_list 156 157 @property 158 def comparison_structure(self): 159 return self._comparison_structure 160 161 @comparison_structure.setter 162 def comparison_structure(self, new_struct:Sara2SecondaryStructure): 163 self._comparison_structure = new_struct 164 165 @property 166 def comp_struct_list_option(self): 167 return self._comp_struct_list_option 168 169 @comp_struct_list_option.setter 170 def comp_struct_list_option(self, new_list:List[Sara2SecondaryStructure]): 171 self._comp_struct_list_option = new_list 172 173 @property 174 def num_groups(self): 175 return self._num_groups 176 177 @num_groups.setter 178 def num_groups(self, new_num:int): 179 self._num_groups = new_num 180 181 @property 182 def group_token(self): 183 return self._group_token 184 185 @group_token.setter 186 def group_token(self, new_token:EVToken): 187 self._group_token = new_token 188 189 @property 190 def EV(self): 191 return self._EV 192 193 #@EV.setter 194 #def EV(self, new_ev:EnsembleVariation): 195 # self._EV = new_ev 196 197 def run_EV(self): 198 self.start_calculations() 199 self.wait_for_finish() 200 #the test should be done now 201 #check foor index that is -1 and if so then use prev value 202 num_groups:int = len(self.group_token.group_results) 203 for index in range(1, num_groups): 204 if self.group_token.group_results[index].ev_normalized == -1: 205 previous_EV = self.group_token.group_results[index-1] 206 self.group_token.group_results[index] = previous_EV 207 self.group_token.group_dict[index] = previous_EV 208 return self.group_token 209 210 def start_calculations(self): 211 comp_structure: Sara2SecondaryStructure = Sara2SecondaryStructure() 212 for thread_index in range(self.num_groups): 213 if len(self.comp_struct_list_option) == self.num_groups: 214 comp_structure = self.comp_struct_list_option[thread_index] 215 else: 216 comp_structure = self.comparison_structure 217 sara2_structs: Sara2StructureList = self.sara2_groups[thread_index] 218 new_shuttle: EVShuttle = EVShuttle(structs_list=sara2_structs, mfe=comp_structure, group_index=thread_index,token=self.group_token) 219 mew_thread = threading.Thread(target=self.EV.thread_ev, args=[new_shuttle]) 220 mew_thread.start() 221 222 223 def wait_for_finish(self): 224 225 stop:bool = False 226 while stop == False: 227 print(f'Checking LMV status at {datetime.now()}') 228 current_status: List[bool] = self.group_token.group_done_status 229 is_done:bool = self.group_token.is_done 230 231 message: str = '' 232 for index in range(self.num_groups): 233 goup_value:str = self.group_token.group_values[index] 234 done_status: bool = self.group_token.group_done_status[index] 235 message = message + f'Group_{index+1}: kcal_group={goup_value}, status={done_status}\n' 236 print(message) 237 238 if is_done == True: 239 stop = True 240 print(f'Its done at {datetime.now()}') 241 else: 242 dwell_time:int = 5 243 print(f'dwelling for {dwell_time} seconds until next check') 244 time.sleep(dwell_time)
EV_ThreadProcessor( stuctures: List[serena.utilities.ensemble_structures.Sara2StructureList], comp_structure: serena.utilities.ensemble_structures.Sara2SecondaryStructure = <serena.utilities.ensemble_structures.Sara2SecondaryStructure object>, comp_struct_list_option: List[serena.utilities.ensemble_structures.Sara2SecondaryStructure] = [])
141 def __init__(self, stuctures: List[Sara2StructureList], comp_structure: Sara2SecondaryStructure = Sara2SecondaryStructure(), comp_struct_list_option: List[Sara2SecondaryStructure] = []) -> None: 142 self._sara2_groups: List[Sara2StructureList] = stuctures 143 num_groups:int = len(stuctures) 144 self._num_groups: int = num_groups 145 self._group_token: EVToken = EVToken(num_groups) 146 self._EV: EnsembleVariation = EnsembleVariation() 147 self._comparison_structure: Sara2SecondaryStructure = comp_structure 148 self._comp_struct_list_option:List[Sara2SecondaryStructure] = comp_struct_list_option
def
run_EV(self):
197 def run_EV(self): 198 self.start_calculations() 199 self.wait_for_finish() 200 #the test should be done now 201 #check foor index that is -1 and if so then use prev value 202 num_groups:int = len(self.group_token.group_results) 203 for index in range(1, num_groups): 204 if self.group_token.group_results[index].ev_normalized == -1: 205 previous_EV = self.group_token.group_results[index-1] 206 self.group_token.group_results[index] = previous_EV 207 self.group_token.group_dict[index] = previous_EV 208 return self.group_token
def
start_calculations(self):
210 def start_calculations(self): 211 comp_structure: Sara2SecondaryStructure = Sara2SecondaryStructure() 212 for thread_index in range(self.num_groups): 213 if len(self.comp_struct_list_option) == self.num_groups: 214 comp_structure = self.comp_struct_list_option[thread_index] 215 else: 216 comp_structure = self.comparison_structure 217 sara2_structs: Sara2StructureList = self.sara2_groups[thread_index] 218 new_shuttle: EVShuttle = EVShuttle(structs_list=sara2_structs, mfe=comp_structure, group_index=thread_index,token=self.group_token) 219 mew_thread = threading.Thread(target=self.EV.thread_ev, args=[new_shuttle]) 220 mew_thread.start()
def
wait_for_finish(self):
223 def wait_for_finish(self): 224 225 stop:bool = False 226 while stop == False: 227 print(f'Checking LMV status at {datetime.now()}') 228 current_status: List[bool] = self.group_token.group_done_status 229 is_done:bool = self.group_token.is_done 230 231 message: str = '' 232 for index in range(self.num_groups): 233 goup_value:str = self.group_token.group_values[index] 234 done_status: bool = self.group_token.group_done_status[index] 235 message = message + f'Group_{index+1}: kcal_group={goup_value}, status={done_status}\n' 236 print(message) 237 238 if is_done == True: 239 stop = True 240 print(f'Its done at {datetime.now()}') 241 else: 242 dwell_time:int = 5 243 print(f'dwelling for {dwell_time} seconds until next check') 244 time.sleep(dwell_time)