solph.components.experimental._link.LinkBlock._create() - Code Metrics - Inspection of "Features/network 0.5" - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#836)

by Uwe

created 2022-06-24 10:01 UTC

LinkBlock._create() C

↳ Parent: solph.components.experimental._link

Complexity

Conditions

Size

Total Lines	97
Code Lines	58

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	58
dl	0
loc	97
rs	6.9745
c	0
b	0
f	0
cc	7
nop	2

How to fix Long Method

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

"""
In-development component to add some intelligence
to connection between two Nodes.

SPDX-FileCopyrightText: Uwe Krien <[email protected]>
SPDX-FileCopyrightText: Simon Hilpert
SPDX-FileCopyrightText: Cord Kaldemeyer
SPDX-FileCopyrightText: Patrik Schönfeldt
SPDX-FileCopyrightText: Johannes Röder
SPDX-FileCopyrightText: jakob-wo
SPDX-FileCopyrightText: gplssm
SPDX-FileCopyrightText: jnnr

SPDX-License-Identifier: MIT

"""
from warnings import warn

from oemof.network import network as on
from oemof.tools import debugging
from pyomo.core import Binary
from pyomo.core import Set
from pyomo.core import Var
from pyomo.core.base.block import ScalarBlock
from pyomo.environ import BuildAction
from pyomo.environ import Constraint

from oemof.solph._plumbing import sequence


