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

17.81

/src/multi_vector_simulator/server.py

"""
This is the main file of the tool "Multi-vector simulation tool".

Tool structure:

(parent)    multi_vector_simulator.py

(child)    constants.py

(child)     --A0_initialization.py

(child)      --B0_data_input.py

(child)     --C0_data_processing.py
(child sub)    --C1_verification.py
(child sub)    --C2_economic_processing.py

(child)     --D0_modelling_and_optimization.py
(child sub)    --D1_model_components.py
(child sub)    --D2_model_constraints.py

(child)     --E0_evaluation.py
(child sub)    --E1_process_results.py
(child sub)    --E2_verification_of_constraints.py
(child sub)    --E4_verification.py

(child)    --F0_output.py
(child sub)    --F1_plotting.py
(child sub)    --F2_autoreport.py
patent:     Main file, all children connected through parent
child:      Child file, one of the main functions of the tool.
            Internal processes, feeds output back to parent
child-sub:  Sub-child function, feeds only back to child functions
"""

import logging
import json
import os
import tempfile

from oemof.tools import logger
import oemof.solph as solph

# Loading all child functions
import multi_vector_simulator.B0_data_input_json as B0
import multi_vector_simulator.C0_data_processing as C0
import multi_vector_simulator.D0_modelling_and_optimization as D0
import multi_vector_simulator.E0_evaluation as E0
import multi_vector_simulator.F0_output as F0
from multi_vector_simulator.version import version_num, version_date
from multi_vector_simulator.utils import (
    data_parser,
    nested_dict_crawler,
    get_nested_value,
)


from multi_vector_simulator.utils.constants_json_strings import (
    SIMULATION_SETTINGS,
    OUTPUT_LP_FILE,
    VALUE,
    UNIT,
    ENERGY_BUSSES,
    AUTO_CREATED_HIGHLIGHT,
    ENERGY_VECTOR,
    TYPE_ASSET,
    OEMOF_ASSET_TYPE,
    ENERGY_PRODUCTION,
    ENERGY_CONSUMPTION,
    ENERGY_CONVERSION,
    ENERGY_PROVIDERS,
    ENERGY_STORAGE,
    TIMESERIES_PEAK,
    OPTIMIZE_CAP,
    OPTIMIZED_ADD_CAP,
    VALUE,
)
from multi_vector_simulator.utils.constants import TYPE_STR
from multi_vector_simulator.utils.helpers import get_asset_types


import pandas as pd


def bus_flow(
    flow_tuple, busses_info, asset_types=None
):  # can work as well with nodes (assets)
    if not isinstance(busses_info, dict):
        raise ValueError("Expected a dict")
    busses_names = [bn for bn in busses_info]
    bus_name = set(busses_names).intersection(set(flow_tuple))
    answer = None
    if len(bus_name) == 1:
        bus_name = bus_name.pop()
        idx_bus = flow_tuple.index(bus_name)
        if idx_bus == 0:
            asset_name = flow_tuple[1]
            answer = (bus_name, busses_info[bus_name][ENERGY_VECTOR], "out", asset_name)
        elif idx_bus == 1:
            asset_name = flow_tuple[0]
            answer = (bus_name, busses_info[bus_name][ENERGY_VECTOR], "in", asset_name)
        if asset_types is not None:
            df_at = pd.DataFrame.from_records(asset_types).set_index("label")
            answer = answer + (
                df_at.loc[asset_name, TYPE_ASSET],
                df_at.loc[asset_name, OEMOF_ASSET_TYPE],
            )
    return answer


