tests.test_processing.TestParameterResult.test_net_storage_flow_empty() - Code Metrics - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

TestParameterResult.test_net_storage_flow_empty() A
last analyzed 2025-11-04 19:41 UTC

↳ Parent: tests.test_processing

Complexity

Conditions

Size

Total Lines	6
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	6
dl	0
loc	6
rs	10
c	0
b	0
f	0
cc	1
nop	1

# -*- coding: utf-8 -

"""Tests the processing module of solph.

This file is part of project oemof (github.com/oemof/oemof). It's copyrighted
by the contributors recorded in the version control history of the file,
available from its original location oemof/tests/test_processing.py

SPDX-License-Identifier: MIT
"""

import pandas
import pytest
from pandas.testing import assert_frame_equal
from pandas.testing import assert_series_equal

from oemof.solph import EnergySystem
from oemof.solph import Investment
from oemof.solph import Model
from oemof.solph import processing
from oemof.solph import views
from oemof.solph.buses import Bus
from oemof.solph.components import Converter
from oemof.solph.components import GenericStorage
from oemof.solph.components import Sink
from oemof.solph.flows import Flow


class TestParameterResult:
    @classmethod
    def setup_class(cls):
        cls.period = 24
        cls.es = EnergySystem(
            timeindex=pandas.date_range(
                "2016-01-01",
                periods=cls.period,
                freq="h",
            ),
            infer_last_interval=True,
        )

        # BUSSES
        b_el1 = Bus(label="b_el1")
        b_el2 = Bus(label="b_el2")
        b_diesel = Bus(label="b_diesel", balanced=False)
        cls.es.add(b_el1, b_el2, b_diesel)

        # TEST DIESEL:
        dg = Converter(
            label="diesel",
            inputs={b_diesel: Flow(variable_costs=2)},
            outputs={
                b_el1: Flow(
                    variable_costs=1, nominal_capacity=Investment(ep_costs=0.5)
                )
            },
            conversion_factors={b_el1: 2},
        )

        batt = GenericStorage(
            label="storage",
            inputs={
                b_el1: Flow(variable_costs=3, nominal_capacity=Investment())
            },
            outputs={
                b_el2: Flow(variable_costs=2.5, nominal_capacity=Investment())
            },
            loss_rate=0.00,
            initial_storage_level=0,
            invest_relation_input_capacity=1 / 6,
            invest_relation_output_capacity=1 / 6,
            inflow_conversion_factor=1,
            outflow_conversion_factor=0.8,
            nominal_capacity=Investment(ep_costs=0.4),
        )

        cls.demand_values = [0.0] + [100] * 23
        demand = Sink(
            label="demand_el",
            inputs={
                b_el2: Flow(
                    nominal_capacity=1,
                    fix=cls.demand_values,
                )
            },
        )
        cls.es.add(dg, batt, demand)
        cls.om = Model(cls.es)
        cls.om.receive_duals()
        cls.om.solve()
        cls.mod = Model(cls.es)
        cls.mod.solve()

    def test_flows_with_none_exclusion(self):
        b_el2 = self.es.groups["b_el2"]
        demand = self.es.groups["demand_el"]
        param_results = processing.parameter_as_dict(
            self.es, exclude_none=True
        )
        assert_series_equal(
            param_results[(b_el2, demand)]["scalars"].sort_index(),
            pandas.Series(
                {
                    "bidirectional": False,
                    "integer": False,
                    "nominal_capacity": 1,
                    "max": 1,
                    "min": 0,
                    "variable_costs": 0,
                    "label": str(b_el2.outputs[demand].label),
                }
            ).sort_index(),
        )
        assert_frame_equal(
            param_results[(b_el2, demand)]["sequences"],
            pandas.DataFrame({"fix": self.demand_values}),
            check_like=True,
        )

    def test_flows_without_none_exclusion(self):
        b_el2 = self.es.groups["b_el2"]
        demand = self.es.groups["demand_el"]
        param_results = processing.parameter_as_dict(
            self.es, exclude_none=False
        )
        default_attributes = {
            "age": None,
            "lifetime": None,
            "integer": False,
            "investment": None,
            "nominal_capacity": 1,
            "nonconvex": None,
            "bidirectional": False,
            "full_load_time_max": None,
            "full_load_time_min": None,
            "max": 1,
            "min": 0,
            "negative_gradient_limit": None,
            "positive_gradient_limit": None,
            "variable_costs": 0,
            "fixed_costs": None,
            "flow": None,
            "values": None,
            "label": str(b_el2.outputs[demand].label),
        }
        assert_series_equal(
            param_results[(b_el2, demand)]["scalars"].sort_index(),
            pandas.Series(default_attributes).sort_index(),
        )
        sequences_attributes = {
            "fix": self.demand_values,
        }

        assert_frame_equal(
            param_results[(b_el2, demand)]["sequences"],
            pandas.DataFrame(sequences_attributes),
            check_like=True,
        )

    def test_nodes_with_none_exclusion(self):

        param_results = processing.parameter_as_dict(
            self.es, exclude_none=True
        )
        param_results = processing.convert_keys_to_strings(param_results)
        assert_series_equal(
            param_results[("storage", "None")]["scalars"],
            pandas.Series(
                {
                    "balanced": True,
                    "depth": 0,
                    "initial_storage_level": 0,
                    "investment_age": 0,
                    "investment_existing": 0,
                    "investment_nonconvex": False,
                    "investment_ep_costs": 0.4,
                    "investment_maximum": float("inf"),
                    "investment_minimum": 0,
                    "investment_nonconvex": False,
                    "investment_offset": 0,
                    "label": "storage",
                    "fixed_costs": 0,
                    "fixed_losses_absolute": 0,
                    "fixed_losses_relative": 0,
                    "inflow_conversion_factor": 1,
                    "invest_relation_input_capacity": 1 / 6,
                    "invest_relation_output_capacity": 1 / 6,
                    "loss_rate": 0,
                    "max_storage_level": 1,
                    "min_storage_level": 0,
                    "outflow_conversion_factor": 0.8,
                }
            ),
        )
        assert_frame_equal(
            param_results[("storage", "None")]["sequences"], pandas.DataFrame()
        )

    def test_nodes_with_none_exclusion_old_name(self):

        param_results = processing.parameter_as_dict(
            self.es, exclude_none=True
        )
        param_results = processing.convert_keys_to_strings(
            param_results, keep_none_type=True
        )
        assert_series_equal(
            param_results[("storage", None)]["scalars"],
            pandas.Series(
                {
                    "balanced": True,
                    "depth": 0,
                    "initial_storage_level": 0,
                    "investment_age": 0,
                    "investment_existing": 0,
                    "investment_nonconvex": False,
                    "investment_ep_costs": 0.4,
                    "investment_maximum": float("inf"),
                    "investment_minimum": 0,
                    "investment_nonconvex": False,
                    "investment_offset": 0,
                    "label": "storage",
                    "fixed_costs": 0,
                    "fixed_losses_absolute": 0,
                    "fixed_losses_relative": 0,
                    "inflow_conversion_factor": 1,
                    "invest_relation_input_capacity": 1 / 6,
                    "invest_relation_output_capacity": 1 / 6,
                    "loss_rate": 0,
                    "max_storage_level": 1,
                    "min_storage_level": 0,
                    "outflow_conversion_factor": 0.8,
                }
            ),
        )
        assert_frame_equal(
            param_results[("storage", None)]["sequences"], pandas.DataFrame()
        )

    def test_nodes_without_none_exclusion(self):
        diesel = self.es.groups["diesel"]
        param_results = processing.parameter_as_dict(
            self.es, exclude_none=False
        )
        assert_series_equal(
            param_results[(diesel, None)]["scalars"],
            pandas.Series(
                {
                    "depth": 0,
                    "label": "diesel",
                    "parent": None,
                    "conversion_factors_b_el1": 2,
                    "conversion_factors_b_diesel": 1,
                }
            ),
        )
        assert_frame_equal(
            param_results[(diesel, None)]["sequences"], pandas.DataFrame()
        )

    def test_nodes_with_excluded_attrs(self):
        diesel = self.es.groups["diesel"]
        param_results = processing.parameter_as_dict(
            self.es, exclude_attrs=["conversion_factors"]
        )
        assert_series_equal(
            param_results[(diesel, None)]["scalars"],
            pandas.Series(
                {
                    "depth": 0,
                    "label": "diesel",
                }
            ),
        )
        assert_frame_equal(
            param_results[(diesel, None)]["sequences"], pandas.DataFrame()
        )

    def test_parameter_with_node_view(self):
        param_results = processing.parameter_as_dict(
            self.es, exclude_none=True
        )
        bel1 = views.node(param_results, "b_el1")
        assert (
            bel1["scalars"][[(("b_el1", "storage"), "variable_costs")]].values
            == 3
        )

        bel1_m = views.node(param_results, "b_el1", multiindex=True)
        assert bel1_m["scalars"][("b_el1", "storage", "variable_costs")] == 3

    def test_multiindex_sequences(self):
        results = processing.results(self.om)
        bel1 = views.node(results, "b_el1", multiindex=True)
        assert (
            int(bel1["sequences"][("diesel", "b_el1", "flow")].sum()) == 2875
        )

    def test_error_from_nan_values(self):
        trsf = self.es.groups["diesel"]
        bus = self.es.groups["b_el1"]
        self.mod.flow[trsf, bus, 5] = float("nan")
        with pytest.raises(ValueError):
            processing.results(self.mod)

    def test_duals(self):
        results = processing.results(self.om)
        bel = views.node(results, "b_el1", multiindex=True)
        assert int(bel["sequences"]["b_el1", "None", "duals"].sum()) == 48

    def test_node_weight_by_type(self):
        results = processing.results(self.om)
        storage_content = views.node_weight_by_type(
            results, node_type=GenericStorage
        )
        assert (
            storage_content.sum().iloc[0] == pytest.approx(1437.5, abs=0.1)
        ).all()

    def test_output_by_type_view(self):
        results = processing.results(self.om)
        converter_output = views.node_output_by_type(
            results, node_type=Converter
        )
        compare = views.node(results, "diesel", multiindex=True)["sequences"][
            ("diesel", "b_el1", "flow")
        ]
        assert converter_output.sum().iloc[0] == pytest.approx(compare.sum())

    def test_input_by_type_view(self):
        results = processing.results(self.om)
        sink_input = views.node_input_by_type(results, node_type=Sink)
        compare = views.node(results, "demand_el", multiindex=True)
        assert sink_input.sum().iloc[0] == pytest.approx(
            compare["sequences"][("b_el2", "demand_el", "flow")].sum()
        )

    def test_net_storage_flow(self):
        results = processing.results(self.om)
        storage_flow = views.net_storage_flow(
            results, node_type=GenericStorage
        )

        compare = views.node(results, "storage", multiindex=True)["sequences"]

        assert (
            (
                (
                    compare[("storage", "b_el2", "flow")]
                    - compare[("b_el1", "storage", "flow")]
                )
                .to_frame()
                .fillna(0)
                == storage_flow.values
            )
            .all()
            .iloc[0]
        )

    def test_output_by_type_view_empty(self):
        results = processing.results(self.om)
        view = views.node_output_by_type(results, node_type=Flow)
        assert view is None

    def test_input_by_type_view_empty(self):
        results = processing.results(self.om)
        view = views.node_input_by_type(results, node_type=Flow)
        assert view is None

    def test_net_storage_flow_empty(self):
        results = processing.results(self.om)
        view = views.net_storage_flow(results, node_type=Sink)
        assert view is None
        view2 = views.net_storage_flow(results, node_type=Flow)
        assert view2 is None

    def test_node_weight_by_type_empty(self):
        results = processing.results(self.om)
        view = views.node_weight_by_type(results, node_type=Flow)
        assert view is None


