-
Notifications
You must be signed in to change notification settings - Fork 3
/
ernesto.py
160 lines (123 loc) · 7.45 KB
/
ernesto.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
import argparse
import logging
import sys
from joblib import Parallel, delayed
from src.utils.logger import CustomFormatter
from src.digital_twin.orchestrator.orchestrator import DTOrchestrator
def get_args():
main_parser = argparse.ArgumentParser(description="Digital Twin of a Battery Energy Storage System (RSE)",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
def get_driven_args():
"""
Parser of arguments for DRIVEN SIMULATION mode
"""
driven_parser.add_argument("--config_files", nargs='*', default=["./data/config/sim_config_example2.yaml"],
help="Specifies the list of files containing parameters for each parallel experiment.")
def get_schedule_args():
"""
Parser of arguments for SCHEDULED SIMULATION mode
"""
schedule_parser.add_argument("--config_files", nargs='*', default=["./data/config/scheduled_config.yaml"],
help="Specifies the list of files containing parameters for each parallel experiment.")
schedule_parser.add_argument("--iterations", default=500, type=int,
help="Specifies the number of iterations of the entire experiment.")
schedule_parser.add_argument("--timestep", default=1., type=float,
help="Specifies the timestep of the simulator in seconds.")
def get_adaptive_args():
"""
Parser of arguments for ADAPTIVE SIMULATION mode
"""
adaptive_parser.add_argument("--config_files", nargs='*', default=["./data/config/adaptive_config.yaml"],
help="Specifies the list of files containing parameters for each parallel experiment.")
adaptive_parser.add_argument("--iterations", default=500, type=int,
help="Specifies the number of iterations of the entire experiment.")
adaptive_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 = ['first_order_thevenin', 'second_order_thevenin']
main_parser.add_argument("--battery_model", nargs=1, choices=electrical_choices, default=['first_order_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=['r2c_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("--interactive", action="store_true",
help="Enable the interaction of the user by a dedicated CLI.")
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)
driven_parser = subparsers.add_parser('driven', help="Driven Simulation Mode",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
get_driven_args()
schedule_parser = subparsers.add_parser('scheduled', help="Scheduled Simulation Mode",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
get_schedule_args()
adaptive_parser = subparsers.add_parser('adaptive', help="Adaptive Simulation Mode",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
get_adaptive_args()
main_args = vars(main_parser.parse_args())
return main_args
if __name__ == '__main__':
args = get_args()
# Setup logger
logging.basicConfig(format='%(asctime)s | %(name)s-%(levelname)s: %(message)s')
logger = logging.getLogger(name="ErNESTO-DT")
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']
def run_experiment(args, config_file):
args['config'] = config_file
orchestrator = DTOrchestrator(**args)
orchestrator.run()
#orchestrator.evaluate()
parallel_exp_config = args['config_files']
del args['config_files']
n_cores = args['n_cores']
del args['n_cores']
try:
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)
except Exception as e:
logger.error("An error occurred during the simulation. Exiting...")
logger.error(e)
sys.exit(1)