class OemofBusResults(pd.DataFrame):  # real results
    def __init__(self, results, busses_info=None, asset_types=None):
        # TODO add a division by timeseries peak
        if isinstance(results, dict):
            ts = []
            investments = []
            flows = []
            for x, res in solph.views.convert_keys_to_strings(results).items():
                if x[1] != "None":
                    col_name = res["sequences"].columns[0]
                    ts.append(
                        res["sequences"].rename(
                            columns={col_name: x, "variable_name": "timesteps"}
                        )
                    )
                    flows.append(bus_flow(x, busses_info, asset_types))
                    invest = (
                        None if res["scalars"].empty is True else res["scalars"].invest
                    )
                    investments.append(invest)
            ts_df = pd.concat(ts, axis=1, join="inner")
            mindex = pd.MultiIndex.from_tuples(
                flows,
                names=[
                    "bus",
                    "energy_vector",
                    "direction",
                    "asset",
                    "asset_type",
                    "oemof_type",
                ],
            )

        elif isinstance(results, str):
            js = json.loads(results)
            mindex = pd.MultiIndex.from_tuples(
                js["columns"],
                names=[
                    "bus",
                    "energy_vector",
                    "direction",
                    "asset",
                    "asset_type",
                    "oemof_type",
                ],
            )
            df = pd.DataFrame(data=js["data"], columns=mindex)

            ts_df = df.iloc[:-1]
            ts_index = pd.to_datetime(js["index"][:-1], unit="ms")
            investments = df.iloc[-1]
            ts_df.index = ts_index
        for extra_var in ["status", "status_nominal"]:
            if extra_var in ts_df:
                ts_df.drop(extra_var, axis=1, inplace=True)
        super().__init__(
            data=ts_df.T.to_dict(orient="split")["data"],
            index=mindex,
            columns=ts_df.index,
        )

        self["investments"] = investments
        self.sort_index(inplace=True)

    def to_json(self, **kwargs):
        kwargs["orient"] = "split"
        return self.T.to_json(**kwargs)

    def bus_flows(self, bus_name):
        return self.loc[bus_name, self.columns != "investments"].T

    def asset_optimized_capacities(self):
        return self.loc[:, "investments"]

    def asset_optimized_capacity(self, asset_name):
        optimized_capacity = self.loc[
            self.index.get_level_values("asset") == asset_name, "investments"
        ].dropna()
        if len(optimized_capacity) == 1:
            optimized_capacity = optimized_capacity[0]
        return optimized_capacity


def run_simulation(json_dict, epa_format=True, verbatim=False, **kwargs):
    r"""
     Starts MVS tool simulation from an input json file

     Parameters
    -----------
     json_dict: dict
         json from http request
     epa_format: bool, optional
         Specifies whether the output is formatted for EPA standards
         Default: True

     Other Parameters
     ----------------
     pdf_report: bool, optional
         Can generate an automatic pdf report of the simulation's results (True) or not (False)
         Default: False.
     display_output : str, optional
         Sets the level of displayed logging messages.
         Options: "debug", "info", "warning", "error". Default: "info".
     lp_file_output : bool, optional
         Specifies whether linear equation system generated is saved as lp file.
         Default: False.

    """
    display_output = kwargs.get("display_output", None)

    if display_output == "debug":
        screen_level = logging.DEBUG
    elif display_output == "info":
        screen_level = logging.INFO
    elif display_output == "warning":
        screen_level = logging.WARNING
    elif display_output == "error":
        screen_level = logging.ERROR
    else:
        screen_level = logging.INFO

    # Define logging settings and path for saving log
    logger.define_logging(screen_level=screen_level)

    welcome_text = (
        "\n \n Multi-Vector Simulation Tool (MVS) V"
        + version_num
        + " "
        + "\n Version: "
        + version_date
        + " "
        + '\n Part of the toolbox of H2020 project "E-LAND", '
        + "Integrated multi-vector management system for Energy isLANDs"
        + "\n Coded at: Reiner Lemoine Institute (Berlin) "
        + "\n Reference: https://zenodo.org/record/4610237 \n \n "
    )

    logging.info(welcome_text)

    logging.debug("Accessing script: B0_data_input_json")
    dict_values = B0.convert_from_json_to_special_types(json_dict)

    # if True will return the lp file's content in dict_values
    lp_file_output = dict_values[SIMULATION_SETTINGS][OUTPUT_LP_FILE][VALUE]
    # to avoid the lp file being saved somewhere on the server
    dict_values[SIMULATION_SETTINGS][OUTPUT_LP_FILE][VALUE] = False

    print("")
    logging.debug("Accessing script: C0_data_processing")
    C0.all(dict_values)

    print("")
    logging.debug("Accessing script: D0_modelling_and_optimization")
    results_meta, results_main, local_energy_system = D0.run_oemof(
        dict_values, return_les=True
    )

    br = OemofBusResults(
        results_main,
        busses_info=dict_values[ENERGY_BUSSES],
        asset_types=get_asset_types(dict_values),
    )  # if AUTO_CREATED_HIGHLIGHT not in bl])

    if lp_file_output is True:
        logging.debug("Saving the content of the model's lp file")
        with tempfile.TemporaryDirectory() as tmpdirname:
            local_energy_system.write(
                os.path.join(tmpdirname, "lp_file.lp"),
                io_options={"symbolic_solver_labels": True},
            )
            with open(os.path.join(tmpdirname, "lp_file.lp")) as fp:
                file_content = fp.read()

        dict_values[SIMULATION_SETTINGS][OUTPUT_LP_FILE][VALUE] = file_content
        dict_values[SIMULATION_SETTINGS][OUTPUT_LP_FILE][UNIT] = TYPE_STR

    print("")
    logging.debug("Accessing script: E0_evaluation")
    E0.evaluate_dict(dict_values, results_main, results_meta)

    # Correct the optimized values
    for asset_group in [
        ENERGY_PRODUCTION,
        ENERGY_CONSUMPTION,
        ENERGY_CONVERSION,
        ENERGY_PROVIDERS,
        ENERGY_STORAGE,
    ]:
        for asset_name, asset in dict_values[asset_group].items():
            if (
                asset.get(OPTIMIZE_CAP, {VALUE: False}).get(VALUE, False) is True
                and TIMESERIES_PEAK in asset
            ):
                corrected_optimized_capacity = asset[OPTIMIZED_ADD_CAP][VALUE]
                br.loc[
                    br.index.get_level_values("asset") == asset_name, "investments"
                ] = corrected_optimized_capacity

    dict_values["raw_results"] = br.to_json()  # to_dict(orient="split") #

    logging.debug("Convert results to json")

    if epa_format is True:
        epa_dict_values = data_parser.convert_mvs_params_to_epa(
            dict_values, verbatim=verbatim
        )

        json_values = F0.store_as_json(epa_dict_values)
        answer = json.loads(json_values)
    else:
        answer = dict_values

    return answer


