nested_subnetwork_example.plot_figures_for() - Code Metrics - Inspection of "Add facade in oemof.solph" - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#1193)

by Patrik

created 2025-09-16 14:42 UTC

nested_subnetwork_example.plot_figures_for() A

↳ Parent: nested_subnetwork_example

Complexity

Conditions

Size

Total Lines	11
Code Lines	10

Duplication

Lines	11
Ratio	100 %

Importance

Changes

Metric	Value
eloc	10
dl	11
loc	11
rs	9.9
c	0
b	0
f	0
cc	1
nop	1

import logging
import os

import matplotlib.pyplot as plt
import pandas as pd
from oemof.tools import logger
from oemof.network import SubNetwork
from oemof import solph

from oemof.solph import EnergySystem
from oemof.solph import Model
from oemof.solph import buses
from oemof.solph import components
from oemof.solph import create_time_index
from oemof.solph import flows
from oemof.solph import helpers
from oemof.solph import Results

STORAGE_LABEL = "battery_storage"


def get_data_from_file_path(file_path: str) -> pd.DataFrame:
    file_dir = os.path.dirname(os.path.abspath(__file__))
    data = pd.read_csv(file_dir + "/" + file_path)
    return data


def plot_figures_for(element: dict) -> None:

    figure, axes = plt.subplots(figsize=(10, 5))
    element["sequences"].plot(ax=axes, kind="line", drawstyle="steps-post")
    plt.legend(
        loc="upper center",
        prop={"size": 8},
        bbox_to_anchor=(0.5, 1.25),
        ncol=2,
    )
    figure.subplots_adjust(top=0.8)
    plt.show()


class Volatile(SubNetwork):

    def __init__(
        self,
        label: str,
        output_bus: solph.Bus,
        timeseries: float | list[float],
        nominal_capacity: float,
        parent_node=None,
    ):
        self.output_bus = output_bus
        self.timeseries = timeseries
        self.nominal_capacity = nominal_capacity

        super().__init__(
            label=label,
            parent_node=parent_node,
        )

        self.subnode(
            solph.components.Source,
            local_name="source",
            outputs={
                self.output_bus: solph.Flow(
                    max=self.timeseries, nominal_capacity=self.nominal_capacity
                ),
            },
        )


