solph.processing.convert_keys_to_strings() - Code Metrics - Inspection of "Feature/result dict wrapper" - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#1195)

by Patrik

created 2025-10-27 16:44 UTC

solph.processing.convert_keys_to_strings() B

↳ Parent: solph.processing

Complexity

Conditions

Size

Total Lines	23
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	12
dl	0
loc	23
rs	8.6666
c	0
b	0
f	0
cc	6
nop	2

# -*- coding: utf-8 -*-

"""Modules for providing a convenient data structure for solph results.

Information about the possible usage is provided within the examples.

SPDX-FileCopyrightText: Uwe Krien <[email protected]>
SPDX-FileCopyrightText: Simon Hilpert
SPDX-FileCopyrightText: Cord Kaldemeyer
SPDX-FileCopyrightText: Stephan Günther
SPDX-FileCopyrightText: henhuy
SPDX-FileCopyrightText: Johannes Kochems
SPDX-FileCopyrightText: Patrik Schönfeldt <[email protected]>

SPDX-License-Identifier: MIT

"""
import numbers
from collections import abc

import pandas as pd

from ._plumbing import _FakeSequence
from ._processing import results as new_results
from .helpers import flatten


def convert_keys_to_strings(result, keep_none_type=False):
    """
    Convert the dictionary keys to strings.

    All (tuple) keys of the result object e.g. results[(pp1, bus1)] are
    converted into strings that represent the object labels
    e.g. results[('pp1','bus1')].
    """
    if keep_none_type:
        converted = {
            (
                tuple([str(e) if e is not None else None for e in k])
                if isinstance(k, tuple)
                else str(k) if k is not None else None
            ): v
            for k, v in result.items()
        }
    else:
        converted = {
            tuple(map(str, k)) if isinstance(k, tuple) else str(k): v
            for k, v in result.items()
        }
    return converted


def results(model, remove_last_time_point=False):
    return new_results(
        model=model,
        remove_last_time_point=remove_last_time_point,
    )


def meta_results(om, undefined=False):
    """
    Fetch some metadata from the Solver. Feel free to add more keys.

    Valid keys of the resulting dictionary are: 'objective', 'problem',
    'solver'.

    om : oemof.solph.Model
        A solved Model.
    undefined : bool
        By default (False) only defined keys can be found in the dictionary.
        Set to True to get also the undefined keys.

    Returns
    -------
    dict
    """
    meta_res = {"objective": om.objective()}

    for k1 in ["Problem", "Solver"]:
        k1 = k1.lower()
        meta_res[k1] = {}
        for k2, v2 in om.es.results[k1][0].items():
            try:
                if str(om.es.results[k1][0][k2]) == "<undefined>":
                    if undefined:
                        meta_res[k1][k2] = str(om.es.results[k1][0][k2])
                else:
                    meta_res[k1][k2] = om.es.results[k1][0][k2]
            except TypeError:
                if undefined:
                    msg = "Cannot fetch meta results of type {0}"
                    meta_res[k1][k2] = msg.format(
                        type(om.es.results[k1][0][k2])
                    )

    return meta_res


