emission_constraint_invest_and_flow.main() - 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-19 09:39 UTC

emission_constraint_invest_and_flow.main() B

↳ Parent: emission_constraint_invest_and_flow

Complexity

Conditions

Size

Total Lines	248
Code Lines	167

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	167
dl	0
loc	248
rs	7
c	0
b	0
f	0
cc	3
nop	1

How to fix Long Method

# -*- coding: utf-8 -*-
r"""
General description
-------------------
Example that shows how to use "Generic Investment Limit With Offset".

There are two supply chains. The energy systems looks like that:

.. code-block:: text

                  bus_a_0          bus_a_1
                   |                 |
    source_a_0 --->|---> trafo_a --->|--->demand_a
                                     |
                       source_a_1--->|
                                     |

                  bus_b_0          bus_b_1
                   |                 |
    source_b_0 --->|---> trafo_b --->|--->demand_b
                                     |
                       source_b_1--->|
                                     |
                                     v
                              generic_storage_b
                               /            \
                              |              |
                              v              v
                            bus_b_1        bus_b_1


We individualized the costs for the sources, the demand, the efficiency
of the Converter. And both Converter have an investment at the output.
The source '\*_1' is in both cases very expensive, so that
a investment is probably done in the converter. The demand and the
maximum investment in Converter b1 is so set, that an investment in a generic
storage is beneficial to not use source b0. Two generic storage are set, since
the storage b_0 is cheaper the preferred investment is there and afterward
in storage b_1.
Now, all investments share a third resource, which is called "emission" in this
example. (This could be anything, and you could use as many additional
resources as you want.) And this resource is limited. In this case, every
converter and storage capacity unit, which might be installed, needs emission
for each installed capacity as well as an offset. In addition to that the flow
through a converter has as well emission.
See what happens, have fun ;)

Code
----
Download source code: :download:`emission_constraint_invest.py </../examples/generic_invest_limit/emission_constraint_invest_and_flow.py>`

.. dropdown:: Click to display code

    .. literalinclude:: /../examples/generic_invest_limit/emission_constraint_invest_and_flow.py
        :language: python
        :lines: 71-

Installation requirements
-------------------------
This example requires oemof.solph (>=0.6.1), install by:

.. code:: bash

    pip install oemof.solph[examples]

License
-------
Maximilian Hillen <[email protected]>

`MIT license <https://github.com/oemof/oemof-solph/blob/dev/LICENSE>`_
"""

import logging
import os

from PIL.ImageChops import offset

try:
    import matplotlib.pyplot as plt
except ModuleNotFoundError:
    plt = None

from oemof import solph