1		# -*- coding: utf-8 -
2
3		"""Tests the processing module of solph.
4
5		This file is part of project oemof (github.com/oemof/oemof). It's copyrighted
6		by the contributors recorded in the version control history of the file,
7		available from its original location oemof/tests/test_processing.py
8
9		SPDX-License-Identifier: MIT
10		"""
11
12		import pandas
13		import pytest
14		from pandas.testing import assert_frame_equal
15		from pandas.testing import assert_series_equal
16
17		from oemof.solph import EnergySystem
18		from oemof.solph import Investment
19		from oemof.solph import Model
20		from oemof.solph import processing
21		from oemof.solph import views
22		from oemof.solph.buses import Bus
23		from oemof.solph.components import Converter
24		from oemof.solph.components import GenericStorage
25		from oemof.solph.components import Sink
26		from oemof.solph.flows import Flow
27
28
29		class TestParameterResult:
30		@classmethod
31		def setup_class(cls):
32		cls.period = 24
33		cls.es = EnergySystem(
34		timeindex=pandas.date_range(
35		"2016-01-01",
36		periods=cls.period,
37		freq="h",
38		),
39		infer_last_interval=True,
40		)
41
42		# BUSSES
43		b_el1 = Bus(label="b_el1")
44		b_el2 = Bus(label="b_el2")
45		b_diesel = Bus(label="b_diesel", balanced=False)
46		cls.es.add(b_el1, b_el2, b_diesel)
47
48		# TEST DIESEL:
49		dg = Converter(
50		label="diesel",
51		inputs={b_diesel: Flow(variable_costs=2)},
52		outputs={
53		b_el1: Flow(
54		variable_costs=1, nominal_capacity=Investment(ep_costs=0.5)
55		)
56		},
57		conversion_factors={b_el1: 2},
58		)
59
60		batt = GenericStorage(
61		label="storage",
62		inputs={
63		b_el1: Flow(variable_costs=3, nominal_capacity=Investment())
64		},
65		outputs={
66		b_el2: Flow(variable_costs=2.5, nominal_capacity=Investment())
67		},
68		loss_rate=0.00,
69		initial_storage_level=0,
70		invest_relation_input_capacity=1 / 6,
71		invest_relation_output_capacity=1 / 6,
72		inflow_conversion_factor=1,
73		outflow_conversion_factor=0.8,
74		nominal_capacity=Investment(ep_costs=0.4),
75		)
76
77		cls.demand_values = [0.0] + [100] * 23
78		demand = Sink(
79		label="demand_el",
80		inputs={
81		b_el2: Flow(
82		nominal_capacity=1,
83		fix=cls.demand_values,
84		)
85		},
86		)
87		cls.es.add(dg, batt, demand)
88		cls.om = Model(cls.es)
89		cls.om.receive_duals()
90		cls.om.solve()
91		cls.mod = Model(cls.es)
92		cls.mod.solve()
93
94		def test_flows_with_none_exclusion(self):
95		b_el2 = self.es.groups["b_el2"]
96		demand = self.es.groups["demand_el"]
97		param_results = processing.parameter_as_dict(
98		self.es, exclude_none=True
99		)
100		assert_series_equal(
101		param_results[(b_el2, demand)]["scalars"].sort_index(),
102		pandas.Series(
103		{
104		"bidirectional": False,
105		"integer": False,
106		"nominal_capacity": 1,
107		"max": 1,
108		"min": 0,
109		"variable_costs": 0,
110		"label": str(b_el2.outputs[demand].label),
111		}
112		).sort_index(),
113		)
114		assert_frame_equal(
115		param_results[(b_el2, demand)]["sequences"],
116		pandas.DataFrame({"fix": self.demand_values}),
117		check_like=True,
118		)
119
120		def test_flows_without_none_exclusion(self):
121		b_el2 = self.es.groups["b_el2"]
122		demand = self.es.groups["demand_el"]
123		param_results = processing.parameter_as_dict(
124		self.es, exclude_none=False
125		)
126		default_attributes = {
127		"age": None,
128		"lifetime": None,
129		"integer": False,
130		"investment": None,
131		"nominal_capacity": 1,
132		"nonconvex": None,
133		"bidirectional": False,
134		"full_load_time_max": None,
135		"full_load_time_min": None,
136		"max": 1,
137		"min": 0,
138		"negative_gradient_limit": None,
139		"positive_gradient_limit": None,
140		"variable_costs": 0,
141		"fixed_costs": None,
142		"flow": None,
143		"values": None,
144		"label": str(b_el2.outputs[demand].label),
145		}
146		assert_series_equal(
147		param_results[(b_el2, demand)]["scalars"].sort_index(),
148		pandas.Series(default_attributes).sort_index(),
149		)
150		sequences_attributes = {
151		"fix": self.demand_values,
152		}
153
154		assert_frame_equal(
155		param_results[(b_el2, demand)]["sequences"],
156		pandas.DataFrame(sequences_attributes),
157		check_like=True,
158		)
159
160	View Code Duplication	def test_nodes_with_none_exclusion(self):
		0 ignored issues – show Duplication introduced 2020-03-31 12:19 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
