nested_subnetwork_example.Volatile.define_subnetwork() - 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)

unknown

created 2025-09-11 22:15 UTC

Volatile.define_subnetwork() A

↳ Parent: nested_subnetwork_example

Complexity

Conditions

Size

Total Lines	7
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	7
dl	0
loc	7
rs	10
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(solph.Facade):

    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, facade_type=type(self)
        )

    def define_subnetwork(self):
        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. This not a Facade it just meant
    # to group components
    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(solph.Facade):
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, parent_node=parent_node, facade_type=type(self)
57		)
58
59		def define_subnetwork(self):
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. This not a Facade it just meant
140		# to group components
141		renewables = SubNetwork("renewables")
142		re_bus = renewables.subnode(buses.Bus, "re_elec")
143
144		# create fixed source object representing wind power plants
145		renewables.subnode(
146		Volatile,
147		local_name="wind",
148		output_bus=re_bus,
149		timeseries=data["wind"],
150		nominal_capacity=1000000,
151		)
152		# create fixed source object representing pv power plants
153		renewables.subnode(
154		Volatile,
155		local_name="pv",
156		output_bus=re_bus,
157		timeseries=data["pv"],
158		nominal_capacity=582000,
159		)
160		renewables.subnode(
161		components.Converter,
162		local_name="connection",
163		outputs={bus_electricity: flows.Flow()},
164		inputs={re_bus: flows.Flow()},
165		)
166		energysystem.add(renewables) # Subnetwork to Energysystem
167		# *************************************************************
168
169		# create simple sink object representing the electrical demand
170		# nominal_value is set to 1 because demand_el is not a normalised series
171		energysystem.add(
172		components.Sink(
173		label="demand",
174		inputs={
175		bus_electricity: flows.Flow(
176		fix=data["demand_el"], nominal_capacity=1
177		)
178		},
179		)
180		)
181
182		# create simple converter object representing a gas power plant
183		energysystem.add(
184		components.Converter(
185		label="pp_gas",
186		inputs={bus_gas: flows.Flow()},
187		outputs={
188		bus_electricity: flows.Flow(
189		nominal_capacity=10e10, variable_costs=50
190		)
191		},
192		conversion_factors={bus_electricity: 0.58},
193		)
194		)
195
196		# create storage object representing a battery
197		nominal_capacity = 10077997
198		nominal_value = nominal_capacity / 6
199
200		battery_storage = components.GenericStorage(
201		nominal_capacity=nominal_capacity,
202		label=STORAGE_LABEL,
203		inputs={bus_electricity: flows.Flow(nominal_capacity=nominal_value)},
204		outputs={
205		bus_electricity: flows.Flow(
206		nominal_capacity=nominal_value, variable_costs=0.001
207		)
208		},
209		loss_rate=0.00,
210		initial_storage_level=None,
211		inflow_conversion_factor=1,
212		outflow_conversion_factor=0.8,
213		)
214
215		energysystem.add(battery_storage)
216
217		##########################################################################
218		# Optimise the energy system and plot the results
219		##########################################################################
220
221		logging.info("Optimise the energy system")
222
223		# initialise the operational model
224		energysystem_model = Model(energysystem)
225
226		# This is for debugging only. It is not(!) necessary to solve the problem
227		# and should be set to False to save time and disc space in normal use. For
228		# debugging the timesteps should be set to 3, to increase the readability
229		# of the lp-file.
230		if debug:
231		file_path = os.path.join(
232		helpers.extend_basic_path("lp_files"), "basic_example.lp"
233		)
234		logging.info(f"Store lp-file in {file_path}.")
235		io_option = {"symbolic_solver_labels": True}
236		energysystem_model.write(file_path, io_options=io_option)
237
238		# if tee_switch is true solver messages will be displayed
239		logging.info("Solve the optimization problem")
240		energysystem_model.solve(
241		solver=solver, solve_kwargs={"tee": solver_verbose}
242		)
243
244		results = Results(energysystem_model)
245
246		# ToDO Implement a filter methode for the Result object to exclude
247		# subcomponents of a facade/sub-network
248		# The following lines are meant to show how the result should look like
249		# in case the subcomponents should be exclude. There should not be a
250		# postprocessing it is better to filter the nodes directly
251
252		# Filter columns that are internal only
253		keep_columns = [
254		c
255		for c in results.flow.columns
256		if getattr(c[1].label, "parent", None)
257		!= getattr(c[0].label, "parent", None)
258		or (
259		getattr(c[0].label, "parent", True) is True
260		and getattr(c[1].label, "parent", True) is True
261		)
262		]
263		flow_results_filtered = results.flow[keep_columns].copy()
264
265		# Replace subcomponent with facade object
266		for level in [0, 1]:
267		flow_results_filtered.rename(
268		columns={
269		c[level]: getattr(c[level].label, "parent", c[level])
270		for c in flow_results_filtered.columns
271		},
272		level=level,
273		inplace=True,
274		)
275
276		print("** All results **")
277		print(results.flow.sum())
278
279		print("** Filtered results **")
280		print(flow_results_filtered.sum())
281
282
283		if __name__ == "__main__":
284		main()
285

oemof / oemof-solph

Pull Request — dev (#1193)

Volatile.define_subnetwork() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like