def main(optimize=True):

    data = [2, 2, 12, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10]
    # create an energy system
    idx = solph.create_time_index(2020, number=len(data))
    es = solph.EnergySystem(timeindex=idx, infer_last_interval=False)

    # Parameter: costs for the sources
    c_0 = 1000
    c_1 = 10000

    epc_invest = 50

    # commodity a
    bus_a_0 = solph.Bus(label="bus_a_0")
    bus_a_1 = solph.Bus(label="bus_a_1")
    es.add(bus_a_0, bus_a_1)

    es.add(
        solph.components.Source(
            label="source_a_0",
            outputs={bus_a_0: solph.Flow(variable_costs=c_0)},
        )
    )

    es.add(
        solph.components.Source(
            label="source_a_1",
            outputs={bus_a_1: solph.Flow(variable_costs=c_1)},
        )
    )

    es.add(
        solph.components.Sink(
            label="demand_a",
            inputs={bus_a_1: solph.Flow(fix=data, nominal_capacity=1)},
        )
    )

    # commodity b
    bus_b_0 = solph.Bus(label="bus_b_0")
    bus_b_1 = solph.Bus(label="bus_b_1")
    es.add(bus_b_0, bus_b_1)
    es.add(
        solph.components.Source(
            label="source_b_0",
            outputs={bus_b_0: solph.Flow(variable_costs=c_0)},
        )
    )

    es.add(
        solph.components.Source(
            label="source_b_1",
            outputs={bus_b_1: solph.Flow(variable_costs=c_1)},
        )
    )

    es.add(
        solph.components.Sink(
            label="demand_b",
            inputs={bus_b_1: solph.Flow(fix=data, nominal_capacity=1)},
        )
    )

    # Converter a
    emission_conv_a_linear = 1
    emission_conv_a_offset = 20
    emission_conv_a_flow = 0.1
    converter_a = solph.components.Converter(
        label="trafo_a",
        inputs={bus_a_0: solph.Flow()},
        outputs={
            bus_a_1: solph.Flow(
                nominal_capacity=solph.Investment(
                    ep_costs=epc_invest,
                    custom_attributes={
                        "emission": {
                            "linear": emission_conv_a_linear,
                            "offset": emission_conv_a_offset,
                        }
                    },
                    nonconvex=True,
                    maximum=20,
                ),
                custom_attributes={"emission": emission_conv_a_flow},
            )
        },
        conversion_factors={bus_a_1: 1},
    )
    es.add(converter_a)

    # Converter b
    emission_conv_b_linear = 1
    emission_conv_b_flow = 0.1
    converter_b = solph.components.Converter(
        label="trafo_b",
        inputs={bus_b_0: solph.Flow()},
        outputs={
            bus_b_1: solph.Flow(
                nominal_capacity=solph.Investment(
                    ep_costs=epc_invest,
                    custom_attributes={
                        "emission": {"linear": emission_conv_b_linear}
                    },
                    nonconvex=True,
                    maximum=10,
                ),
                custom_attributes={"emission": emission_conv_b_flow},
            )
        },
    )
    es.add(converter_b)

    # Generic Storage b_0
    emission_storage_b_0_linear = 0.5
    emission_storage_b_0_offset = 1
    generic_storage_b_0 = solph.components.GenericStorage(
        label="generic_storage_b_0",
        inputs={bus_b_1: solph.Flow()},
        outputs={bus_b_1: solph.Flow()},
        inflow_conversion_factor=1,
        nominal_capacity=solph.Investment(
            ep_costs=epc_invest,
            nonconvex=True,
            maximum=1,
            custom_attributes={
                "emission": {
                    "linear": emission_storage_b_0_linear,
                    "offset": emission_storage_b_0_offset,
                }
            },
        ),
        invest_relation_input_capacity=0.5,
        invest_relation_output_capacity=0.5,
    )

    es.add(generic_storage_b_0)

    # Generic Storage b_1
    emission_storage_b_1_linear = 1
    emission_storage_b_1_offset = 5
    generic_storage_b_1 = solph.components.GenericStorage(
        label="generic_storage_b_1",
        inputs={bus_b_1: solph.Flow()},
        outputs={bus_b_1: solph.Flow()},
        inflow_conversion_factor=1,
        nominal_capacity=solph.Investment(
            ep_costs=epc_invest * 100,
            nonconvex=True,
            maximum=2,
            custom_attributes={
                "emission": {
                    "linear": emission_storage_b_1_linear,
                    "offset": emission_storage_b_1_offset,
                }
            },
        ),
    )

    es.add(generic_storage_b_1)
    if optimize is False:
        return es

    # create an optimization problem and solve it
    om = solph.Model(es)

    # add constraint for generic investment limit
    om = solph.constraints.additional_total_limit(om, "emission", limit=100)
    # export lp file
    filename = os.path.join(
        solph.helpers.extend_basic_path("lp_files"), "GenericInvest.lp"
    )
    logging.info("Store lp-file in {0}.".format(filename))
    om.write(filename, io_options={"symbolic_solver_labels": True})

    # solve model
    om.solve(solver="cbc", solve_kwargs={"tee": True})

    # create result object
    results = solph.processing.results(om)

    bus1 = solph.views.node(results, "bus_a_1")["sequences"]
    bus2 = solph.views.node(results, "bus_b_1")["sequences"]

    # plot the time series (sequences) of a specific component/bus
    if plt is not None:
        bus1.plot(kind="line", drawstyle="steps-mid")
        plt.legend()
        plt.show()
        bus2.plot(kind="line", drawstyle="steps-mid")
        plt.legend()
        plt.show()

    emission_used = om.total_limit_emission()
    print("emission value: ", emission_used)
    print(
        "Investment trafo_a: ",
        solph.views.node(results, "trafo_a")["scalars"][0],
    )
    print(
        "Emission investment of trafo_a: ",
        solph.views.node(results, "trafo_a")["scalars"][0]
        * emission_conv_a_linear
        + emission_conv_a_offset,
    )
    print(
        "Emission flow through trafo_a: ",
        results[converter_a, bus_a_1]["sequences"]["flow"].sum()
        * emission_conv_a_flow,
    )

    print(
        "Investment trafo_b: ",
        solph.views.node(results, "trafo_b")["scalars"][0],
    )
    print(
        "Emission investment of trafo_b: ",
        solph.views.node(results, "trafo_b")["scalars"][0]
        * emission_conv_b_linear,
    )
    print(
        "Emission flow through trafo_b: ",
        results[converter_b, bus_b_1]["sequences"]["flow"].sum()
        * emission_conv_b_flow,
    )

    print(
        "Investment generic_storage_b_0: ",
        results[generic_storage_b_0, None]["scalars"]["total"],
    )
    print(
        "Emission investment generic_storage_b_0: ",
        results[generic_storage_b_0, None]["scalars"]["total"]
        * emission_storage_b_0_linear
        + results[generic_storage_b_0, None]["scalars"]["invest_status"]
        * emission_storage_b_0_offset,
    )

    print(
        "Investment generic_storage_b_1: ",
        results[generic_storage_b_1, None]["scalars"]["total"],
    )
    print(
        "Emission investment generic_storage_b_1: ",
        results[generic_storage_b_1, None]["scalars"]["total"]
        * emission_storage_b_1_linear
        + results[generic_storage_b_1, None]["scalars"]["invest_status"]
        * emission_storage_b_1_offset,
    )


