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

51.81

/src/multi_vector_simulator/utils/helpers.py

"""
Helper functions
================

Util functions that are useful throughout the MVS

Including:
- find_valvue_by_key(): Finds value of a key in a nested dictionary.
"""

from copy import deepcopy
import os

from multi_vector_simulator.utils.constants_json_strings import (
    DSO_FEEDIN_CAP,
    AUTO_CREATED_HIGHLIGHT,
    DSO_CONSUMPTION,
    DSO_FEEDIN,
    DSO_PEAK_DEMAND_PERIOD,
    DSO_PEAK_DEMAND_SUFFIX,
    ENERGY_CONSUMPTION,
    ENERGY_CONVERSION,
    ENERGY_STORAGE,
    ENERGY_PRODUCTION,
    ENERGY_PROVIDERS,
    ENERGY_BUSSES,
    OEMOF_ASSET_TYPE,
    TYPE_ASSET,
    INFLOW_DIRECTION,
    OUTFLOW_DIRECTION,
    ENERGY_VECTOR,
    SUFFIX_CRITICAL,
    SUFFIX_NONCRITICAL,
    REDUCABLE_DEMAND,
)


def find_value_by_key(data, target, result=None):
    """
    Finds value of a key in a nested dictionary.

    Parameters
    ----------
    data: dict
        Dict to be searched for target key

    target: str
        Key for which the value should be found in data

    result: None, value or list
        Only provided if function loops in itself

    Returns
    -------
    value if the key is only once in data
    list of values if it appears multiple times.
    """
    # check each item-value pair in the level
    for k, v in data.items():
        # if target in keys of level
        if k == target:
            if result is None:
                result = v
            elif isinstance(result, list):
                # Expands list of key finds
                result.append(v)
            else:
                # creates list for multiple key finds
                previous_result = result
                result = []
                result.append(previous_result)
                result.append(v)
        # Check next level for target
        if isinstance(v, dict):
            result = find_value_by_key(data=v, target=target, result=result)
    return result


def translates_epa_strings_to_mvs_readable(folder_name, file_name):
    """
    This function translates the json file generated by the EPA to a file readable by the MVS.
    This is necessary as there are some parameter names whose string representative differs in both tools.

    Parameters
    ----------
    folder_name: path
        Path to the folder with the json file
    file_name: json file name with extension
        Json to be converted

    Returns
    -------
    Stores converted json file to current dict

    Usage:
        `import multi_vector_simulator.utils.helpers as helpers`
        `helpers.translates_epa_strings_to_mvs_readable("./epa_benchmark", "epa_benchmark.json-original")`
    """
    import json
    from multi_vector_simulator.utils.data_parser import convert_epa_params_to_mvs

    with open(os.path.join(folder_name, file_name)) as json_file:
        epa_dict = json.load(json_file)

    dict_values = convert_epa_params_to_mvs(epa_dict)

    with open(os.path.join(folder_name, "mvs_config.json"), "w") as json_file:
        json.dump(dict_values, json_file, indent=4)


def get_item_if_list(list_or_float, index):
    if isinstance(list_or_float, list):
        answer = list_or_float[index]
    else:
        answer = list_or_float
    return answer


def get_length_if_list(list_or_float):
    if isinstance(list_or_float, list):
        answer = len(list_or_float)
    else:
        answer = 0
    return answer


def reducable_demand_name(demand_name: str, critical: bool = False):
    """Name for auto created bus related to peak demand pricing period"""

    if critical is False:
        suffix = SUFFIX_NONCRITICAL
    else:
        suffix = SUFFIX_CRITICAL

    return f"{demand_name}_{suffix} {AUTO_CREATED_HIGHLIGHT}"


def peak_demand_bus_name(dso_name: str, feedin: bool = False):
    """Name for auto created bus related to peak demand pricing period"""

    if feedin is False:
        dso_direction = DSO_CONSUMPTION
    else:
        dso_direction = DSO_FEEDIN

    return (
        f"{dso_name}{dso_direction}_{DSO_PEAK_DEMAND_SUFFIX} {AUTO_CREATED_HIGHLIGHT}"
    )


def peak_demand_transformer_name(
    dso_name: str, peak_number: int = None, feedin: bool = False
):
    """Name for auto created bus related to peak demand pricing period"""
    if feedin is False:
        dso_direction = DSO_CONSUMPTION
    else:
        dso_direction = DSO_FEEDIN
    transformer_name = f"{dso_name}{dso_direction}{DSO_PEAK_DEMAND_PERIOD}"
    if peak_number is not None:
        transformer_name = f"{transformer_name}_{str(peak_number)}"

    return f"{transformer_name} {AUTO_CREATED_HIGHLIGHT}"


