solph.flows._flow.FlowBlock._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

solph.flows._flow.FlowBlock._create() F

↳ Parent: solph.flows._flow

Complexity

Conditions

Size

Total Lines	160
Code Lines	95

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	95
dl	0
loc	160
rs	2.2636
c	0
b	0
f	0
cc	15
nop	2

How to fix Long Method Complexity

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

"""
solph version of oemof.network.Edge including base constraints

SPDX-FileCopyrightText: Uwe Krien <[email protected]>
SPDX-FileCopyrightText: Simon Hilpert
SPDX-FileCopyrightText: Cord Kaldemeyer
SPDX-FileCopyrightText: Stephan Günther
SPDX-FileCopyrightText: Birgit Schachler
SPDX-FileCopyrightText: jnnr
SPDX-FileCopyrightText: jmloenneberga

SPDX-License-Identifier: MIT

"""
import math
from warnings import warn

from oemof.network import network as on
from oemof.tools import debugging
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 ScalarBlock

from oemof.solph._plumbing import sequence


class Flow(on.Edge):
    r"""Defines a flow between two nodes.

    Keyword arguments are used to set the attributes of this flow. Parameters
    which are handled specially are noted below.
    For the case where a parameter can be either a scalar or an iterable, a
    scalar value will be converted to a sequence containing the scalar value at
    every index. This sequence is then stored under the paramter's key.

    Parameters
    ----------
    nominal_value : numeric, :math:`P_{nom}`
        The nominal value of the flow. If this value is set the corresponding
        optimization variable of the flow object will be bounded by this value
        multiplied with min(lower bound)/max(upper bound).
    max : numeric (iterable or scalar), :math:`f_{max}`
        Normed maximum value of the flow. The flow absolute maximum will be
        calculated by multiplying :attr:`nominal_value` with :attr:`max`
    min : numeric (iterable or scalar), :math:`f_{min}`
        Normed minimum value of the flow (see :attr:`max`).
    fix : numeric (iterable or scalar), :math:`f_{fix}`
        Normed fixed value for the flow variable. Will be multiplied with the
        :attr:`nominal_value` to get the absolute value.
    positive_gradient : :obj:`dict`, default: `{'ub': None}`
        A dictionary containing the following key:

         * `'ub'`: numeric (iterable, scalar or None), the normed *upper
           bound* on the positive difference (`flow[t-1] < flow[t]`) of
           two consecutive flow values.

    negative_gradient : :obj:`dict`, default: `{'ub': None}`

        A dictionary containing the following key:

          * `'ub'`: numeric (iterable, scalar or None), the normed *upper
            bound* on the negative difference (`flow[t-1] > flow[t]`) of
            two consecutive flow values.

    full_load_time_max : numeric, :math:`t_{full\_load,max}`
        Upper bound on the summed flow expressed as the equivalent time that
        the flow would have to run at full capacity to yield the same sum. The
        value will be multiplied with the nominal_value to get the absolute
        limit.
    full_load_time_min : numeric, :math:`t_{full\_load,min}`
        Lower bound on the summed flow expressed as the equivalent time that
        the flow would have to run at full capacity to yield the same sum. The
        value will be multiplied with the nominal_value to get the absolute
        limit.
    variable_costs : numeric (iterable or scalar)
        The costs associated with one unit of the flow. If this is set the
        costs will be added to the objective expression of the optimization
        problem.
    fixed : boolean
        Boolean value indicating if a flow is fixed during the optimization
        problem to its ex-ante set value. Used in combination with the
        :attr:`fix`.
    integer : boolean
        Set True to bound the flow values to integers.

    Notes
    -----
    See :py:class:`~oemof.solph.flows._flow.FlowBlock` for the variables,
    constraints and objective parts, that are created for a FLow object.

    Examples
    --------
    Creating a fixed flow object:

    >>> f = Flow(nominal_value=2, fix=[10, 4, 4], variable_costs=5)
    >>> f.variable_costs[2]
    5
    >>> f.fix[2]
    4

    Creating a flow object with time-depended lower and upper bounds:

    >>> f1 = Flow(min=[0.2, 0.3], max=0.99, nominal_value=100)
    >>> f1.max[1]
    0.99
    """

    def __init__(self, **kwargs):
        # TODO: Check if we can inherit from pyomo.core.base.var _VarData
        # then we need to create the var object with
        # pyomo.core.base.IndexedVarWithDomain before any FlowBlock is created.
        # E.g. create the variable in the energy system and populate with
        # information afterwards when creating objects.

        # --- BEGIN: The following code can be removed for versions >= v0.6 ---
        msg = (
            "\nThe parameter 'summed_{0}' ist deprecated and will be removed "
            "in version v0.6.\nRename the parameter to 'full_load_time_{0}', "
            "to avoid this warning and future problems. "
        )
        if "summed_max" in kwargs:
            warn(msg.format("max"), FutureWarning)
            kwargs["full_load_time_max"] = kwargs["summed_max"]
        if "summed_min" in kwargs:
            warn(msg.format("min"), FutureWarning)
            kwargs["full_load_time_min"] = kwargs["summed_min"]
        # --- END ---

        super().__init__()

        scalars = [
            "nominal_value",
            "full_load_time_max",
            "full_load_time_min",
            "investment",
            "nonconvex",
            "integer",
        ]
        sequences = ["fix", "variable_costs", "min", "max"]
        dictionaries = ["positive_gradient", "negative_gradient"]
        defaults = {
            "variable_costs": 0,
            "positive_gradient": {"ub": None},
            "negative_gradient": {"ub": None},
        }
        need_nominal_value = [
            "fix",
            "full_load_time_max",
            "full_load_time_min",
            "max",
            "min",
            # --- BEGIN: To be removed for versions >= v0.6 ---
            "summed_max",
            "summed_min",
            # --- END ---
        ]
        keys = [k for k in kwargs if k != "label"]

        if "fixed_costs" in keys:
            raise AttributeError(
                "The `fixed_costs` attribute has been removed" " with v0.2!"
            )

        if "actual_value" in keys:
            raise AttributeError(
                "The `actual_value` attribute has been renamed"
                " to `fix` with v0.4. The attribute `fixed` is"
                " set to True automatically when passing `fix`."
            )

        if "fixed" in keys:
            msg = (
                "The `fixed` attribute is deprecated.\nIf you have defined "
                "the `fix` attribute the flow variable will be fixed.\n"
                "The `fixed` attribute does not change anything."
            )
            warn(msg, debugging.SuspiciousUsageWarning)

        # It is not allowed to define min or max if fix is defined.
        if kwargs.get("fix") is not None and (
            kwargs.get("min") is not None or kwargs.get("max") is not None
        ):
            raise AttributeError(
                "It is not allowed to define `min`/`max` if `fix` is defined."
            )

        # Set default value for min and max
        if kwargs.get("min") is None:
            if "bidirectional" in keys:
                defaults["min"] = -1
            else:
                defaults["min"] = 0
        if kwargs.get("max") is None:
            defaults["max"] = 1

        # Check gradient dictionaries for non-valid keys
        for gradient_dict in ["negative_gradient", "positive_gradient"]:
            if gradient_dict in kwargs:
                if list(kwargs[gradient_dict].keys()) != list(
                    defaults[gradient_dict].keys()
                ):
                    msg = (
                        "Only the key 'ub' is allowed for the '{0}' attribute"
                    )
                    raise AttributeError(msg.format(gradient_dict))

        for attribute in set(scalars + sequences + dictionaries + keys):
            value = kwargs.get(attribute, defaults.get(attribute))
            if attribute in dictionaries:

                setattr(
                    self,
                    attribute,
                    {"ub": sequence(value["ub"])},
                )

            else:
                setattr(
                    self,
                    attribute,
                    sequence(value) if attribute in sequences else value,
                )

        # Checking for impossible attribute combinations
        if self.investment and self.nominal_value is not None:
            raise ValueError(
                "Using the investment object the nominal_value"
                " has to be set to None."
            )
        if self.investment and self.nonconvex:
            raise ValueError(
                "Investment flows cannot be combined with "
                + "nonconvex flows!"
            )

        infinite_error_msg = (
            "{} must be a finite value. Passing an infinite "
            "value is not allowed."
        )
        if not self.investment:
            if self.nominal_value is None:
                for attr in need_nominal_value:
                    if kwargs.get(attr) is not None:
                        raise AttributeError(
                            "If {} is set in a flow (except InvestmentFlow), "
                            "nominal_value must be set as well.\n"
                            "Otherwise, it won't have any effect.".format(attr)
                        )
            elif not math.isfinite(self.nominal_value):
                raise ValueError(infinite_error_msg.format("nominal_value"))
        if not math.isfinite(self.max[0]):
            raise ValueError(infinite_error_msg.format("max"))

        # Checking for impossible gradient combinations
        if self.nonconvex:

            if self.nonconvex.positive_gradient["ub"][0] is not None and (
                self.positive_gradient["ub"][0] is not None
                or self.negative_gradient["ub"][0] is not None
            ):
                raise ValueError(
                    "You specified a positive gradient in your nonconvex "
                    "option. This cannot be combined with a positive or a "
                    "negative gradient for a standard flow!"
                )

        if self.nonconvex:

            if self.nonconvex.negative_gradient["ub"][0] is not None and (
                self.positive_gradient["ub"][0] is not None
                or self.negative_gradient["ub"][0] is not None
            ):
                raise ValueError(
                    "You specified a negative gradient in your nonconvex "
                    "option. This cannot be combined with a positive or a "
                    "negative gradient for a standard flow!"
                )


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

    See :class:`~oemof.solph.flows._flow.Flow` class for all parameters of the *Flow*.

    **Variables**

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

    * :math:`P(t)`
        Actual flow value (created in :class:`~oemof.solph._models.Model`).
        The variable is bound to: :math:`f_{min}(t) \cdot P_{nom} \ge P(t) \le f_{max}(t) \cdot P_{nom}`.

        If `Flow.fix` is not None the variable is bound to
        :math:`P(t) = f_{fix}`.

    * :math:`ve_n` (`Flow.negative_gradient` is not `None`)
        Difference of a flow in consecutive timesteps if flow is reduced. The
        variable is bound to: :math:`0 \ge ve_n \ge ve_n^{max}`.

    * :math:`ve_p` (`Flow.positive_gradient` is not `None`)
        Difference of a flow in consecutive timesteps if flow is increased. The
        variable is bound to: :math:`0 \ge ve_p \ge ve_p^{max}`.

    The following variable is build for Flows with the attribute
    `integer_flows` being not None.

    * :math:`i`(`Flow.integer` is `True`)
        All flow values are integers. Variable is bound to non-negative
        integers.

    **Constraints**

    The following constraints are created, if the appropriate attribute of the
    *Flow* (see :class:`oemof.solph.network.Flow`) object is set:

    * `Flow.full_load_time_max` is not `None` (full_load_time_max_constr):
        .. math::
            \sum_t P(t) \cdot \tau \leq F_{max} \cdot P_{nom}

    * `Flow.full_load_time_min` is not `None` (full_load_time_min_constr):
        .. math::
            \sum_t P(t) \cdot \tau \geq F_{min} \cdot P_{nom}


    * `Flow.negative_gradient` is not `None` (negative_gradient_constr):
        .. math::
          P(t-1) - P(t) \geq ve_n(t)

    * `Flow.positive_gradient` is not `None` (positive_gradient_constr):
        .. math::
          P(t) - P(t-1) \geq ve_p(t)

    * `Flow.integer` is `True`
        .. math::
          P(t) = i(t)

    **Objective function**

    Depending on the attributes of the `Flow` object the following parts of
    the objective function are created:

    * `Flow.variable_costs` is not `None`:
        .. math::
          \sum_{(i,o)} \sum_t P(t) \cdot c_{var}(i, o, t)

    .. csv-table:: List of Variables
        :header: "symbol", "attribute", "explanation"
        :widths: 1, 1, 1

        ":math:`P(t)`", ":command:`flow[i, o][t]`", "Actual flow value"
        ":math:`ve_n`", ":command:`negative_gradient[n, o, t]`", "Negative gradient of the flow"
        ":math:`ve_p`", ":command:`positive_gradient[n, o, t]`", "Positive gradient of the flow"
        ":math:`i`", ":command:`integer_flow[i, o, t]`","Integer flow"


    .. csv-table:: List of Parameters
        :header: "symbol", "attribute", "explanation"
        :widths: 1, 1, 1

        ":math:`P_{nom}`", ":command:`flows[i, o].nominal_value`","Nominal value of the flow"
        ":math:`F_{max}`",":command:`flow[i, o].full_load_time_max`", "Maximal full
        load time"
        ":math:`F_{min}`",":command:`flow[i, o].full_load_time_min`", "Minimal full
        load time"
        ":math:`c_{var}`", ":command:`variable\_costs[t]`", "Variable cost of the flow"
        ":math:`f_{max}`", ":command:`flows[i, o].max[t]`", "Normed maximum value of the flow, the absolute maximum is :math:`f_{max} \cdot P_{nom}`"
        ":math:`f_{min}`", ":command:`flows[i, o].min[t]`", "Normed minimum value of the flow, the absolute minimum is :math:`f_{min} \cdot P_{nom}`"
        ":math:`f_{fix}`", ":command:`flows[i, o].min[t]`", "Normed fixed value of the flow, the absolute fixed value is :math:`f_{fix} \cdot P_{nom}`"
        ":math:`ve_n^{max}`",":command:`flows[i, o].negative_gradient`","Normed maximal negative gradient of the flow, the absolute maximum gradient is :math:`ve_n^{max} \cdot P_{nom}`"
        ":math:`ve_p^{max}`",":command:`flows[i, o].positive_gradient`","Normed maximal positive gradient of the flow, the absolute maximum gradient is :math:`ve_n^{max} \cdot P_{nom}`"

    Note
    ----
    See the :class:`~oemof.solph.flows._flow.Flow` class for the definition of
    all parameters from the "List of Parameters above.

    """  # noqa: E501

    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.FULL_LOAD_TIME_MAX_FLOWS = Set(
            initialize=[
                (g[0], g[1])
                for g in group
                if g[2].full_load_time_max is not None
                and g[2].nominal_value is not None
            ]
        )

        self.FULL_LOAD_TIME_MIN_FLOWS = Set(
            initialize=[
                (g[0], g[1])
                for g in group
                if g[2].full_load_time_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_full_load_time_max_rule(model):
            """Rule definition for build action of max. sum flow constraint."""
            for inp, out in self.FULL_LOAD_TIME_MAX_FLOWS:
                lhs = sum(
                    m.flow[inp, out, ts] * m.timeincrement[ts]

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

        self.full_load_time_max_constr = Constraint(
            self.FULL_LOAD_TIME_MAX_FLOWS, noruleinit=True
        )
        self.full_load_time_max_build = BuildAction(
            rule=_flow_full_load_time_max_rule
        )

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

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

        self.full_load_time_min_constr = Constraint(
            self.FULL_LOAD_TIME_MIN_FLOWS, noruleinit=True
        )
        self.full_load_time_min_build = BuildAction(
            rule=_flow_full_load_time_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
                        )

        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
                        )

        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		"""
