tests.test_processing

Complexity

Total Complexity

20

Size/Duplication

Total Lines	370
Duplicated Lines	19.46 %

Importance

Changes

0

19 Methods

Rating	Name	Duplication	Size	Complexity
A	TestParameterResult.test_flows_without_none_exclusion()	0	38	1
A	TestParameterResult.test_multiindex_sequences()	0	5	1
A	TestParameterResult.test_duals()	0	4	1
A	TestParameterResult.test_node_weight_by_type_empty()	0	4	1
A	TestParameterResult.test_output_by_type_view()	0	9	1
A	TestParameterResult.test_net_storage_flow_empty()	0	6	1
A	TestParameterResult.test_output_by_type_view_empty()	0	4	1
B	TestParameterResult.setup_class()	0	59	1
A	TestParameterResult.test_net_storage_flow()	0	20	1
A	TestParameterResult.test_nodes_with_none_exclusion()	35	35	1
A	TestParameterResult.test_node_weight_by_type()	0	8	1
A	TestParameterResult.test_nodes_with_none_exclusion_old_name()	37	37	1
A	TestParameterResult.test_flows_with_none_exclusion()	0	24	1
A	TestParameterResult.test_error_from_nan_values()	0	6	2
A	TestParameterResult.test_input_by_type_view_empty()	0	4	1
A	TestParameterResult.test_nodes_without_none_exclusion()	0	17	1
A	TestParameterResult.test_parameter_with_node_view()	0	12	1
A	TestParameterResult.test_nodes_with_excluded_attrs()	0	15	1
A	TestParameterResult.test_input_by_type_view()	0	6	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)},
            outputs={b_el2: Flow(variable_costs=2.5)},
            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):

This code seems to be duplicated in your project.

        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,
                    "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):

This code seems to be duplicated in your project.

        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,
                    "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(
                {
                    "label": "diesel",
                    "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(
                {
                    "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