def get_asset_types(dict_values):
    """Function which returns records of assets in the energy system"""
    asset_types = []
    for asset_group in (
        ENERGY_CONSUMPTION,
        ENERGY_CONVERSION,
        ENERGY_STORAGE,
        ENERGY_PRODUCTION,
        ENERGY_PROVIDERS,
    ):
        for asset_name, asset_params in dict_values.get(asset_group, {}).items():
            asset_type = {"label": asset_name}
            for param in (OEMOF_ASSET_TYPE, TYPE_ASSET):
                asset_type[param] = asset_params.get(param)
            asset_busses = {}
            input_bus = asset_params.get(INFLOW_DIRECTION)
            if input_bus is not None:
                if not isinstance(input_bus, list):
                    # print("not a list :", input_bus)
                    input_bus = [input_bus]
            else:
                input_bus = []

            output_bus = asset_params.get(OUTFLOW_DIRECTION)
            if output_bus is not None:
                if not isinstance(output_bus, list):
                    # print("not a list: ", output_bus)
                    output_bus = [output_bus]
            else:
                output_bus = []

            for bus in input_bus + output_bus:
                asset_busses[bus] = dict_values[ENERGY_BUSSES][bus].get(ENERGY_VECTOR)
            asset_type["busses"] = asset_busses
            if asset_type[TYPE_ASSET] == REDUCABLE_DEMAND:

                asset_label = asset_type["label"]
                asset_type["label"] = reducable_demand_name(asset_label)
                asset_types.append(asset_type)
                crit_asset_type = deepcopy(asset_type)
                crit_asset_type["label"] = reducable_demand_name(
                    asset_label, critical=True
                )
                asset_types.append(crit_asset_type)
            else:
                asset_types.append(asset_type)
    return asset_types

