8870538658

Committed 28 Apr 2024 09:31PM UTC coverage: 75.582% (-1.4%) from 76.96%

Build # 8870538658

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #971 from rl-institut/fix/black-vulnerability

Fix/black vulnerability

Run Details

26 of 29 new or added lines in 15 files covered. (89.66%)

826 existing lines in 21 files now uncovered.

5977 of 7908 relevant lines covered (75.58%)

0.76 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.74

/src/multi_vector_simulator/F0_output.py

"""
Module F0 - Output
==================

The model F0 output defines all functions that store evaluation results to file.
- Aggregate demand profiles to a total demand profile
- Plot all energy flows for both 14 and 365 days for each energy bus
- Store timeseries of all energy flows to excel (one sheet = one energy bus)
- Execute function: plot optimised capacities as a barchart (F1)
- Execute function: plot all annuities as a barchart (F1)
- Store scalars/KPI to excel
- Process dictionary so that it can be stored to Json
- Store dictionary to Json
"""

import json
import logging
import os

import pandas as pd

from multi_vector_simulator.B0_data_input_json import convert_from_special_types_to_json
from multi_vector_simulator.E1_process_results import get_units_of_cost_matrix_entries
import multi_vector_simulator.F1_plotting as F1_plots

try:
    import multi_vector_simulator.F2_autoreport as autoreport

    AUTOREPORT = True
except ModuleNotFoundError:
    logging.warning("The reporting feature is disabled")
    AUTOREPORT = False

from multi_vector_simulator.utils.constants import (
    SIMULATION_SETTINGS,
    PATH_OUTPUT_FOLDER,
    OUTPUT_FOLDER,
    LOGFILE,
    PATHS_TO_PLOTS,
)

from multi_vector_simulator.utils.constants import (
    JSON_WITH_RESULTS,
    JSON_FILE_EXTENSION,
)
from multi_vector_simulator.utils.constants_json_strings import (
    UNIT,
    KPI,
    OPTIMIZED_FLOWS,
    DEMANDS,
    RESOURCES,
    LES_ENERGY_VECTOR_S,
    KPI_SCALARS_DICT,
    KPI_SCALAR_MATRIX,
    KPI_COST_MATRIX,
    PROJECT_DATA,
    ECONOMIC_DATA,
    SIMULATION_RESULTS,
    LOGS,
    ERRORS,
    WARNINGS,
    FIX_COST,
    ENERGY_BUSSES,
)


def evaluate_dict(dict_values, path_pdf_report=None, path_png_figs=None):
    """This is the main function of F0. It calls all functions that prepare the simulation output, ie. Storing all simulation output into excellent files, bar charts, and graphs.

    Parameters
    ----------
    dict_values :
        dict Of all input and output parameters up to F0

    path_pdf_report : (str)
        if provided, generate a pdf report of the simulation to the given path

    path_png_figs : (str)
        if provided, generate png figures of the simulation's results to the given path

    Returns
    -------
    type
        NA

    """

    logging.info(
        "Summarizing simulation results to results_timeseries and results_scalars_assets."
    )

    parse_simulation_log(
        path_log_file=os.path.join(
            dict_values[SIMULATION_SETTINGS][PATH_OUTPUT_FOLDER], LOGFILE
        ),
        dict_values=dict_values,
    )

    # storing all flows to exel.
    store_timeseries_all_busses_to_excel(dict_values)

    # Write everything to file with multiple tabs
    store_scalars_to_excel(dict_values)

    store_as_json(
        dict_values,
        dict_values[SIMULATION_SETTINGS][PATH_OUTPUT_FOLDER],
        JSON_WITH_RESULTS,
    )

    # generate png figures
    if path_png_figs is not None:
        # plot demand timeseries
        F1_plots.plot_timeseries(
            dict_values, data_type=DEMANDS, file_path=path_png_figs
        )
        # plot demand timeseries for the first 2 weeks only
        F1_plots.plot_timeseries(
            dict_values, data_type=DEMANDS, max_days=14, file_path=path_png_figs
        )

        # plot supply timeseries
        F1_plots.plot_timeseries(
            dict_values, data_type=RESOURCES, file_path=path_png_figs
        )
        # plot supply timeseries for the first 2 weeks only
        F1_plots.plot_timeseries(
            dict_values, data_type=RESOURCES, max_days=14, file_path=path_png_figs
        )

        # plot power flows in the energy system
        F1_plots.plot_instant_power(dict_values, file_path=path_png_figs)

        # plot optimal capacities if there are optimized assets
        F1_plots.plot_optimized_capacities(dict_values, file_path=path_png_figs)

        # plot annuity, first-investment and om costs
        F1_plots.plot_piecharts_of_costs(dict_values, file_path=path_png_figs)

    # generate a pdf report
    if path_pdf_report is not None:
        app = autoreport.create_app(dict_values)
        autoreport.print_pdf(app, path_pdf_report=path_pdf_report)
        logging.info(
            "Generating PDF report of the simulation: {}".format(path_pdf_report)
        )