def run_sensitivity_analysis_step(
    json_input, step_idx, output_variables, epa_format=True, **kwargs
):
    r"""
     Starts MVS tool simulation from an input json file

     Parameters
    -----------
    json_input: dict
        json from http request
    step_idx: int
        step of the sensitivity analysis
    output_variables: tuple of str
        collection of output variables names
    epa_format: bool, optional
        Specifies whether the output is formatted for EPA standards
        Default: True

     Other Parameters
     ----------------
     pdf_report: bool, optional
         Can generate an automatic pdf report of the simulation's results (True) or not (False)
         Default: False.
     display_output : str, optional
         Sets the level of displayed logging messages.
         Options: "debug", "info", "warning", "error". Default: "info".
     lp_file_output : bool, optional
         Specifies whether linear equation system generated is saved as lp file.
         Default: False.

    """

    # Process the argument json_input based on its type
    if isinstance(json_input, str):
        # load the file if it is a path
        simulation_input = B0.load_json(json_input)
    elif isinstance(json_input, dict):
        # this is already a json variable
        simulation_input = json_input
    else:
        simulation_input = None
        logging.error(
            f"Simulation input `{json_input}` is neither a file path, nor a json dict. "
            f"It can therefore not be processed."
        )

    sim_output_json = run_simulation(
        simulation_input, display_output="error", epa_format=epa_format
    )
    output_variables_paths = nested_dict_crawler(sim_output_json)

    output_parameters = {}
    # for each of the output parameter path, add the value located under this path in
    # the final json dict, that could also be applied to the full json dict as
    # post-processing
    for output_param in output_variables:
        output_param_pathes = output_variables_paths.get(output_param, [])

        if len(output_param_pathes) == 0:
            output_parameters[output_param] = dict(
                value=None,
                path=f"Not found in mvs results (version {version_num}), check if you have typos in output parameter name",
            )

        for output_param_path in output_param_pathes:
            if output_param not in output_parameters:
                output_parameters[output_param] = dict(
                    value=[get_nested_value(sim_output_json, output_param_path)],
                    path=[".".join(output_param_path)],
                )
            else:
                output_parameters[output_param]["value"].append(
                    get_nested_value(sim_output_json, output_param_path)
                )
                output_parameters[output_param]["path"].append(
                    ".".join(output_param_path)
                )

    return {"step_idx": step_idx, "output_values": output_parameters}

