solph.blocks.flow.Flow._create() - Code Metrics - Inspection of "Remove gradient costs due to bugs and implementati..." - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#789)

by Uwe

created 2021-09-23 07:32 UTC

solph.blocks.flow.Flow._create() F

↳ Parent: solph.blocks.flow

Complexity

Conditions

Size

Total Lines	156
Code Lines	93

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	93
dl	0
loc	156
rs	2.329
c	0
b	0
f	0
cc	15
nop	2

How to fix Long Method Complexity

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

"""Creating sets, variables, constraints and parts of the objective function
for Flow objects.

SPDX-FileCopyrightText: Uwe Krien <[email protected]>
SPDX-FileCopyrightText: Simon Hilpert
SPDX-FileCopyrightText: Cord Kaldemeyer
SPDX-FileCopyrightText: Patrik Schönfeldt
SPDX-FileCopyrightText: Birgit Schachler
SPDX-FileCopyrightText: jnnr
SPDX-FileCopyrightText: jmloenneberga

SPDX-License-Identifier: MIT

"""

from pyomo.core import BuildAction
from pyomo.core import Constraint
from pyomo.core import NonNegativeIntegers
from pyomo.core import Set
from pyomo.core import Var
from pyomo.core.base.block import SimpleBlock


class Flow(SimpleBlock):
    r""" Flow block with definitions for standard flows.

    **The following variables are created**:

    negative_gradient :
        Difference of a flow in consecutive timesteps if flow is reduced
        indexed by NEGATIVE_GRADIENT_FLOWS, TIMESTEPS.

    positive_gradient :
        Difference of a flow in consecutive timesteps if flow is increased
        indexed by NEGATIVE_GRADIENT_FLOWS, TIMESTEPS.

    **The following sets are created:** (-> see basic sets at :class:`.Model` )

    SUMMED_MAX_FLOWS
        A set of flows with the attribute :attr:`summed_max` being not None.
    SUMMED_MIN_FLOWS
        A set of flows with the attribute :attr:`summed_min` being not None.
    NEGATIVE_GRADIENT_FLOWS
        A set of flows with the attribute :attr:`negative_gradient` being not
        None.
    POSITIVE_GRADIENT_FLOWS
        A set of flows with the attribute :attr:`positive_gradient` being not
        None
    INTEGER_FLOWS
        A set of flows where the attribute :attr:`integer` is True (forces flow
        to only take integer values)

    **The following constraints are build:**

    Flow max sum :attr:`om.Flow.summed_max[i, o]`
      .. math::
        \sum_t flow(i, o, t) \cdot \tau
            \leq summed\_max(i, o) \cdot nominal\_value(i, o), \\
        \forall (i, o) \in \textrm{SUMMED\_MAX\_FLOWS}.

    Flow min sum :attr:`om.Flow.summed_min[i, o]`
      .. math::
        \sum_t flow(i, o, t) \cdot \tau
            \geq summed\_min(i, o) \cdot nominal\_value(i, o), \\
        \forall (i, o) \in \textrm{SUMMED\_MIN\_FLOWS}.

    Negative gradient constraint
      :attr:`om.Flow.negative_gradient_constr[i, o]`:
        .. math::
          flow(i, o, t-1) - flow(i, o, t) \geq \
          negative\_gradient(i, o, t), \\
          \forall (i, o) \in \textrm{NEGATIVE\_GRADIENT\_FLOWS}, \\
          \forall t \in \textrm{TIMESTEPS}.

    Positive gradient constraint
      :attr:`om.Flow.positive_gradient_constr[i, o]`:
        .. math:: flow(i, o, t) - flow(i, o, t-1) \geq \
          positive\__gradient(i, o, t), \\
          \forall (i, o) \in \textrm{POSITIVE\_GRADIENT\_FLOWS}, \\
          \forall t \in \textrm{TIMESTEPS}.

    **The following parts of the objective function are created:**

    If :attr:`variable_costs` are set by the user:
      .. math::
          \sum_{(i,o)} \sum_t flow(i, o, t) \cdot variable\_costs(i, o, t)

    The expression can be accessed by :attr:`om.Flow.variable_costs` and
    their value after optimization by :meth:`om.Flow.variable_costs()` .

    """

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

    def _create(self, group=None):
        r"""Creates sets, variables and constraints for all standard flows.

        Parameters
        ----------
        group : list
            List containing tuples containing flow (f) objects and the
            associated source (s) and target (t)
            of flow e.g. groups=[(s1, t1, f1), (s2, t2, f2),..]
        """
        if group is None:
            return None

        m = self.parent_block()

        # ########################## SETS #################################
        # set for all flows with an global limit on the flow over time
        self.SUMMED_MAX_FLOWS = Set(
            initialize=[
                (g[0], g[1])
                for g in group
                if g[2].summed_max is not None
                and g[2].nominal_value is not None
            ]
        )

        self.SUMMED_MIN_FLOWS = Set(
            initialize=[
                (g[0], g[1])
                for g in group
                if g[2].summed_min is not None
                and g[2].nominal_value is not None
            ]
        )

        self.NEGATIVE_GRADIENT_FLOWS = Set(
            initialize=[
                (g[0], g[1])
                for g in group
                if g[2].negative_gradient["ub"][0] is not None
            ]
        )

        self.POSITIVE_GRADIENT_FLOWS = Set(
            initialize=[
                (g[0], g[1])
                for g in group
                if g[2].positive_gradient["ub"][0] is not None
            ]
        )

        self.INTEGER_FLOWS = Set(
            initialize=[(g[0], g[1]) for g in group if g[2].integer]
        )
        # ######################### Variables  ################################

        self.positive_gradient = Var(self.POSITIVE_GRADIENT_FLOWS, m.TIMESTEPS)

        self.negative_gradient = Var(self.NEGATIVE_GRADIENT_FLOWS, m.TIMESTEPS)

        self.integer_flow = Var(
            self.INTEGER_FLOWS, m.TIMESTEPS, within=NonNegativeIntegers
        )
        # set upper bound of gradient variable
        for i, o, f in group:
            if m.flows[i, o].positive_gradient["ub"][0] is not None:
                for t in m.TIMESTEPS:
                    self.positive_gradient[i, o, t].setub(
                        f.positive_gradient["ub"][t] * f.nominal_value
                    )
            if m.flows[i, o].negative_gradient["ub"][0] is not None:
                for t in m.TIMESTEPS:
                    self.negative_gradient[i, o, t].setub(
                        f.negative_gradient["ub"][t] * f.nominal_value
                    )

        # ######################### CONSTRAINTS ###############################

        def _flow_summed_max_rule(model):
            """Rule definition for build action of max. sum flow constraint."""
            for inp, out in self.SUMMED_MAX_FLOWS:
                lhs = sum(
                    m.flow[inp, out, ts] * m.timeincrement[ts]

                    for ts in m.TIMESTEPS
                )
                rhs = (
                    m.flows[inp, out].summed_max
                    * m.flows[inp, out].nominal_value
                )
                self.summed_max.add((inp, out), lhs <= rhs)

        self.summed_max = Constraint(self.SUMMED_MAX_FLOWS, noruleinit=True)
        self.summed_max_build = BuildAction(rule=_flow_summed_max_rule)

        def _flow_summed_min_rule(model):
            """Rule definition for build action of min. sum flow constraint."""
            for inp, out in self.SUMMED_MIN_FLOWS:
                lhs = sum(
                    m.flow[inp, out, ts] * m.timeincrement[ts]

                    for ts in m.TIMESTEPS
                )
                rhs = (
                    m.flows[inp, out].summed_min
                    * m.flows[inp, out].nominal_value
                )
                self.summed_min.add((inp, out), lhs >= rhs)

        self.summed_min = Constraint(self.SUMMED_MIN_FLOWS, noruleinit=True)
        self.summed_min_build = BuildAction(rule=_flow_summed_min_rule)

        def _positive_gradient_flow_rule(model):
            """Rule definition for positive gradient constraint."""
            for inp, out in self.POSITIVE_GRADIENT_FLOWS:
                for ts in m.TIMESTEPS:

                    if ts > 0:
                        lhs = m.flow[inp, out, ts] - m.flow[inp, out, ts - 1]
                        rhs = self.positive_gradient[inp, out, ts]
                        self.positive_gradient_constr.add(
                            (inp, out, ts), lhs <= rhs
                        )
                    else:
                        pass  # return(Constraint.Skip)

        self.positive_gradient_constr = Constraint(
            self.POSITIVE_GRADIENT_FLOWS, m.TIMESTEPS, noruleinit=True
        )
        self.positive_gradient_build = BuildAction(
            rule=_positive_gradient_flow_rule
        )

        def _negative_gradient_flow_rule(model):
            """Rule definition for negative gradient constraint."""
            for inp, out in self.NEGATIVE_GRADIENT_FLOWS:
                for ts in m.TIMESTEPS:

                    if ts > 0:
                        lhs = m.flow[inp, out, ts - 1] - m.flow[inp, out, ts]
                        rhs = self.negative_gradient[inp, out, ts]
                        self.negative_gradient_constr.add(
                            (inp, out, ts), lhs <= rhs
                        )
                    else:
                        pass  # return(Constraint.Skip)

        self.negative_gradient_constr = Constraint(
            self.NEGATIVE_GRADIENT_FLOWS, m.TIMESTEPS, noruleinit=True
        )
        self.negative_gradient_build = BuildAction(
            rule=_negative_gradient_flow_rule
        )

        def _integer_flow_rule(block, ii, oi, ti):
            """Force flow variable to NonNegativeInteger values."""
            return self.integer_flow[ii, oi, ti] == m.flow[ii, oi, ti]


        self.integer_flow_constr = Constraint(
            self.INTEGER_FLOWS, m.TIMESTEPS, rule=_integer_flow_rule
        )

    def _objective_expression(self):
        r"""Objective expression for all standard flows with fixed costs
        and variable costs.
        """
        m = self.parent_block()

        variable_costs = 0

        for i, o in m.FLOWS:
            if m.flows[i, o].variable_costs[0] is not None:
                for t in m.TIMESTEPS:
                    variable_costs += (
                        m.flow[i, o, t]
                        * m.objective_weighting[t]
                        * m.flows[i, o].variable_costs[t]
                    )

        return variable_costs