161		param_results = processing.parameter_as_dict(
162		self.es, exclude_none=True
163		)
164		param_results = processing.convert_keys_to_strings(param_results)
165		assert_series_equal(
166		param_results[("storage", "None")]["scalars"],
167		pandas.Series(
168		{
169		"balanced": True,
170		"depth": 0,
171		"initial_storage_level": 0,
172		"investment_age": 0,
173		"investment_existing": 0,
174		"investment_nonconvex": False,
175		"investment_ep_costs": 0.4,
176		"investment_maximum": float("inf"),
177		"investment_minimum": 0,
178		"investment_nonconvex": False,
179		"investment_offset": 0,
180		"label": "storage",
181		"fixed_costs": 0,
182		"fixed_losses_absolute": 0,
183		"fixed_losses_relative": 0,
184		"inflow_conversion_factor": 1,
185		"invest_relation_input_capacity": 1 / 6,
186		"invest_relation_output_capacity": 1 / 6,
187		"loss_rate": 0,
188		"max_storage_level": 1,
189		"min_storage_level": 0,
190		"outflow_conversion_factor": 0.8,
191		}
192		),
193		)
194		assert_frame_equal(
195		param_results[("storage", "None")]["sequences"], pandas.DataFrame()
196		)
197
198	View Code Duplication	def test_nodes_with_none_exclusion_old_name(self):
		0 ignored issues – show Duplication introduced 2024-02-13 11:54 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
