forked from Daveonwave/ErNESTO-DT
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathernesto_DTSDA.py
179 lines (142 loc) · 8.01 KB
/
ernesto_DTSDA.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
67
68
69
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
99
100
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
import argparse
import logging
from joblib import Parallel, delayed
from src.utils.logger import CustomFormatter
from src.digital_twin.orchestrator import GeneralPurposeManager
def get_args():
main_parser = argparse.ArgumentParser(description="Digital Twin of a Battery Energy Storage System (RSE)",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
def get_sim_args():
"""
Parser of arguments for SIMULATION mode
"""
sim_parser.add_argument("--config_files", nargs='*', default=["./data/config/sim_config_example.yaml"],
help="Specifies the list of files containing parameters for each parallel experiment.")
def get_whatif_args():
"""
Parser of arguments for WHAT-IF mode
"""
whatif_parser.add_argument("--config", action="store", default="./data/config/whatif_config.yaml",
type=str, help="Specifies the file containing parameters for what-if mode.")
whatif_parser.add_argument("--iterations", default=500, type=int,
help="Specifies the number of iterations of the entire experiment.")
whatif_parser.add_argument("--timestep", default=1., type=float,
help="Specifies the timestep of the simulator in seconds.")
def get_learn_args():
"""
Parser of arguments for LEARNING mode
"""
learn_parser.add_argument("--iterations", default=500, type=int,
help="Specifies the number of iterations of the entire experiment.")
learn_parser.add_argument("--timestep", default=1., type=float,
help="Specifies the timestep of the simulator in seconds.")
def get_optim_parser():
"""
Parser of arguments for OPTIMIZATION mode
"""
optim_parser.add_argument("--iterations", default=500, type=int,
help="Specifies the number of iterations of the entire experiment.")
optim_parser.add_argument("--timestep", default=1., type=float,
help="Specifies the timestep of the simulator in seconds.")
def get_generic_args():
"""
Arguments of the main parser that can be useful to all the kind of modes
"""
main_parser.add_argument("--config_folder", action="store", default="./data/config", type=str,
help="Specifies the folder which we retrieve preprocessing from.")
main_parser.add_argument("--output_folder", action="store", default="./data/output", type=str,
help="Specifies the name of the folder where to store the output results.")
main_parser.add_argument("--ground_folder", action="store", default="./data/ground", type=str,
help="Specifies the folder which we retrieve preprocessing from.")
main_parser.add_argument("--assets", action="store", default="./data/config/assets.yaml",
type=str, help="Specifies the file containing parameters useful for the experiment.")
electrical_choices = ['thevenin', 'data_driven']
main_parser.add_argument("--battery_model", nargs=1, choices=electrical_choices, default=['thevenin'],
help="Specifies the name of the core model of the battery, electrical or data driven.")
thermal_choices = ['rc_thermal', 'r2c_thermal', 'dummy_thermal', 'mlp_thermal']
main_parser.add_argument("--thermal_model", nargs=1, choices=thermal_choices, default=['dummy_thermal'],
help="Specifies the name of the thermal model that has to be used.")
aging_choices = ['bolun']
main_parser.add_argument("--aging_model", nargs=1, choices=aging_choices,
help="Specifies the name of the aging model that has to be used.")
main_parser.add_argument("--save_results", action="store_true",
help="Specifies if save computed results at the end of the experiment.")
main_parser.add_argument("--save_metrics", action="store_true",
help="Specifies if save computed metrics at the end of the experiment.")
main_parser.add_argument("--plot", action="store_true",
help="Specifies if plot computed results at the end of the experiment.")
main_parser.add_argument("--n_cores", action="store", default=-1, type=int,
help="Specifies the number of cores to use for parallel simulations. If save_results "
"is set, cores will be override to 1 to limit RAM consumption.")
main_parser.add_argument("--verbose", action="store_true",
help="Increases logged information, but slows down the computation.")
get_generic_args()
subparsers = main_parser.add_subparsers(title="Mode", dest='mode', description="Experiment mode",
help="Working mode of the Digital Twin", required=True)
sim_parser = subparsers.add_parser('simulation', help="Simulation Mode",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
get_sim_args()
whatif_parser = subparsers.add_parser('whatif', help="What-If Mode",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
get_whatif_args()
learn_parser = subparsers.add_parser('learning', help="Learning Mode",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
get_learn_args()
optim_parser = subparsers.add_parser('optimization', help="Optimization Mode",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
get_optim_parser()
main_args = vars(main_parser.parse_args())
return main_args
def run_experiment(args, config_file):
dt_manager = GeneralPurposeManager.get_instance(args['mode'])
args['config'] = config_file
orchestrator= dt_manager(**args)
# orchestrator.run()
orchestrator.run_step()
orchestrator.run_step()
# orchestrator.evaluate()
if __name__ == '__main__':
# args = get_args()
args={ 'config_folder': './data/config',
'output_folder': './data/output',
'ground_folder': './data/ground',
'assets': './data/config/assets.yaml',
'battery_model': ['thevenin'],
'thermal_model': ['mlp_thermal'],
'aging_model': None,
'save_results': False,
'save_metrics': False,
'plot': False,
'n_cores': 1,
'verbose': True,
'mode': 'simulation',
'config_files': ['./data/config/sim_config_example.yaml']
}
# Setup logger
#logging.basicConfig(format='%(asctime)s | %(name)s-%(levelname)s: %(message)s')
logger = logging.getLogger(name="DT_ernesto")
ch = logging.StreamHandler()
if args['verbose']:
logger.setLevel(logging.DEBUG)
ch.setLevel(logging.DEBUG)
ch.setFormatter(CustomFormatter())
logger.addHandler(ch)
# Parsing of models employed in the current experiment
args['models'] = []
if args['battery_model']:
args['models'].extend(args['battery_model'])
del args['battery_model']
if args['thermal_model']:
args['models'].extend(args['thermal_model'])
del args['thermal_model']
if args['aging_model']:
args['models'].extend(args['aging_model'])
del args['aging_model']
parallel_exp_config = args['config_files']
del args['config_files']
n_cores = args['n_cores']
del args['n_cores']
if n_cores == 1:
run_experiment(args, parallel_exp_config[0])
else:
Parallel(n_jobs=n_cores)(delayed(run_experiment)(args, config) for config in parallel_exp_config)