if __name__ == "__main__":
    main()


1			# -- coding: utf-8 --
2			r"""
3			General description
4			-------------------
5			Example that shows how to use "Generic Investment Limit With Offset".
6
7			There are two supply chains. The energy systems looks like that:
8
9			.. code-block:: text
10
11			bus_a_0 bus_a_1
12			\| \|
13			source_a_0 --->\|---> trafo_a --->\|--->demand_a
14			\|
15			source_a_1--->\|
16			\|
17
18			bus_b_0 bus_b_1
19			\| \|
20			source_b_0 --->\|---> trafo_b --->\|--->demand_b
21			\|
22			source_b_1--->\|
23			\|
24			v
25			generic_storage_b
26			/ \
27			\| \|
28			v v
29			bus_b_1 bus_b_1
30
31
32			We individualized the costs for the sources, the demand, the efficiency
33			of the Converter. And both Converter have an investment at the output.
34			The source '\*_1' is in both cases very expensive, so that
35			a investment is probably done in the converter. The demand and the
36			maximum investment in Converter b1 is so set, that an investment in a generic
37			storage is beneficial to not use source b0. Two generic storage are set, since
38			the storage b_0 is cheaper the preferred investment is there and afterward
39			in storage b_1.
40			Now, all investments share a third resource, which is called "emission" in this
41			example. (This could be anything, and you could use as many additional
42			resources as you want.) And this resource is limited. In this case, every
43			converter and storage capacity unit, which might be installed, needs emission
44			for each installed capacity as well as an offset. In addition to that the flow
45			through a converter has as well emission.
46			See what happens, have fun ;)
47
48			Code
49			----
50			Download source code: :download:`emission_constraint_invest.py </../examples/generic_invest_limit/emission_constraint_invest_and_flow.py>`
51
52			.. dropdown:: Click to display code
53
54			.. literalinclude:: /../examples/generic_invest_limit/emission_constraint_invest_and_flow.py
55			:language: python
56			:lines: 71-
57
58			Installation requirements
59			-------------------------
60			This example requires oemof.solph (>=0.6.1), install by:
61
62			.. code:: bash
63
64			pip install oemof.solph[examples]
65
66			License
67			-------
68			Maximilian Hillen <[email protected]>
69
70			`MIT license <https://github.com/oemof/oemof-solph/blob/dev/LICENSE>`_
71			"""
72
73			import logging
74			import os
75
76			from PIL.ImageChops import offset
77
78			try:
79			import matplotlib.pyplot as plt
80			except ModuleNotFoundError:
81			plt = None
82
83			from oemof import solph
84
85
86			def main(optimize=True):
87
88			data = [2, 2, 12, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10]
89			# create an energy system
90			idx = solph.create_time_index(2020, number=len(data))
91			es = solph.EnergySystem(timeindex=idx, infer_last_interval=False)
92
93			# Parameter: costs for the sources
94			c_0 = 1000
95			c_1 = 10000
96
97			epc_invest = 50
98
99			# commodity a
100			bus_a_0 = solph.Bus(label="bus_a_0")
101			bus_a_1 = solph.Bus(label="bus_a_1")
102			es.add(bus_a_0, bus_a_1)
103
104			es.add(
105			solph.components.Source(
106			label="source_a_0",
107			outputs={bus_a_0: solph.Flow(variable_costs=c_0)},
108			)
109			)
110
111			es.add(
112			solph.components.Source(
113			label="source_a_1",
114			outputs={bus_a_1: solph.Flow(variable_costs=c_1)},
115			)
116			)
117
118			es.add(
119			solph.components.Sink(
120			label="demand_a",
121			inputs={bus_a_1: solph.Flow(fix=data, nominal_capacity=1)},
122			)
123			)
124
125			# commodity b
126			bus_b_0 = solph.Bus(label="bus_b_0")
127			bus_b_1 = solph.Bus(label="bus_b_1")
128			es.add(bus_b_0, bus_b_1)
129			es.add(
130			solph.components.Source(
131			label="source_b_0",
132			outputs={bus_b_0: solph.Flow(variable_costs=c_0)},
133			)
134			)
135
136			es.add(
137			solph.components.Source(
138			label="source_b_1",
139			outputs={bus_b_1: solph.Flow(variable_costs=c_1)},
140			)
141			)
142
143			es.add(
144			solph.components.Sink(
145			label="demand_b",
146			inputs={bus_b_1: solph.Flow(fix=data, nominal_capacity=1)},
147			)
148			)
149
150			# Converter a
151			emission_conv_a_linear = 1
152			emission_conv_a_offset = 20
153			emission_conv_a_flow = 0.1
154			converter_a = solph.components.Converter(
155			label="trafo_a",
156			inputs={bus_a_0: solph.Flow()},
157			outputs={
158			bus_a_1: solph.Flow(
159			nominal_capacity=solph.Investment(
160			ep_costs=epc_invest,
161			custom_attributes={
162			"emission": {
163			"linear": emission_conv_a_linear,
164			"offset": emission_conv_a_offset,
165			}
166			},
167			nonconvex=True,
168			maximum=20,
169			),
170			custom_attributes={"emission": emission_conv_a_flow},
171			)
172			},
173			conversion_factors={bus_a_1: 1},
174			)
175			es.add(converter_a)
176
177			# Converter b
178			emission_conv_b_linear = 1
179			emission_conv_b_flow = 0.1
180			converter_b = solph.components.Converter(
181			label="trafo_b",
182			inputs={bus_b_0: solph.Flow()},
183			outputs={
184			bus_b_1: solph.Flow(
185			nominal_capacity=solph.Investment(
186			ep_costs=epc_invest,
187			custom_attributes={
188			"emission": {"linear": emission_conv_b_linear}
189			},
190			nonconvex=True,
191			maximum=10,
192			),
193			custom_attributes={"emission": emission_conv_b_flow},
194			)
195			},
196			)
197			es.add(converter_b)
198
199			# Generic Storage b_0
200			emission_storage_b_0_linear = 0.5
201			emission_storage_b_0_offset = 1
202			generic_storage_b_0 = solph.components.GenericStorage(
203			label="generic_storage_b_0",
204			inputs={bus_b_1: solph.Flow()},
205			outputs={bus_b_1: solph.Flow()},
206			inflow_conversion_factor=1,
207			nominal_capacity=solph.Investment(
208			ep_costs=epc_invest,
209			nonconvex=True,
210			maximum=1,
211			custom_attributes={
212			"emission": {
213			"linear": emission_storage_b_0_linear,
214			"offset": emission_storage_b_0_offset,
215			}
216			},
217			),
218			invest_relation_input_capacity=0.5,
219			invest_relation_output_capacity=0.5,
220			)
221
222			es.add(generic_storage_b_0)
223
224			# Generic Storage b_1
225			emission_storage_b_1_linear = 1
226			emission_storage_b_1_offset = 5
227			generic_storage_b_1 = solph.components.GenericStorage(
228			label="generic_storage_b_1",
229			inputs={bus_b_1: solph.Flow()},
230			outputs={bus_b_1: solph.Flow()},
231			inflow_conversion_factor=1,
232			nominal_capacity=solph.Investment(
233			ep_costs=epc_invest * 100,
234			nonconvex=True,
235			maximum=2,
236			custom_attributes={
237			"emission": {
238			"linear": emission_storage_b_1_linear,
239			"offset": emission_storage_b_1_offset,
240			}
241			},
242			),
243			)
244
245			es.add(generic_storage_b_1)
246			if optimize is False:
247			return es
248
249			# create an optimization problem and solve it
250			om = solph.Model(es)
251
252			# add constraint for generic investment limit
253			om = solph.constraints.additional_total_limit(om, "emission", limit=100)
254			# export lp file
255			filename = os.path.join(
256			solph.helpers.extend_basic_path("lp_files"), "GenericInvest.lp"
257			)
258			logging.info("Store lp-file in {0}.".format(filename))
259			om.write(filename, io_options={"symbolic_solver_labels": True})
260
261			# solve model
262			om.solve(solver="cbc", solve_kwargs={"tee": True})
263
264			# create result object
265			results = solph.processing.results(om)
266
267			bus1 = solph.views.node(results, "bus_a_1")["sequences"]
268			bus2 = solph.views.node(results, "bus_b_1")["sequences"]
269
270			# plot the time series (sequences) of a specific component/bus
271			if plt is not None:
272			bus1.plot(kind="line", drawstyle="steps-mid")
273			plt.legend()
274			plt.show()
275			bus2.plot(kind="line", drawstyle="steps-mid")
276			plt.legend()
277			plt.show()
278
279			emission_used = om.total_limit_emission()
280			print("emission value: ", emission_used)
281			print(
282			"Investment trafo_a: ",
283			solph.views.node(results, "trafo_a")["scalars"][0],
284			)
285			print(
286			"Emission investment of trafo_a: ",
287			solph.views.node(results, "trafo_a")["scalars"][0]
288			* emission_conv_a_linear
289			+ emission_conv_a_offset,
290			)
291			print(
292			"Emission flow through trafo_a: ",
293			results[converter_a, bus_a_1]["sequences"]["flow"].sum()
294			* emission_conv_a_flow,
295			)
296
297			print(
298			"Investment trafo_b: ",
299			solph.views.node(results, "trafo_b")["scalars"][0],
300			)
301			print(
302			"Emission investment of trafo_b: ",
303			solph.views.node(results, "trafo_b")["scalars"][0]
304			* emission_conv_b_linear,
305			)
306			print(
307			"Emission flow through trafo_b: ",
308			results[converter_b, bus_b_1]["sequences"]["flow"].sum()
309			* emission_conv_b_flow,
310			)
311
312			print(
313			"Investment generic_storage_b_0: ",
314			results[generic_storage_b_0, None]["scalars"]["total"],
315			)
316			print(
317			"Emission investment generic_storage_b_0: ",
318			results[generic_storage_b_0, None]["scalars"]["total"]
319			* emission_storage_b_0_linear
320			+ results[generic_storage_b_0, None]["scalars"]["invest_status"]
321			* emission_storage_b_0_offset,
322			)
323
324			print(
325			"Investment generic_storage_b_1: ",
326			results[generic_storage_b_1, None]["scalars"]["total"],
327			)
328			print(
329			"Emission investment generic_storage_b_1: ",
330			results[generic_storage_b_1, None]["scalars"]["total"]
331			* emission_storage_b_1_linear
332			+ results[generic_storage_b_1, None]["scalars"]["invest_status"]
333			* emission_storage_b_1_offset,
334			)
335
336
337			if __name__ == "__main__":
338			main()
339

oemof / oemof-solph

Pull Request — dev (#1174)

emission_constraint_invest_and_flow.main() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like