solph.flows._investment_flow_block.InvestmentFlowBlock._create_constraints() - Code Metrics - Inspection of "Steamline investment periods" - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#1183)

by Patrik

created 2025-08-29 13:17 UTC

InvestmentFlowBlock._create_constraints() F

↳ Parent: solph.flows._investment_flow_block

Complexity

Conditions

Size

Total Lines	327
Code Lines	94

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	94
dl	0
loc	327
rs	3.1363
c	0
b	0
f	0
cc	13
nop	1

How to fix Long Method Complexity

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

If many parameters/temporary variables are present:
- Replace temporary variables with Query
- Introduce parameter object; often combined with preserve whole object
- If the above two are insufficient: Replace method with method object
If you have long conditionals: Decompose Conditional
Otherwise: Extract method

Complexity

Complex classes like solph.flows._investment_flow_block.InvestmentFlowBlock._create_constraints() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

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

"""Creating sets, variables, constraints and parts of the objective function
for Flow objects with investment but without nonconvex option.

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-FileCopyrightText: Johannes Kochems

SPDX-License-Identifier: MIT

"""
from pyomo.core import Binary
from pyomo.core import BuildAction
from pyomo.core import Constraint
from pyomo.core import Expression
from pyomo.core import NonNegativeReals
from pyomo.core import Set
from pyomo.core import Var
from pyomo.core.base.block import ScalarBlock