1			# -- coding: utf-8 --
2
3			"""Creating sets, variables, constraints and parts of the objective function
4			for Flow objects.
5
6			SPDX-FileCopyrightText: Uwe Krien <[email protected]>
7			SPDX-FileCopyrightText: Simon Hilpert
8			SPDX-FileCopyrightText: Cord Kaldemeyer
9			SPDX-FileCopyrightText: Patrik Schönfeldt
10			SPDX-FileCopyrightText: Birgit Schachler
11			SPDX-FileCopyrightText: jnnr
12			SPDX-FileCopyrightText: jmloenneberga
13
14			SPDX-License-Identifier: MIT
15
16			"""
17
18			from pyomo.core import BuildAction
19			from pyomo.core import Constraint
20			from pyomo.core import NonNegativeIntegers
21			from pyomo.core import Set
22			from pyomo.core import Var
23			from pyomo.core.base.block import SimpleBlock
24
25
26			class Flow(SimpleBlock):
27			r""" Flow block with definitions for standard flows.
28
29			The following variables are created:
30
31			negative_gradient :
32			Difference of a flow in consecutive timesteps if flow is reduced
33			indexed by NEGATIVE_GRADIENT_FLOWS, TIMESTEPS.
34
35			positive_gradient :
36			Difference of a flow in consecutive timesteps if flow is increased
37			indexed by NEGATIVE_GRADIENT_FLOWS, TIMESTEPS.
38
39			The following sets are created: (-> see basic sets at :class:`.Model` )
40
41			SUMMED_MAX_FLOWS
42			A set of flows with the attribute :attr:`summed_max` being not None.
43			SUMMED_MIN_FLOWS
44			A set of flows with the attribute :attr:`summed_min` being not None.
45			NEGATIVE_GRADIENT_FLOWS
46			A set of flows with the attribute :attr:`negative_gradient` being not
47			None.
48			POSITIVE_GRADIENT_FLOWS
49			A set of flows with the attribute :attr:`positive_gradient` being not
50			None
51			INTEGER_FLOWS
52			A set of flows where the attribute :attr:`integer` is True (forces flow
53			to only take integer values)
54
55			The following constraints are build:
56
57			Flow max sum :attr:`om.Flow.summed_max[i, o]`
58			.. math::
59			\sum_t flow(i, o, t) \cdot \tau
60			\leq summed\_max(i, o) \cdot nominal\_value(i, o), \\
61			\forall (i, o) \in \textrm{SUMMED\_MAX\_FLOWS}.
62
63			Flow min sum :attr:`om.Flow.summed_min[i, o]`
64			.. math::
65			\sum_t flow(i, o, t) \cdot \tau
66			\geq summed\_min(i, o) \cdot nominal\_value(i, o), \\
67			\forall (i, o) \in \textrm{SUMMED\_MIN\_FLOWS}.
68
69			Negative gradient constraint
70			:attr:`om.Flow.negative_gradient_constr[i, o]`:
71			.. math::
72			flow(i, o, t-1) - flow(i, o, t) \geq \
73			negative\_gradient(i, o, t), \\
74			\forall (i, o) \in \textrm{NEGATIVE\_GRADIENT\_FLOWS}, \\
75			\forall t \in \textrm{TIMESTEPS}.
76
77			Positive gradient constraint
78			:attr:`om.Flow.positive_gradient_constr[i, o]`:
79			.. math:: flow(i, o, t) - flow(i, o, t-1) \geq \
80			positive\__gradient(i, o, t), \\
81			\forall (i, o) \in \textrm{POSITIVE\_GRADIENT\_FLOWS}, \\
82			\forall t \in \textrm{TIMESTEPS}.
83
84			The following parts of the objective function are created:
85
86			If :attr:`variable_costs` are set by the user:
87			.. math::
88			\sum_{(i,o)} \sum_t flow(i, o, t) \cdot variable\_costs(i, o, t)
89
90			The expression can be accessed by :attr:`om.Flow.variable_costs` and
91			their value after optimization by :meth:`om.Flow.variable_costs()` .
92
93			"""
94
95			def __init__(self, args, *kwargs):
96			super().__init__(args, *kwargs)
97
98			def _create(self, group=None):
99			r"""Creates sets, variables and constraints for all standard flows.
100
101			Parameters
102			----------
103			group : list
104			List containing tuples containing flow (f) objects and the
105			associated source (s) and target (t)
106			of flow e.g. groups=[(s1, t1, f1), (s2, t2, f2),..]
107			"""
108			if group is None:
109			return None
110
111			m = self.parent_block()
112
113			# ########################## SETS #################################
114			# set for all flows with an global limit on the flow over time
115			self.SUMMED_MAX_FLOWS = Set(
116			initialize=[
117			(g[0], g[1])
118			for g in group
119			if g[2].summed_max is not None
120			and g[2].nominal_value is not None
121			]
122			)
123
124			self.SUMMED_MIN_FLOWS = Set(
125			initialize=[
126			(g[0], g[1])
127			for g in group
128			if g[2].summed_min is not None
129			and g[2].nominal_value is not None
130			]
131			)
132
133			self.NEGATIVE_GRADIENT_FLOWS = Set(
134			initialize=[
135			(g[0], g[1])
136			for g in group
137			if g[2].negative_gradient["ub"][0] is not None
138			]
139			)
140
141			self.POSITIVE_GRADIENT_FLOWS = Set(
142			initialize=[
143			(g[0], g[1])
144			for g in group
145			if g[2].positive_gradient["ub"][0] is not None
146			]
147			)
148
149			self.INTEGER_FLOWS = Set(
150			initialize=[(g[0], g[1]) for g in group if g[2].integer]
151			)
152			# ######################### Variables ################################
153
154			self.positive_gradient = Var(self.POSITIVE_GRADIENT_FLOWS, m.TIMESTEPS)
155
156			self.negative_gradient = Var(self.NEGATIVE_GRADIENT_FLOWS, m.TIMESTEPS)
157
158			self.integer_flow = Var(
159			self.INTEGER_FLOWS, m.TIMESTEPS, within=NonNegativeIntegers
160			)
161			# set upper bound of gradient variable
162			for i, o, f in group:
163			if m.flows[i, o].positive_gradient["ub"][0] is not None:
164			for t in m.TIMESTEPS:
165			self.positive_gradient[i, o, t].setub(
166			f.positive_gradient["ub"][t] * f.nominal_value
167			)
168			if m.flows[i, o].negative_gradient["ub"][0] is not None:
169			for t in m.TIMESTEPS:
170			self.negative_gradient[i, o, t].setub(
171			f.negative_gradient["ub"][t] * f.nominal_value
172			)
173
174			# ######################### CONSTRAINTS ###############################
175
176			def _flow_summed_max_rule(model):
177			"""Rule definition for build action of max. sum flow constraint."""
178			for inp, out in self.SUMMED_MAX_FLOWS:
179			lhs = sum(
180			m.flow[inp, out, ts] * m.timeincrement[ts]
			0 ignored issues – show introduced 2021-02-16 18:34 UTC by Report Bug Copy Issue Report The variable `m` does not seem to be defined for all execution paths. Loading history...
181			for ts in m.TIMESTEPS
182			)
183			rhs = (
184			m.flows[inp, out].summed_max
185			* m.flows[inp, out].nominal_value
186			)
187			self.summed_max.add((inp, out), lhs <= rhs)
188
189			self.summed_max = Constraint(self.SUMMED_MAX_FLOWS, noruleinit=True)
190			self.summed_max_build = BuildAction(rule=_flow_summed_max_rule)
191
192			def _flow_summed_min_rule(model):
193			"""Rule definition for build action of min. sum flow constraint."""
194			for inp, out in self.SUMMED_MIN_FLOWS:
195			lhs = sum(
196			m.flow[inp, out, ts] * m.timeincrement[ts]
			0 ignored issues – show introduced 2021-02-16 18:34 UTC by Report Bug Copy Issue Report The variable `m` does not seem to be defined for all execution paths. Loading history...
197			for ts in m.TIMESTEPS
198			)
199			rhs = (
200			m.flows[inp, out].summed_min
201			* m.flows[inp, out].nominal_value
202			)
203			self.summed_min.add((inp, out), lhs >= rhs)
204
205			self.summed_min = Constraint(self.SUMMED_MIN_FLOWS, noruleinit=True)
206			self.summed_min_build = BuildAction(rule=_flow_summed_min_rule)
207
208			def _positive_gradient_flow_rule(model):
209			"""Rule definition for positive gradient constraint."""
210			for inp, out in self.POSITIVE_GRADIENT_FLOWS:
211			for ts in m.TIMESTEPS:
			0 ignored issues – show introduced 2020-05-27 09:25 UTC by Report Bug Copy Issue Report The variable `m` does not seem to be defined for all execution paths. Loading history...
