time_series_un_even.solve_model() - Code Metrics - Inspection of "Examples timeseries retcon paper" - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#1226)

by Uwe

created 2025-12-19 18:02 UTC

time_series_un_even.solve_model() B

↳ Parent: time_series_un_even

Complexity

Conditions

Size

Total Lines	154
Code Lines	97

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	97
dl	0
loc	154
rs	7.0981
c	0
b	0
f	0
cc	4
nop	5

How to fix Long Method

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

"""
SPDX-FileCopyrightText: Patrik Schönfeldt
SPDX-FileCopyrightText: DLR e.V.

SPDX-License-Identifier: MIT
"""

import logging
import warnings
from collections import namedtuple
from datetime import datetime
from pathlib import Path

import pandas as pd
import pytz
from cost_data import discounted_average_price
from cost_data import energy_prices
from cost_data import investment_costs
from create_timeseries import reshape_unevenly
from matplotlib import pyplot as plt
from oemof.network import graph
from oemof.tools import debugging
from oemof.tools import logger
from oemof.tools.economics import annuity
from shared import prepare_input_data

from oemof.solph import Bus
from oemof.solph import EnergySystem
from oemof.solph import Flow
from oemof.solph import Investment
from oemof.solph import Model
from oemof.solph import Results
from oemof.solph import components as cmp

warnings.filterwarnings(
    "ignore", category=debugging.ExperimentalFeatureWarning
)
logger.define_logging()


def calculate_annuity(value):
    return value / 20


def calculate_fix_cost(value):
    return value / 20


def prepare_technical_data(minutes):
    data = namedtuple("data", "even uneven")
    df = prepare_input_data().resample(f"{minutes} min").mean()
    df["ev charge (kW)"] = 0
    df_un = reshape_unevenly(df)
    return data(even=df, uneven=df_un)


def prepare_cost_data():
    pass


def solve_model(data, year=2025, es=None, n=20, r=0.05):
    if es is None:
        es = EnergySystem(timeindex=data.index)

    var_cost = discounted_average_price(energy_prices(), r, n, year)
    invest_cost = investment_costs().loc[year]

    # Create Investment objects from cost data
    investments = {}
    for key in ["gas boiler", "heat pump", "battery", "pv"]:
        try:
            epc = annuity(
                invest_cost[(key, "specific_costs [Eur/kW]")], n, r
            )
        except KeyError:
            epc = annuity(
                invest_cost[(key, "specific_costs [Eur/kWh]")], n, r
            )
        fix_cost = calculate_fix_cost(invest_cost[(key, "fixed_costs [Eur]")])
        investments[key] = Investment(ep_costs=epc, fixed_costs=fix_cost)

    # Buses
    bus_el = Bus(label="electricity")
    bus_heat = Bus(label="heat")
    bus_gas = Bus(label="gas")
    es.add(bus_el, bus_heat, bus_gas)

    # Sources
    es.add(
        cmp.Source(
            label="PV",
            outputs={
                bus_el: Flow(
                    fix=data["PV (kW/kWp)"],
                    nominal_capacity=investments["pv"],
                )
            },
        )
    )
    es.add(
        cmp.Source(
            label="Shortage_heat", outputs={bus_heat: Flow(variable_costs=99)}
        )
    )
    es.add(
        cmp.Source(
            label="Grid import",
            outputs={
                bus_el: Flow(
                    variable_costs=var_cost["electricity_prices [Eur/kWh]"]
                )
            },
        )
    )
    es.add(
        cmp.Source(
            label="Gas import",
            outputs={
                bus_el: Flow(variable_costs=var_cost["gas_prices [Eur/kWh]"])
            },
        )
    )

    # Battery
    es.add(
        cmp.GenericStorage(
            label="Battery",
            inputs={bus_el: Flow()},
            outputs={bus_el: Flow()},
            nominal_capacity=investments["battery"],  # kWh
            min_storage_level=0.0,
            max_storage_level=1.0,
            balanced=True,
            loss_rate=0.001,  # 0.1%/h
            inflow_conversion_factor=0.95,  # Lade-Wirkungsgrad
            outflow_conversion_factor=0.95,  # Entlade-Wirkungsgrad
        )
    )

    # Sinks
    es.add(cmp.Sink(label="Excess_el", inputs={bus_el: Flow()}))
    es.add(cmp.Sink(label="Excess_heat", inputs={bus_heat: Flow()}))
    es.add(
        cmp.Sink(
            label="Heat demand",
            inputs={
                bus_heat: Flow(
                    fix=data["heat demand (kW)"], nominal_capacity=5.0
                )
            },
        )
    )
    es.add(
        cmp.Sink(
            label="Electricity demand",
            inputs={
                bus_el: Flow(
                    fix=data["electricity demand (kW)"], nominal_capacity=1.0
                )
            },
        )
    )
    es.add(
        cmp.Sink(
            label="Electric Vehicle",
            inputs={
                bus_el: Flow(fix=data["ev charge (kW)"], nominal_capacity=1.0)
            },
        )
    )
    es.add(
        cmp.Sink(
            label="Grid Feed-in",
            inputs={
                bus_el: Flow(
                    variable_costs=-var_cost["pv_feed_in [Eur/kWh]"] / 1000
                )
            },
        )
    )

    # Heat Pump
    es.add(
        cmp.Converter(
            label="Heat pump",
            inputs={bus_el: Flow()},
            outputs={
                bus_heat: Flow(nominal_capacity=investments["heat pump"])
            },
            conversion_factors={bus_heat: data["cop"]},
        )
    )
    # Gas Boiler
    es.add(
        cmp.Converter(
            label="Gas Boiler",
            inputs={bus_gas: Flow()},
            outputs={
                bus_heat: Flow(nominal_capacity=investments["gas boiler"])
            },
            conversion_factors={bus_heat: data["cop"]},
        )
    )

    graph.create_nx_graph(es, filename=Path(Path.home(), "test_graph.graphml"))

    # Create Model and solve it
    logging.info("Creating Model...")
    m = Model(es)
    logging.info("Solving Model...")
    m.solve(solver="cbc", solve_kwargs={"tee": False})

    # Create Results
    return Results(m)


