tuple_as_label.main() - Code Metrics - Inspection of "Add examples from oemof-examples" - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#850)

by Uwe

created 2022-11-10 19:41 UTC

tuple_as_label.main() B

↳ Parent: tuple_as_label

Complexity

Conditions

Size

Total Lines	213
Code Lines	114

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	114
dl	0
loc	213
rs	7
c	0
b	0
f	0
cc	3
nop	0

How to fix Long Method

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

"""
General description
-------------------

You should have grasped the basic_example to understand this one.

This is an example to show how the label attribute can be used with tuples to
manage the results of large energy system. Even though, the feature is
introduced in a small example it is made for large system.

In small energy system you normally address the node, you want your results
from, directly. In large systems you may want to group your results and collect
all power plants of a specific region or pv feed-in of all regions.

Therefore you can use named tuples as label. In a named tuple you need to
specify the fields:

>>> label = namedtuple('solph_label', ['region', 'tag1', 'tag2'])

>>> pv_label = label('region_1', 'renewable_source', 'pv')
>>> pp_gas_label = label('region_2', 'power_plant', 'natural_gas')
>>> demand_label = label('region_3', 'electricity', 'demand')

You always have to address all fields but you can use empty strings or None as
place holders.

>>> elec_bus = label('region_4', 'electricity', '')
>>> print(elec_bus)
solph_label(region='region_4', tag1='electricity', tag2='')

>>> elec_bus = label('region_4', 'electricity', None)
>>> print(elec_bus)
solph_label(region='region_4', tag1='electricity', tag2=None)

Now you can filter the results using the label or the instance:

>>> for key, value in results.items():  # Loop results (keys are tuples!)
...     if isinstance(key[0], comp.Sink) & (key[0].label.tag2 == 'demand'):
...         print("elec demand {0}: {1}".format(key[0].label.region,
...                                             value['sequences'].sum()))

elec demand region_1: 3456
elec demand region_2: 2467
...

In the example below a subclass is created to define ones own string output.
By default the output of a namedtuple is `field1=value1, field2=value2,...`:

>>> print(str(pv_label))
solph_label(region='region_1', tag1='renewable_source', tag2='pv')

With the subclass we created below the output is different, because we defined
our own string representation:

>>> new_pv_label = Label('region_1', 'renewable_source', 'pv')
>>> print(str(new_pv_label))
region_1_renewable_source_pv

You still will be able to get the original string using `repr`:

>>> print(repr(new_pv_label))
Label(tag1='region_1', tag2='renewable_source', tag3='pv')

This a helpful adaption for automatic plots etc..

Afterwards you can use `format` to define your own custom string.:
>>> print('{0}+{1}-{2}'.format(pv_label.region, pv_label.tag2, pv_label.tag1))
region_1+pv-renewable_source

Data
----
basic_example.csv


Installation requirements
-------------------------
This example requires oemof.solph (v0.5.x), install by:

    pip install oemof.solph[examples]


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

"""

# ****************************************************************************
# ********** PART 1 - Define and optimise the energy system ******************
# ****************************************************************************

import logging
import os
import warnings
from collections import namedtuple

import pandas as pd
from oemof.tools import logger

from oemof.solph import EnergySystem
from oemof.solph import Model
from oemof.solph import buses
from oemof.solph import components as comp
from oemof.solph import create_time_index
from oemof.solph import flows
from oemof.solph import helpers
from oemof.solph import processing


# Subclass of the named tuple with its own __str__ method.
# You can add as many tags as you like
# For tag1, tag2 you can define your own fields like region, fuel, type...
class Label(namedtuple("solph_label", ["tag1", "tag2", "tag3"])):
    __slots__ = ()

    def __str__(self):
        """The string is used within solph as an ID, so it hast to be unique"""
        return "_".join(map(str, self._asdict().values()))