212			if ts > 0:
213			lhs = m.flow[inp, out, ts] - m.flow[inp, out, ts - 1]
214			rhs = self.positive_gradient[inp, out, ts]
215			self.positive_gradient_constr.add(
216			(inp, out, ts), lhs <= rhs
217			)
218			else:
219			pass # return(Constraint.Skip)
220
221			self.positive_gradient_constr = Constraint(
222			self.POSITIVE_GRADIENT_FLOWS, m.TIMESTEPS, noruleinit=True
223			)
224			self.positive_gradient_build = BuildAction(
225			rule=_positive_gradient_flow_rule
226			)
227
228			def _negative_gradient_flow_rule(model):
229			"""Rule definition for negative gradient constraint."""
230			for inp, out in self.NEGATIVE_GRADIENT_FLOWS:
231			for ts 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...
232			if ts > 0:
233			lhs = m.flow[inp, out, ts - 1] - m.flow[inp, out, ts]
234			rhs = self.negative_gradient[inp, out, ts]
235			self.negative_gradient_constr.add(
236			(inp, out, ts), lhs <= rhs
237			)
238			else:
239			pass # return(Constraint.Skip)
240
241			self.negative_gradient_constr = Constraint(
242			self.NEGATIVE_GRADIENT_FLOWS, m.TIMESTEPS, noruleinit=True
243			)
244			self.negative_gradient_build = BuildAction(
245			rule=_negative_gradient_flow_rule
246			)
247
248			def _integer_flow_rule(block, ii, oi, ti):
249			"""Force flow variable to NonNegativeInteger values."""
250			return self.integer_flow[ii, oi, ti] == m.flow[ii, oi, ti]
			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...
251
252			self.integer_flow_constr = Constraint(
253			self.INTEGER_FLOWS, m.TIMESTEPS, rule=_integer_flow_rule
254			)
255
256			def _objective_expression(self):
257			r"""Objective expression for all standard flows with fixed costs
258			and variable costs.
259			"""
260			m = self.parent_block()
261
262			variable_costs = 0
263
264			for i, o in m.FLOWS:
265			if m.flows[i, o].variable_costs[0] is not None:
266			for t in m.TIMESTEPS:
267			variable_costs += (
268			m.flow[i, o, t]
269			* m.objective_weighting[t]
270			* m.flows[i, o].variable_costs[t]
271			)
272
273			return variable_costs
274

oemof / oemof-solph

Pull Request — dev (#789)

solph.blocks.flow.Flow._create() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like