def process_results(results):
    flow = results["flow"]
    year = flow.index[0].year

    end_time = pytz.utc.localize(
        datetime.strptime(f"{year + 1}-01-01 00:00", "%Y-%m-%d %H:%M")
    )
    intervals = pd.Series(
        flow.index.diff().seconds / 3600, index=flow.index
    ).shift(-1)
    intervals.iloc[-1] = (end_time - flow.index[-2]).seconds / 3600 - 1

    # print(flow.mul(intervals, axis=0).sum())

    soc = results["storage_content"]
    soc.name = "Battery SOC [kWh]"
    investments = results["invest"].rename(
        columns={
            c: c[0].label
            for c in results["invest"].columns
            if isinstance(c, tuple)
        },
    )
    print(investments)


def compare_results(even, uneven):
    flow_e = even["flow"]
    flow_u = uneven["flow"]
    # print(flow_e)
    # print(flow_u)


#
# # interval_hours = df.groupby(buckets).size().sort_index()
# # interval_hours.name = 'interval_hours'
#
#
# print("Energy Balance")
# print(flow.sum())
# print("")
# print("Investment")
# print(investments.squeeze())

# investments.squeeze().plot(kind="bar")
#
# day = 186  # day of the year
# n = 2  # number of days to plot
# flow = flow[day * 24 * 6 : day * 24 * 6 + n * 24 * 6]
# soc = soc[day * 24 * 6 : day * 24 * 6 + 48 * 6]
#
# supply = flow[[c for c in flow.columns if c[1].label == "electricity"]]
# supply = supply.droplevel(1, axis=1)
# supply.rename(columns={c: c.label for c in supply.columns}, inplace=True)
# demand = flow[[c for c in flow.columns if c[0].label == "electricity"]]
# demand = demand.droplevel(0, axis=1)
# demand.rename(columns={c: c.label for c in demand.columns}, inplace=True)


if __name__ == "__main__":
    my_data = prepare_technical_data(10)
    start = datetime.now()
    results_even = solve_model(my_data.even)
    time_even = datetime.now() - start
    start = datetime.now()
    results_uneven = solve_model(my_data.uneven)
    time_uneven = datetime.now() - start
    process_results(results_even)
    process_results(results_uneven)
    compare_results(results_even, results_uneven)
    print("*** Times ****")
    print("even", time_even)
    print("uneven", time_uneven)