def store_scalars_to_excel(dict_values):
    """All output data that is a scalar is storage to an excellent file tab. This could for example be economical data or technical data.

    Parameters
    ----------
    dict_values :
        dict Of all input and output parameters up to F0

    Returns
    -------
    type
        Excel file with scalar data

    """
    results_scalar_output_file = "/scalars" + ".xlsx"

    with pd.ExcelWriter(
        dict_values[SIMULATION_SETTINGS][PATH_OUTPUT_FOLDER]
        + results_scalar_output_file
    ) as open_file:  # doctest: +SKIP
        for kpi_set in dict_values[KPI]:
            if isinstance(dict_values[KPI][kpi_set], dict):
                data = pd.DataFrame([dict_values[KPI][kpi_set]])
            else:
                data = dict_values[KPI][kpi_set]

            # Transpose results and add units to the entries
            if kpi_set == KPI_SCALARS_DICT:
                data = data.transpose()
                units_cost_kpi = get_units_of_cost_matrix_entries(
                    dict_values[ECONOMIC_DATA], dict_values[KPI][kpi_set]
                )
                data[UNIT] = units_cost_kpi

            data.to_excel(open_file, sheet_name=kpi_set)
            logging.debug(
                "Saved scalar results to: %s, tab %s.",
                results_scalar_output_file,
                kpi_set,
            )


def store_timeseries_all_busses_to_excel(dict_values):
    """This function plots the energy flows of each single bus and the energy system and saves it as PNG and additionally as a tab and an Excel sheet.

    Parameters
    ----------
    dict_values :
        dict Of all input and output parameters up to F0

    Returns
    -------
    type
        Plots and excel with all timeseries of each bus

    """

    timeseries_output_file = "/timeseries_all_busses" + ".xlsx"
    with pd.ExcelWriter(
        dict_values[SIMULATION_SETTINGS][PATH_OUTPUT_FOLDER] + timeseries_output_file
    ) as open_file:  # doctest: +SKIP
        for bus in dict_values[OPTIMIZED_FLOWS]:
            dict_values[OPTIMIZED_FLOWS][bus].to_excel(open_file, sheet_name=bus)

    logging.debug("Saved flows at busses to: %s.", timeseries_output_file)


def parse_simulation_log(path_log_file, dict_values):
    """Gather a log file with several log messages, this function gathers them all and inputs them into the dict with
    all input and output parameters up to F0

    Parameters
    ----------
    path_log_file: str/None
        path to the mvs log file
        Default: None

    dict_values :
        dict Of all input and output parameters up to F0

    Returns
    -------
    Updates the results dictionary with the log messages of the simulation

    Notes
    -----
    This function is tested with:
    - test_F0_output.TestLogCreation.test_parse_simulation_log

    """
    # Dictionaries to gather non-fatal warning and error messages that appear during the simulation

    error_dict, warning_dict = {}, {}

    if path_log_file is None:
        path_log_file = os.path.join(OUTPUT_FOLDER, LOGFILE)

    with open(path_log_file) as log_messages:
        log_messages = log_messages.readlines()

    i = j = 0

    # Loop through the list of lines of the log file to check for the relevant log messages and gather them in dicts
    for line in log_messages:
        if "ERROR" in line:
            i = i + 1
            substrings = line.split(" - ")
            message_string = substrings[-1]
            error_dict.update({i: message_string})
        elif "WARNING" in line:
            j = j + 1
            substrings = line.split(" - ")
            message_string = substrings[-1]
            warning_dict.update({j: message_string})

    log_dict = {ERRORS: error_dict, WARNINGS: warning_dict}

    dict_values[SIMULATION_RESULTS].update({LOGS: log_dict})