1	"""
2	Helper functions
3	================
4
5	Util functions that are useful throughout the MVS
6
7	Including:
8	- find_valvue_by_key(): Finds value of a key in a nested dictionary.
9	"""
10
11	from copy import deepcopy	1✔
12	import os	1✔
13
14	from multi_vector_simulator.utils.constants_json_strings import (	1✔
15	DSO_FEEDIN_CAP,
16	AUTO_CREATED_HIGHLIGHT,
17	DSO_CONSUMPTION,
18	DSO_FEEDIN,
19	DSO_PEAK_DEMAND_PERIOD,
20	DSO_PEAK_DEMAND_SUFFIX,
21	ENERGY_CONSUMPTION,
22	ENERGY_CONVERSION,
23	ENERGY_STORAGE,
24	ENERGY_PRODUCTION,
25	ENERGY_PROVIDERS,
26	ENERGY_BUSSES,
27	OEMOF_ASSET_TYPE,
28	TYPE_ASSET,
29	INFLOW_DIRECTION,
30	OUTFLOW_DIRECTION,
31	ENERGY_VECTOR,
32	SUFFIX_CRITICAL,
33	SUFFIX_NONCRITICAL,
34	REDUCABLE_DEMAND,
35	)
36
37
38	def find_value_by_key(data, target, result=None):	1✔
39	"""
40	Finds value of a key in a nested dictionary.
41
42	Parameters
43	----------
44	data: dict
45	Dict to be searched for target key
46
47	target: str
48	Key for which the value should be found in data
49
50	result: None, value or list
51	Only provided if function loops in itself
52
53	Returns
54	-------
55	value if the key is only once in data
56	list of values if it appears multiple times.
57	"""
58	# check each item-value pair in the level
59	for k, v in data.items():	1✔
60	# if target in keys of level
61	if k == target:	1✔
62	if result is None:	1✔
63	result = v	1✔
64	elif isinstance(result, list):	1✔
65	# Expands list of key finds
66	result.append(v)	1✔
67	else:
68	# creates list for multiple key finds
69	previous_result = result	1✔
70	result = []	1✔
71	result.append(previous_result)	1✔
72	result.append(v)	1✔
73	# Check next level for target
74	if isinstance(v, dict):	1✔
75	result = find_value_by_key(data=v, target=target, result=result)	1✔
76	return result	1✔
77
78
79	def translates_epa_strings_to_mvs_readable(folder_name, file_name):	1✔
80	"""
81	This function translates the json file generated by the EPA to a file readable by the MVS.
82	This is necessary as there are some parameter names whose string representative differs in both tools.
83
84	Parameters
85	----------
86	folder_name: path
87	Path to the folder with the json file
88	file_name: json file name with extension
89	Json to be converted
90
91	Returns
92	-------
93	Stores converted json file to current dict
94
95	Usage:
96	`import multi_vector_simulator.utils.helpers as helpers`
97	`helpers.translates_epa_strings_to_mvs_readable("./epa_benchmark", "epa_benchmark.json-original")`
98	"""
UNCOV 99	import json	×
UNCOV 100	from multi_vector_simulator.utils.data_parser import convert_epa_params_to_mvs	×
101
UNCOV 102	with open(os.path.join(folder_name, file_name)) as json_file:	×
UNCOV 103	epa_dict = json.load(json_file)	×
104
UNCOV 105	dict_values = convert_epa_params_to_mvs(epa_dict)	×
106
UNCOV 107	with open(os.path.join(folder_name, "mvs_config.json"), "w") as json_file:	×
UNCOV 108	json.dump(dict_values, json_file, indent=4)	×
109
110
111	def get_item_if_list(list_or_float, index):	1✔
112	if isinstance(list_or_float, list):	1✔
113	answer = list_or_float[index]	1✔
114	else:
115	answer = list_or_float	1✔
116	return answer	1✔
117
118
119	def get_length_if_list(list_or_float):	1✔
120	if isinstance(list_or_float, list):	1✔
121	answer = len(list_or_float)	1✔
122	else:
123	answer = 0	1✔
124	return answer	1✔
125
126
127	def reducable_demand_name(demand_name: str, critical: bool = False):	1✔
128	"""Name for auto created bus related to peak demand pricing period"""
129
UNCOV 130	if critical is False:	×
UNCOV 131	suffix = SUFFIX_NONCRITICAL	×
132	else:
UNCOV 133	suffix = SUFFIX_CRITICAL	×
134
UNCOV 135	return f"{demand_name}_{suffix} {AUTO_CREATED_HIGHLIGHT}"	×
136
137
138	def peak_demand_bus_name(dso_name: str, feedin: bool = False):	1✔
139	"""Name for auto created bus related to peak demand pricing period"""
140
141	if feedin is False:	1✔
142	dso_direction = DSO_CONSUMPTION	1✔
143	else:
144	dso_direction = DSO_FEEDIN	1✔
145
146	return (	1✔
147	f"{dso_name}{dso_direction}_{DSO_PEAK_DEMAND_SUFFIX} {AUTO_CREATED_HIGHLIGHT}"
148	)
149
150
151	def peak_demand_transformer_name(	1✔
152	dso_name: str, peak_number: int = None, feedin: bool = False
153	):
154	"""Name for auto created bus related to peak demand pricing period"""
155	if feedin is False:	1✔
156	dso_direction = DSO_CONSUMPTION	1✔
157	else:
158	dso_direction = DSO_FEEDIN	1✔
159	transformer_name = f"{dso_name}{dso_direction}{DSO_PEAK_DEMAND_PERIOD}"	1✔
160	if peak_number is not None:	1✔
161	transformer_name = f"{transformer_name}_{str(peak_number)}"	1✔
162
163	return f"{transformer_name} {AUTO_CREATED_HIGHLIGHT}"	1✔
164
165
166	def get_asset_types(dict_values):	1✔
167	"""Function which returns records of assets in the energy system"""
UNCOV 168	asset_types = []	×
UNCOV 169	for asset_group in (	×
170	ENERGY_CONSUMPTION,
171	ENERGY_CONVERSION,
172	ENERGY_STORAGE,
173	ENERGY_PRODUCTION,
174	ENERGY_PROVIDERS,
175	):
UNCOV 176	for asset_name, asset_params in dict_values.get(asset_group, {}).items():	×
UNCOV 177	asset_type = {"label": asset_name}	×
UNCOV 178	for param in (OEMOF_ASSET_TYPE, TYPE_ASSET):	×
UNCOV 179	asset_type[param] = asset_params.get(param)	×
UNCOV 180	asset_busses = {}	×
UNCOV 181	input_bus = asset_params.get(INFLOW_DIRECTION)	×
UNCOV 182	if input_bus is not None:	×
UNCOV 183	if not isinstance(input_bus, list):	×
184	# print("not a list :", input_bus)
UNCOV 185	input_bus = [input_bus]	×
186	else:
UNCOV 187	input_bus = []	×
188
UNCOV 189	output_bus = asset_params.get(OUTFLOW_DIRECTION)	×
UNCOV 190	if output_bus is not None:	×
UNCOV 191	if not isinstance(output_bus, list):	×
192	# print("not a list: ", output_bus)
UNCOV 193	output_bus = [output_bus]	×
194	else:
UNCOV 195	output_bus = []	×
196
UNCOV 197	for bus in input_bus + output_bus:	×
UNCOV 198	asset_busses[bus] = dict_values[ENERGY_BUSSES][bus].get(ENERGY_VECTOR)	×
UNCOV 199	asset_type["busses"] = asset_busses	×
UNCOV 200	if asset_type[TYPE_ASSET] == REDUCABLE_DEMAND:	×
201
UNCOV 202	asset_label = asset_type["label"]	×
UNCOV 203	asset_type["label"] = reducable_demand_name(asset_label)	×
UNCOV 204	asset_types.append(asset_type)	×
UNCOV 205	crit_asset_type = deepcopy(asset_type)	×
UNCOV 206	crit_asset_type["label"] = reducable_demand_name(	×
207	asset_label, critical=True
208	)
UNCOV 209	asset_types.append(crit_asset_type)	×
210	else:
UNCOV 211	asset_types.append(asset_type)	×
UNCOV 212	return asset_types	×

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