def main():

    # For models that need a long time to optimise, saving and loading the
    # EnergySystem might be advised. By default, we do not do this here. Feel
    # free to experiment with this once you understood the rest of the code.

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

    # Read data file
    file_name = "subnetwork_example.csv"
    data = get_data_from_file_path(file_name)

    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.INFO,
    )

    logging.info("Initialize the energy system")
    date_time_index = create_time_index(2012, number=number_of_time_steps)

    # create the energysystem and assign the time index
    energysystem = EnergySystem(
        timeindex=date_time_index, infer_last_interval=False
    )

    ##########################################################################
    # Create oemof objects
    ##########################################################################

    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
    bus_gas = buses.Bus(label="natural_gas")

    # create electricity bus
    bus_electricity = buses.Bus(label="electricity")

    # adding the buses to the energy system
    energysystem.add(bus_gas, bus_electricity)

    # create excess component for the electricity bus to allow overproduction
    energysystem.add(
        components.Sink(
            label="excess_bus_electricity",
            inputs={bus_electricity: flows.Flow()},
        )
    )

    # create source object representing the gas commodity
    energysystem.add(
        components.Source(
            label="rgas",
            outputs={bus_gas: flows.Flow()},
        )
    )

    # *** SUB-NETWORK ***************************
    # Add a subnetwork for Renewable Energies.
    renewables = SubNetwork("renewables")
    re_bus = renewables.subnode(buses.Bus, "re_elec")

    # create fixed source object representing wind power plants
    renewables.subnode(
        Volatile,
        local_name="wind",
        output_bus=re_bus,
        timeseries=data["wind"],
        nominal_capacity=1000000,
    )
    # create fixed source object representing pv power plants
    renewables.subnode(
        Volatile,
        local_name="pv",
        output_bus=re_bus,
        timeseries=data["pv"],
        nominal_capacity=582000,
    )
    renewables.subnode(
        components.Converter,
        local_name="connection",
        outputs={bus_electricity: flows.Flow()},
        inputs={re_bus: flows.Flow()},
    )
    energysystem.add(renewables)  # Subnetwork to Energysystem
    # *************************************************************

    # create simple sink object representing the electrical demand
    # nominal_value is set to 1 because demand_el is not a normalised series
    energysystem.add(
        components.Sink(
            label="demand",
            inputs={
                bus_electricity: flows.Flow(
                    fix=data["demand_el"], nominal_capacity=1
                )
            },
        )
    )

    # create simple converter object representing a gas power plant
    energysystem.add(
        components.Converter(
            label="pp_gas",
            inputs={bus_gas: flows.Flow()},
            outputs={
                bus_electricity: flows.Flow(
                    nominal_capacity=10e10, variable_costs=50
                )
            },
            conversion_factors={bus_electricity: 0.58},
        )
    )

    # create storage object representing a battery
    nominal_capacity = 10077997
    nominal_value = nominal_capacity / 6

    battery_storage = components.GenericStorage(
        nominal_capacity=nominal_capacity,
        label=STORAGE_LABEL,
        inputs={bus_electricity: flows.Flow(nominal_capacity=nominal_value)},
        outputs={
            bus_electricity: flows.Flow(
                nominal_capacity=nominal_value, variable_costs=0.001
            )
        },
        loss_rate=0.00,
        initial_storage_level=None,
        inflow_conversion_factor=1,
        outflow_conversion_factor=0.8,
    )

    energysystem.add(battery_storage)

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

    logging.info("Optimise the energy system")

    # initialise the operational model
    energysystem_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:
        file_path = os.path.join(
            helpers.extend_basic_path("lp_files"), "basic_example.lp"
        )
        logging.info(f"Store lp-file in {file_path}.")
        io_option = {"symbolic_solver_labels": True}
        energysystem_model.write(file_path, io_options=io_option)

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

    results = Results(energysystem_model)

    # ToDO Implement a filter methode for the Result object to exclude
    #  subcomponents of a facade/sub-network
    # The following lines are meant to show how the result should look like
    # in case the subcomponents should be exclude. There should not be a
    # postprocessing it is better to filter the nodes directly

    # Filter columns that are internal only
    keep_columns = [
        c
        for c in results.flow.columns
        if getattr(c[1].label, "parent", None)
        != getattr(c[0].label, "parent", None)
        or (
            getattr(c[0].label, "parent", True) is True
            and getattr(c[1].label, "parent", True) is True
        )
    ]
    flow_results_filtered = results.flow[keep_columns].copy()

    # Replace subcomponent with facade object
    for level in [0, 1]:
        flow_results_filtered.rename(
            columns={
                c[level]: getattr(c[level].label, "parent", c[level])
                for c in flow_results_filtered.columns
            },
            level=level,
            inplace=True,
        )

    print("**** All results ****")
    print(results.flow.sum())

    print("**** Filtered results ****")
    print(flow_results_filtered.sum())


if __name__ == "__main__":
    main()


1		import logging
2		import os
3
4		import matplotlib.pyplot as plt
5		import pandas as pd
6		from oemof.tools import logger
7		from oemof.network import SubNetwork
8		from oemof import solph
9
10		from oemof.solph import EnergySystem
11		from oemof.solph import Model
12		from oemof.solph import buses
13		from oemof.solph import components
14		from oemof.solph import create_time_index
15		from oemof.solph import flows
16		from oemof.solph import helpers
17		from oemof.solph import Results
18
19		STORAGE_LABEL = "battery_storage"
20
21
22		def get_data_from_file_path(file_path: str) -> pd.DataFrame:
23		file_dir = os.path.dirname(os.path.abspath(__file__))
24		data = pd.read_csv(file_dir + "/" + file_path)
25		return data
26
27
28	View Code Duplication	def plot_figures_for(element: dict) -> None:
		0 ignored issues – show Duplication introduced 2025-05-14 12:50 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