def __separate_attrs(
    system, exclude_attrs, get_flows=False, exclude_none=True
):
    """
    Create a dictionary with flow scalars and series.

    The dictionary is structured with flows as tuples and nested dictionaries
    holding the scalars and series e.g.
    {(node1, node2): {'scalars': {'attr1': scalar, 'attr2': 'text'},
    'sequences': {'attr1': iterable, 'attr2': iterable}}}

    system:
        A solved oemof.solph.Model or oemof.solph.Energysystem
    exclude_attrs: List[str]
        List of additional attributes which shall be excluded from
        parameter dict
    get_flows: bool
        Whether to include flow values or not
    exclude_none: bool
        If set, scalars and sequences containing None values are excluded

    Returns
    -------
    dict
    """

    def detect_scalars_and_sequences(com):
        scalars = {}
        sequences = {}

        default_exclusions = [
            "__",
            "_",
            "registry",
            "inputs",
            "outputs",
            "Label",
            "input",
            "output",
            "constraint_group",
        ]
        # Must be tuple in order to work with `str.startswith()`:
        exclusions = tuple(default_exclusions + exclude_attrs)
        attrs = [
            i
            for i in dir(com)
            if not (i.startswith(exclusions) or callable(getattr(com, i)))
        ]

        for a in attrs:
            attr_value = getattr(com, a)

            # Iterate trough investment and add scalars and sequences with
            # "investment" prefix to component data:
            if attr_value.__class__.__name__ == "Investment":
                invest_data = detect_scalars_and_sequences(attr_value)
                scalars.update(
                    {
                        "investment_" + str(k): v
                        for k, v in invest_data["scalars"].items()
                    }
                )
                sequences.update(
                    {
                        "investment_" + str(k): v
                        for k, v in invest_data["sequences"].items()
                    }
                )
                continue

            if isinstance(attr_value, str):
                scalars[a] = attr_value
                continue

            # If the label is a tuple it is iterable, therefore it should be
            # converted to a string. Otherwise, it will be a sequence.
            if a == "label":
                attr_value = str(attr_value)

            if isinstance(attr_value, abc.Iterable):
                sequences[a] = attr_value
            elif isinstance(attr_value, _FakeSequence):
                scalars[a] = attr_value.value
            else:
                scalars[a] = attr_value

        sequences = flatten(sequences)

        com_data = {
            "scalars": scalars,
            "sequences": sequences,
        }
        move_undetected_scalars(com_data)
        if exclude_none:
            remove_nones(com_data)

        com_data = {
            "scalars": pd.Series(com_data["scalars"]),
            "sequences": pd.DataFrame(com_data["sequences"]),
        }
        return com_data

    def move_undetected_scalars(com):
        for ckey, value in list(com["sequences"].items()):
            if isinstance(value, (str, numbers.Number)):
                com["scalars"][ckey] = value
                del com["sequences"][ckey]
            elif isinstance(value, _FakeSequence):
                com["scalars"][ckey] = value.value
                del com["sequences"][ckey]
            elif len(value) == 0:
                del com["sequences"][ckey]

    def remove_nones(com):
        for ckey, value in list(com["scalars"].items()):
            if value is None:
                del com["scalars"][ckey]
        for ckey, value in list(com["sequences"].items()):
            if len(value) == 0 or value[0] is None:
                del com["sequences"][ckey]

    # Check if system is es or om:
    if system.__class__.__name__ == "EnergySystem":
        components = system.flows() if get_flows else system.nodes
    else:
        components = system.flows if get_flows else system.es.nodes

    data = {}
    for com_key in components:
        component = components[com_key] if get_flows else com_key
        component_data = detect_scalars_and_sequences(component)
        comkey = com_key if get_flows else (com_key, None)
        data[comkey] = component_data
    return data


def parameter_as_dict(system, exclude_none=True, exclude_attrs=None):
    """
    Create a result dictionary containing node parameters.

    Results are written into a dictionary of pandas objects where
    a Series holds all scalar values and a dataframe all sequences for nodes
    and flows.
    The dictionary is keyed by flows (n, n) and nodes (n, None), e.g.
    `parameter[(n, n)]['sequences']` or `parameter[(n, n)]['scalars']`.

    Parameters
    ----------
    system: energy_system.EnergySystem
        A populated energy system.
    exclude_none: bool
        If True, all scalars and sequences containing None values are excluded
    exclude_attrs: Optional[List[str]]
        Optional list of additional attributes which shall be excluded from
        parameter dict

    Returns
    -------
    dict: Parameters for all nodes and flows
    """

    if exclude_attrs is None:
        exclude_attrs = []

    flow_data = __separate_attrs(
        system, exclude_attrs, get_flows=True, exclude_none=exclude_none
    )
    node_data = __separate_attrs(
        system, exclude_attrs, get_flows=False, exclude_none=exclude_none
    )

    flow_data.update(node_data)
    return flow_data