def main():
    # Read data file
    filename = os.path.join(os.getcwd(), "tuple_as_label.csv")
    try:
        data = pd.read_csv(filename)
    except FileNotFoundError:
        msg = "Data file not found: {0}. Only one value used!"
        warnings.warn(msg.format(filename), UserWarning)
        data = pd.DataFrame({"pv": [0.3], "wind": [0.6], "demand_el": [500]})

    solver = "cbc"  # 'glpk', 'gurobi',....
    debug = False  # Set number_of_timesteps to 3 to get a readable lp-file.
    number_of_time_steps = len(data)
    solver_verbose = False  # show/hide solver output

    # initiate the logger (see the API docs for more information)
    logger.define_logging(
        logfile="oemof_example.log",
        screen_level=logging.INFO,
        file_level=logging.WARNING,
    )

    logging.info("Initialize the energy system")
    energysystem = EnergySystem(
        timeindex=create_time_index(2012, number=number_of_time_steps),
        infer_last_interval=False,
    )

    ##########################################################################
    # Create oemof object
    ##########################################################################

    logging.info("Create oemof objects")

    # The bus objects were assigned to variables which makes it easier to
    # connect components to these buses (see below).

    # create natural gas bus
    bgas = buses.Bus(label=Label("bus", "gas", None))

    # create electricity bus
    bel = buses.Bus(label=Label("bus", "electricity", None))

    # adding the buses to the energy system
    energysystem.add(bgas, bel)

    # create excess component for the electricity bus to allow overproduction
    energysystem.add(
        comp.Sink(
            label=Label("sink", "electricity", "excess"),
            inputs={bel: flows.Flow()},
        )
    )

    # create source object representing the gas commodity (annual limit)
    energysystem.add(
        comp.Source(
            label=Label("commodity_source", "gas", "commodity"),
            outputs={bgas: flows.Flow()},
        )
    )

    # create fixed source object representing wind pow er plants
    energysystem.add(
        comp.Source(
            label=Label("ee_source", "electricity", "wind"),
            outputs={bel: flows.Flow(fix=data["wind"], nominal_value=2000)},
        )
    )

    # create fixed source object representing pv power plants
    energysystem.add(
        comp.Source(
            label=Label("ee_source", "electricity", "pv"),
            outputs={bel: flows.Flow(fix=data["pv"], nominal_value=3000)},
        )
    )

    # create simple sink object representing the electrical demand
    energysystem.add(
        comp.Sink(
            label=Label("sink", "electricity", "demand"),
            inputs={
                bel: flows.Flow(fix=data["demand_el"] / 1000, nominal_value=1)
            },
        )
    )

    # create simple transformer object representing a gas power plant
    energysystem.add(
        comp.Transformer(
            label=Label("power plant", "electricity", "gas"),
            inputs={bgas: flows.Flow()},
            outputs={bel: flows.Flow(nominal_value=10000, variable_costs=50)},
            conversion_factors={bel: 0.58},
        )
    )

    # create storage object representing a battery
    nominal_storage_capacity = 5000
    storage = comp.GenericStorage(
        nominal_storage_capacity=nominal_storage_capacity,
        label=Label("storage", "electricity", "battery"),
        inputs={bel: flows.Flow(nominal_value=nominal_storage_capacity / 6)},
        outputs={bel: flows.Flow(nominal_value=nominal_storage_capacity / 6)},
        loss_rate=0.00,
        initial_storage_level=None,
        inflow_conversion_factor=1,
        outflow_conversion_factor=0.8,
    )

    energysystem.add(storage)

    ##########################################################################
    # Optimise the energy system and plot the results
    ##########################################################################

    logging.info("Optimise the energy system")

    # initialise the operational model
    model = Model(energysystem)

    # This is for debugging only. It is not(!) necessary to solve the problem
    # and should be set to False to save time and disc space in normal use. For
    # debugging the timesteps should be set to 3, to increase the readability
    # of the lp-file.
    if debug:
        filename = os.path.join(
            helpers.extend_basic_path("lp_files"), "basic_example.lp"
        )
        logging.info("Store lp-file in {0}.".format(filename))
        model.write(filename, io_options={"symbolic_solver_labels": True})

    # if tee_switch is true solver messages will be displayed
    logging.info("Solve the optimization problem")
    model.receive_duals()
    model.solve(solver=solver, solve_kwargs={"tee": solver_verbose})

    logging.info("Store the energy system with the results.")

    # The processing module of the outputlib can be used to extract the results
    # from the model transfer them into a homogeneous structured dictionary.

    results = processing.results(model)

    # ** Create a table with all sequences and store it into a file (csv/xlsx)
    flows_to_bus = pd.DataFrame(
        {
            str(k[0].label): v["sequences"]["flow"]
            for k, v in results.items()
            if k[1] is not None and k[1] == bel
        }
    )
    flows_from_bus = pd.DataFrame(
        {
            str(k[1].label): v["sequences"]["flow"]
            for k, v in results.items()
            if k[1] is not None and k[0] == bel
        }
    )

    storage = pd.DataFrame(
        {
            str(k[0].label): v["sequences"]["storage_content"]
            for k, v in results.items()
            if k[1] is None and k[0] == storage
        }
    )

    duals = pd.DataFrame(
        {
            str(k[0].label): v["sequences"]["duals"]
            for k, v in results.items()
            if k[1] is None and isinstance(k[0], buses.Bus)
        }
    )

    my_flows = pd.concat(
        [flows_to_bus, flows_from_bus, storage, duals],
        keys=["to_bus", "from_bus", "content", "duals"],
        axis=1,
    )

    # Store the table to csv or excel file:
    home_path = os.path.expanduser("~")
    my_flows.to_csv(os.path.join(home_path, "my_flows.csv"))
    # my_flows.to_excel(os.path.join(home_path, "my_flows.xlsx"))
    print(my_flows.sum())

    # ********* Use your tuple labels to filter the components
    ee_sources = [
        str(f[0].label)
        for f in results.keys()
        if f[0].label.tag1 == "ee_source"
    ]
    print(ee_sources)

    # It is possible to filter components by the label tags and the class, so
    # the label concepts is a result of the postprocessing. If it is necessary
    # to get all components of a region, "region" should be a field of the
    # label. To filter only by tags you can add a tag named "class" with the
    # name of the class as value.
    electricity_buses = list(
        set(
            [
                str(f[0].label)
                for f in results.keys()
                if f[0].label.tag2 == "electricity"
                and isinstance(f[0], buses.Bus)
            ]
        )
    )
    print(electricity_buses)