1	"""
2	This is the main file of the tool "Multi-vector simulation tool".
3
4	Tool structure:
5
6	(parent) multi_vector_simulator.py
7
8	(child) constants.py
9
10	(child) --A0_initialization.py
11
12	(child) --B0_data_input.py
13
14	(child) --C0_data_processing.py
15	(child sub) --C1_verification.py
16	(child sub) --C2_economic_processing.py
17
18	(child) --D0_modelling_and_optimization.py
19	(child sub) --D1_model_components.py
20	(child sub) --D2_model_constraints.py
21
22	(child) --E0_evaluation.py
23	(child sub) --E1_process_results.py
24	(child sub) --E2_verification_of_constraints.py
25	(child sub) --E4_verification.py
26
27	(child) --F0_output.py
28	(child sub) --F1_plotting.py
29	(child sub) --F2_autoreport.py
30	patent: Main file, all children connected through parent
31	child: Child file, one of the main functions of the tool.
32	Internal processes, feeds output back to parent
33	child-sub: Sub-child function, feeds only back to child functions
34	"""
35
36	import logging	1✔
37	import json	1✔
38	import os	1✔
39	import tempfile	1✔
40
41	from oemof.tools import logger	1✔
42	import oemof.solph as solph	1✔
43
44	# Loading all child functions
45	import multi_vector_simulator.B0_data_input_json as B0	1✔
46	import multi_vector_simulator.C0_data_processing as C0	1✔
47	import multi_vector_simulator.D0_modelling_and_optimization as D0	1✔
48	import multi_vector_simulator.E0_evaluation as E0	1✔
49	import multi_vector_simulator.F0_output as F0	1✔
50	from multi_vector_simulator.version import version_num, version_date	1✔
51	from multi_vector_simulator.utils import (	1✔
52	data_parser,
53	nested_dict_crawler,
54	get_nested_value,
55	)
56
57
58	from multi_vector_simulator.utils.constants_json_strings import (	1✔
59	SIMULATION_SETTINGS,
60	OUTPUT_LP_FILE,
61	VALUE,
62	UNIT,
63	ENERGY_BUSSES,
64	AUTO_CREATED_HIGHLIGHT,
65	ENERGY_VECTOR,
66	TYPE_ASSET,
67	OEMOF_ASSET_TYPE,
68	ENERGY_PRODUCTION,
69	ENERGY_CONSUMPTION,
70	ENERGY_CONVERSION,
71	ENERGY_PROVIDERS,
72	ENERGY_STORAGE,
73	TIMESERIES_PEAK,
74	OPTIMIZE_CAP,
75	OPTIMIZED_ADD_CAP,
76	VALUE,
77	)
78	from multi_vector_simulator.utils.constants import TYPE_STR	1✔
79	from multi_vector_simulator.utils.helpers import get_asset_types	1✔
80
81
82	import pandas as pd	1✔
83
84
85	def bus_flow(	1✔
86	flow_tuple, busses_info, asset_types=None
87	): # can work as well with nodes (assets)
UNCOV 88	if not isinstance(busses_info, dict):	×
UNCOV 89	raise ValueError("Expected a dict")	×
UNCOV 90	busses_names = [bn for bn in busses_info]	×
91	bus_name = set(busses_names).intersection(set(flow_tuple))	×
UNCOV 92	answer = None	×
93	if len(bus_name) == 1:	×
94	bus_name = bus_name.pop()	×
95	idx_bus = flow_tuple.index(bus_name)	×
96	if idx_bus == 0:	×
97	asset_name = flow_tuple[1]	×
98	answer = (bus_name, busses_info[bus_name][ENERGY_VECTOR], "out", asset_name)	×
99	elif idx_bus == 1:	×
100	asset_name = flow_tuple[0]	×
UNCOV 101	answer = (bus_name, busses_info[bus_name][ENERGY_VECTOR], "in", asset_name)	×
102	if asset_types is not None:	×
UNCOV 103	df_at = pd.DataFrame.from_records(asset_types).set_index("label")	×
UNCOV 104	answer = answer + (	×
105	df_at.loc[asset_name, TYPE_ASSET],
106	df_at.loc[asset_name, OEMOF_ASSET_TYPE],
107	)
UNCOV 108	return answer	×
109
110
111	class OemofBusResults(pd.DataFrame): # real results	1✔
112	def __init__(self, results, busses_info=None, asset_types=None):	1✔
113	# TODO add a division by timeseries peak
UNCOV 114	if isinstance(results, dict):	×
UNCOV 115	ts = []	×
UNCOV 116	investments = []	×
UNCOV 117	flows = []	×
UNCOV 118	for x, res in solph.views.convert_keys_to_strings(results).items():	×
UNCOV 119	if x[1] != "None":	×
120	col_name = res["sequences"].columns[0]	×
UNCOV 121	ts.append(	×
122	res["sequences"].rename(
123	columns={col_name: x, "variable_name": "timesteps"}
124	)
125	)
126	flows.append(bus_flow(x, busses_info, asset_types))	×
UNCOV 127	invest = (	×
128	None if res["scalars"].empty is True else res["scalars"].invest
129	)
130	investments.append(invest)	×
131	ts_df = pd.concat(ts, axis=1, join="inner")	×
132	mindex = pd.MultiIndex.from_tuples(	×
133	flows,
134	names=[
135	"bus",
136	"energy_vector",
137	"direction",
138	"asset",
139	"asset_type",
140	"oemof_type",
141	],
142	)
143
UNCOV 144	elif isinstance(results, str):	×
UNCOV 145	js = json.loads(results)	×
UNCOV 146	mindex = pd.MultiIndex.from_tuples(	×
147	js["columns"],
148	names=[
149	"bus",
150	"energy_vector",
151	"direction",
152	"asset",
153	"asset_type",
154	"oemof_type",
155	],
156	)
157	df = pd.DataFrame(data=js["data"], columns=mindex)	×
158
159	ts_df = df.iloc[:-1]	×
160	ts_index = pd.to_datetime(js["index"][:-1], unit="ms")	×
UNCOV 161	investments = df.iloc[-1]	×
162	ts_df.index = ts_index	×
163	for extra_var in ["status", "status_nominal"]:	×
UNCOV 164	if extra_var in ts_df:	×
UNCOV 165	ts_df.drop(extra_var, axis=1, inplace=True)	×
166	super().__init__(	×
167	data=ts_df.T.to_dict(orient="split")["data"],
168	index=mindex,
169	columns=ts_df.index,
170	)
171
UNCOV 172	self["investments"] = investments	×
UNCOV 173	self.sort_index(inplace=True)	×
174
175	def to_json(self, **kwargs):	1✔
UNCOV 176	kwargs["orient"] = "split"	×
UNCOV 177	return self.T.to_json(**kwargs)	×
178
179	def bus_flows(self, bus_name):	1✔
UNCOV 180	return self.loc[bus_name, self.columns != "investments"].T	×
181
182	def asset_optimized_capacities(self):	1✔
UNCOV 183	return self.loc[:, "investments"]	×
184
185	def asset_optimized_capacity(self, asset_name):	1✔
UNCOV 186	optimized_capacity = self.loc[	×
187	self.index.get_level_values("asset") == asset_name, "investments"
188	].dropna()
UNCOV 189	if len(optimized_capacity) == 1:	×
UNCOV 190	optimized_capacity = optimized_capacity[0]	×
UNCOV 191	return optimized_capacity	×
192
193
194	def run_simulation(json_dict, epa_format=True, verbatim=False, **kwargs):	1✔
195	r"""
196	Starts MVS tool simulation from an input json file
197
198	Parameters
199	-----------
200	json_dict: dict
201	json from http request
202	epa_format: bool, optional
203	Specifies whether the output is formatted for EPA standards
204	Default: True
205
206	Other Parameters
207	----------------
208	pdf_report: bool, optional
209	Can generate an automatic pdf report of the simulation's results (True) or not (False)
210	Default: False.
211	display_output : str, optional
212	Sets the level of displayed logging messages.
213	Options: "debug", "info", "warning", "error". Default: "info".
214	lp_file_output : bool, optional
215	Specifies whether linear equation system generated is saved as lp file.
216	Default: False.
217
218	"""
UNCOV 219	display_output = kwargs.get("display_output", None)	×
220
UNCOV 221	if display_output == "debug":	×
222	screen_level = logging.DEBUG	×
UNCOV 223	elif display_output == "info":	×
UNCOV 224	screen_level = logging.INFO	×
UNCOV 225	elif display_output == "warning":	×
226	screen_level = logging.WARNING	×
227	elif display_output == "error":	×
UNCOV 228	screen_level = logging.ERROR	×
229	else:
230	screen_level = logging.INFO	×
231
232	# Define logging settings and path for saving log
UNCOV 233	logger.define_logging(screen_level=screen_level)	×
234
235	welcome_text = (	×
236	"\n \n Multi-Vector Simulation Tool (MVS) V"
237	+ version_num
238	+ " "
239	+ "\n Version: "
240	+ version_date
241	+ " "
242	+ '\n Part of the toolbox of H2020 project "E-LAND", '
243	+ "Integrated multi-vector management system for Energy isLANDs"
244	+ "\n Coded at: Reiner Lemoine Institute (Berlin) "
245	+ "\n Reference: https://zenodo.org/record/4610237 \n \n "
246	)
247
UNCOV 248	logging.info(welcome_text)	×
249
UNCOV 250	logging.debug("Accessing script: B0_data_input_json")	×
UNCOV 251	dict_values = B0.convert_from_json_to_special_types(json_dict)	×
252
253	# if True will return the lp file's content in dict_values
UNCOV 254	lp_file_output = dict_values[SIMULATION_SETTINGS][OUTPUT_LP_FILE][VALUE]	×
255	# to avoid the lp file being saved somewhere on the server
UNCOV 256	dict_values[SIMULATION_SETTINGS][OUTPUT_LP_FILE][VALUE] = False	×
257
UNCOV 258	print("")	×
UNCOV 259	logging.debug("Accessing script: C0_data_processing")	×
UNCOV 260	C0.all(dict_values)	×
261
UNCOV 262	print("")	×
UNCOV 263	logging.debug("Accessing script: D0_modelling_and_optimization")	×
UNCOV 264	results_meta, results_main, local_energy_system = D0.run_oemof(	×
265	dict_values, return_les=True
266	)
267
UNCOV 268	br = OemofBusResults(	×
269	results_main,
270	busses_info=dict_values[ENERGY_BUSSES],
271	asset_types=get_asset_types(dict_values),
272	) # if AUTO_CREATED_HIGHLIGHT not in bl])
273
UNCOV 274	if lp_file_output is True:	×
UNCOV 275	logging.debug("Saving the content of the model's lp file")	×
UNCOV 276	with tempfile.TemporaryDirectory() as tmpdirname:	×
UNCOV 277	local_energy_system.write(	×
278	os.path.join(tmpdirname, "lp_file.lp"),
279	io_options={"symbolic_solver_labels": True},
280	)
UNCOV 281	with open(os.path.join(tmpdirname, "lp_file.lp")) as fp:	×
UNCOV 282	file_content = fp.read()	×
283
UNCOV 284	dict_values[SIMULATION_SETTINGS][OUTPUT_LP_FILE][VALUE] = file_content	×
UNCOV 285	dict_values[SIMULATION_SETTINGS][OUTPUT_LP_FILE][UNIT] = TYPE_STR	×
286
UNCOV 287	print("")	×
UNCOV 288	logging.debug("Accessing script: E0_evaluation")	×
UNCOV 289	E0.evaluate_dict(dict_values, results_main, results_meta)	×
290
291	# Correct the optimized values
UNCOV 292	for asset_group in [	×
293	ENERGY_PRODUCTION,
294	ENERGY_CONSUMPTION,
295	ENERGY_CONVERSION,
296	ENERGY_PROVIDERS,
297	ENERGY_STORAGE,
298	]:
UNCOV 299	for asset_name, asset in dict_values[asset_group].items():	×
UNCOV 300	if (	×
301	asset.get(OPTIMIZE_CAP, {VALUE: False}).get(VALUE, False) is True
302	and TIMESERIES_PEAK in asset
303	):
UNCOV 304	corrected_optimized_capacity = asset[OPTIMIZED_ADD_CAP][VALUE]	×
UNCOV 305	br.loc[	×
306	br.index.get_level_values("asset") == asset_name, "investments"
307	] = corrected_optimized_capacity
308
UNCOV 309	dict_values["raw_results"] = br.to_json() # to_dict(orient="split") #	×
310
UNCOV 311	logging.debug("Convert results to json")	×
312
UNCOV 313	if epa_format is True:	×
UNCOV 314	epa_dict_values = data_parser.convert_mvs_params_to_epa(	×
315	dict_values, verbatim=verbatim
316	)
317
UNCOV 318	json_values = F0.store_as_json(epa_dict_values)	×
UNCOV 319	answer = json.loads(json_values)	×
320	else:
UNCOV 321	answer = dict_values	×
322
UNCOV 323	return answer	×
324
325
326	def run_sensitivity_analysis_step(	1✔
327	json_input, step_idx, output_variables, epa_format=True, **kwargs
328	):
329	r"""
330	Starts MVS tool simulation from an input json file
331
332	Parameters
333	-----------
334	json_input: dict
335	json from http request
336	step_idx: int
337	step of the sensitivity analysis
338	output_variables: tuple of str
339	collection of output variables names
340	epa_format: bool, optional
341	Specifies whether the output is formatted for EPA standards
342	Default: True
343
344	Other Parameters
345	----------------
346	pdf_report: bool, optional
347	Can generate an automatic pdf report of the simulation's results (True) or not (False)
348	Default: False.
349	display_output : str, optional
350	Sets the level of displayed logging messages.
351	Options: "debug", "info", "warning", "error". Default: "info".
352	lp_file_output : bool, optional
353	Specifies whether linear equation system generated is saved as lp file.
354	Default: False.
355
356	"""
357
358	# Process the argument json_input based on its type
UNCOV 359	if isinstance(json_input, str):	×
360	# load the file if it is a path
UNCOV 361	simulation_input = B0.load_json(json_input)	×
UNCOV 362	elif isinstance(json_input, dict):	×
363	# this is already a json variable
UNCOV 364	simulation_input = json_input	×
365	else:
UNCOV 366	simulation_input = None	×
UNCOV 367	logging.error(	×
368	f"Simulation input `{json_input}` is neither a file path, nor a json dict. "
369	f"It can therefore not be processed."
370	)
371
UNCOV 372	sim_output_json = run_simulation(	×
373	simulation_input, display_output="error", epa_format=epa_format
374	)
UNCOV 375	output_variables_paths = nested_dict_crawler(sim_output_json)	×
376
UNCOV 377	output_parameters = {}	×
378	# for each of the output parameter path, add the value located under this path in
379	# the final json dict, that could also be applied to the full json dict as
380	# post-processing
UNCOV 381	for output_param in output_variables:	×
UNCOV 382	output_param_pathes = output_variables_paths.get(output_param, [])	×
383
UNCOV 384	if len(output_param_pathes) == 0:	×
UNCOV 385	output_parameters[output_param] = dict(	×
386	value=None,
387	path=f"Not found in mvs results (version {version_num}), check if you have typos in output parameter name",
388	)
389
UNCOV 390	for output_param_path in output_param_pathes:	×
UNCOV 391	if output_param not in output_parameters:	×
UNCOV 392	output_parameters[output_param] = dict(	×
393	value=[get_nested_value(sim_output_json, output_param_path)],
394	path=[".".join(output_param_path)],
395	)
396	else:
UNCOV 397	output_parameters[output_param]["value"].append(	×
398	get_nested_value(sim_output_json, output_param_path)
399	)
UNCOV 400	output_parameters[output_param]["path"].append(	×
401	".".join(output_param_path)
402	)
403
UNCOV 404	return {"step_idx": step_idx, "output_values": output_parameters}	×

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