class Link(on.Transformer):
    """A Link object with 1...2 inputs and 1...2 outputs.

    Parameters
    ----------
    conversion_factors : dict
        Dictionary containing conversion factors for conversion of each flow.
        Keys are the connected tuples (input, output) bus objects.
        The dictionary values can either be a scalar or an iterable with length
        of time horizon for simulation.
    limit_direction : boolean, default: True
        Wether direction constraint should be set for Link component

    Note: This component is experimental. Use it with care.

    Notes
    -----
    The sets, variables, constraints and objective parts are created
     * :py:class:`~oemof.solph.components.experimental._link.LinkBlock`

    Examples
    --------

    >>> from oemof import solph
    >>> bel0 = solph.buses.Bus(label="el0")
    >>> bel1 = solph.buses.Bus(label="el1")

    >>> link = solph.components.experimental.Link(
    ...    label="transshipment_link",
    ...    inputs={bel0: solph.flows.Flow(nominal_value=4),
    ...            bel1: solph.flows.Flow(nominal_value=2)},
    ...    outputs={bel0: solph.flows.Flow(),
    ...             bel1: solph.flows.Flow()},
    ...    conversion_factors={(bel0, bel1): 0.8, (bel1, bel0): 0.9})
    >>> print(sorted([x[1][5] for x in link.conversion_factors.items()]))
    [0.8, 0.9]

    >>> type(link)
    <class 'oemof.solph.components.experimental._link.Link'>

    >>> sorted([str(i) for i in link.inputs])
    ['el0', 'el1']

    >>> link.conversion_factors[(bel0, bel1)][3]
    0.8
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.conversion_factors = {
            k: sequence(v)
            for k, v in kwargs.get("conversion_factors", {}).items()
        }
        self.limit_direction = kwargs.get("limit_direction", True)

        msg = (
            "Component `Link` should have exactly "
            + "2 inputs, 2 outputs, and 2 "
            + "conversion factors connecting these. You are initializing "
            + "a `Link`without obeying this specification. "
            + "If this is intended and you know what you are doing you can "
            + "disable the SuspiciousUsageWarning globally."
        )

        if (
            len(self.inputs) != 2
            or len(self.outputs) != 2
            or len(self.conversion_factors) != 2
        ):
            warn(msg, debugging.SuspiciousUsageWarning)

    def constraint_group(self):
        return LinkBlock


class LinkBlock(ScalarBlock):
    r"""Block for the relation of nodes with type
    :class:`~oemof.solph.components.experimental.Link`

    Note: This component is experimental. Use it with care.

    **The following constraints are created:**
    (Equation 2&3 are only implemented, if `limit_direction` is enabled)

    .. _Link-equations:

    .. math::
        &
        (1) \qquad P_{\mathrm{in},n}(t) = c_n(t) \times P_{\mathrm{out},n}(t)
            \quad \forall t \in T, \forall n in {1,2} \\
        &
        (2) \qquad 1 \ge \hat{S} + P_{\mathrm{in},1}(t)
                                 / P_{\mathrm{in},1,\mathrm{max}}
            \quad \forall t \in T \\
        &
        (3) \qquad 0 \le \hat{S} - P_{\mathrm{in},2}(t)
                                 / P_{2\mathrm{in},2,\mathrm{max}}
            \quad \forall t \in T \\
        &

    """
    CONSTRAINT_GROUP = True

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def _create(self, group=None):
        """Creates the relation for the class:`Link`.

        Parameters
        ----------
        group : list
            List of oemof.solph.components.experimental.Link objects for which
            the relation of inputs and outputs is createdBuildAction
            e.g. group = [link1, link2, link3, ...]. The components inside
            the list need to hold an attribute `conversion_factors` of type
            dict containing the conversion factors for all inputs to outputs.
        """
        if group is None:
            return None

        m = self.parent_block()

        all_conversions = {}
        for n in group:
            all_conversions[n] = {
                k: v for k, v in n.conversion_factors.items()
            }

        self.LINKS = Set(initialize=[g for g in group])

        directed_conversions = {
            n: n.conversion_factors for n in group if n.limit_direction
        }
        self.LINK_1ST_INFLOWS = Set(
            initialize=[
                (list(c)[0][0], n) for n, c in directed_conversions.items()
            ]
        )
        self.LINK_2ND_INFLOWS = Set(
            initialize=[
                (list(c)[1][0], n) for n, c in directed_conversions.items()
            ]
        )

        #  0: Flows 1 connected; 1: Flows 2 connected
        self.direction = Var(self.LINKS, m.TIMESTEPS, within=Binary)

        def _input_output_relation(block):
            for t in m.TIMESTEPS:

                for n, conversion in all_conversions.items():

                    for cidx, c in conversion.items():
                        try:
                            expr = (
                                m.flow[n, cidx[1], t]
                                == c[t] * m.flow[cidx[0], n, t]
                            )
                        except ValueError:
                            raise ValueError(
                                "Error in constraint creation",
                                "from: {0}, to: {1}, via: {2}".format(
                                    cidx[0], cidx[1], n
                                ),
                            )
                        block.relation.add((n, cidx[0], cidx[1], t), (expr))

        self.relation = Constraint(
            [
                (n, cidx[0], cidx[1], t)
                for t in m.TIMESTEPS
                for n, conversion in all_conversions.items()
                for cidx, c in conversion.items()
            ],
            noruleinit=True,
        )
        self.relation_build = BuildAction(rule=_input_output_relation)

        def _flow1_rule(block, i, link, t):
            """Rule definition for Eq. (2)."""
            expr = 1 >= (
                self.direction[link, t]
                + m.flow[i, link, t]

                * m.flows[i, link].max[t]
                * m.flows[i, link].nominal_value
            )
            return expr

        self.flow1 = Constraint(
            self.LINK_1ST_INFLOWS, m.TIMESTEPS, rule=_flow1_rule
        )

        def _flow2_rule(block, i, link, t):
            """Rule definition for Eq. (3)."""
            expr = 0 <= (
                self.direction[link, t]
                - m.flow[i, link, t]

                * m.flows[i, link].max[t]
                * m.flows[i, link].nominal_value
            )
            return expr

        self.flow2 = Constraint(
            self.LINK_2ND_INFLOWS, m.TIMESTEPS, rule=_flow2_rule
        )


1			# -- coding: utf-8 --
2
3			"""
4			In-development component to add some intelligence
5			to connection between two Nodes.
6
7			SPDX-FileCopyrightText: Uwe Krien <[email protected]>
8			SPDX-FileCopyrightText: Simon Hilpert
9			SPDX-FileCopyrightText: Cord Kaldemeyer
10			SPDX-FileCopyrightText: Patrik Schönfeldt
11			SPDX-FileCopyrightText: Johannes Röder
12			SPDX-FileCopyrightText: jakob-wo
13			SPDX-FileCopyrightText: gplssm
14			SPDX-FileCopyrightText: jnnr
15
16			SPDX-License-Identifier: MIT
17
18			"""
19			from warnings import warn
20
21			from oemof.network import network as on
22			from oemof.tools import debugging
23			from pyomo.core import Binary
24			from pyomo.core import Set
25			from pyomo.core import Var
26			from pyomo.core.base.block import ScalarBlock
27			from pyomo.environ import BuildAction
28			from pyomo.environ import Constraint
29
30			from oemof.solph._plumbing import sequence
31
32
33			class Link(on.Transformer):
34			"""A Link object with 1...2 inputs and 1...2 outputs.
35
36			Parameters
37			----------
38			conversion_factors : dict
39			Dictionary containing conversion factors for conversion of each flow.
40			Keys are the connected tuples (input, output) bus objects.
41			The dictionary values can either be a scalar or an iterable with length
42			of time horizon for simulation.
43			limit_direction : boolean, default: True
44			Wether direction constraint should be set for Link component
45
46			Note: This component is experimental. Use it with care.
47
48			Notes
49			-----
50			The sets, variables, constraints and objective parts are created
51			* :py:class:`~oemof.solph.components.experimental._link.LinkBlock`
52
53			Examples
54			--------
55
56			>>> from oemof import solph
57			>>> bel0 = solph.buses.Bus(label="el0")
58			>>> bel1 = solph.buses.Bus(label="el1")
59
60			>>> link = solph.components.experimental.Link(
61			... label="transshipment_link",
62			... inputs={bel0: solph.flows.Flow(nominal_value=4),
63			... bel1: solph.flows.Flow(nominal_value=2)},
64			... outputs={bel0: solph.flows.Flow(),
65			... bel1: solph.flows.Flow()},
66			... conversion_factors={(bel0, bel1): 0.8, (bel1, bel0): 0.9})
67			>>> print(sorted([x[1][5] for x in link.conversion_factors.items()]))
68			[0.8, 0.9]
69
70			>>> type(link)
71			<class 'oemof.solph.components.experimental._link.Link'>
72
73			>>> sorted([str(i) for i in link.inputs])
74			['el0', 'el1']
75
76			>>> link.conversion_factors[(bel0, bel1)][3]
77			0.8
78			"""
79
80			def __init__(self, args, *kwargs):
81			super().__init__(args, *kwargs)
82
83			self.conversion_factors = {
84			k: sequence(v)
85			for k, v in kwargs.get("conversion_factors", {}).items()
86			}
87			self.limit_direction = kwargs.get("limit_direction", True)
88
89			msg = (
90			"Component `Link` should have exactly "
91			+ "2 inputs, 2 outputs, and 2 "
92			+ "conversion factors connecting these. You are initializing "
93			+ "a `Link`without obeying this specification. "
94			+ "If this is intended and you know what you are doing you can "
95			+ "disable the SuspiciousUsageWarning globally."
96			)
97
98			if (
99			len(self.inputs) != 2
100			or len(self.outputs) != 2
101			or len(self.conversion_factors) != 2
102			):
103			warn(msg, debugging.SuspiciousUsageWarning)
104
105			def constraint_group(self):
106			return LinkBlock
107
108
109			class LinkBlock(ScalarBlock):
110			r"""Block for the relation of nodes with type
111			:class:`~oemof.solph.components.experimental.Link`
112
113			Note: This component is experimental. Use it with care.
114
115			The following constraints are created:
116			(Equation 2&3 are only implemented, if `limit_direction` is enabled)
117
118			.. _Link-equations:
119
120			.. math::
121			&
122			(1) \qquad P_{\mathrm{in},n}(t) = c_n(t) \times P_{\mathrm{out},n}(t)
123			\quad \forall t \in T, \forall n in {1,2} \\
124			&
125			(2) \qquad 1 \ge \hat{S} + P_{\mathrm{in},1}(t)
126			/ P_{\mathrm{in},1,\mathrm{max}}
127			\quad \forall t \in T \\
128			&
129			(3) \qquad 0 \le \hat{S} - P_{\mathrm{in},2}(t)
130			/ P_{2\mathrm{in},2,\mathrm{max}}
131			\quad \forall t \in T \\
132			&
133
134			"""
135			CONSTRAINT_GROUP = True
136
137			def __init__(self, args, *kwargs):
138			super().__init__(args, *kwargs)
139
140			def _create(self, group=None):
141			"""Creates the relation for the class:`Link`.
142
143			Parameters
144			----------
145			group : list
146			List of oemof.solph.components.experimental.Link objects for which
147			the relation of inputs and outputs is createdBuildAction
148			e.g. group = [link1, link2, link3, ...]. The components inside
149			the list need to hold an attribute `conversion_factors` of type
150			dict containing the conversion factors for all inputs to outputs.
151			"""
152			if group is None:
153			return None
154
155			m = self.parent_block()
156
157			all_conversions = {}
158			for n in group:
159			all_conversions[n] = {
160			k: v for k, v in n.conversion_factors.items()
161			}
162
163			self.LINKS = Set(initialize=[g for g in group])
164
165			directed_conversions = {
166			n: n.conversion_factors for n in group if n.limit_direction
167			}
168			self.LINK_1ST_INFLOWS = Set(
169			initialize=[
170			(list(c)[0][0], n) for n, c in directed_conversions.items()
171			]
172			)
173			self.LINK_2ND_INFLOWS = Set(
174			initialize=[
175			(list(c)[1][0], n) for n, c in directed_conversions.items()
176			]
177			)
178
179			# 0: Flows 1 connected; 1: Flows 2 connected
180			self.direction = Var(self.LINKS, m.TIMESTEPS, within=Binary)
181
182			def _input_output_relation(block):
183			for t in m.TIMESTEPS:
			0 ignored issues – show introduced 2020-03-31 12:19 UTC by Report Bug Copy Issue Report The variable `m` does not seem to be defined for all execution paths. Loading history...
184			for n, conversion in all_conversions.items():
			0 ignored issues – show introduced 2020-03-31 12:19 UTC by Report Bug Copy Issue Report The variable `all_conversions` does not seem to be defined for all execution paths. Loading history...
185			for cidx, c in conversion.items():
186			try:
187			expr = (
188			m.flow[n, cidx[1], t]
189			== c[t] * m.flow[cidx[0], n, t]
190			)
191			except ValueError:
192			raise ValueError(
193			"Error in constraint creation",
194			"from: {0}, to: {1}, via: {2}".format(
195			cidx[0], cidx[1], n
196			),
197			)
198			block.relation.add((n, cidx[0], cidx[1], t), (expr))
199
200			self.relation = Constraint(
201			[
202			(n, cidx[0], cidx[1], t)
203			for t in m.TIMESTEPS
204			for n, conversion in all_conversions.items()
205			for cidx, c in conversion.items()
206			],
207			noruleinit=True,
208			)
209			self.relation_build = BuildAction(rule=_input_output_relation)
210
211			def _flow1_rule(block, i, link, t):
212			"""Rule definition for Eq. (2)."""
213			expr = 1 >= (
214			self.direction[link, t]
215			+ m.flow[i, link, t]
			0 ignored issues – show introduced 2021-03-24 18:20 UTC by Report Bug Copy Issue Report The variable `m` does not seem to be defined for all execution paths. Loading history...
216			* m.flows[i, link].max[t]
217			* m.flows[i, link].nominal_value
218			)
219			return expr
220
221			self.flow1 = Constraint(
222			self.LINK_1ST_INFLOWS, m.TIMESTEPS, rule=_flow1_rule
223			)
224
225			def _flow2_rule(block, i, link, t):
226			"""Rule definition for Eq. (3)."""
227			expr = 0 <= (
228			self.direction[link, t]
229			- m.flow[i, link, t]
			0 ignored issues – show introduced 2021-03-24 18:20 UTC by Report Bug Copy Issue Report The variable `m` does not seem to be defined for all execution paths. Loading history...
230			* m.flows[i, link].max[t]
231			* m.flows[i, link].nominal_value
232			)
233			return expr
234
235			self.flow2 = Constraint(
236			self.LINK_2ND_INFLOWS, m.TIMESTEPS, rule=_flow2_rule
237			)
238

oemof / oemof-solph

Pull Request — dev (#836)

LinkBlock._create() C

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like