1			# -- coding: utf-8 --
2
3			"""Modules for providing a convenient data structure for solph results.
4
5			Information about the possible usage is provided within the examples.
6
7			SPDX-FileCopyrightText: Uwe Krien <[email protected]>
8			SPDX-FileCopyrightText: Simon Hilpert
9			SPDX-FileCopyrightText: Cord Kaldemeyer
10			SPDX-FileCopyrightText: Stephan Günther
11			SPDX-FileCopyrightText: henhuy
12			SPDX-FileCopyrightText: Johannes Kochems
13			SPDX-FileCopyrightText: Patrik Schönfeldt <[email protected]>
14
15			SPDX-License-Identifier: MIT
16
17			"""
18			import numbers
19			from collections import abc
20
21			import pandas as pd
22
23			from ._plumbing import _FakeSequence
24			from ._processing import results as new_results
25			from .helpers import flatten
26
27
28			def convert_keys_to_strings(result, keep_none_type=False):
29			"""
30			Convert the dictionary keys to strings.
31
32			All (tuple) keys of the result object e.g. results[(pp1, bus1)] are
33			converted into strings that represent the object labels
34			e.g. results[('pp1','bus1')].
35			"""
36			if keep_none_type:
37			converted = {
38			(
39			tuple([str(e) if e is not None else None for e in k])
40			if isinstance(k, tuple)
41			else str(k) if k is not None else None
42			): v
43			for k, v in result.items()
44			}
45			else:
46			converted = {
47			tuple(map(str, k)) if isinstance(k, tuple) else str(k): v
48			for k, v in result.items()
49			}
50			return converted
51
52
53			def results(model, remove_last_time_point=False):
54			return new_results(
55			model=model,
56			remove_last_time_point=remove_last_time_point,
57			)
58
59
60			def meta_results(om, undefined=False):
61			"""
62			Fetch some metadata from the Solver. Feel free to add more keys.
63
64			Valid keys of the resulting dictionary are: 'objective', 'problem',
65			'solver'.
66
67			om : oemof.solph.Model
68			A solved Model.
69			undefined : bool
70			By default (False) only defined keys can be found in the dictionary.
71			Set to True to get also the undefined keys.
72
73			Returns
74			-------
75			dict
76			"""
77			meta_res = {"objective": om.objective()}
78
79			for k1 in ["Problem", "Solver"]:
80			k1 = k1.lower()
81			meta_res[k1] = {}
82			for k2, v2 in om.es.results[k1][0].items():
83			try:
84			if str(om.es.results[k1][0][k2]) == "<undefined>":
85			if undefined:
86			meta_res[k1][k2] = str(om.es.results[k1][0][k2])
87			else:
88			meta_res[k1][k2] = om.es.results[k1][0][k2]
89			except TypeError:
90			if undefined:
91			msg = "Cannot fetch meta results of type {0}"
92			meta_res[k1][k2] = msg.format(
93			type(om.es.results[k1][0][k2])
94			)
95
96			return meta_res
97
98
99			def __separate_attrs(
100			system, exclude_attrs, get_flows=False, exclude_none=True
101			):
102			"""
103			Create a dictionary with flow scalars and series.
104
105			The dictionary is structured with flows as tuples and nested dictionaries
106			holding the scalars and series e.g.
107			{(node1, node2): {'scalars': {'attr1': scalar, 'attr2': 'text'},
108			'sequences': {'attr1': iterable, 'attr2': iterable}}}
109
110			system:
111			A solved oemof.solph.Model or oemof.solph.Energysystem
112			exclude_attrs: List[str]
113			List of additional attributes which shall be excluded from
114			parameter dict
115			get_flows: bool
116			Whether to include flow values or not
117			exclude_none: bool
118			If set, scalars and sequences containing None values are excluded
119
120			Returns
121			-------
122			dict
123			"""
124
125			def detect_scalars_and_sequences(com):
126			scalars = {}
127			sequences = {}
128
129			default_exclusions = [
130			"__",
131			"_",
132			"registry",
133			"inputs",
134			"outputs",
135			"Label",
136			"input",
137			"output",
138			"constraint_group",
139			]
140			# Must be tuple in order to work with `str.startswith()`:
141			exclusions = tuple(default_exclusions + exclude_attrs)
142			attrs = [
143			i
144			for i in dir(com)
145			if not (i.startswith(exclusions) or callable(getattr(com, i)))
146			]
147
148			for a in attrs:
149			attr_value = getattr(com, a)
150
151			# Iterate trough investment and add scalars and sequences with
152			# "investment" prefix to component data:
153			if attr_value.__class__.__name__ == "Investment":
154			invest_data = detect_scalars_and_sequences(attr_value)
155			scalars.update(
156			{
157			"investment_" + str(k): v
158			for k, v in invest_data["scalars"].items()
159			}
160			)
161			sequences.update(
162			{
163			"investment_" + str(k): v
164			for k, v in invest_data["sequences"].items()
165			}
166			)
167			continue
168
169			if isinstance(attr_value, str):
170			scalars[a] = attr_value
171			continue
172
173			# If the label is a tuple it is iterable, therefore it should be
174			# converted to a string. Otherwise, it will be a sequence.
175			if a == "label":
176			attr_value = str(attr_value)
177
178			if isinstance(attr_value, abc.Iterable):
179			sequences[a] = attr_value
180			elif isinstance(attr_value, _FakeSequence):
181			scalars[a] = attr_value.value
182			else:
183			scalars[a] = attr_value
184
185			sequences = flatten(sequences)
186
187			com_data = {
188			"scalars": scalars,
189			"sequences": sequences,
190			}
191			move_undetected_scalars(com_data)
192			if exclude_none:
193			remove_nones(com_data)
194
195			com_data = {
196			"scalars": pd.Series(com_data["scalars"]),
197			"sequences": pd.DataFrame(com_data["sequences"]),
198			}
199			return com_data
200
201			def move_undetected_scalars(com):
202			for ckey, value in list(com["sequences"].items()):
203			if isinstance(value, (str, numbers.Number)):
204			com["scalars"][ckey] = value
205			del com["sequences"][ckey]
206			elif isinstance(value, _FakeSequence):
207			com["scalars"][ckey] = value.value
208			del com["sequences"][ckey]
209			elif len(value) == 0:
210			del com["sequences"][ckey]
211
212			def remove_nones(com):
213			for ckey, value in list(com["scalars"].items()):
214			if value is None:
215			del com["scalars"][ckey]
216			for ckey, value in list(com["sequences"].items()):
217			if len(value) == 0 or value[0] is None:
218			del com["sequences"][ckey]
219
220			# Check if system is es or om:
221			if system.__class__.__name__ == "EnergySystem":
222			components = system.flows() if get_flows else system.nodes
223			else:
224			components = system.flows if get_flows else system.es.nodes
225
226			data = {}
227			for com_key in components:
228			component = components[com_key] if get_flows else com_key
229			component_data = detect_scalars_and_sequences(component)
230			comkey = com_key if get_flows else (com_key, None)
231			data[comkey] = component_data
232			return data
233
234
235			def parameter_as_dict(system, exclude_none=True, exclude_attrs=None):
236			"""
237			Create a result dictionary containing node parameters.
238
239			Results are written into a dictionary of pandas objects where
240			a Series holds all scalar values and a dataframe all sequences for nodes
241			and flows.
242			The dictionary is keyed by flows (n, n) and nodes (n, None), e.g.
243			`parameter[(n, n)]['sequences']` or `parameter[(n, n)]['scalars']`.
244
245			Parameters
246			----------
247			system: energy_system.EnergySystem
248			A populated energy system.
249			exclude_none: bool
250			If True, all scalars and sequences containing None values are excluded
251			exclude_attrs: Optional[List[str]]
252			Optional list of additional attributes which shall be excluded from
253			parameter dict
254
255			Returns
256			-------
257			dict: Parameters for all nodes and flows
258			"""
259
260			if exclude_attrs is None:
261			exclude_attrs = []
262
263			flow_data = __separate_attrs(
264			system, exclude_attrs, get_flows=True, exclude_none=exclude_none
265			)
266			node_data = __separate_attrs(
267			system, exclude_attrs, get_flows=False, exclude_none=exclude_none
268			)
269
270			flow_data.update(node_data)
271			return flow_data
272

oemof / oemof-solph

Pull Request — dev (#1195)

solph.processing.convert_keys_to_strings() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like