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-18 21:24 UTC

time_series_un_even.solve_model() B

↳ Parent: time_series_un_even

Complexity

Conditions

Size

Total Lines	173
Code Lines	101

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	101
dl	0
loc	173
rs	7
c	0
b	0
f	0
cc	3
nop	2

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 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 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_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 solve_model(data, year=2025):
    es = EnergySystem(timeindex=data.index)

    var_cost = energy_prices()

    # ToDo: Hier muss jetzt noch die Berechnung mit barwert und Annuität her.
    var_cost = var_cost.loc[year]

    start_year = 2025
    idx = pd.period_range(start=f"{start_year}", periods=20, freq="Y")
    df = pd.DataFrame(index=idx)
    print(df.index)

    # ToDo: Wenn wir von einer Lebensdauer von 20 Jahren ausgehen, dann würde
    #  hier eine einfache Annuität ohne Ersatzbeschaffung reichen
    invest_cost = investment_costs().loc[year]
    # MultiIndex([(  'gas boiler',  'specific_costs [Eur/kW]'),
    #             (  'gas boiler',        'fixed_costs [Eur]'),
    #             (   'heat pump',  'specific_costs [Eur/kW]'),
    #             (   'heat pump',        'fixed_costs [Eur]'),
    #             ('heat storage',  'specific_costs [Eur/m3]'),
    #             ('heat storage',        'fixed_costs [Eur]'),
    #             (          'pv',  'specific_costs [Eur/kW]'),
    #             (          'pv',        'fixed_costs [Eur]'),
    #             (     'battery', 'specific_costs [Eur/kWh]'),
    #             (     'battery',        'fixed_costs [Eur]')],
    #            )
    investments = {}
    for key in ["gas boiler", "heat pump", "battery", "pv"]:
        try:
            annuity = calculate_annuity(
                invest_cost[(key, "specific_costs [Eur/kW]")]
            )
        except KeyError:
            annuity = calculate_annuity(
                invest_cost[(key, "specific_costs [Eur/kWh]")]
            )
        fix_cost = calculate_fix_cost(invest_cost[(key, "fixed_costs [Eur]")])
        investments[key] = Investment(ep_costs=annuity, 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": True})

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

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