199		param_results = processing.parameter_as_dict(
200		self.es, exclude_none=True
201		)
202		param_results = processing.convert_keys_to_strings(
203		param_results, keep_none_type=True
204		)
205		assert_series_equal(
206		param_results[("storage", None)]["scalars"],
207		pandas.Series(
208		{
209		"balanced": True,
210		"depth": 0,
211		"initial_storage_level": 0,
212		"investment_age": 0,
213		"investment_existing": 0,
214		"investment_nonconvex": False,
215		"investment_ep_costs": 0.4,
216		"investment_maximum": float("inf"),
217		"investment_minimum": 0,
218		"investment_nonconvex": False,
219		"investment_offset": 0,
220		"label": "storage",
221		"fixed_costs": 0,
222		"fixed_losses_absolute": 0,
223		"fixed_losses_relative": 0,
224		"inflow_conversion_factor": 1,
225		"invest_relation_input_capacity": 1 / 6,
226		"invest_relation_output_capacity": 1 / 6,
227		"loss_rate": 0,
228		"max_storage_level": 1,
229		"min_storage_level": 0,
230		"outflow_conversion_factor": 0.8,
231		}
232		),
233		)
234		assert_frame_equal(
235		param_results[("storage", None)]["sequences"], pandas.DataFrame()
236		)
237
238		def test_nodes_without_none_exclusion(self):
239		diesel = self.es.groups["diesel"]
240		param_results = processing.parameter_as_dict(
241		self.es, exclude_none=False
242		)
243		assert_series_equal(
244		param_results[(diesel, None)]["scalars"],
245		pandas.Series(
246		{
247		"depth": 0,
248		"label": "diesel",
249		"parent": None,
250		"conversion_factors_b_el1": 2,
251		"conversion_factors_b_diesel": 1,
252		}
253		),
254		)
255		assert_frame_equal(
256		param_results[(diesel, None)]["sequences"], pandas.DataFrame()
257		)
258
259		def test_nodes_with_excluded_attrs(self):
260		diesel = self.es.groups["diesel"]
261		param_results = processing.parameter_as_dict(
262		self.es, exclude_attrs=["conversion_factors"]
263		)
264		assert_series_equal(
265		param_results[(diesel, None)]["scalars"],
266		pandas.Series(
267		{
268		"depth": 0,
269		"label": "diesel",
270		}
271		),
272		)
273		assert_frame_equal(
274		param_results[(diesel, None)]["sequences"], pandas.DataFrame()
275		)
276
277		def test_parameter_with_node_view(self):
278		param_results = processing.parameter_as_dict(
279		self.es, exclude_none=True
280		)
281		bel1 = views.node(param_results, "b_el1")
282		assert (
283		bel1["scalars"][[(("b_el1", "storage"), "variable_costs")]].values
284		== 3
285		)
286
287		bel1_m = views.node(param_results, "b_el1", multiindex=True)
288		assert bel1_m["scalars"][("b_el1", "storage", "variable_costs")] == 3
289
290		def test_multiindex_sequences(self):
291		results = processing.results(self.om)
292		bel1 = views.node(results, "b_el1", multiindex=True)
293		assert (
294		int(bel1["sequences"][("diesel", "b_el1", "flow")].sum()) == 2875
295		)
296
297		def test_error_from_nan_values(self):
298		trsf = self.es.groups["diesel"]
299		bus = self.es.groups["b_el1"]
300		self.mod.flow[trsf, bus, 5] = float("nan")
301		with pytest.raises(ValueError):
302		processing.results(self.mod)
303
304		def test_duals(self):
305		results = processing.results(self.om)
306		bel = views.node(results, "b_el1", multiindex=True)
307		assert int(bel["sequences"]["b_el1", "None", "duals"].sum()) == 48
308
309		def test_node_weight_by_type(self):
310		results = processing.results(self.om)
311		storage_content = views.node_weight_by_type(
312		results, node_type=GenericStorage
313		)
314		assert (
315		storage_content.sum().iloc[0] == pytest.approx(1437.5, abs=0.1)
316		).all()
317
318		def test_output_by_type_view(self):
319		results = processing.results(self.om)
320		converter_output = views.node_output_by_type(
321		results, node_type=Converter
322		)
323		compare = views.node(results, "diesel", multiindex=True)["sequences"][
324		("diesel", "b_el1", "flow")
325		]
326		assert converter_output.sum().iloc[0] == pytest.approx(compare.sum())
327
328		def test_input_by_type_view(self):
329		results = processing.results(self.om)
330		sink_input = views.node_input_by_type(results, node_type=Sink)
331		compare = views.node(results, "demand_el", multiindex=True)
332		assert sink_input.sum().iloc[0] == pytest.approx(
333		compare["sequences"][("b_el2", "demand_el", "flow")].sum()
334		)
335
336		def test_net_storage_flow(self):
337		results = processing.results(self.om)
338		storage_flow = views.net_storage_flow(
339		results, node_type=GenericStorage
340		)
341
342		compare = views.node(results, "storage", multiindex=True)["sequences"]
343
344		assert (
345		(
346		(
347		compare[("storage", "b_el2", "flow")]
348		- compare[("b_el1", "storage", "flow")]
349		)
350		.to_frame()
351		.fillna(0)
352		== storage_flow.values
353		)
354		.all()
355		.iloc[0]
356		)
357
358		def test_output_by_type_view_empty(self):
359		results = processing.results(self.om)
360		view = views.node_output_by_type(results, node_type=Flow)
361		assert view is None
362
363		def test_input_by_type_view_empty(self):
364		results = processing.results(self.om)
365		view = views.node_input_by_type(results, node_type=Flow)
366		assert view is None
367
368		def test_net_storage_flow_empty(self):
369		results = processing.results(self.om)
370		view = views.net_storage_flow(results, node_type=Sink)
371		assert view is None
372		view2 = views.net_storage_flow(results, node_type=Flow)
373		assert view2 is None
374
375		def test_node_weight_by_type_empty(self):
376		results = processing.results(self.om)
377		view = views.node_weight_by_type(results, node_type=Flow)
378		assert view is None
379

oemof / oemof-solph

TestParameterResult.test_net_storage_flow_empty() A last analyzed 2025-11-04 19:41 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

TestParameterResult.test_net_storage_flow_empty() A
last analyzed 2025-11-04 19:41 UTC