1			# -- coding: utf-8 --
2
3			"""
4			SPDX-FileCopyrightText: Patrik Schönfeldt
5			SPDX-FileCopyrightText: DLR e.V.
6
7			SPDX-License-Identifier: MIT
8			"""
9
10			import logging
11			import warnings
12			from collections import namedtuple
13			from datetime import datetime
14			from pathlib import Path
15
16			import pandas as pd
17			import pytz
18			from cost_data import discounted_average_price
19			from cost_data import energy_prices
20			from cost_data import investment_costs
21			from create_timeseries import reshape_unevenly
22			from matplotlib import pyplot as plt
23			from oemof.network import graph
24			from oemof.tools import debugging
25			from oemof.tools import logger
26			from oemof.tools.economics import annuity
27			from shared import prepare_input_data
28
29			from oemof.solph import Bus
30			from oemof.solph import EnergySystem
31			from oemof.solph import Flow
32			from oemof.solph import Investment
33			from oemof.solph import Model
34			from oemof.solph import Results
35			from oemof.solph import components as cmp
36
37			warnings.filterwarnings(
38			"ignore", category=debugging.ExperimentalFeatureWarning
39			)
40			logger.define_logging()
41
42
43			def calculate_annuity(value):
44			return value / 20
45
46
47			def calculate_fix_cost(value):
48			return value / 20
49
50
51			def prepare_technical_data(minutes):
52			data = namedtuple("data", "even uneven")
53			df = prepare_input_data().resample(f"{minutes} min").mean()
54			df["ev charge (kW)"] = 0
55			df_un = reshape_unevenly(df)
56			return data(even=df, uneven=df_un)
57
58
59			def prepare_cost_data():
60			pass
61
62
63			def solve_model(data, year=2025, es=None, n=20, r=0.05):
64			if es is None:
65			es = EnergySystem(timeindex=data.index)
66
67			var_cost = discounted_average_price(energy_prices(), r, n, year)
68			invest_cost = investment_costs().loc[year]
69
70			# Create Investment objects from cost data
71			investments = {}
72			for key in ["gas boiler", "heat pump", "battery", "pv"]:
73			try:
74			epc = annuity(
75			invest_cost[(key, "specific_costs [Eur/kW]")], n, r
76			)
77			except KeyError:
78			epc = annuity(
79			invest_cost[(key, "specific_costs [Eur/kWh]")], n, r
80			)
81			fix_cost = calculate_fix_cost(invest_cost[(key, "fixed_costs [Eur]")])
82			investments[key] = Investment(ep_costs=epc, fixed_costs=fix_cost)
83
84			# Buses
85			bus_el = Bus(label="electricity")
86			bus_heat = Bus(label="heat")
87			bus_gas = Bus(label="gas")
88			es.add(bus_el, bus_heat, bus_gas)
89
90			# Sources
91			es.add(
92			cmp.Source(
93			label="PV",
94			outputs={
95			bus_el: Flow(
96			fix=data["PV (kW/kWp)"],
97			nominal_capacity=investments["pv"],
98			)
99			},
100			)
101			)
102			es.add(
103			cmp.Source(
104			label="Shortage_heat", outputs={bus_heat: Flow(variable_costs=99)}
105			)
106			)
107			es.add(
108			cmp.Source(
109			label="Grid import",
110			outputs={
111			bus_el: Flow(
112			variable_costs=var_cost["electricity_prices [Eur/kWh]"]
113			)
114			},
115			)
116			)
117			es.add(
118			cmp.Source(
119			label="Gas import",
120			outputs={
121			bus_el: Flow(variable_costs=var_cost["gas_prices [Eur/kWh]"])
122			},
123			)
124			)
125
126			# Battery
127			es.add(
128			cmp.GenericStorage(
129			label="Battery",
130			inputs={bus_el: Flow()},
131			outputs={bus_el: Flow()},
132			nominal_capacity=investments["battery"], # kWh
133			min_storage_level=0.0,
134			max_storage_level=1.0,
135			balanced=True,
136			loss_rate=0.001, # 0.1%/h
137			inflow_conversion_factor=0.95, # Lade-Wirkungsgrad
138			outflow_conversion_factor=0.95, # Entlade-Wirkungsgrad
139			)
140			)
141
142			# Sinks
143			es.add(cmp.Sink(label="Excess_el", inputs={bus_el: Flow()}))
144			es.add(cmp.Sink(label="Excess_heat", inputs={bus_heat: Flow()}))
145			es.add(
146			cmp.Sink(
147			label="Heat demand",
148			inputs={
149			bus_heat: Flow(
150			fix=data["heat demand (kW)"], nominal_capacity=5.0
151			)
152			},
153			)
154			)
155			es.add(
156			cmp.Sink(
157			label="Electricity demand",
158			inputs={
159			bus_el: Flow(
160			fix=data["electricity demand (kW)"], nominal_capacity=1.0
161			)
162			},
163			)
164			)
165			es.add(
166			cmp.Sink(
167			label="Electric Vehicle",
168			inputs={
169			bus_el: Flow(fix=data["ev charge (kW)"], nominal_capacity=1.0)
170			},
171			)
172			)
173			es.add(
174			cmp.Sink(
175			label="Grid Feed-in",
176			inputs={
177			bus_el: Flow(
178			variable_costs=-var_cost["pv_feed_in [Eur/kWh]"] / 1000
179			)
180			},
181			)
182			)
183
184			# Heat Pump
185			es.add(
186			cmp.Converter(
187			label="Heat pump",
188			inputs={bus_el: Flow()},
189			outputs={
190			bus_heat: Flow(nominal_capacity=investments["heat pump"])
191			},
192			conversion_factors={bus_heat: data["cop"]},
193			)
194			)
195			# Gas Boiler
196			es.add(
197			cmp.Converter(
198			label="Gas Boiler",
199			inputs={bus_gas: Flow()},
200			outputs={
201			bus_heat: Flow(nominal_capacity=investments["gas boiler"])
202			},
203			conversion_factors={bus_heat: data["cop"]},
204			)
205			)
206
207			graph.create_nx_graph(es, filename=Path(Path.home(), "test_graph.graphml"))
208
209			# Create Model and solve it
210			logging.info("Creating Model...")
211			m = Model(es)
212			logging.info("Solving Model...")
213			m.solve(solver="cbc", solve_kwargs={"tee": False})
214
215			# Create Results
216			return Results(m)
217
218
219			def process_results(results):
220			flow = results["flow"]
221			year = flow.index[0].year
222
223			end_time = pytz.utc.localize(
224			datetime.strptime(f"{year + 1}-01-01 00:00", "%Y-%m-%d %H:%M")
225			)
226			intervals = pd.Series(
227			flow.index.diff().seconds / 3600, index=flow.index
228			).shift(-1)
229			intervals.iloc[-1] = (end_time - flow.index[-2]).seconds / 3600 - 1
230
231			# print(flow.mul(intervals, axis=0).sum())
232
233			soc = results["storage_content"]
234			soc.name = "Battery SOC [kWh]"
235			investments = results["invest"].rename(
236			columns={
237			c: c[0].label
238			for c in results["invest"].columns
239			if isinstance(c, tuple)
240			},
241			)
242			print(investments)
243
244
245			def compare_results(even, uneven):
246			flow_e = even["flow"]
247			flow_u = uneven["flow"]
248			# print(flow_e)
249			# print(flow_u)
250
251
252			#
253			# # interval_hours = df.groupby(buckets).size().sort_index()
254			# # interval_hours.name = 'interval_hours'
255			#
256			#
257			# print("Energy Balance")
258			# print(flow.sum())
259			# print("")
260			# print("Investment")
261			# print(investments.squeeze())
262
263			# investments.squeeze().plot(kind="bar")
264			#
265			# day = 186 # day of the year
266			# n = 2 # number of days to plot
267			# flow = flow[day * 24 * 6 : day * 24 * 6 + n * 24 * 6]
268			# soc = soc[day * 24 * 6 : day * 24 * 6 + 48 * 6]
269			#
270			# supply = flow[[c for c in flow.columns if c[1].label == "electricity"]]
271			# supply = supply.droplevel(1, axis=1)
272			# supply.rename(columns={c: c.label for c in supply.columns}, inplace=True)
273			# demand = flow[[c for c in flow.columns if c[0].label == "electricity"]]
274			# demand = demand.droplevel(0, axis=1)
275			# demand.rename(columns={c: c.label for c in demand.columns}, inplace=True)
276
277
278			if __name__ == "__main__":
279			my_data = prepare_technical_data(10)
280			start = datetime.now()
281			results_even = solve_model(my_data.even)
282			time_even = datetime.now() - start
283			start = datetime.now()
284			results_uneven = solve_model(my_data.uneven)
285			time_uneven = datetime.now() - start
286			process_results(results_even)
287			process_results(results_uneven)
288			compare_results(results_even, results_uneven)
289			print("* Times **")
290			print("even", time_even)
291			print("uneven", time_uneven)
292

oemof / oemof-solph

Pull Request — dev (#1226)

time_series_un_even.solve_model() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like