def store_as_json(dict_values, output_folder=None, file_name=None):
    """Converts dict_values to JSON format and saves dict_values as a JSON file or return json

    Parameters
    ----------
    dict_values : (dict)
        dict to be stored as json
    output_folder : (path)
        Folder into which json should be stored
        Default None
    file_name : (str)
        Name of the file the json should be stored as
        Default None

    Returns
    -------
    If file_name is provided, the json variable converted from the dict_values is saved under
    this file_name, otherwise the json variable is returned
    """
    json_data = json.dumps(
        dict_values,
        skipkeys=False,
        sort_keys=True,
        default=convert_from_special_types_to_json,
        indent=4,
    )
    if file_name is not None:
        file_path = os.path.abspath(os.path.join(output_folder, file_name + ".json"))

        if os.path.exists(os.path.dirname(file_path)):
            myfile = open(file_path, "w")

            myfile.write(json_data)
            myfile.close()
            logging.info(
                "Converted and stored processed simulation data to json: %s", file_path
            )
            answer = file_path
        else:
            answer = None
    else:
        answer = json_data

    return answer

1	"""
2	Module F0 - Output
3	==================
4
5	The model F0 output defines all functions that store evaluation results to file.
6	- Aggregate demand profiles to a total demand profile
7	- Plot all energy flows for both 14 and 365 days for each energy bus
8	- Store timeseries of all energy flows to excel (one sheet = one energy bus)
9	- Execute function: plot optimised capacities as a barchart (F1)
10	- Execute function: plot all annuities as a barchart (F1)
11	- Store scalars/KPI to excel
12	- Process dictionary so that it can be stored to Json
13	- Store dictionary to Json
14	"""
15
16	import json	1✔
17	import logging	1✔
18	import os	1✔
19
20	import pandas as pd	1✔
21
22	from multi_vector_simulator.B0_data_input_json import convert_from_special_types_to_json	1✔
23	from multi_vector_simulator.E1_process_results import get_units_of_cost_matrix_entries	1✔
24	import multi_vector_simulator.F1_plotting as F1_plots	1✔
25
26	try:	1✔
27	import multi_vector_simulator.F2_autoreport as autoreport	1✔
28
29	AUTOREPORT = True	1✔
30	except ModuleNotFoundError:	×
31	logging.warning("The reporting feature is disabled")	×
32	AUTOREPORT = False	×
33
34	from multi_vector_simulator.utils.constants import (	1✔
35	SIMULATION_SETTINGS,
36	PATH_OUTPUT_FOLDER,
37	OUTPUT_FOLDER,
38	LOGFILE,
39	PATHS_TO_PLOTS,
40	)
41
42	from multi_vector_simulator.utils.constants import (	1✔
43	JSON_WITH_RESULTS,
44	JSON_FILE_EXTENSION,
45	)
46	from multi_vector_simulator.utils.constants_json_strings import (	1✔
47	UNIT,
48	KPI,
49	OPTIMIZED_FLOWS,
50	DEMANDS,
51	RESOURCES,
52	LES_ENERGY_VECTOR_S,
53	KPI_SCALARS_DICT,
54	KPI_SCALAR_MATRIX,
55	KPI_COST_MATRIX,
56	PROJECT_DATA,
57	ECONOMIC_DATA,
58	SIMULATION_RESULTS,
59	LOGS,
60	ERRORS,
61	WARNINGS,
62	FIX_COST,
63	ENERGY_BUSSES,
64	)
65
66
67	def evaluate_dict(dict_values, path_pdf_report=None, path_png_figs=None):	1✔
68	"""This is the main function of F0. It calls all functions that prepare the simulation output, ie. Storing all simulation output into excellent files, bar charts, and graphs.
69
70	Parameters
71	----------
72	dict_values :
73	dict Of all input and output parameters up to F0
74
75	path_pdf_report : (str)
76	if provided, generate a pdf report of the simulation to the given path
77
78	path_png_figs : (str)
79	if provided, generate png figures of the simulation's results to the given path
80
81	Returns
82	-------
83	type
84	NA
85
86	"""
87
88	logging.info(	1✔
89	"Summarizing simulation results to results_timeseries and results_scalars_assets."
90	)
91
92	parse_simulation_log(	1✔
93	path_log_file=os.path.join(
94	dict_values[SIMULATION_SETTINGS][PATH_OUTPUT_FOLDER], LOGFILE
95	),
96	dict_values=dict_values,
97	)
98
99	# storing all flows to exel.
100	store_timeseries_all_busses_to_excel(dict_values)	1✔
101
102	# Write everything to file with multiple tabs
103	store_scalars_to_excel(dict_values)	1✔
104
105	store_as_json(	1✔
106	dict_values,
107	dict_values[SIMULATION_SETTINGS][PATH_OUTPUT_FOLDER],
108	JSON_WITH_RESULTS,
109	)
110
111	# generate png figures
112	if path_png_figs is not None:	1✔
113	# plot demand timeseries
114	F1_plots.plot_timeseries(	×
115	dict_values, data_type=DEMANDS, file_path=path_png_figs
116	)
117	# plot demand timeseries for the first 2 weeks only
118	F1_plots.plot_timeseries(	×
119	dict_values, data_type=DEMANDS, max_days=14, file_path=path_png_figs
120	)
121
122	# plot supply timeseries
123	F1_plots.plot_timeseries(	×
124	dict_values, data_type=RESOURCES, file_path=path_png_figs
125	)
126	# plot supply timeseries for the first 2 weeks only
127	F1_plots.plot_timeseries(	×
128	dict_values, data_type=RESOURCES, max_days=14, file_path=path_png_figs
129	)
130
131	# plot power flows in the energy system
132	F1_plots.plot_instant_power(dict_values, file_path=path_png_figs)	×
133
134	# plot optimal capacities if there are optimized assets
135	F1_plots.plot_optimized_capacities(dict_values, file_path=path_png_figs)	×
136
137	# plot annuity, first-investment and om costs
138	F1_plots.plot_piecharts_of_costs(dict_values, file_path=path_png_figs)	×
139
140	# generate a pdf report
141	if path_pdf_report is not None:	1✔
142	app = autoreport.create_app(dict_values)	×
143	autoreport.print_pdf(app, path_pdf_report=path_pdf_report)	×
144	logging.info(	×
145	"Generating PDF report of the simulation: {}".format(path_pdf_report)
146	)
147
148
149	def store_scalars_to_excel(dict_values):	1✔
150	"""All output data that is a scalar is storage to an excellent file tab. This could for example be economical data or technical data.
151
152	Parameters
153	----------
154	dict_values :
155	dict Of all input and output parameters up to F0
156
157	Returns
158	-------
159	type
160	Excel file with scalar data
161
162	"""
163	results_scalar_output_file = "/scalars" + ".xlsx"	1✔
164
165	with pd.ExcelWriter(	1✔
166	dict_values[SIMULATION_SETTINGS][PATH_OUTPUT_FOLDER]
167	+ results_scalar_output_file
168	) as open_file: # doctest: +SKIP
169	for kpi_set in dict_values[KPI]:	1✔
170	if isinstance(dict_values[KPI][kpi_set], dict):	1✔
171	data = pd.DataFrame([dict_values[KPI][kpi_set]])	1✔
172	else:
173	data = dict_values[KPI][kpi_set]	1✔
174
175	# Transpose results and add units to the entries
176	if kpi_set == KPI_SCALARS_DICT:	1✔
177	data = data.transpose()	1✔
178	units_cost_kpi = get_units_of_cost_matrix_entries(	1✔
179	dict_values[ECONOMIC_DATA], dict_values[KPI][kpi_set]
180	)
181	data[UNIT] = units_cost_kpi	1✔
182
183	data.to_excel(open_file, sheet_name=kpi_set)	1✔
184	logging.debug(	1✔
185	"Saved scalar results to: %s, tab %s.",
186	results_scalar_output_file,
187	kpi_set,
188	)
189
190
191	def store_timeseries_all_busses_to_excel(dict_values):	1✔
192	"""This function plots the energy flows of each single bus and the energy system and saves it as PNG and additionally as a tab and an Excel sheet.
193
194	Parameters
195	----------
196	dict_values :
197	dict Of all input and output parameters up to F0
198
199	Returns
200	-------
201	type
202	Plots and excel with all timeseries of each bus
203
204	"""
205
206	timeseries_output_file = "/timeseries_all_busses" + ".xlsx"	1✔
207	with pd.ExcelWriter(	1✔
208	dict_values[SIMULATION_SETTINGS][PATH_OUTPUT_FOLDER] + timeseries_output_file
209	) as open_file: # doctest: +SKIP
210	for bus in dict_values[OPTIMIZED_FLOWS]:	1✔
211	dict_values[OPTIMIZED_FLOWS][bus].to_excel(open_file, sheet_name=bus)	1✔
212
213	logging.debug("Saved flows at busses to: %s.", timeseries_output_file)	1✔
214
215
216	def parse_simulation_log(path_log_file, dict_values):	1✔
217	"""Gather a log file with several log messages, this function gathers them all and inputs them into the dict with
218	all input and output parameters up to F0
219
220	Parameters
221	----------
222	path_log_file: str/None
223	path to the mvs log file
224	Default: None
225
226	dict_values :
227	dict Of all input and output parameters up to F0
228
229	Returns
230	-------
231	Updates the results dictionary with the log messages of the simulation
232
233	Notes
234	-----
235	This function is tested with:
236	- test_F0_output.TestLogCreation.test_parse_simulation_log
237
238	"""
239	# Dictionaries to gather non-fatal warning and error messages that appear during the simulation
240
241	error_dict, warning_dict = {}, {}	1✔
242
243	if path_log_file is None:	1✔
244	path_log_file = os.path.join(OUTPUT_FOLDER, LOGFILE)	×
245
246	with open(path_log_file) as log_messages:	1✔
247	log_messages = log_messages.readlines()	1✔
248
249	i = j = 0	1✔
250
251	# Loop through the list of lines of the log file to check for the relevant log messages and gather them in dicts
252	for line in log_messages:	1✔
253	if "ERROR" in line:	1✔
UNCOV 254	i = i + 1	×
UNCOV 255	substrings = line.split(" - ")	×
UNCOV 256	message_string = substrings[-1]	×
UNCOV 257	error_dict.update({i: message_string})	×
258	elif "WARNING" in line:	1✔
259	j = j + 1	1✔
260	substrings = line.split(" - ")	1✔
261	message_string = substrings[-1]	1✔
262	warning_dict.update({j: message_string})	1✔
263
264	log_dict = {ERRORS: error_dict, WARNINGS: warning_dict}	1✔
265
266	dict_values[SIMULATION_RESULTS].update({LOGS: log_dict})	1✔
267
268
269	def store_as_json(dict_values, output_folder=None, file_name=None):	1✔
270	"""Converts dict_values to JSON format and saves dict_values as a JSON file or return json
271
272	Parameters
273	----------
274	dict_values : (dict)
275	dict to be stored as json
276	output_folder : (path)
277	Folder into which json should be stored
278	Default None
279	file_name : (str)
280	Name of the file the json should be stored as
281	Default None
282
283	Returns
284	-------
285	If file_name is provided, the json variable converted from the dict_values is saved under
286	this file_name, otherwise the json variable is returned
287	"""
288	json_data = json.dumps(	1✔
289	dict_values,
290	skipkeys=False,
291	sort_keys=True,
292	default=convert_from_special_types_to_json,
293	indent=4,
294	)
295	if file_name is not None:	1✔
296	file_path = os.path.abspath(os.path.join(output_folder, file_name + ".json"))	1✔
297
298	if os.path.exists(os.path.dirname(file_path)):	1✔
299	myfile = open(file_path, "w")	1✔
300
301	myfile.write(json_data)	1✔
302	myfile.close()	1✔
303	logging.info(	1✔
304	"Converted and stored processed simulation data to json: %s", file_path
305	)
306	answer = file_path	1✔
307	else:
308	answer = None	×
309	else:
310	answer = json_data	×
311
312	return answer	1✔

rl-institut / multi-vector-simulator / 8870538658

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous