solph.constraints.investment_limit.additional_total_limit() - Code Metrics - Inspection of "Feature/custom attributes for investments" - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#1174)

unknown

created 2025-09-17 14:08 UTC

additional_total_limit() C

↳ Parent: solph.constraints.investment_limit

Complexity

Conditions

Size

Total Lines	159
Code Lines	56

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	56
dl	0
loc	159
rs	6.1066
c	0
b	0
f	0
cc	9
nop	3

How to fix Long Method

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

"""Limits for investments.

SPDX-FileCopyrightText: Uwe Krien <[email protected]>
SPDX-FileCopyrightText: Simon Hilpert
SPDX-FileCopyrightText: Patrik Schönfeldt
SPDX-FileCopyrightText: Johannes Röder
SPDX-FileCopyrightText: Johannes Kochems
SPDX-FileCopyrightText: Johannes Giehl

SPDX-License-Identifier: MIT

"""

from pyomo import environ as po

from oemof.solph._plumbing import sequence


def investment_limit(model, limit=None):
    r"""Set an absolute limit for the total investment costs of an investment
    optimization problem (over all periods in case of a multi-period model):

    .. math:: \sum_{investment\_costs} \leq limit

    Parameters
    ----------
    model : oemof.solph._models.Model
        Model to which the constraint is added
    limit : float
        Absolute limit of the investment (i.e. RHS of constraint)
    """

    def investment_rule(m):
        expr = 0

        if hasattr(m, "InvestmentFlowBlock"):
            expr += m.InvestmentFlowBlock.investment_costs

        if hasattr(m, "GenericInvestmentStorageBlock"):
            expr += m.GenericInvestmentStorageBlock.investment_costs

        if hasattr(m, "SinkDSMOemofInvestmentBlock"):
            expr += m.SinkDSMOemofInvestmentBlock.investment_costs

        if hasattr(m, "SinkDSMDIWInvestmentBlock"):
            expr += m.SinkDSMDIWInvestmentBlock.investment_costs

        if hasattr(m, "SinkDSMDLRInvestmentBlock"):
            expr += m.SinkDSMDLRInvestmentBlock.investment_costs

        return expr <= limit

    model.investment_limit = po.Constraint(rule=investment_rule)

    return model


def investment_limit_per_period(model, limit=None):
    r"""Set an absolute limit for the total investment costs of a
    investment optimization problem for each period
    of the multi-period problem.

    .. math::
        \sum_{investment\_costs(p)} \leq limit(p)
        \forall p \in \textrm{PERIODS}

    Parameters
    ----------
    model : oemof.solph.Model
        Model to which the constraint is added
    limit : sequence of float, :math:`limit(p)`
        Absolute limit of the investment for each period
        (i.e. RHS of constraint)
    """

    if model.es.periods is None:
        msg = (
            "investment_limit_per_period is only applicable "
            "for multi-period models.\nIn order to create such a model, "
            "explicitly set attribute `periods` of your energy system."
        )
        raise ValueError(msg)

    if limit is not None:
        limit = sequence(limit)
    else:
        msg = (
            "You have to provide an investment limit for each period!\n"
            "If you provide a scalar value, this will be applied as a "
            "limit for each period."
        )
        raise ValueError(msg)

    def investment_period_rule(m, p):
        expr = 0

        if hasattr(m, "InvestmentFlowBlock"):
            expr += m.InvestmentFlowBlock.period_investment_costs[p]

        if hasattr(m, "GenericInvestmentStorageBlock"):
            expr += m.GenericInvestmentStorageBlock.period_investment_costs[p]

        if hasattr(m, "SinkDSMOemofInvestmentBlock"):
            expr += m.SinkDSMOemofInvestmentBlock.period_investment_costs[p]

        if hasattr(m, "SinkDSMDIWInvestmentBlock"):
            expr += m.SinkDSMDIWInvestmentBlock.period_investment_costs[p]

        if hasattr(m, "SinkDSMDLRInvestmentBlock"):
            expr += m.SinkDSMDLRInvestmentBlock.period_investment_costs[p]

        return expr <= limit[p]

    model.investment_limit_per_period = po.Constraint(
        model.PERIODS, rule=investment_period_rule
    )

    return model


def additional_investment_flow_limit(model, keyword, limit=None):
    r"""
    Global limit for investment flows weighted by an attribute keyword.

    This constraint is only valid for Flows not for components such as an
    investment storage.

    The attribute named by keyword has to be added to every Investment
    attribute of the flow you want to take into account.
    Total value of keyword attributes after optimization can be retrieved
    calling the `oemof.solph._models.Model.invest_limit_${keyword}()`.

    .. math::
        \sum_{p \in \textrm{PERIODS}}
        \sum_{i \in IF}  P_{i}(p) \cdot w_i \leq limit

    With `IF` being the set of InvestmentFlows considered for the integral
    limit.

    The symbols used are defined as follows
    (with Variables (V) and Parameters (P)):

    +------------------+---------------------------------------+------+--------------------------------------------------------------+
    | symbol           | attribute                             | type | explanation                                                  |
    +==================+=======================================+======+==============================================================+
    | :math:`P_{i}(p)` | `InvestmentFlowBlock.invest[i, o, p]` | V    | invested capacity of investment flow in period p             |
    +------------------+---------------------------------------+------+--------------------------------------------------------------+
    | :math:`w_i`      | `keyword`                             | P    | weight given to investment flow named according to `keyword` |
    +------------------+---------------------------------------+------+--------------------------------------------------------------+
    | :math:`limit`    | `limit`                               | P    | global limit given by keyword `limit`                        |
    +------------------+---------------------------------------+------+--------------------------------------------------------------+

    Parameters
    ----------
    model : oemof.solph.Model
        Model to which constraints are added.
    keyword : attribute to consider
        All flows with Investment attribute containing the keyword will be
        used.
    limit : numeric
        Global limit of keyword attribute for the energy system.

    Note
    ----
    The Investment attribute of the considered (Investment-)flows requires an
    attribute named like keyword!

    Examples
    --------
    >>> import pandas as pd
    >>> from oemof import solph
    >>> date_time_index = pd.date_range('1/1/2020', periods=6, freq='h')
    >>> es = solph.EnergySystem(
    ...     timeindex=date_time_index,
    ...     infer_last_interval=False,
    ... )
    >>> bus = solph.buses.Bus(label='bus_1')
    >>> sink = solph.components.Sink(label="sink", inputs={bus:
    ...     solph.flows.Flow(nominal_capacity=10, fix=[10, 20, 30, 40, 50])})
    >>> src1 = solph.components.Source(
    ...     label='source_0', outputs={bus: solph.flows.Flow(
    ...         nominal_capacity=solph.Investment(
    ...             ep_costs=50, custom_attributes={"space": 4},
    ...         ))
    ...     })
    >>> src2 = solph.components.Source(
    ...     label='source_1', outputs={bus: solph.flows.Flow(
    ...         nominal_capacity=solph.Investment(
    ...              ep_costs=100, custom_attributes={"space": 1},
    ...         ))
    ...     })
    >>> es.add(bus, sink, src1, src2)
    >>> model = solph.Model(es)
    >>> model = solph.constraints.additional_investment_flow_limit(
    ...     model, "space", limit=1500)
    >>> a = model.solve(solver="cbc")
    >>> int(round(model.invest_limit_space()))
    1500
    """  # noqa: E501
    invest_flows = {}

    for i, o in model.flows:
        if hasattr(model.flows[i, o].investment, keyword):
            invest_flows[(i, o)] = model.flows[i, o].investment

    limit_name = "invest_limit_" + keyword

    setattr(
        model,
        limit_name,
        po.Expression(
            expr=sum(
                model.InvestmentFlowBlock.invest[inflow, outflow, p]
                * getattr(invest_flows[inflow, outflow], keyword)
                for (inflow, outflow) in invest_flows
                for p in model.PERIODS
            )
        ),
    )

    setattr(
        model,
        limit_name + "_constraint",
        po.Constraint(expr=(getattr(model, limit_name) <= limit)),
    )

    return model


def additional_total_limit(model, keyword, limit=None):
    r"""
    Global limit for investment flows and operation flows
    weighted by an attribute keyword.

    This constraint is  valid for Flows and for an
    investment storage.

    The attribute named by keyword has to be added to every Investment
    attribute of the flow you want to take into account.
    Total value of keyword attributes after optimization can be retrieved
    calling the `oemof.solph._models.Model.total_limit_${keyword}()`.

    .. math::
        \sum_{p \in \textrm{PERIODS}}
        \sum_{i \in IF}  P_{i}(p) \cdot w_i
        \sum_{i \in F_E} \sum_{t \in T} P_i(p, t) \cdot w_i(t)
               \cdot \tau(t) \leq limit

    With `IF` being the set of InvestmentFlows considered for the integral
    limit,  `F_I` being the set of flows considered for the integral limit and
    `T` being the set of time steps.

    The symbols used are defined as follows
    (with Variables (V) and Parameters (P)):

    +------------------+---------------------------------------+------+--------------------------------------------------------------+
    | symbol           | attribute                             | type | explanation                                                  |
    +==================+=======================================+======+==============================================================+
    | :math:`P_{i}(p)` | `InvestmentFlowBlock.invest[i, o, p]` | V    | invested capacity of investment flow in period p             |
    +------------------+---------------------------------------+------+--------------------------------------------------------------+
    | :math:`w_i`      | `keyword`                             | P    | weight given to investment flow named according to `keyword` |
    +------------------+---------------------------------------+------+--------------------------------------------------------------+
    | :math:`limit`    | `limit`                               | P    | global limit given by keyword `limit`                        |
    +------------------+---------------------------------------+------+--------------------------------------------------------------+
    | :math:`P_n(p, t)`| `limit`                               | P    | power flow :math:`n` at time index :math:`p, t`              |
    +------------------+---------------------------------------+------+--------------------------------------------------------------+
    | :math:`w_N(t)`   | `limit`                               | P    | weight given to Flow named according to `keyword`            |
    +------------------+---------------------------------------+------+--------------------------------------------------------------+
    | :math:`\tau(t)`  | `limit`                               | P    | width of time step :math:`t`                                 |
    +------------------+---------------------------------------+------+--------------------------------------------------------------+

    Parameters
    ----------
    model : oemof.solph.Model
        Model to which constraints are added.
    keyword : attribute to consider
        All flows with Investment attribute containing the keyword will be
        used.
    limit : numeric
        Global limit of keyword attribute for the energy system.

    Note
    ----
    The Investment attribute of the considered (Investment-)flows requires an
    attribute named like keyword!

    Examples
    --------
    >>> import pandas as pd
    >>> from oemof import solph
    >>> date_time_index = pd.date_range('1/1/2020', periods=6, freq='h')
    >>> es = solph.EnergySystem(
    ...     timeindex=date_time_index,
    ...     infer_last_interval=False,
    ... )
    >>> bus = solph.buses.Bus(label='bus_1')
    >>> sink = solph.components.Sink(label="sink", inputs={bus:
    ...     solph.flows.Flow(nominal_capacity=10, fix=[10, 20, 30, 40, 50])})
    >>> src1 = solph.components.Source(
    ...     label='source_0', outputs={bus: solph.flows.Flow(
    ...         nominal_capacity=solph.Investment(
    ...             ep_costs=50, custom_attributes={"space": 4},
    ...         ))
    ...     })
    >>> src2 = solph.components.Source(
    ...     label='source_1', outputs={bus: solph.flows.Flow(
    ...         nominal_capacity=solph.Investment(
    ...              ep_costs=100, custom_attributes={"space": 1},
    ...         ))
    ...     })
    >>> es.add(bus, sink, src1, src2)
    >>> model = solph.Model(es)
    >>> model = solph.constraints.additional_investment_flow_limit(
    ...     model, "space", limit=1500)
    >>> a = model.solve(solver="cbc")
    >>> int(round(model.invest_limit_space()))
    1500
    """  # noqa: E501
    invest_flows = {}
    operational_flows = {}
    storages = {}
    for i, o in model.flows:
        if hasattr(model.flows[i, o].investment, keyword):
            invest_flows[(i, o)] = model.flows[i, o].investment
        if hasattr(model.flows[i, o], keyword):
            operational_flows[(i, o)] = model.flows[i, o]
    limit_name = "total_limit_" + keyword

    if hasattr(model, "GenericInvestmentStorageBlock"):
        for st, _ in model.GenericInvestmentStorageBlock.invest:
            storages[st] = [st]

    setattr(
        model,
        limit_name,
        po.Expression(
            rule=lambda m: sum(
                model.InvestmentFlowBlock.invest[inflow, outflow, p]
                * getattr(invest_flows[inflow, outflow], keyword).get(
                    "cost", 0
                )
                + (
                    model.InvestmentFlowBlock.invest_status[inflow, outflow, p]
                    * getattr(invest_flows[inflow, outflow], keyword).get("offset", 0)
                    if (inflow, outflow, p)
                    in model.InvestmentFlowBlock.invest_status
                    else 0
                )
                for (inflow, outflow) in invest_flows
                for p in model.PERIODS
            ) +
            sum(
                model.flow[inflow, outflow, t]
                * model.tsam_weighting[t]
                * model.timeincrement[t]
                * sequence(
                    getattr(operational_flows[inflow, outflow], keyword)
                )[t]
                for (inflow, outflow) in operational_flows
                for p in model.PERIODS
                for t in model.TIMESTEPS_IN_PERIOD[p]
            ) +
            sum(
                model.GenericInvestmentStorageBlock.invest[st, p]
                * getattr(storages[st][0].investment, keyword).get(
                    "cost", 0
                )
                + model.GenericInvestmentStorageBlock.invest_status[st, p]
                * getattr(storages[st][0].investment, keyword).get(
                    "offset", 0
                )
                if (st, p)
                in model.GenericInvestmentStorageBlock.invest_status
                else 0
                for st in storages
                for p in model.PERIODS
                if hasattr(model, "GenericInvestmentStorageBlock")
            )
        ),
    )

    setattr(
        model,
        limit_name + "_constraint",
        po.Constraint(expr=(getattr(model, limit_name) <= limit)),
    )

    return model