if __name__ == "__main__":
    main()


1			# -- coding: utf-8 --
2
3			"""
4			General description
5			-------------------
6
7			You should have grasped the basic_example to understand this one.
8
9			This is an example to show how the label attribute can be used with tuples to
10			manage the results of large energy system. Even though, the feature is
11			introduced in a small example it is made for large system.
12
13			In small energy system you normally address the node, you want your results
14			from, directly. In large systems you may want to group your results and collect
15			all power plants of a specific region or pv feed-in of all regions.
16
17			Therefore you can use named tuples as label. In a named tuple you need to
18			specify the fields:
19
20			>>> label = namedtuple('solph_label', ['region', 'tag1', 'tag2'])
21
22			>>> pv_label = label('region_1', 'renewable_source', 'pv')
23			>>> pp_gas_label = label('region_2', 'power_plant', 'natural_gas')
24			>>> demand_label = label('region_3', 'electricity', 'demand')
25
26			You always have to address all fields but you can use empty strings or None as
27			place holders.
28
29			>>> elec_bus = label('region_4', 'electricity', '')
30			>>> print(elec_bus)
31			solph_label(region='region_4', tag1='electricity', tag2='')
32
33			>>> elec_bus = label('region_4', 'electricity', None)
34			>>> print(elec_bus)
35			solph_label(region='region_4', tag1='electricity', tag2=None)
36
37			Now you can filter the results using the label or the instance:
38
39			>>> for key, value in results.items(): # Loop results (keys are tuples!)
40			... if isinstance(key[0], comp.Sink) & (key[0].label.tag2 == 'demand'):
41			... print("elec demand {0}: {1}".format(key[0].label.region,
42			... value['sequences'].sum()))
43
44			elec demand region_1: 3456
45			elec demand region_2: 2467
46			...
47
48			In the example below a subclass is created to define ones own string output.
49			By default the output of a namedtuple is `field1=value1, field2=value2,...`:
50
51			>>> print(str(pv_label))
52			solph_label(region='region_1', tag1='renewable_source', tag2='pv')
53
54			With the subclass we created below the output is different, because we defined
55			our own string representation:
56
57			>>> new_pv_label = Label('region_1', 'renewable_source', 'pv')
58			>>> print(str(new_pv_label))
59			region_1_renewable_source_pv
60
61			You still will be able to get the original string using `repr`:
62
63			>>> print(repr(new_pv_label))
64			Label(tag1='region_1', tag2='renewable_source', tag3='pv')
65
66			This a helpful adaption for automatic plots etc..
67
68			Afterwards you can use `format` to define your own custom string.:
69			>>> print('{0}+{1}-{2}'.format(pv_label.region, pv_label.tag2, pv_label.tag1))
70			region_1+pv-renewable_source
71
72			Data
73			----
74			basic_example.csv
75
76
77			Installation requirements
78			-------------------------
79			This example requires oemof.solph (v0.5.x), install by:
80
81			pip install oemof.solph[examples]
82
83
84			License
85			-------
86			`MIT license <https://github.com/oemof/oemof-solph/blob/dev/LICENSE>`_
87
88			"""
89
90			# ****************************************************************************
91			# ******** PART 1 - Define and optimise the energy system ****************
92			# ****************************************************************************
93
94			import logging
95			import os
96			import warnings
97			from collections import namedtuple
98
99			import pandas as pd
100			from oemof.tools import logger
101
102			from oemof.solph import EnergySystem
103			from oemof.solph import Model
104			from oemof.solph import buses
105			from oemof.solph import components as comp
106			from oemof.solph import create_time_index
107			from oemof.solph import flows
108			from oemof.solph import helpers
109			from oemof.solph import processing
110
111
112			# Subclass of the named tuple with its own __str__ method.
113			# You can add as many tags as you like
114			# For tag1, tag2 you can define your own fields like region, fuel, type...
115			class Label(namedtuple("solph_label", ["tag1", "tag2", "tag3"])):
116			__slots__ = ()
117
118			def __str__(self):
119			"""The string is used within solph as an ID, so it hast to be unique"""
120			return "_".join(map(str, self._asdict().values()))
121
122
123			def main():
124			# Read data file
125			filename = os.path.join(os.getcwd(), "tuple_as_label.csv")
126			try:
127			data = pd.read_csv(filename)
128			except FileNotFoundError:
129			msg = "Data file not found: {0}. Only one value used!"
130			warnings.warn(msg.format(filename), UserWarning)
131			data = pd.DataFrame({"pv": [0.3], "wind": [0.6], "demand_el": [500]})
132
133			solver = "cbc" # 'glpk', 'gurobi',....
134			debug = False # Set number_of_timesteps to 3 to get a readable lp-file.
135			number_of_time_steps = len(data)
136			solver_verbose = False # show/hide solver output
137
138			# initiate the logger (see the API docs for more information)
139			logger.define_logging(
140			logfile="oemof_example.log",
141			screen_level=logging.INFO,
142			file_level=logging.WARNING,
143			)
144
145			logging.info("Initialize the energy system")
146			energysystem = EnergySystem(
147			timeindex=create_time_index(2012, number=number_of_time_steps),
148			infer_last_interval=False,
149			)
150
151			##########################################################################
152			# Create oemof object
153			##########################################################################
154
155			logging.info("Create oemof objects")
156
157			# The bus objects were assigned to variables which makes it easier to
158			# connect components to these buses (see below).
159
160			# create natural gas bus
161			bgas = buses.Bus(label=Label("bus", "gas", None))
162
163			# create electricity bus
164			bel = buses.Bus(label=Label("bus", "electricity", None))
165
166			# adding the buses to the energy system
167			energysystem.add(bgas, bel)
168
169			# create excess component for the electricity bus to allow overproduction
170			energysystem.add(
171			comp.Sink(
172			label=Label("sink", "electricity", "excess"),
173			inputs={bel: flows.Flow()},
174			)
175			)
176
177			# create source object representing the gas commodity (annual limit)
178			energysystem.add(
179			comp.Source(
180			label=Label("commodity_source", "gas", "commodity"),
181			outputs={bgas: flows.Flow()},
182			)
183			)
184
185			# create fixed source object representing wind pow er plants
186			energysystem.add(
187			comp.Source(
188			label=Label("ee_source", "electricity", "wind"),
189			outputs={bel: flows.Flow(fix=data["wind"], nominal_value=2000)},
190			)
191			)
192
193			# create fixed source object representing pv power plants
194			energysystem.add(
195			comp.Source(
196			label=Label("ee_source", "electricity", "pv"),
197			outputs={bel: flows.Flow(fix=data["pv"], nominal_value=3000)},
198			)
199			)
200
201			# create simple sink object representing the electrical demand
202			energysystem.add(
203			comp.Sink(
204			label=Label("sink", "electricity", "demand"),
205			inputs={
206			bel: flows.Flow(fix=data["demand_el"] / 1000, nominal_value=1)
207			},
208			)
209			)
210
211			# create simple transformer object representing a gas power plant
212			energysystem.add(
213			comp.Transformer(
214			label=Label("power plant", "electricity", "gas"),
215			inputs={bgas: flows.Flow()},
216			outputs={bel: flows.Flow(nominal_value=10000, variable_costs=50)},
217			conversion_factors={bel: 0.58},
218			)
219			)
220
221			# create storage object representing a battery
222			nominal_storage_capacity = 5000
223			storage = comp.GenericStorage(
224			nominal_storage_capacity=nominal_storage_capacity,
225			label=Label("storage", "electricity", "battery"),
226			inputs={bel: flows.Flow(nominal_value=nominal_storage_capacity / 6)},
227			outputs={bel: flows.Flow(nominal_value=nominal_storage_capacity / 6)},
228			loss_rate=0.00,
229			initial_storage_level=None,
230			inflow_conversion_factor=1,
231			outflow_conversion_factor=0.8,
232			)
233
234			energysystem.add(storage)
235
236			##########################################################################
237			# Optimise the energy system and plot the results
238			##########################################################################
239
240			logging.info("Optimise the energy system")
241
242			# initialise the operational model
243			model = Model(energysystem)
244
245			# This is for debugging only. It is not(!) necessary to solve the problem
246			# and should be set to False to save time and disc space in normal use. For
247			# debugging the timesteps should be set to 3, to increase the readability
248			# of the lp-file.
249			if debug:
250			filename = os.path.join(
251			helpers.extend_basic_path("lp_files"), "basic_example.lp"
252			)
253			logging.info("Store lp-file in {0}.".format(filename))
254			model.write(filename, io_options={"symbolic_solver_labels": True})
255
256			# if tee_switch is true solver messages will be displayed
257			logging.info("Solve the optimization problem")
258			model.receive_duals()
259			model.solve(solver=solver, solve_kwargs={"tee": solver_verbose})
260
261			logging.info("Store the energy system with the results.")
262
263			# The processing module of the outputlib can be used to extract the results
264			# from the model transfer them into a homogeneous structured dictionary.
265
266			results = processing.results(model)
267
268			# ** Create a table with all sequences and store it into a file (csv/xlsx)
269			flows_to_bus = pd.DataFrame(
270			{
271			str(k[0].label): v["sequences"]["flow"]
272			for k, v in results.items()
273			if k[1] is not None and k[1] == bel
274			}
275			)
276			flows_from_bus = pd.DataFrame(
277			{
278			str(k[1].label): v["sequences"]["flow"]
279			for k, v in results.items()
280			if k[1] is not None and k[0] == bel
281			}
282			)
283
284			storage = pd.DataFrame(
285			{
286			str(k[0].label): v["sequences"]["storage_content"]
287			for k, v in results.items()
288			if k[1] is None and k[0] == storage
289			}
290			)
291
292			duals = pd.DataFrame(
293			{
294			str(k[0].label): v["sequences"]["duals"]
295			for k, v in results.items()
296			if k[1] is None and isinstance(k[0], buses.Bus)
297			}
298			)
299
300			my_flows = pd.concat(
301			[flows_to_bus, flows_from_bus, storage, duals],
302			keys=["to_bus", "from_bus", "content", "duals"],
303			axis=1,
304			)
305
306			# Store the table to csv or excel file:
307			home_path = os.path.expanduser("~")
308			my_flows.to_csv(os.path.join(home_path, "my_flows.csv"))
309			# my_flows.to_excel(os.path.join(home_path, "my_flows.xlsx"))
310			print(my_flows.sum())
311
312			# ********* Use your tuple labels to filter the components
313			ee_sources = [
314			str(f[0].label)
315			for f in results.keys()
316			if f[0].label.tag1 == "ee_source"
317			]
318			print(ee_sources)
319
320			# It is possible to filter components by the label tags and the class, so
321			# the label concepts is a result of the postprocessing. If it is necessary
322			# to get all components of a region, "region" should be a field of the
323			# label. To filter only by tags you can add a tag named "class" with the
324			# name of the class as value.
325			electricity_buses = list(
326			set(
327			[
328			str(f[0].label)
329			for f in results.keys()
330			if f[0].label.tag2 == "electricity"
331			and isinstance(f[0], buses.Bus)
332			]
333			)
334			)
335			print(electricity_buses)
336
337
338			if __name__ == "__main__":
339			main()
340

oemof / oemof-solph

Pull Request — dev (#850)

tuple_as_label.main() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like