4		solph version of oemof.network.Edge including base constraints
5
6		SPDX-FileCopyrightText: Uwe Krien <[email protected]>
7		SPDX-FileCopyrightText: Simon Hilpert
8		SPDX-FileCopyrightText: Cord Kaldemeyer
9		SPDX-FileCopyrightText: Stephan Günther
10		SPDX-FileCopyrightText: Birgit Schachler
11		SPDX-FileCopyrightText: jnnr
12		SPDX-FileCopyrightText: jmloenneberga
13
14		SPDX-License-Identifier: MIT
15
16		"""
17		import math
18		from warnings import warn
19
20		from oemof.network import network as on
21		from oemof.tools import debugging
22		from pyomo.core import BuildAction
23		from pyomo.core import Constraint
24		from pyomo.core import NonNegativeIntegers
25		from pyomo.core import Set
26		from pyomo.core import Var
27		from pyomo.core.base.block import ScalarBlock
28
29		from oemof.solph._plumbing import sequence
30
31
32		class Flow(on.Edge):
33		r"""Defines a flow between two nodes.
34
35		Keyword arguments are used to set the attributes of this flow. Parameters
36		which are handled specially are noted below.
37		For the case where a parameter can be either a scalar or an iterable, a
38		scalar value will be converted to a sequence containing the scalar value at
39		every index. This sequence is then stored under the paramter's key.
40
41		Parameters
42		----------
43		nominal_value : numeric, :math:`P_{nom}`
44		The nominal value of the flow. If this value is set the corresponding
45		optimization variable of the flow object will be bounded by this value
46		multiplied with min(lower bound)/max(upper bound).
47		max : numeric (iterable or scalar), :math:`f_{max}`
48		Normed maximum value of the flow. The flow absolute maximum will be
49		calculated by multiplying :attr:`nominal_value` with :attr:`max`
50		min : numeric (iterable or scalar), :math:`f_{min}`
51		Normed minimum value of the flow (see :attr:`max`).
52		fix : numeric (iterable or scalar), :math:`f_{fix}`
53		Normed fixed value for the flow variable. Will be multiplied with the
54		:attr:`nominal_value` to get the absolute value.
55		positive_gradient : :obj:`dict`, default: `{'ub': None}`
56		A dictionary containing the following key:
57
58		* `'ub'`: numeric (iterable, scalar or None), the normed *upper
59		bound* on the positive difference (`flow[t-1] < flow[t]`) of
60		two consecutive flow values.
61
62		negative_gradient : :obj:`dict`, default: `{'ub': None}`
63
64		A dictionary containing the following key:
65
66		* `'ub'`: numeric (iterable, scalar or None), the normed *upper
67		bound* on the negative difference (`flow[t-1] > flow[t]`) of
68		two consecutive flow values.
69
70		full_load_time_max : numeric, :math:`t_{full\_load,max}`
71		Upper bound on the summed flow expressed as the equivalent time that
72		the flow would have to run at full capacity to yield the same sum. The
73		value will be multiplied with the nominal_value to get the absolute
74		limit.
75		full_load_time_min : numeric, :math:`t_{full\_load,min}`
76		Lower bound on the summed flow expressed as the equivalent time that
77		the flow would have to run at full capacity to yield the same sum. The
78		value will be multiplied with the nominal_value to get the absolute
79		limit.
80		variable_costs : numeric (iterable or scalar)
81		The costs associated with one unit of the flow. If this is set the
82		costs will be added to the objective expression of the optimization
83		problem.
84		fixed : boolean
85		Boolean value indicating if a flow is fixed during the optimization
86		problem to its ex-ante set value. Used in combination with the
87		:attr:`fix`.
88		integer : boolean
89		Set True to bound the flow values to integers.
90
91		Notes
92		-----
93		See :py:class:`~oemof.solph.flows._flow.FlowBlock` for the variables,
94		constraints and objective parts, that are created for a FLow object.
95
96		Examples
97		--------
98		Creating a fixed flow object:
99
100		>>> f = Flow(nominal_value=2, fix=[10, 4, 4], variable_costs=5)
101		>>> f.variable_costs[2]
102		5
103		>>> f.fix[2]
104		4
105
106		Creating a flow object with time-depended lower and upper bounds:
107
108		>>> f1 = Flow(min=[0.2, 0.3], max=0.99, nominal_value=100)
109		>>> f1.max[1]
110		0.99
111		"""
112
113		def __init__(self, **kwargs):
114		# TODO: Check if we can inherit from pyomo.core.base.var _VarData
115		# then we need to create the var object with
116		# pyomo.core.base.IndexedVarWithDomain before any FlowBlock is created.
117		# E.g. create the variable in the energy system and populate with
118		# information afterwards when creating objects.
119
120		# --- BEGIN: The following code can be removed for versions >= v0.6 ---
121		msg = (
122		"\nThe parameter 'summed_{0}' ist deprecated and will be removed "
123		"in version v0.6.\nRename the parameter to 'full_load_time_{0}', "
124		"to avoid this warning and future problems. "
125		)
126		if "summed_max" in kwargs:
127		warn(msg.format("max"), FutureWarning)
128		kwargs["full_load_time_max"] = kwargs["summed_max"]
129		if "summed_min" in kwargs:
130		warn(msg.format("min"), FutureWarning)
131		kwargs["full_load_time_min"] = kwargs["summed_min"]
132		# --- END ---
133
134		super().__init__()
135
136		scalars = [
137		"nominal_value",
138		"full_load_time_max",
139		"full_load_time_min",
140		"investment",
141		"nonconvex",
142		"integer",
143		]
144		sequences = ["fix", "variable_costs", "min", "max"]
145		dictionaries = ["positive_gradient", "negative_gradient"]
146		defaults = {
147		"variable_costs": 0,
148		"positive_gradient": {"ub": None},
149		"negative_gradient": {"ub": None},
150		}
151		need_nominal_value = [
152		"fix",
153		"full_load_time_max",
154		"full_load_time_min",
155		"max",
156		"min",
157		# --- BEGIN: To be removed for versions >= v0.6 ---
158		"summed_max",
159		"summed_min",
160		# --- END ---
161		]
162		keys = [k for k in kwargs if k != "label"]
163
164		if "fixed_costs" in keys:
165		raise AttributeError(
166		"The `fixed_costs` attribute has been removed" " with v0.2!"
167		)
168
169		if "actual_value" in keys:
170		raise AttributeError(
171		"The `actual_value` attribute has been renamed"
172		" to `fix` with v0.4. The attribute `fixed` is"
173		" set to True automatically when passing `fix`."
174		)
175
176		if "fixed" in keys:
177		msg = (
178		"The `fixed` attribute is deprecated.\nIf you have defined "
179		"the `fix` attribute the flow variable will be fixed.\n"
180		"The `fixed` attribute does not change anything."
181		)
182		warn(msg, debugging.SuspiciousUsageWarning)
183
184		# It is not allowed to define min or max if fix is defined.
185		if kwargs.get("fix") is not None and (
186		kwargs.get("min") is not None or kwargs.get("max") is not None
187		):
188		raise AttributeError(
189		"It is not allowed to define `min`/`max` if `fix` is defined."
190		)
191
192		# Set default value for min and max
193		if kwargs.get("min") is None:
194		if "bidirectional" in keys:
195		defaults["min"] = -1
196		else:
197		defaults["min"] = 0
198		if kwargs.get("max") is None:
199		defaults["max"] = 1
200
201		# Check gradient dictionaries for non-valid keys
202		for gradient_dict in ["negative_gradient", "positive_gradient"]:
203		if gradient_dict in kwargs:
204		if list(kwargs[gradient_dict].keys()) != list(
205		defaults[gradient_dict].keys()
206		):
207		msg = (
208		"Only the key 'ub' is allowed for the '{0}' attribute"
209		)
210		raise AttributeError(msg.format(gradient_dict))
211
212		for attribute in set(scalars + sequences + dictionaries + keys):
213		value = kwargs.get(attribute, defaults.get(attribute))
214	View Code Duplication	if attribute in dictionaries:
		0 ignored issues – show Duplication introduced 2021-05-29 13:08 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
215		setattr(
216		self,
217		attribute,
218		{"ub": sequence(value["ub"])},
219		)
220
221		else:
222		setattr(
223		self,
224		attribute,
225		sequence(value) if attribute in sequences else value,
226		)
227
228		# Checking for impossible attribute combinations
229		if self.investment and self.nominal_value is not None:
230		raise ValueError(
231		"Using the investment object the nominal_value"
232		" has to be set to None."
233		)
234		if self.investment and self.nonconvex:
235		raise ValueError(
236		"Investment flows cannot be combined with "
237		+ "nonconvex flows!"
238		)
239
240		infinite_error_msg = (
241		"{} must be a finite value. Passing an infinite "
242		"value is not allowed."
243		)
244		if not self.investment:
245		if self.nominal_value is None:
246		for attr in need_nominal_value:
247		if kwargs.get(attr) is not None:
248		raise AttributeError(
249		"If {} is set in a flow (except InvestmentFlow), "
250		"nominal_value must be set as well.\n"
251		"Otherwise, it won't have any effect.".format(attr)
252		)
253		elif not math.isfinite(self.nominal_value):
254		raise ValueError(infinite_error_msg.format("nominal_value"))
255		if not math.isfinite(self.max[0]):
256		raise ValueError(infinite_error_msg.format("max"))
257
258		# Checking for impossible gradient combinations
259	View Code Duplication	if self.nonconvex:
		0 ignored issues – show Duplication introduced 2021-05-31 13:01 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
260		if self.nonconvex.positive_gradient["ub"][0] is not None and (
261		self.positive_gradient["ub"][0] is not None
262		or self.negative_gradient["ub"][0] is not None
263		):
264		raise ValueError(
265		"You specified a positive gradient in your nonconvex "
266		"option. This cannot be combined with a positive or a "
267		"negative gradient for a standard flow!"
268		)
269
270	View Code Duplication	if self.nonconvex:
		0 ignored issues – show Duplication introduced 2021-05-31 13:01 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
271		if self.nonconvex.negative_gradient["ub"][0] is not None and (
272		self.positive_gradient["ub"][0] is not None
273		or self.negative_gradient["ub"][0] is not None
274		):
275		raise ValueError(
276		"You specified a negative gradient in your nonconvex "
277		"option. This cannot be combined with a positive or a "
278		"negative gradient for a standard flow!"
279		)
280
281
282		class FlowBlock(ScalarBlock):
283		r"""Flow block with definitions for standard flows.
284
285		See :class:`~oemof.solph.flows._flow.Flow` class for all parameters of the Flow.
286
287		Variables
288
289		All Flow objects are indexed by a starting and ending node
290		:math:`(i, o)`, which is omitted in the following for the sake of
291		convenience. The creation of some variables depend on the values of
292		Flow attributes. The following variables are created:
293
294		* :math:`P(t)`
295		Actual flow value (created in :class:`~oemof.solph._models.Model`).
296		The variable is bound to: :math:`f_{min}(t) \cdot P_{nom} \ge P(t) \le f_{max}(t) \cdot P_{nom}`.
297
298		If `Flow.fix` is not None the variable is bound to
299		:math:`P(t) = f_{fix}`.
300
301		* :math:`ve_n` (`Flow.negative_gradient` is not `None`)
302		Difference of a flow in consecutive timesteps if flow is reduced. The
303		variable is bound to: :math:`0 \ge ve_n \ge ve_n^{max}`.
304
305		* :math:`ve_p` (`Flow.positive_gradient` is not `None`)
306		Difference of a flow in consecutive timesteps if flow is increased. The
307		variable is bound to: :math:`0 \ge ve_p \ge ve_p^{max}`.
308
309		The following variable is build for Flows with the attribute
310		`integer_flows` being not None.
311
312		* :math:`i`(`Flow.integer` is `True`)
313		All flow values are integers. Variable is bound to non-negative
314		integers.
315
316		Constraints
317
318		The following constraints are created, if the appropriate attribute of the
319		Flow (see :class:`oemof.solph.network.Flow`) object is set:
320
321		* `Flow.full_load_time_max` is not `None` (full_load_time_max_constr):
322		.. math::
323		\sum_t P(t) \cdot \tau \leq F_{max} \cdot P_{nom}
324
325		* `Flow.full_load_time_min` is not `None` (full_load_time_min_constr):
326		.. math::
327		\sum_t P(t) \cdot \tau \geq F_{min} \cdot P_{nom}
328
329
330		* `Flow.negative_gradient` is not `None` (negative_gradient_constr):
331		.. math::
332		P(t-1) - P(t) \geq ve_n(t)
333
334		* `Flow.positive_gradient` is not `None` (positive_gradient_constr):
335		.. math::
336		P(t) - P(t-1) \geq ve_p(t)
337
338		* `Flow.integer` is `True`
339		.. math::
340		P(t) = i(t)
341
342		Objective function
343
344		Depending on the attributes of the `Flow` object the following parts of
345		the objective function are created:
346
347		* `Flow.variable_costs` is not `None`:
348		.. math::
349		\sum_{(i,o)} \sum_t P(t) \cdot c_{var}(i, o, t)
350
351		.. csv-table:: List of Variables
352		:header: "symbol", "attribute", "explanation"
353		:widths: 1, 1, 1
354
355		":math:`P(t)`", ":command:`flow[i, o][t]`", "Actual flow value"
356		":math:`ve_n`", ":command:`negative_gradient[n, o, t]`", "Negative gradient of the flow"
357		":math:`ve_p`", ":command:`positive_gradient[n, o, t]`", "Positive gradient of the flow"
358		":math:`i`", ":command:`integer_flow[i, o, t]`","Integer flow"
359
360
361		.. csv-table:: List of Parameters
362		:header: "symbol", "attribute", "explanation"
363		:widths: 1, 1, 1
364
365		":math:`P_{nom}`", ":command:`flows[i, o].nominal_value`","Nominal value of the flow"
366		":math:`F_{max}`",":command:`flow[i, o].full_load_time_max`", "Maximal full
367		load time"
368		":math:`F_{min}`",":command:`flow[i, o].full_load_time_min`", "Minimal full
369		load time"
370		":math:`c_{var}`", ":command:`variable\_costs[t]`", "Variable cost of the flow"
371		":math:`f_{max}`", ":command:`flows[i, o].max[t]`", "Normed maximum value of the flow, the absolute maximum is :math:`f_{max} \cdot P_{nom}`"
372		":math:`f_{min}`", ":command:`flows[i, o].min[t]`", "Normed minimum value of the flow, the absolute minimum is :math:`f_{min} \cdot P_{nom}`"
373		":math:`f_{fix}`", ":command:`flows[i, o].min[t]`", "Normed fixed value of the flow, the absolute fixed value is :math:`f_{fix} \cdot P_{nom}`"
374		":math:`ve_n^{max}`",":command:`flows[i, o].negative_gradient`","Normed maximal negative gradient of the flow, the absolute maximum gradient is :math:`ve_n^{max} \cdot P_{nom}`"
375		":math:`ve_p^{max}`",":command:`flows[i, o].positive_gradient`","Normed maximal positive gradient of the flow, the absolute maximum gradient is :math:`ve_n^{max} \cdot P_{nom}`"
376
377		Note
378		----
379		See the :class:`~oemof.solph.flows._flow.Flow` class for the definition of
380		all parameters from the "List of Parameters above.
381
382		""" # noqa: E501
383
384		def __init__(self, args, *kwargs):
385		super().__init__(args, *kwargs)
386
387		def _create(self, group=None):
388		r"""Creates sets, variables and constraints for all standard flows.
389
390		Parameters
391		----------
392		group : list
393		List containing tuples containing flow (f) objects and the
394		associated source (s) and target (t)
395		of flow e.g. groups=[(s1, t1, f1), (s2, t2, f2),..]
396		"""
397		if group is None:
398		return None
399
400		m = self.parent_block()
401
402		# ########################## SETS #################################
403		# set for all flows with an global limit on the flow over time
404		self.FULL_LOAD_TIME_MAX_FLOWS = Set(
405		initialize=[
406		(g[0], g[1])
407		for g in group
408		if g[2].full_load_time_max is not None
409		and g[2].nominal_value is not None
410		]
411		)
412
413		self.FULL_LOAD_TIME_MIN_FLOWS = Set(
414		initialize=[
415		(g[0], g[1])
416		for g in group
417		if g[2].full_load_time_min is not None
418		and g[2].nominal_value is not None
419		]
420		)
421
422		self.NEGATIVE_GRADIENT_FLOWS = Set(
423		initialize=[
424		(g[0], g[1])
425		for g in group
426		if g[2].negative_gradient["ub"][0] is not None
427		]
428		)
429
430		self.POSITIVE_GRADIENT_FLOWS = Set(
431		initialize=[
432		(g[0], g[1])
433		for g in group
434		if g[2].positive_gradient["ub"][0] is not None
435		]
436		)
437
438		self.INTEGER_FLOWS = Set(
439		initialize=[(g[0], g[1]) for g in group if g[2].integer]
440		)
441		# ######################### Variables ################################
442
443		self.positive_gradient = Var(self.POSITIVE_GRADIENT_FLOWS, m.TIMESTEPS)
444
445		self.negative_gradient = Var(self.NEGATIVE_GRADIENT_FLOWS, m.TIMESTEPS)
446
447		self.integer_flow = Var(
448		self.INTEGER_FLOWS, m.TIMESTEPS, within=NonNegativeIntegers
449		)
450		# set upper bound of gradient variable
451		for i, o, f in group:
452		if m.flows[i, o].positive_gradient["ub"][0] is not None:
453		for t in m.TIMESTEPS:
454		self.positive_gradient[i, o, t].setub(
455		f.positive_gradient["ub"][t] * f.nominal_value
456		)
457		if m.flows[i, o].negative_gradient["ub"][0] is not None:
458		for t in m.TIMESTEPS:
459		self.negative_gradient[i, o, t].setub(
460		f.negative_gradient["ub"][t] * f.nominal_value
461		)
462
463		# ######################### CONSTRAINTS ###############################
464
465		def _flow_full_load_time_max_rule(model):
466		"""Rule definition for build action of max. sum flow constraint."""
467		for inp, out in self.FULL_LOAD_TIME_MAX_FLOWS:
468		lhs = sum(
469		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...
470		for ts in m.TIMESTEPS
471		)
472		rhs = (
473		m.flows[inp, out].full_load_time_max
474		* m.flows[inp, out].nominal_value
475		)
476		self.full_load_time_max_constr.add((inp, out), lhs <= rhs)
477
478		self.full_load_time_max_constr = Constraint(
479		self.FULL_LOAD_TIME_MAX_FLOWS, noruleinit=True
480		)
481		self.full_load_time_max_build = BuildAction(
482		rule=_flow_full_load_time_max_rule
483		)
484
485		def _flow_full_load_time_min_rule(model):
486		"""Rule definition for build action of min. sum flow constraint."""
487		for inp, out in self.FULL_LOAD_TIME_MIN_FLOWS:
488		lhs = sum(
489		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...
490		for ts in m.TIMESTEPS
491		)
492		rhs = (
493		m.flows[inp, out].full_load_time_min
494		* m.flows[inp, out].nominal_value
495		)
496		self.full_load_time_min_constr.add((inp, out), lhs >= rhs)
497
498		self.full_load_time_min_constr = Constraint(
499		self.FULL_LOAD_TIME_MIN_FLOWS, noruleinit=True
500		)
501		self.full_load_time_min_build = BuildAction(
502		rule=_flow_full_load_time_min_rule
503		)
504
505		def _positive_gradient_flow_rule(model):
506		"""Rule definition for positive gradient constraint."""
507		for inp, out in self.POSITIVE_GRADIENT_FLOWS:
508		for ts in m.TIMESTEPS:
		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...
509		if ts > 0:
510		lhs = m.flow[inp, out, ts] - m.flow[inp, out, ts - 1]
511		rhs = self.positive_gradient[inp, out, ts]
512		self.positive_gradient_constr.add(
513		(inp, out, ts), lhs <= rhs
514		)
515
516		self.positive_gradient_constr = Constraint(
517		self.POSITIVE_GRADIENT_FLOWS, m.TIMESTEPS, noruleinit=True
518		)
519		self.positive_gradient_build = BuildAction(
520		rule=_positive_gradient_flow_rule
521		)
522
523		def _negative_gradient_flow_rule(model):
524		"""Rule definition for negative gradient constraint."""
525		for inp, out in self.NEGATIVE_GRADIENT_FLOWS:
526		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...
527		if ts > 0:
528		lhs = m.flow[inp, out, ts - 1] - m.flow[inp, out, ts]
529		rhs = self.negative_gradient[inp, out, ts]
530		self.negative_gradient_constr.add(
531		(inp, out, ts), lhs <= rhs
532		)
533
534		self.negative_gradient_constr = Constraint(
535		self.NEGATIVE_GRADIENT_FLOWS, m.TIMESTEPS, noruleinit=True
536		)
537		self.negative_gradient_build = BuildAction(
538		rule=_negative_gradient_flow_rule
539		)
540
541		def _integer_flow_rule(block, ii, oi, ti):
542		"""Force flow variable to NonNegativeInteger values."""
543		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...
544
545		self.integer_flow_constr = Constraint(
546		self.INTEGER_FLOWS, m.TIMESTEPS, rule=_integer_flow_rule
547		)
548
549		def _objective_expression(self):
550		r"""Objective expression for all standard flows with fixed costs
551		and variable costs.
552		"""
553		m = self.parent_block()
554
555		variable_costs = 0
556
557		for i, o in m.FLOWS:
558		if m.flows[i, o].variable_costs[0] is not None:
559		for t in m.TIMESTEPS:
560		variable_costs += (
561		m.flow[i, o, t]
562		* m.objective_weighting[t]
563		* m.flows[i, o].variable_costs[t]
564		)
565
566		return variable_costs
567

oemof / oemof-solph

Pull Request — dev (#836)

solph.flows._flow.FlowBlock._create() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like