class InvestmentFlowBlock(ScalarBlock):
    r"""Block for all flows with :attr:`Investment` being not None.

    .. automethod:: _create_constraints
    .. automethod:: _create_variables
    .. automethod:: _create_sets

    .. automethod:: _objective_expression

    See :class:`oemof.solph.options.Investment` for all parameters of the
    *Investment* class.

    See :class:`oemof.solph.flows._simple_flow_block.SimpleFlowBlock`
    for all parameters of the *SimpleFlowBlock* class.

    The overall summed cost expressions for all *InvestmentFlowBlock* objects
    can be accessed by

    * :attr:`om.InvestmentFlowBlock.investment_costs`,
    * :attr:`om.InvestmentFlowBlock.costs`.

    Note
    ----
    In case of a nonconvex investment flow (:attr:`nonconvex=True`),
    the existing flow capacity :math:`P_{exist}` needs to be zero.

    Note
    ----
    See also :class:`~oemof.solph.flows._flow.Flow`,
    :class:`~oemof.solph.flows._simple_flow_block.SimpleFlowBlock` and
    :class:`~oemof.solph._options.Investment`

    """  # noqa: E501

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

    def _create(self, group=None):
        r"""Creates sets, variables and constraints for SimpleFlowBlock
        with investment attribute of type class:`.Investment`.

        Parameters
        ----------
        group : list
            List containing tuples containing flow (f) objects that have an
            attribute investment 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

        self._create_sets(group)
        self._create_variables(group)
        self._create_constraints()

    def _create_sets(self, group):
        """
        Creates all sets for investment flows.
        """
        self.INVESTFLOWS = Set(initialize=[(g[0], g[1]) for g in group])

        self.CONVEX_INVESTFLOWS = Set(
            initialize=[
                (g[0], g[1])
                for g in group
                if g[2].investment.nonconvex is False
            ]
        )

        self.NON_CONVEX_INVESTFLOWS = Set(
            initialize=[
                (g[0], g[1])
                for g in group
                if g[2].investment.nonconvex is True
            ]
        )

        self.FIXED_INVESTFLOWS = Set(
            initialize=[(g[0], g[1]) for g in group if g[2].fix[0] is not None]
        )

        self.NON_FIXED_INVESTFLOWS = Set(
            initialize=[(g[0], g[1]) for g in group if g[2].fix[0] is None]
        )

        self.FULL_LOAD_TIME_MAX_INVESTFLOWS = Set(
            initialize=[
                (g[0], g[1])
                for g in group
                if g[2].full_load_time_max is not None
            ]
        )

        self.FULL_LOAD_TIME_MIN_INVESTFLOWS = Set(
            initialize=[
                (g[0], g[1])
                for g in group
                if g[2].full_load_time_min is not None
            ]
        )

        self.MIN_INVESTFLOWS = Set(
            initialize=[(g[0], g[1]) for g in group if g[2].min.min() != 0]
        )

        self.EXISTING_INVESTFLOWS = Set(
            initialize=[
                (g[0], g[1])
                for g in group
                if g[2].investment.existing is not None
            ]
        )

        self.OVERALL_MAXIMUM_INVESTFLOWS = Set(
            initialize=[
                (g[0], g[1])
                for g in group
                if g[2].investment.overall_maximum is not None
            ]
        )

        self.OVERALL_MINIMUM_INVESTFLOWS = Set(
            initialize=[
                (g[0], g[1])
                for g in group
                if g[2].investment.overall_minimum is not None
            ]
        )

    def _create_variables(self, _):
        r"""Creates all variables for investment flows.

        All *InvestmentFlowBlock* objects are indexed by a starting and
        ending node :math:`(i, o)`, which is omitted in the following
        for the sake of convenience. The following variables are created:

        * :math:`P(p, t)`

            Actual flow value
            (created in :class:`oemof.solph.models.Model`),
            indexed by tuple of periods p and timestep t

        * :math:`P_{invest}(p)`

            Value of the investment variable in period p,
            equal to what is being invested and equivalent resp. similar to
            the nominal capacity of the flows after optimization.

        * :math:`P_{total}(p)`

            Total installed capacity / energy in period p,
            equivalent to the nominal capacity of the flows after optimization.

        * :math:`P_{old}(p)`

            Old capacity / energy to be decommissioned in period p
            due to reaching its lifetime; applicable only
            for multi-period models.

        * :math:`P_{old,exo}(p)`

            Old exogenous capacity / energy to be decommissioned in period p
            due to reaching its lifetime, i.e. the amount that has
            been specified by :attr:`existing` when it is decommisioned;
            applicable only for multi-period models.

        * :math:`P_{old,end}(p)`

            Old endogenous capacity / energy to be decommissioned in period p
            due to reaching its lifetime, i.e. the amount that has been
            invested in by the model itself that is decommissioned in
            a later period because of reaching its lifetime;
            applicable only for multi-period models.

        * :math:`Y_{invest}(p)`

            Binary variable for the status of the investment, if
            :attr:`nonconvex` is `True`.
        """
        m = self.parent_block()

        def _investvar_bound_rule(block, i, o, p):
            """Rule definition for bounds of invest variable."""
            if (i, o) in self.CONVEX_INVESTFLOWS:
                return (
                    m.flows[i, o].investment.minimum[p],
                    m.flows[i, o].investment.maximum[p],
                )
            elif (i, o) in self.NON_CONVEX_INVESTFLOWS:
                return 0, m.flows[i, o].investment.maximum[p]

        # create invest variable for an investment flow
        self.invest = Var(
            self.INVESTFLOWS,
            m.CAPACITY_PERIODS,
            within=NonNegativeReals,
            bounds=_investvar_bound_rule,
        )

        # Total capacity
        self.total = Var(
            self.INVESTFLOWS, m.CAPACITY_PERIODS, within=NonNegativeReals
        )

        if m.es.investment_times is not None:
            self.old = Var(
                self.INVESTFLOWS, m.CAPACITY_PERIODS, within=NonNegativeReals
            )

            # Old endogenous capacity to be decommissioned (due to lifetime)
            self.old_end = Var(
                self.INVESTFLOWS, m.CAPACITY_PERIODS, within=NonNegativeReals
            )

            # Old exogenous capacity to be decommissioned (due to lifetime)
            self.old_exo = Var(
                self.INVESTFLOWS, m.CAPACITY_PERIODS, within=NonNegativeReals
            )

        # create status variable for a non-convex investment flow
        self.invest_status = Var(
            self.NON_CONVEX_INVESTFLOWS, m.CAPACITY_PERIODS, within=Binary
        )

    def _create_constraints(self):
        r"""Creates all constraints for standard flows.

        Depending on the attributes of the *InvestmentFlowBlock*
        and *SimpleFlowBlock*, different constraints are created.
        The following constraints are created
        for all *InvestmentFlowBlock* objects:\

            Total capacity / energy

            .. math::
                &
                if \quad p=0:\\
                &
                P_{total}(p) = P_{invest}(p) + P_{exist}(p) \\
                &\\
                &
                else:\\
                &
                P_{total}(p) = P_{total}(p-1) + P_{invest}(p) - P_{old}(p) \\
                &\\
                &
                \forall p \in \textrm{CAPACITY_PERIODS}

            Upper bound for the flow value

            .. math::
                &
                P(p, t) \le ( P_{total}(p) ) \cdot f_{max}(t) \\
                &
                \forall p, t \in \textrm{TIMEINDEX}

        For a multi-period model, the old capacity is defined as follows:

            .. math::
                &
                P_{old}(p) = P_{old,exo}(p) + P_{old,end}(p)\\
                &\\
                &
                if \quad p=0:\\
                &
                P_{old,end}(p) = 0\\
                &\\
                &
                else \quad if \quad l \leq year(p):\\
                &
                P_{old,end}(p) = P_{invest}(p_{comm})\\
                &\\
                &
                else:\\
                &
                P_{old,end}(p) = 0\\
                &\\
                &
                if \quad p=0:\\
                &
                P_{old,exo}(p) = 0\\
                &\\
                &
                else \quad if \quad l - a \leq year(p):\\
                &
                P_{old,exo}(p) = P_{exist} (*)\\
                &\\
                &
                else:\\
                &
                P_{old,exo}(p) = 0\\
                &\\
                &
                \forall p \in \textrm{CAPACITY_PERIODS}

            where:

            * (*) is only performed for the first period the condition
              is True. A decommissioning flag is then set to True
              to prevent having falsely added old capacity in future periods.
            * :math:`year(p)` is the year corresponding to period p
            * :math:`p_{comm}` is the commissioning period of the flow
              (which is determined by the model itself)

        Depending on the attribute :attr:`nonconvex`, the constraints for the
        bounds of the decision variable :math:`P_{invest}(p)` are different:\

            * :attr:`nonconvex = False`

            .. math::
                &
                P_{invest, min}(p) \le P_{invest}(p) \le P_{invest, max}(p) \\
                &
                \forall p \in \textrm{CAPACITY_PERIODS}

            * :attr:`nonconvex = True`

            .. math::
                &
                P_{invest, min}(p) \cdot Y_{invest}(p) \le P_{invest}(p)\\
                &
                P_{invest}(p) \le P_{invest, max}(p) \cdot Y_{invest}(p)\\
                &\\
                &
                \forall p \in \textrm{CAPACITY_PERIODS}

        For all *InvestmentFlowBlock* objects
        (independent of the attribute :attr:`nonconvex`),
        the following additional constraints are created, if the appropriate
        attribute of the *SimpleFlowBlock*
        (see :class:`oemof.solph.flows._simple_flow_block.SimpleFlowBlock`)
        is set:

            * :attr:`fix` is not None

                Actual value constraint for investments with fixed flow values

            .. math::
                &
                P(p, t) = P_{total}(p) \cdot f_{fix}(t) \\
                &\\
                &
                \forall p, t \in \textrm{TIMEINDEX}

            * :attr:`min != 0`

                Lower bound for the flow values

            .. math::
                &
                P(p, t) \geq P_{total}(p) \cdot f_{min}(t) \\
                &\\
                &
                \forall p, t \in \textrm{TIMEINDEX}

            * :attr:`full_load_time_max is not None`

                Upper bound for the sum of all flow values
                (e.g. maximum full load hours)

            .. math::
                \sum_{p, t} P(p, t) \cdot \tau(t) \leq P_{total}(p)
                \cdot t_{full\_load, min}

            * :attr:`full_load_time_min is not None`

                Lower bound for the sum of all flow values
                (e.g. minimum full load hours)

            .. math::
                \sum_{p, t} P(t) \cdot \tau(t) \geq P_{total}
                \cdot t_{full\_load, min}

            * :attr:`overall_maximum` is not None
              (for multi-period model only)

                Overall maximum of total installed capacity / energy for flow

            .. math::
                &
                P_{total}(p) \leq P_{overall,max} \\
                &\\
                &
                \forall p \in \textrm{CAPACITY_PERIODS}

            * :attr:`overall_minimum` is not None
              (for multi-period model only)

                Overall minimum of total installed capacity / energy for flow;
                applicable only in last period

            .. math::
                P_{total}(p_{last}) \geq P_{overall,min}
        """
        m = self.parent_block()

        self.minimum_rule = self._minimum_investment_constraint()
        self.maximum_rule = self._maximum_investment_constraint()

        # Handle unit lifetimes
        def _total_capacity_rule(block):
            """Rule definition for determining total installed
            capacity (taking decommissioning into account)
            """
            for i, o in self.INVESTFLOWS:
                for p in m.CAPACITY_PERIODS:
                    if p == 0:
                        expr = (
                            self.total[i, o, p]
                            == self.invest[i, o, p]
                            + m.flows[i, o].investment.existing
                        )
                        self.total_rule.add((i, o, p), expr)
                    # applicable for multi-period model only
                    else:
                        expr = (
                            self.total[i, o, p]
                            == self.invest[i, o, p]
                            + self.total[i, o, p - 1]
                            - self.old[i, o, p]
                        )
                        self.total_rule.add((i, o, p), expr)

        self.total_rule = Constraint(
            self.INVESTFLOWS, m.CAPACITY_PERIODS, noruleinit=True
        )
        self.total_rule_build = BuildAction(rule=_total_capacity_rule)

        def _investflow_fixed_rule(block):
            """Rule definition of constraint to fix flow variable
            of investment flow to (normed) actual value
            """
            for i, o in self.FIXED_INVESTFLOWS:
                for p, t in m.TIMEINDEX:
                    expr = (
                        m.flow[i, o, t]
                        == self.total[i, o, p] * m.flows[i, o].fix[t]
                    )
                    self.fixed.add((i, o, p, t), expr)

        self.fixed = Constraint(
            self.FIXED_INVESTFLOWS, m.TIMEINDEX, noruleinit=True
        )
        self.fixed_build = BuildAction(rule=_investflow_fixed_rule)

        def _max_investflow_rule(block):
            """Rule definition of constraint setting an upper bound of flow
            variable in investment case.
            """
            for i, o in self.NON_FIXED_INVESTFLOWS:
                for p, t in m.TIMEINDEX:
                    expr = (
                        m.flow[i, o, t]
                        <= self.total[i, o, p] * m.flows[i, o].max[t]
                    )
                    self.max.add((i, o, p, t), expr)

        self.max = Constraint(
            self.NON_FIXED_INVESTFLOWS, m.TIMEINDEX, noruleinit=True
        )
        self.max_build = BuildAction(rule=_max_investflow_rule)

        def _min_investflow_rule(block):
            """Rule definition of constraint setting a lower bound on flow
            variable in investment case.
            """
            for i, o in self.MIN_INVESTFLOWS:
                for p, t in m.TIMEINDEX:
                    expr = (
                        m.flow[i, o, t]
                        >= self.total[i, o, p] * m.flows[i, o].min[t]
                    )
                    self.min.add((i, o, p, t), expr)

        self.min = Constraint(
            self.MIN_INVESTFLOWS, m.TIMEINDEX, noruleinit=True
        )
        self.min_build = BuildAction(rule=_min_investflow_rule)

        def _full_load_time_max_investflow_rule(_, i, o):
            """Rule definition for build action of max. sum flow constraint
            in investment case.
            """
            expr = sum(
                m.flow[i, o, t] * m.timeincrement[t] for t in m.TIMESTEPS
            ) <= (
                m.flows[i, o].full_load_time_max
                * sum(self.total[i, o, p] for p in m.CAPACITY_PERIODS)
            )
            return expr

        self.full_load_time_max = Constraint(
            self.FULL_LOAD_TIME_MAX_INVESTFLOWS,
            rule=_full_load_time_max_investflow_rule,
        )

        def _full_load_time_min_investflow_rule(_, i, o):
            """Rule definition for build action of min. sum flow constraint
            in investment case.
            """
            expr = sum(
                m.flow[i, o, t] * m.timeincrement[t] for t in m.TIMESTEPS
            ) >= (
                sum(self.total[i, o, p] for p in m.CAPACITY_PERIODS)
                * m.flows[i, o].full_load_time_min
            )
            return expr

        self.full_load_time_min = Constraint(
            self.FULL_LOAD_TIME_MIN_INVESTFLOWS,
            rule=_full_load_time_min_investflow_rule,
        )

        if m.es.investment_times is not None:

            def _overall_maximum_investflow_rule(block):
                """Rule definition for maximum overall investment
                in investment case.
                """
                for i, o in self.OVERALL_MAXIMUM_INVESTFLOWS:
                    for p in m.CAPACITY_PERIODS:
                        expr = (
                            self.total[i, o, p]
                            <= m.flows[i, o].investment.overall_maximum
                        )
                        self.overall_maximum.add((i, o, p), expr)

            self.overall_maximum = Constraint(
                self.OVERALL_MAXIMUM_INVESTFLOWS,
                m.CAPACITY_PERIODS,
                noruleinit=True,
            )
            self.overall_maximum_build = BuildAction(
                rule=_overall_maximum_investflow_rule
            )

            def _overall_minimum_investflow_rule(block, i, o):
                """Rule definition for minimum overall investment
                in investment case.

                Note: This is only applicable for the last period
                """
                expr = (
                    m.flows[i, o].investment.overall_minimum
                    <= self.total[i, o, m.CAPACITY_PERIODS[-1]]
                )
                return expr

            self.overall_minimum = Constraint(
                self.OVERALL_MINIMUM_INVESTFLOWS,
                rule=_overall_minimum_investflow_rule,
            )

    def _objective_expression(self):
        r"""Objective expression for flows with investment attribute of type
        class:`.Investment`. The returned costs are fixed and
        investment costs. Variable costs are added from the standard flow
        objective expression.

        Objective terms for a standard model and a multi-period model differ
        quite strongly. Besides, the part of the objective function added by
        the *InvestmentFlowBlock* also depends on whether a convex
        or nonconvex *InvestmentFlowBlock* is selected.
        The following parts of the objective function are created:


        * :attr:`nonconvex = False`

            .. math::
                \sum_p P_{invest}(p) \cdot c_{invest,var}(p)

        * :attr:`nonconvex = True`

            .. math::
                \sum_p \left(P_{invest}(p) \cdot c_{invest,var}(p)
                + c_{invest,fix}(p) \cdot Y_{invest}(p) \right)\\

        Where p denotes the capacity period.

        """
        if not hasattr(self, "INVESTFLOWS"):
            return 0

        m = self.parent_block()
        investment_costs = 0

        for i, o in self.CONVEX_INVESTFLOWS:
            for p in m.CAPACITY_PERIODS:
                investment_costs += (
                    self.invest[i, o, p] * m.flows[i, o].investment.ep_costs[p]
                )

        for i, o in self.NON_CONVEX_INVESTFLOWS:
            for p in m.CAPACITY_PERIODS:
                investment_costs += (
                    self.invest[i, o, p] * m.flows[i, o].investment.ep_costs[p]
                    + self.invest_status[i, o, p]
                    * m.flows[i, o].investment.offset[p]
                )

        self.investment_costs = Expression(expr=investment_costs)
        self.costs = Expression(expr=investment_costs)

        return self.costs

    def _minimum_investment_constraint(self):

        """Constraint factory for a minimum investment"""
        m = self.parent_block()

        def _min_invest_rule(_):
            """Rule definition for applying a minimum investment"""
            for i, o in self.NON_CONVEX_INVESTFLOWS:
                for p in m.CAPACITY_PERIODS:
                    expr = (
                        m.flows[i, o].investment.minimum[p]
                        * self.invest_status[i, o, p]
                        <= self.invest[i, o, p]
                    )
                    self.minimum_rule.add((i, o, p), expr)

        self.minimum_rule = Constraint(
            self.NON_CONVEX_INVESTFLOWS, m.CAPACITY_PERIODS, noruleinit=True
        )
        self.minimum_rule_build = BuildAction(rule=_min_invest_rule)

        return self.minimum_rule

    def _maximum_investment_constraint(self):

        """Constraint factory for a maximum investment"""
        m = self.parent_block()

        def _max_invest_rule(_):
            """Rule definition for applying a minimum investment"""
            for i, o in self.NON_CONVEX_INVESTFLOWS:
                for p in m.CAPACITY_PERIODS:
                    expr = self.invest[i, o, p] <= (
                        m.flows[i, o].investment.maximum[p]
                        * self.invest_status[i, o, p]
                    )
                    self.maximum_rule.add((i, o, p), expr)

        self.maximum_rule = Constraint(
            self.NON_CONVEX_INVESTFLOWS, m.CAPACITY_PERIODS, noruleinit=True
        )
        self.maximum_rule_build = BuildAction(rule=_max_invest_rule)

        return self.maximum_rule


1		# -- coding: utf-8 --
2
3		"""Creating sets, variables, constraints and parts of the objective function
4		for Flow objects with investment but without nonconvex option.
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		SPDX-FileCopyrightText: Johannes Kochems
14
15		SPDX-License-Identifier: MIT
16
17		"""
18		from pyomo.core import Binary
19		from pyomo.core import BuildAction
20		from pyomo.core import Constraint
21		from pyomo.core import Expression
22		from pyomo.core import NonNegativeReals
23		from pyomo.core import Set
24		from pyomo.core import Var
25		from pyomo.core.base.block import ScalarBlock
26
27
28		class InvestmentFlowBlock(ScalarBlock):
29		r"""Block for all flows with :attr:`Investment` being not None.
30
31		.. automethod:: _create_constraints
32		.. automethod:: _create_variables
33		.. automethod:: _create_sets
34
35		.. automethod:: _objective_expression
36
37		See :class:`oemof.solph.options.Investment` for all parameters of the
38		Investment class.
39
40		See :class:`oemof.solph.flows._simple_flow_block.SimpleFlowBlock`
41		for all parameters of the SimpleFlowBlock class.
42
43		The overall summed cost expressions for all InvestmentFlowBlock objects
44		can be accessed by
45
46		* :attr:`om.InvestmentFlowBlock.investment_costs`,
47		* :attr:`om.InvestmentFlowBlock.costs`.
48
49		Note
50		----
51		In case of a nonconvex investment flow (:attr:`nonconvex=True`),
52		the existing flow capacity :math:`P_{exist}` needs to be zero.
53
54		Note
55		----
56		See also :class:`~oemof.solph.flows._flow.Flow`,
57		:class:`~oemof.solph.flows._simple_flow_block.SimpleFlowBlock` and
58		:class:`~oemof.solph._options.Investment`
59
60		""" # noqa: E501
61
62		def __init__(self, args, *kwargs):
63		super().__init__(args, *kwargs)
64
65		def _create(self, group=None):
66		r"""Creates sets, variables and constraints for SimpleFlowBlock
67		with investment attribute of type class:`.Investment`.
68
69		Parameters
70		----------
71		group : list
72		List containing tuples containing flow (f) objects that have an
73		attribute investment and the associated source (s) and target (t)
74		of flow e.g. groups=[(s1, t1, f1), (s2, t2, f2),..]
75		"""
76		if group is None:
77		return None
78
79		self._create_sets(group)
80		self._create_variables(group)
81		self._create_constraints()
82
83		def _create_sets(self, group):
84		"""
85		Creates all sets for investment flows.
86		"""
87		self.INVESTFLOWS = Set(initialize=[(g[0], g[1]) for g in group])
88
89		self.CONVEX_INVESTFLOWS = Set(
90		initialize=[
91		(g[0], g[1])
92		for g in group
93		if g[2].investment.nonconvex is False
94		]
95		)
96
97		self.NON_CONVEX_INVESTFLOWS = Set(
98		initialize=[
99		(g[0], g[1])
100		for g in group
101		if g[2].investment.nonconvex is True
102		]
103		)
104
105		self.FIXED_INVESTFLOWS = Set(
106		initialize=[(g[0], g[1]) for g in group if g[2].fix[0] is not None]
107		)
108
109		self.NON_FIXED_INVESTFLOWS = Set(
110		initialize=[(g[0], g[1]) for g in group if g[2].fix[0] is None]
111		)
112
113		self.FULL_LOAD_TIME_MAX_INVESTFLOWS = Set(
114		initialize=[
115		(g[0], g[1])
116		for g in group
117		if g[2].full_load_time_max is not None
118		]
119		)
120
121		self.FULL_LOAD_TIME_MIN_INVESTFLOWS = Set(
122		initialize=[
123		(g[0], g[1])
124		for g in group
125		if g[2].full_load_time_min is not None
126		]
127		)
128
129		self.MIN_INVESTFLOWS = Set(
130		initialize=[(g[0], g[1]) for g in group if g[2].min.min() != 0]
131		)
132
133		self.EXISTING_INVESTFLOWS = Set(
134		initialize=[
135		(g[0], g[1])
136		for g in group
137		if g[2].investment.existing is not None
138		]
139		)
140
141		self.OVERALL_MAXIMUM_INVESTFLOWS = Set(
142		initialize=[
143		(g[0], g[1])
144		for g in group
145		if g[2].investment.overall_maximum is not None
146		]
147		)
148
149		self.OVERALL_MINIMUM_INVESTFLOWS = Set(
150		initialize=[
151		(g[0], g[1])
152		for g in group
153		if g[2].investment.overall_minimum is not None
154		]
155		)
156
157		def _create_variables(self, _):
158		r"""Creates all variables for investment flows.
159
160		All InvestmentFlowBlock objects are indexed by a starting and
161		ending node :math:`(i, o)`, which is omitted in the following
162		for the sake of convenience. The following variables are created:
163
164		* :math:`P(p, t)`
165
166		Actual flow value
167		(created in :class:`oemof.solph.models.Model`),
168		indexed by tuple of periods p and timestep t
169
170		* :math:`P_{invest}(p)`
171
172		Value of the investment variable in period p,
173		equal to what is being invested and equivalent resp. similar to
174		the nominal capacity of the flows after optimization.
175
176		* :math:`P_{total}(p)`
177
178		Total installed capacity / energy in period p,
179		equivalent to the nominal capacity of the flows after optimization.
180
181		* :math:`P_{old}(p)`
182
183		Old capacity / energy to be decommissioned in period p
184		due to reaching its lifetime; applicable only
185		for multi-period models.
186
187		* :math:`P_{old,exo}(p)`
188
189		Old exogenous capacity / energy to be decommissioned in period p
190		due to reaching its lifetime, i.e. the amount that has
191		been specified by :attr:`existing` when it is decommisioned;
192		applicable only for multi-period models.
193
194		* :math:`P_{old,end}(p)`
195
196		Old endogenous capacity / energy to be decommissioned in period p
197		due to reaching its lifetime, i.e. the amount that has been
198		invested in by the model itself that is decommissioned in
199		a later period because of reaching its lifetime;
200		applicable only for multi-period models.
201
202		* :math:`Y_{invest}(p)`
203
204		Binary variable for the status of the investment, if
205		:attr:`nonconvex` is `True`.
206		"""
207		m = self.parent_block()
208
209		def _investvar_bound_rule(block, i, o, p):
210		"""Rule definition for bounds of invest variable."""
211		if (i, o) in self.CONVEX_INVESTFLOWS:
212		return (
213		m.flows[i, o].investment.minimum[p],
214		m.flows[i, o].investment.maximum[p],
215		)
216		elif (i, o) in self.NON_CONVEX_INVESTFLOWS:
217		return 0, m.flows[i, o].investment.maximum[p]
218
219		# create invest variable for an investment flow
220		self.invest = Var(
221		self.INVESTFLOWS,
222		m.CAPACITY_PERIODS,
223		within=NonNegativeReals,
224		bounds=_investvar_bound_rule,
225		)
226
227		# Total capacity
228		self.total = Var(
229		self.INVESTFLOWS, m.CAPACITY_PERIODS, within=NonNegativeReals
230		)
231
232		if m.es.investment_times is not None:
233		self.old = Var(
234		self.INVESTFLOWS, m.CAPACITY_PERIODS, within=NonNegativeReals
235		)
236
237		# Old endogenous capacity to be decommissioned (due to lifetime)
238		self.old_end = Var(
239		self.INVESTFLOWS, m.CAPACITY_PERIODS, within=NonNegativeReals
240		)
241
242		# Old exogenous capacity to be decommissioned (due to lifetime)
243		self.old_exo = Var(
244		self.INVESTFLOWS, m.CAPACITY_PERIODS, within=NonNegativeReals
245		)
246
247		# create status variable for a non-convex investment flow
248		self.invest_status = Var(
249		self.NON_CONVEX_INVESTFLOWS, m.CAPACITY_PERIODS, within=Binary
250		)
251
252		def _create_constraints(self):
253		r"""Creates all constraints for standard flows.
254
255		Depending on the attributes of the InvestmentFlowBlock
256		and SimpleFlowBlock, different constraints are created.
257		The following constraints are created
258		for all InvestmentFlowBlock objects:\
259
260		Total capacity / energy
261
262		.. math::
263		&
264		if \quad p=0:\\
265		&
266		P_{total}(p) = P_{invest}(p) + P_{exist}(p) \\
267		&\\
268		&
269		else:\\
270		&
271		P_{total}(p) = P_{total}(p-1) + P_{invest}(p) - P_{old}(p) \\
272		&\\
273		&
274		\forall p \in \textrm{CAPACITY_PERIODS}
275
276		Upper bound for the flow value
277
278		.. math::
279		&
280		P(p, t) \le ( P_{total}(p) ) \cdot f_{max}(t) \\
281		&
282		\forall p, t \in \textrm{TIMEINDEX}
283
284		For a multi-period model, the old capacity is defined as follows:
285
286		.. math::
287		&
288		P_{old}(p) = P_{old,exo}(p) + P_{old,end}(p)\\
289		&\\
290		&
291		if \quad p=0:\\
292		&
293		P_{old,end}(p) = 0\\
294		&\\
295		&
296		else \quad if \quad l \leq year(p):\\
297		&
298		P_{old,end}(p) = P_{invest}(p_{comm})\\
299		&\\
300		&
301		else:\\
302		&
303		P_{old,end}(p) = 0\\
304		&\\
305		&
306		if \quad p=0:\\
307		&
308		P_{old,exo}(p) = 0\\
309		&\\
310		&
311		else \quad if \quad l - a \leq year(p):\\
312		&
313		P_{old,exo}(p) = P_{exist} (*)\\
314		&\\
315		&
316		else:\\
317		&
318		P_{old,exo}(p) = 0\\
319		&\\
320		&
321		\forall p \in \textrm{CAPACITY_PERIODS}
322
323		where:
324
325		* (*) is only performed for the first period the condition
326		is True. A decommissioning flag is then set to True
327		to prevent having falsely added old capacity in future periods.
328		* :math:`year(p)` is the year corresponding to period p
329		* :math:`p_{comm}` is the commissioning period of the flow
330		(which is determined by the model itself)
331
332		Depending on the attribute :attr:`nonconvex`, the constraints for the
333		bounds of the decision variable :math:`P_{invest}(p)` are different:\
334
335		* :attr:`nonconvex = False`
336
337		.. math::
338		&
339		P_{invest, min}(p) \le P_{invest}(p) \le P_{invest, max}(p) \\
340		&
341		\forall p \in \textrm{CAPACITY_PERIODS}
342
343		* :attr:`nonconvex = True`
344
345		.. math::
346		&
347		P_{invest, min}(p) \cdot Y_{invest}(p) \le P_{invest}(p)\\
348		&
349		P_{invest}(p) \le P_{invest, max}(p) \cdot Y_{invest}(p)\\
350		&\\
351		&
352		\forall p \in \textrm{CAPACITY_PERIODS}
353
354		For all InvestmentFlowBlock objects
355		(independent of the attribute :attr:`nonconvex`),
356		the following additional constraints are created, if the appropriate
357		attribute of the SimpleFlowBlock
358		(see :class:`oemof.solph.flows._simple_flow_block.SimpleFlowBlock`)
359		is set:
360
361		* :attr:`fix` is not None
362
363		Actual value constraint for investments with fixed flow values
364
365		.. math::
366		&
367		P(p, t) = P_{total}(p) \cdot f_{fix}(t) \\
368		&\\
369		&
370		\forall p, t \in \textrm{TIMEINDEX}
371
372		* :attr:`min != 0`
373
374		Lower bound for the flow values
375
376		.. math::
377		&
378		P(p, t) \geq P_{total}(p) \cdot f_{min}(t) \\
379		&\\
380		&
381		\forall p, t \in \textrm{TIMEINDEX}
382
383		* :attr:`full_load_time_max is not None`
384
385		Upper bound for the sum of all flow values
386		(e.g. maximum full load hours)
387
388		.. math::
389		\sum_{p, t} P(p, t) \cdot \tau(t) \leq P_{total}(p)
390		\cdot t_{full\_load, min}
391
392		* :attr:`full_load_time_min is not None`
393
394		Lower bound for the sum of all flow values
395		(e.g. minimum full load hours)
396
397		.. math::
398		\sum_{p, t} P(t) \cdot \tau(t) \geq P_{total}
399		\cdot t_{full\_load, min}
400
401		* :attr:`overall_maximum` is not None
402		(for multi-period model only)
403
404		Overall maximum of total installed capacity / energy for flow
405
406		.. math::
407		&
408		P_{total}(p) \leq P_{overall,max} \\
409		&\\
410		&
411		\forall p \in \textrm{CAPACITY_PERIODS}
412
413		* :attr:`overall_minimum` is not None
414		(for multi-period model only)
415
416		Overall minimum of total installed capacity / energy for flow;
417		applicable only in last period
418
419		.. math::
420		P_{total}(p_{last}) \geq P_{overall,min}
421		"""
422		m = self.parent_block()
423
424		self.minimum_rule = self._minimum_investment_constraint()
425		self.maximum_rule = self._maximum_investment_constraint()
426
427		# Handle unit lifetimes
428		def _total_capacity_rule(block):
429		"""Rule definition for determining total installed
430		capacity (taking decommissioning into account)
431		"""
432		for i, o in self.INVESTFLOWS:
433		for p in m.CAPACITY_PERIODS:
434		if p == 0:
435		expr = (
436		self.total[i, o, p]
437		== self.invest[i, o, p]
438		+ m.flows[i, o].investment.existing
439		)
440		self.total_rule.add((i, o, p), expr)
441		# applicable for multi-period model only
442		else:
443		expr = (
444		self.total[i, o, p]
445		== self.invest[i, o, p]
446		+ self.total[i, o, p - 1]
447		- self.old[i, o, p]
448		)
449		self.total_rule.add((i, o, p), expr)
450
451		self.total_rule = Constraint(
452		self.INVESTFLOWS, m.CAPACITY_PERIODS, noruleinit=True
453		)
454		self.total_rule_build = BuildAction(rule=_total_capacity_rule)
455
456		def _investflow_fixed_rule(block):
457		"""Rule definition of constraint to fix flow variable
458		of investment flow to (normed) actual value
459		"""
460		for i, o in self.FIXED_INVESTFLOWS:
461		for p, t in m.TIMEINDEX:
462		expr = (
463		m.flow[i, o, t]
464		== self.total[i, o, p] * m.flows[i, o].fix[t]
465		)
466		self.fixed.add((i, o, p, t), expr)
467
468		self.fixed = Constraint(
469		self.FIXED_INVESTFLOWS, m.TIMEINDEX, noruleinit=True
470		)
471		self.fixed_build = BuildAction(rule=_investflow_fixed_rule)
472
473		def _max_investflow_rule(block):
474		"""Rule definition of constraint setting an upper bound of flow
475		variable in investment case.
476		"""
477		for i, o in self.NON_FIXED_INVESTFLOWS:
478		for p, t in m.TIMEINDEX:
479		expr = (
480		m.flow[i, o, t]
481		<= self.total[i, o, p] * m.flows[i, o].max[t]
482		)
483		self.max.add((i, o, p, t), expr)
484
485		self.max = Constraint(
486		self.NON_FIXED_INVESTFLOWS, m.TIMEINDEX, noruleinit=True
487		)
488		self.max_build = BuildAction(rule=_max_investflow_rule)
489
490		def _min_investflow_rule(block):
491		"""Rule definition of constraint setting a lower bound on flow
492		variable in investment case.
493		"""
494		for i, o in self.MIN_INVESTFLOWS:
495		for p, t in m.TIMEINDEX:
496		expr = (
497		m.flow[i, o, t]
498		>= self.total[i, o, p] * m.flows[i, o].min[t]
499		)
500		self.min.add((i, o, p, t), expr)
501
502		self.min = Constraint(
503		self.MIN_INVESTFLOWS, m.TIMEINDEX, noruleinit=True
504		)
505		self.min_build = BuildAction(rule=_min_investflow_rule)
506
507		def _full_load_time_max_investflow_rule(_, i, o):
508		"""Rule definition for build action of max. sum flow constraint
509		in investment case.
510		"""
511		expr = sum(
512		m.flow[i, o, t] * m.timeincrement[t] for t in m.TIMESTEPS
513		) <= (
514		m.flows[i, o].full_load_time_max
515		* sum(self.total[i, o, p] for p in m.CAPACITY_PERIODS)
516		)
517		return expr
518
519		self.full_load_time_max = Constraint(
520		self.FULL_LOAD_TIME_MAX_INVESTFLOWS,
521		rule=_full_load_time_max_investflow_rule,
522		)
523
524		def _full_load_time_min_investflow_rule(_, i, o):
525		"""Rule definition for build action of min. sum flow constraint
526		in investment case.
527		"""
528		expr = sum(
529		m.flow[i, o, t] * m.timeincrement[t] for t in m.TIMESTEPS
530		) >= (
531		sum(self.total[i, o, p] for p in m.CAPACITY_PERIODS)
532		* m.flows[i, o].full_load_time_min
533		)
534		return expr
535
536		self.full_load_time_min = Constraint(
537		self.FULL_LOAD_TIME_MIN_INVESTFLOWS,
538		rule=_full_load_time_min_investflow_rule,
539		)
540
541		if m.es.investment_times is not None:
542
543		def _overall_maximum_investflow_rule(block):
544		"""Rule definition for maximum overall investment
545		in investment case.
546		"""
547		for i, o in self.OVERALL_MAXIMUM_INVESTFLOWS:
548		for p in m.CAPACITY_PERIODS:
549		expr = (
550		self.total[i, o, p]
551		<= m.flows[i, o].investment.overall_maximum
552		)
553		self.overall_maximum.add((i, o, p), expr)
554
555		self.overall_maximum = Constraint(
556		self.OVERALL_MAXIMUM_INVESTFLOWS,
557		m.CAPACITY_PERIODS,
558		noruleinit=True,
559		)
560		self.overall_maximum_build = BuildAction(
561		rule=_overall_maximum_investflow_rule
562		)
563
564		def _overall_minimum_investflow_rule(block, i, o):
565		"""Rule definition for minimum overall investment
566		in investment case.
567
568		Note: This is only applicable for the last period
569		"""
570		expr = (
571		m.flows[i, o].investment.overall_minimum
572		<= self.total[i, o, m.CAPACITY_PERIODS[-1]]
573		)
574		return expr
575
576		self.overall_minimum = Constraint(
577		self.OVERALL_MINIMUM_INVESTFLOWS,
578		rule=_overall_minimum_investflow_rule,
579		)
580
581		def _objective_expression(self):
582		r"""Objective expression for flows with investment attribute of type
583		class:`.Investment`. The returned costs are fixed and
584		investment costs. Variable costs are added from the standard flow
585		objective expression.
586
587		Objective terms for a standard model and a multi-period model differ
588		quite strongly. Besides, the part of the objective function added by
589		the InvestmentFlowBlock also depends on whether a convex
590		or nonconvex InvestmentFlowBlock is selected.
591		The following parts of the objective function are created:
592
593
594		* :attr:`nonconvex = False`
595
596		.. math::
597		\sum_p P_{invest}(p) \cdot c_{invest,var}(p)
598
599		* :attr:`nonconvex = True`
600
601		.. math::
602		\sum_p \left(P_{invest}(p) \cdot c_{invest,var}(p)
603		+ c_{invest,fix}(p) \cdot Y_{invest}(p) \right)\\
604
605		Where p denotes the capacity period.
606
607		"""
608		if not hasattr(self, "INVESTFLOWS"):
609		return 0
610
611		m = self.parent_block()
612		investment_costs = 0
613
614		for i, o in self.CONVEX_INVESTFLOWS:
615		for p in m.CAPACITY_PERIODS:
616		investment_costs += (
617		self.invest[i, o, p] * m.flows[i, o].investment.ep_costs[p]
618		)
619
620		for i, o in self.NON_CONVEX_INVESTFLOWS:
621		for p in m.CAPACITY_PERIODS:
622		investment_costs += (
623		self.invest[i, o, p] * m.flows[i, o].investment.ep_costs[p]
624		+ self.invest_status[i, o, p]
625		* m.flows[i, o].investment.offset[p]
626		)
627
628		self.investment_costs = Expression(expr=investment_costs)
629		self.costs = Expression(expr=investment_costs)
630
631		return self.costs
632
633	View Code Duplication	def _minimum_investment_constraint(self):
		0 ignored issues – show Duplication introduced 2024-08-26 11:00 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
634		"""Constraint factory for a minimum investment"""
635		m = self.parent_block()
636
637		def _min_invest_rule(_):
638		"""Rule definition for applying a minimum investment"""
639		for i, o in self.NON_CONVEX_INVESTFLOWS:
640		for p in m.CAPACITY_PERIODS:
641		expr = (
642		m.flows[i, o].investment.minimum[p]
643		* self.invest_status[i, o, p]
644		<= self.invest[i, o, p]
645		)
646		self.minimum_rule.add((i, o, p), expr)
647
648		self.minimum_rule = Constraint(
649		self.NON_CONVEX_INVESTFLOWS, m.CAPACITY_PERIODS, noruleinit=True
650		)
651		self.minimum_rule_build = BuildAction(rule=_min_invest_rule)
652
653		return self.minimum_rule
654
655	View Code Duplication	def _maximum_investment_constraint(self):
		0 ignored issues – show Duplication introduced 2024-08-26 11:00 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
656		"""Constraint factory for a maximum investment"""
657		m = self.parent_block()
658
659		def _max_invest_rule(_):
660		"""Rule definition for applying a minimum investment"""
661		for i, o in self.NON_CONVEX_INVESTFLOWS:
662		for p in m.CAPACITY_PERIODS:
663		expr = self.invest[i, o, p] <= (
664		m.flows[i, o].investment.maximum[p]
665		* self.invest_status[i, o, p]
666		)
667		self.maximum_rule.add((i, o, p), expr)
668
669		self.maximum_rule = Constraint(
670		self.NON_CONVEX_INVESTFLOWS, m.CAPACITY_PERIODS, noruleinit=True
671		)
672		self.maximum_rule_build = BuildAction(rule=_max_invest_rule)
673
674		return self.maximum_rule
675

oemof / oemof-solph

Pull Request — dev (#1183)

InvestmentFlowBlock._create_constraints() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like