29		figure, axes = plt.subplots(figsize=(10, 5))
30		element["sequences"].plot(ax=axes, kind="line", drawstyle="steps-post")
31		plt.legend(
32		loc="upper center",
33		prop={"size": 8},
34		bbox_to_anchor=(0.5, 1.25),
35		ncol=2,
36		)
37		figure.subplots_adjust(top=0.8)
38		plt.show()
39
40
41		class Volatile(SubNetwork):
42
43		def __init__(
44		self,
45		label: str,
46		output_bus: solph.Bus,
47		timeseries: float \| list[float],
48		nominal_capacity: float,
49		parent_node=None,
50		):
51		self.output_bus = output_bus
52		self.timeseries = timeseries
53		self.nominal_capacity = nominal_capacity
54
55		super().__init__(
56		label=label,
57		parent_node=parent_node,
58		)
59
60		self.subnode(
61		solph.components.Source,
62		local_name="source",
63		outputs={
64		self.output_bus: solph.Flow(
65		max=self.timeseries, nominal_capacity=self.nominal_capacity
66		),
67		},
68		)
69
70
71	View Code Duplication	def main():
		0 ignored issues – show Duplication introduced 2025-07-15 12:09 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
72		# For models that need a long time to optimise, saving and loading the
73		# EnergySystem might be advised. By default, we do not do this here. Feel
74		# free to experiment with this once you understood the rest of the code.
75
76		# *************************************************************************
77		# ******** PART 1 - Define and optimise the energy system *************
78		# *************************************************************************
79
80		# Read data file
81		file_name = "subnetwork_example.csv"
82		data = get_data_from_file_path(file_name)
83
84		solver = "cbc" # 'glpk', 'gurobi',....
85		debug = False # Set number_of_timesteps to 3 to get a readable lp-file.
86		number_of_time_steps = len(data)
87		solver_verbose = False # show/hide solver output
88
89		# initiate the logger (see the API docs for more information)
90		logger.define_logging(
91		logfile="oemof_example.log",
92		screen_level=logging.INFO,
93		file_level=logging.INFO,
94		)
95
96		logging.info("Initialize the energy system")
97		date_time_index = create_time_index(2012, number=number_of_time_steps)
98
99		# create the energysystem and assign the time index
100		energysystem = EnergySystem(
101		timeindex=date_time_index, infer_last_interval=False
102		)
103
104		##########################################################################
105		# Create oemof objects
106		##########################################################################
107
108		logging.info("Create oemof objects")
109
110		# The bus objects were assigned to variables which makes it easier to
111		# connect components to these buses (see below).
112
113		# create natural gas bus
114		bus_gas = buses.Bus(label="natural_gas")
115
116		# create electricity bus
117		bus_electricity = buses.Bus(label="electricity")
118
119		# adding the buses to the energy system
120		energysystem.add(bus_gas, bus_electricity)
121
122		# create excess component for the electricity bus to allow overproduction
123		energysystem.add(
124		components.Sink(
125		label="excess_bus_electricity",
126		inputs={bus_electricity: flows.Flow()},
127		)
128		)
129
130		# create source object representing the gas commodity
131		energysystem.add(
132		components.Source(
133		label="rgas",
134		outputs={bus_gas: flows.Flow()},
135		)
136		)
137
138		# * SUB-NETWORK *************************
139		# Add a subnetwork for Renewable Energies.
140		renewables = SubNetwork("renewables")
141		re_bus = renewables.subnode(buses.Bus, "re_elec")
142
143		# create fixed source object representing wind power plants
144		renewables.subnode(
145		Volatile,
146		local_name="wind",
147		output_bus=re_bus,
148		timeseries=data["wind"],
149		nominal_capacity=1000000,
150		)
151		# create fixed source object representing pv power plants
152		renewables.subnode(
153		Volatile,
154		local_name="pv",
155		output_bus=re_bus,
156		timeseries=data["pv"],
157		nominal_capacity=582000,
158		)
159		renewables.subnode(
160		components.Converter,
161		local_name="connection",
162		outputs={bus_electricity: flows.Flow()},
163		inputs={re_bus: flows.Flow()},
164		)
165		energysystem.add(renewables) # Subnetwork to Energysystem
166		# *************************************************************
167
168		# create simple sink object representing the electrical demand
169		# nominal_value is set to 1 because demand_el is not a normalised series
170		energysystem.add(
171		components.Sink(
172		label="demand",
173		inputs={
174		bus_electricity: flows.Flow(
175		fix=data["demand_el"], nominal_capacity=1
176		)
177		},
178		)
179		)
180
181		# create simple converter object representing a gas power plant
182		energysystem.add(
183		components.Converter(
184		label="pp_gas",
185		inputs={bus_gas: flows.Flow()},
186		outputs={
187		bus_electricity: flows.Flow(
188		nominal_capacity=10e10, variable_costs=50
189		)
190		},
191		conversion_factors={bus_electricity: 0.58},
192		)
193		)
194
195		# create storage object representing a battery
196		nominal_capacity = 10077997
197		nominal_value = nominal_capacity / 6
198
199		battery_storage = components.GenericStorage(
200		nominal_capacity=nominal_capacity,
201		label=STORAGE_LABEL,
202		inputs={bus_electricity: flows.Flow(nominal_capacity=nominal_value)},
203		outputs={
204		bus_electricity: flows.Flow(
205		nominal_capacity=nominal_value, variable_costs=0.001
206		)
207		},
208		loss_rate=0.00,
209		initial_storage_level=None,
210		inflow_conversion_factor=1,
211		outflow_conversion_factor=0.8,
212		)
213
214		energysystem.add(battery_storage)
215
216		##########################################################################
217		# Optimise the energy system and plot the results
218		##########################################################################
219
220		logging.info("Optimise the energy system")
221
222		# initialise the operational model
223		energysystem_model = Model(energysystem)
224
225		# This is for debugging only. It is not(!) necessary to solve the problem
226		# and should be set to False to save time and disc space in normal use. For
227		# debugging the timesteps should be set to 3, to increase the readability
228		# of the lp-file.
229		if debug:
230		file_path = os.path.join(
231		helpers.extend_basic_path("lp_files"), "basic_example.lp"
232		)
233		logging.info(f"Store lp-file in {file_path}.")
234		io_option = {"symbolic_solver_labels": True}
235		energysystem_model.write(file_path, io_options=io_option)
236
237		# if tee_switch is true solver messages will be displayed
238		logging.info("Solve the optimization problem")
239		energysystem_model.solve(
240		solver=solver, solve_kwargs={"tee": solver_verbose}
241		)
242
243		results = Results(energysystem_model)
244
245		# ToDO Implement a filter methode for the Result object to exclude
246		# subcomponents of a facade/sub-network
247		# The following lines are meant to show how the result should look like
248		# in case the subcomponents should be exclude. There should not be a
249		# postprocessing it is better to filter the nodes directly
250
251		# Filter columns that are internal only
252		keep_columns = [
253		c
254		for c in results.flow.columns
255		if getattr(c[1].label, "parent", None)
256		!= getattr(c[0].label, "parent", None)
257		or (
258		getattr(c[0].label, "parent", True) is True
259		and getattr(c[1].label, "parent", True) is True
260		)
261		]
262		flow_results_filtered = results.flow[keep_columns].copy()
263
264		# Replace subcomponent with facade object
265		for level in [0, 1]:
266		flow_results_filtered.rename(
267		columns={
268		c[level]: getattr(c[level].label, "parent", c[level])
269		for c in flow_results_filtered.columns
270		},
271		level=level,
272		inplace=True,
273		)
274
275		print("** All results **")
276		print(results.flow.sum())
277
278		print("** Filtered results **")
279		print(flow_results_filtered.sum())
280
281
282		if __name__ == "__main__":
283		main()
284

oemof / oemof-solph

Pull Request — dev (#1193)

nested_subnetwork_example.plot_figures_for() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like