1			# -- coding: utf-8 --
2
3			"""Limits for investments.
4
5			SPDX-FileCopyrightText: Uwe Krien <[email protected]>
6			SPDX-FileCopyrightText: Simon Hilpert
7			SPDX-FileCopyrightText: Patrik Schönfeldt
8			SPDX-FileCopyrightText: Johannes Röder
9			SPDX-FileCopyrightText: Johannes Kochems
10			SPDX-FileCopyrightText: Johannes Giehl
11
12			SPDX-License-Identifier: MIT
13
14			"""
15
16			from pyomo import environ as po
17
18			from oemof.solph._plumbing import sequence
19
20
21			def investment_limit(model, limit=None):
22			r"""Set an absolute limit for the total investment costs of an investment
23			optimization problem (over all periods in case of a multi-period model):
24
25			.. math:: \sum_{investment\_costs} \leq limit
26
27			Parameters
28			----------
29			model : oemof.solph._models.Model
30			Model to which the constraint is added
31			limit : float
32			Absolute limit of the investment (i.e. RHS of constraint)
33			"""
34
35			def investment_rule(m):
36			expr = 0
37
38			if hasattr(m, "InvestmentFlowBlock"):
39			expr += m.InvestmentFlowBlock.investment_costs
40
41			if hasattr(m, "GenericInvestmentStorageBlock"):
42			expr += m.GenericInvestmentStorageBlock.investment_costs
43
44			if hasattr(m, "SinkDSMOemofInvestmentBlock"):
45			expr += m.SinkDSMOemofInvestmentBlock.investment_costs
46
47			if hasattr(m, "SinkDSMDIWInvestmentBlock"):
48			expr += m.SinkDSMDIWInvestmentBlock.investment_costs
49
50			if hasattr(m, "SinkDSMDLRInvestmentBlock"):
51			expr += m.SinkDSMDLRInvestmentBlock.investment_costs
52
53			return expr <= limit
54
55			model.investment_limit = po.Constraint(rule=investment_rule)
56
57			return model
58
59
60			def investment_limit_per_period(model, limit=None):
61			r"""Set an absolute limit for the total investment costs of a
62			investment optimization problem for each period
63			of the multi-period problem.
64
65			.. math::
66			\sum_{investment\_costs(p)} \leq limit(p)
67			\forall p \in \textrm{PERIODS}
68
69			Parameters
70			----------
71			model : oemof.solph.Model
72			Model to which the constraint is added
73			limit : sequence of float, :math:`limit(p)`
74			Absolute limit of the investment for each period
75			(i.e. RHS of constraint)
76			"""
77
78			if model.es.periods is None:
79			msg = (
80			"investment_limit_per_period is only applicable "
81			"for multi-period models.\nIn order to create such a model, "
82			"explicitly set attribute `periods` of your energy system."
83			)
84			raise ValueError(msg)
85
86			if limit is not None:
87			limit = sequence(limit)
88			else:
89			msg = (
90			"You have to provide an investment limit for each period!\n"
91			"If you provide a scalar value, this will be applied as a "
92			"limit for each period."
93			)
94			raise ValueError(msg)
95
96			def investment_period_rule(m, p):
97			expr = 0
98
99			if hasattr(m, "InvestmentFlowBlock"):
100			expr += m.InvestmentFlowBlock.period_investment_costs[p]
101
102			if hasattr(m, "GenericInvestmentStorageBlock"):
103			expr += m.GenericInvestmentStorageBlock.period_investment_costs[p]
104
105			if hasattr(m, "SinkDSMOemofInvestmentBlock"):
106			expr += m.SinkDSMOemofInvestmentBlock.period_investment_costs[p]
107
108			if hasattr(m, "SinkDSMDIWInvestmentBlock"):
109			expr += m.SinkDSMDIWInvestmentBlock.period_investment_costs[p]
110
111			if hasattr(m, "SinkDSMDLRInvestmentBlock"):
112			expr += m.SinkDSMDLRInvestmentBlock.period_investment_costs[p]
113
114			return expr <= limit[p]
115
116			model.investment_limit_per_period = po.Constraint(
117			model.PERIODS, rule=investment_period_rule
118			)
119
120			return model
121
122
123			def additional_investment_flow_limit(model, keyword, limit=None):
124			r"""
125			Global limit for investment flows weighted by an attribute keyword.
126
127			This constraint is only valid for Flows not for components such as an
128			investment storage.
129
130			The attribute named by keyword has to be added to every Investment
131			attribute of the flow you want to take into account.
132			Total value of keyword attributes after optimization can be retrieved
133			calling the `oemof.solph._models.Model.invest_limit_${keyword}()`.
134
135			.. math::
136			\sum_{p \in \textrm{PERIODS}}
137			\sum_{i \in IF} P_{i}(p) \cdot w_i \leq limit
138
139			With `IF` being the set of InvestmentFlows considered for the integral
140			limit.
141
142			The symbols used are defined as follows
143			(with Variables (V) and Parameters (P)):
144
145			+------------------+---------------------------------------+------+--------------------------------------------------------------+
146			\| symbol \| attribute \| type \| explanation \|
147			+==================+=======================================+======+==============================================================+
148			\| :math:`P_{i}(p)` \| `InvestmentFlowBlock.invest[i, o, p]` \| V \| invested capacity of investment flow in period p \|
149			+------------------+---------------------------------------+------+--------------------------------------------------------------+
150			\| :math:`w_i` \| `keyword` \| P \| weight given to investment flow named according to `keyword` \|
151			+------------------+---------------------------------------+------+--------------------------------------------------------------+
152			\| :math:`limit` \| `limit` \| P \| global limit given by keyword `limit` \|
153			+------------------+---------------------------------------+------+--------------------------------------------------------------+
154
155			Parameters
156			----------
157			model : oemof.solph.Model
158			Model to which constraints are added.
159			keyword : attribute to consider
160			All flows with Investment attribute containing the keyword will be
161			used.
162			limit : numeric
163			Global limit of keyword attribute for the energy system.
164
165			Note
166			----
167			The Investment attribute of the considered (Investment-)flows requires an
168			attribute named like keyword!
169
170			Examples
171			--------
172			>>> import pandas as pd
173			>>> from oemof import solph
174			>>> date_time_index = pd.date_range('1/1/2020', periods=6, freq='h')
175			>>> es = solph.EnergySystem(
176			... timeindex=date_time_index,
177			... infer_last_interval=False,
178			... )
179			>>> bus = solph.buses.Bus(label='bus_1')
180			>>> sink = solph.components.Sink(label="sink", inputs={bus:
181			... solph.flows.Flow(nominal_capacity=10, fix=[10, 20, 30, 40, 50])})
182			>>> src1 = solph.components.Source(
183			... label='source_0', outputs={bus: solph.flows.Flow(
184			... nominal_capacity=solph.Investment(
185			... ep_costs=50, custom_attributes={"space": 4},
186			... ))
187			... })
188			>>> src2 = solph.components.Source(
189			... label='source_1', outputs={bus: solph.flows.Flow(
190			... nominal_capacity=solph.Investment(
191			... ep_costs=100, custom_attributes={"space": 1},
192			... ))
193			... })
194			>>> es.add(bus, sink, src1, src2)
195			>>> model = solph.Model(es)
196			>>> model = solph.constraints.additional_investment_flow_limit(
197			... model, "space", limit=1500)
198			>>> a = model.solve(solver="cbc")
199			>>> int(round(model.invest_limit_space()))
200			1500
201			""" # noqa: E501
202			invest_flows = {}
203
204			for i, o in model.flows:
205			if hasattr(model.flows[i, o].investment, keyword):
206			invest_flows[(i, o)] = model.flows[i, o].investment
207
208			limit_name = "invest_limit_" + keyword
209
210			setattr(
211			model,
212			limit_name,
213			po.Expression(
214			expr=sum(
215			model.InvestmentFlowBlock.invest[inflow, outflow, p]
216			* getattr(invest_flows[inflow, outflow], keyword)
217			for (inflow, outflow) in invest_flows
218			for p in model.PERIODS
219			)
220			),
221			)
222
223			setattr(
224			model,
225			limit_name + "_constraint",
226			po.Constraint(expr=(getattr(model, limit_name) <= limit)),
227			)
228
229			return model
230
231
232			def additional_total_limit(model, keyword, limit=None):
233			r"""
234			Global limit for investment flows and operation flows
235			weighted by an attribute keyword.
236
237			This constraint is valid for Flows and for an
238			investment storage.
239
240			The attribute named by keyword has to be added to every Investment
241			attribute of the flow you want to take into account.
242			Total value of keyword attributes after optimization can be retrieved
243			calling the `oemof.solph._models.Model.total_limit_${keyword}()`.
244
245			.. math::
246			\sum_{p \in \textrm{PERIODS}}
247			\sum_{i \in IF} P_{i}(p) \cdot w_i
248			\sum_{i \in F_E} \sum_{t \in T} P_i(p, t) \cdot w_i(t)
249			\cdot \tau(t) \leq limit
250
251			With `IF` being the set of InvestmentFlows considered for the integral
252			limit, `F_I` being the set of flows considered for the integral limit and
253			`T` being the set of time steps.
254
255			The symbols used are defined as follows
256			(with Variables (V) and Parameters (P)):
257
258			+------------------+---------------------------------------+------+--------------------------------------------------------------+
259			\| symbol \| attribute \| type \| explanation \|
260			+==================+=======================================+======+==============================================================+
261			\| :math:`P_{i}(p)` \| `InvestmentFlowBlock.invest[i, o, p]` \| V \| invested capacity of investment flow in period p \|
262			+------------------+---------------------------------------+------+--------------------------------------------------------------+
263			\| :math:`w_i` \| `keyword` \| P \| weight given to investment flow named according to `keyword` \|
264			+------------------+---------------------------------------+------+--------------------------------------------------------------+
265			\| :math:`limit` \| `limit` \| P \| global limit given by keyword `limit` \|
266			+------------------+---------------------------------------+------+--------------------------------------------------------------+
267			\| :math:`P_n(p, t)`\| `limit` \| P \| power flow :math:`n` at time index :math:`p, t` \|
268			+------------------+---------------------------------------+------+--------------------------------------------------------------+
269			\| :math:`w_N(t)` \| `limit` \| P \| weight given to Flow named according to `keyword` \|
270			+------------------+---------------------------------------+------+--------------------------------------------------------------+
271			\| :math:`\tau(t)` \| `limit` \| P \| width of time step :math:`t` \|
272			+------------------+---------------------------------------+------+--------------------------------------------------------------+
273
274			Parameters
275			----------
276			model : oemof.solph.Model
277			Model to which constraints are added.
278			keyword : attribute to consider
279			All flows with Investment attribute containing the keyword will be
280			used.
281			limit : numeric
282			Global limit of keyword attribute for the energy system.
283
284			Note
285			----
286			The Investment attribute of the considered (Investment-)flows requires an
287			attribute named like keyword!
288
289			Examples
290			--------
291			>>> import pandas as pd
292			>>> from oemof import solph
293			>>> date_time_index = pd.date_range('1/1/2020', periods=6, freq='h')
294			>>> es = solph.EnergySystem(
295			... timeindex=date_time_index,
296			... infer_last_interval=False,
297			... )
298			>>> bus = solph.buses.Bus(label='bus_1')
299			>>> sink = solph.components.Sink(label="sink", inputs={bus:
300			... solph.flows.Flow(nominal_capacity=10, fix=[10, 20, 30, 40, 50])})
301			>>> src1 = solph.components.Source(
302			... label='source_0', outputs={bus: solph.flows.Flow(
303			... nominal_capacity=solph.Investment(
304			... ep_costs=50, custom_attributes={"space": 4},
305			... ))
306			... })
307			>>> src2 = solph.components.Source(
308			... label='source_1', outputs={bus: solph.flows.Flow(
309			... nominal_capacity=solph.Investment(
310			... ep_costs=100, custom_attributes={"space": 1},
311			... ))
312			... })
313			>>> es.add(bus, sink, src1, src2)
314			>>> model = solph.Model(es)
315			>>> model = solph.constraints.additional_investment_flow_limit(
316			... model, "space", limit=1500)
317			>>> a = model.solve(solver="cbc")
318			>>> int(round(model.invest_limit_space()))
319			1500
320			""" # noqa: E501
321			invest_flows = {}
322			operational_flows = {}
323			storages = {}
324			for i, o in model.flows:
325			if hasattr(model.flows[i, o].investment, keyword):
326			invest_flows[(i, o)] = model.flows[i, o].investment
327			if hasattr(model.flows[i, o], keyword):
328			operational_flows[(i, o)] = model.flows[i, o]
329			limit_name = "total_limit_" + keyword
330
331			if hasattr(model, "GenericInvestmentStorageBlock"):
332			for st, _ in model.GenericInvestmentStorageBlock.invest:
333			storages[st] = [st]
334
335			setattr(
336			model,
337			limit_name,
338			po.Expression(
339			rule=lambda m: sum(
340			model.InvestmentFlowBlock.invest[inflow, outflow, p]
341			* getattr(invest_flows[inflow, outflow], keyword).get(
342			"cost", 0
343			)
344			+ (
345			model.InvestmentFlowBlock.invest_status[inflow, outflow, p]
346			* getattr(invest_flows[inflow, outflow], keyword).get("offset", 0)
347			if (inflow, outflow, p)
348			in model.InvestmentFlowBlock.invest_status
349			else 0
350			)
351			for (inflow, outflow) in invest_flows
352			for p in model.PERIODS
353			) +
354			sum(
355			model.flow[inflow, outflow, t]
356			* model.tsam_weighting[t]
357			* model.timeincrement[t]
358			* sequence(
359			getattr(operational_flows[inflow, outflow], keyword)
360			)[t]
361			for (inflow, outflow) in operational_flows
362			for p in model.PERIODS
363			for t in model.TIMESTEPS_IN_PERIOD[p]
364			) +
365			sum(
366			model.GenericInvestmentStorageBlock.invest[st, p]
367			* getattr(storages[st][0].investment, keyword).get(
368			"cost", 0
369			)
370			+ model.GenericInvestmentStorageBlock.invest_status[st, p]
371			* getattr(storages[st][0].investment, keyword).get(
372			"offset", 0
373			)
374			if (st, p)
375			in model.GenericInvestmentStorageBlock.invest_status
376			else 0
377			for st in storages
378			for p in model.PERIODS
379			if hasattr(model, "GenericInvestmentStorageBlock")
380			)
381			),
382			)
383
384			setattr(
385			model,
386			limit_name + "_constraint",
387			po.Constraint(expr=(getattr(model, limit_name) <= limit)),
388			)
389
390			return model
391

oemof / oemof-solph

Pull Request — dev (#1174)

additional_total_limit() C

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like