solph._energy_system.EnergySystem.get_period_duration() - Code Metrics - Inspection of "Steamline investment periods" - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#1183)

by Patrik

created 2025-08-29 13:17 UTC

solph._energy_system.EnergySystem.get_period_duration() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	17
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	5
dl	0
loc	17
rs	10
c	0
b	0
f	0
cc	1
nop	2

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

"""
solph version of oemof.network.energy_system

SPDX-FileCopyrightText: Uwe Krien <[email protected]>
SPDX-FileCopyrightText: Simon Hilpert
SPDX-FileCopyrightText: Cord Kaldemeyer
SPDX-FileCopyrightText: Stephan Günther
SPDX-FileCopyrightText: Birgit Schachler
SPDX-FileCopyrightText: Johannes Kochems

SPDX-License-Identifier: MIT

"""

import collections
import itertools
import warnings

import numpy as np
import pandas as pd
from oemof.network import energy_system as es
from oemof.tools import debugging


class EnergySystem(es.EnergySystem):
    """A variant of the class EnergySystem from
    <oemof.network.network.energy_system.EnergySystem> specially tailored to
    solph.

    In order to work in tandem with solph, instances of this class always use
    solph.GROUPINGS <oemof.solph.GROUPINGS>. If custom groupings are
    supplied via the `groupings` keyword argument, solph.GROUPINGS
    <oemof.solph.GROUPINGS> is prepended to those.

    If you know what you are doing and want to use solph without
    solph.GROUPINGS <oemof.solph.GROUPINGS>, you can just use
    EnergySystem <oemof.network.network.energy_system.EnergySystem>` of
    oemof.network directly.

    Parameters
    ----------
    timeindex : sequence of ascending numeric values
        Typically a pandas.DatetimeIndex is used,
        but for example also a list of floats works.

    infer_last_interval : bool
        Add an interval to the last time point. The end time of this interval
        is unknown so it does only work for an equidistant DatetimeIndex with
        a 'freq' attribute that is not None. The parameter has no effect on the
        timeincrement parameter.

    investment_times : list or None
        The points in time investments can be made. Defaults to timeindex[0].
        If multiple times are specified, it leads to creating a multi-period
        model.

    tsa_parameters : list of dicts, dict or None
        Parameter can be set in order to use aggregated timeseries from TSAM.
        If multi-period model is used, one dict per period has to be set.
        If no multi-period (aka single period) approach is selected, a single
        dict can be provided.
        If parameter is None, model is set up as usual.

        Dict must contain keys `timesteps_per_period`
        (from TSAMs `hoursPerPeriod`), `order` (from TSAMs `clusterOrder`) and
        `occurrences` (from TSAMs `clusterPeriodNoOccur`).
        When activated, storage equations and flow rules for full_load_time
        will be adapted. Note that timeseries for components have to
        be set up as already aggregated timeseries.

    kwargs
    """

    def __init__(
        self,
        timeindex=None,
        timeincrement=None,
        infer_last_interval=False,
        investment_times=None,
        tsa_parameters=None,
        groupings=None,
    ):
        # Doing imports at runtime is generally frowned upon, but should work
        # for now. See the TODO in :func:`constraint_grouping
        # <oemof.solph.groupings.constraint_grouping>` for more information.
        from oemof.solph import GROUPINGS

        if groupings is None:
            groupings = []
        groupings = GROUPINGS + groupings

        if infer_last_interval is True and timeindex is not None:
            # Add one time interval to the timeindex by adding one time point.
            if timeindex.freq is None:
                msg = (
                    "You cannot infer the last interval if the 'freq' "
                    "attribute of your DatetimeIndex is None. Set "
                    " 'infer_last_interval=False' or specify a DatetimeIndex "
                    "with a valid frequency."
                )
                raise AttributeError(msg)

            timeindex = timeindex.union(
                pd.date_range(
                    timeindex[-1] + timeindex.freq,
                    periods=1,
                    freq=timeindex.freq,
                )
            )

        # catch wrong combinations and infer timeincrement from timeindex.
        if timeincrement is not None:
            if timeindex is None:
                msg = (
                    "Initialising an EnergySystem using a timeincrement"
                    " is deprecated. Please give a timeindex instead."
                )
                warnings.warn(msg, FutureWarning)
            else:
                msg = (
                    "The timeincrement is infered from the given timeindex."
                    " As both parameters might be conflicting to each other,"
                    " you cannot define both at the same time."
                    " Please give only a timeindex."
                )
                raise AttributeError(msg)

        elif timeindex is None and timeincrement is not None:
            timeindex = pd.Index(
                np.append(np.array([0]), np.cumsum(timeincrement))
            )
        elif timeindex is not None and timeincrement is None:
            if tsa_parameters is not None:
                pass
            else:
                try:
                    df = pd.DataFrame(timeindex)
                except ValueError:
                    raise ValueError("Invalid timeindex.")
                timedelta = df.diff()
                if isinstance(timeindex, pd.DatetimeIndex):
                    timeincrement = timedelta / np.timedelta64(1, "h")
                else:
                    timeincrement = timedelta
                # we want a series (squeeze)
                # without the first item (no delta defined for first entry)
                # but starting with index 0 (reset)
                timeincrement = timeincrement.squeeze()[1:].reset_index(
                    drop=True
                )

        if timeincrement is not None and (pd.Series(timeincrement) <= 0).any():
            msg = (
                "The time increment is inconsistent. Negative values and zero "
                "are not allowed.\nThis is caused by a inconsistent "
                "timeincrement parameter or an incorrect timeindex."
            )
            raise TypeError(msg)
        if tsa_parameters is not None:
            msg = (
                "CAUTION! You specified the 'tsa_parameters' attribute for "
                "your energy system.\n This will lead to setting up "
                "energysystem with aggregated timeseries. "
                "Storages and flows will be adapted accordingly.\n"
                "Please be aware that the feature is experimental as of "
                "now. If you find anything suspicious or any bugs, "
                "please report them."
            )
            warnings.warn(msg, debugging.SuspiciousUsageWarning)

            if isinstance(tsa_parameters, dict):
                # Set up tsa_parameters for single period:
                tsa_parameters = [tsa_parameters]

            # Construct occurrences of typical periods
            if investment_times is not None:
                for p in range(len(investment_times)):
                    tsa_parameters[p]["occurrences"] = collections.Counter(
                        tsa_parameters[p]["order"]
                    )
            else:
                tsa_parameters[0]["occurrences"] = collections.Counter(
                    tsa_parameters[0]["order"]
                )

            # If segmentation is used, timesteps is set to number of
            # segmentations per period.
            # Otherwise, default timesteps_per_period is used.
            for params in tsa_parameters:
                if "segments" in params:
                    params["timesteps"] = int(
                        len(params["segments"]) / len(params["occurrences"])
                    )
                else:
                    params["timesteps"] = params["timesteps_per_period"]
        self.tsa_parameters = tsa_parameters

        timeincrement = self._init_timeincrement(
            timeincrement, timeindex, investment_times, tsa_parameters
        )
        super().__init__(
            groupings=groupings,
            timeindex=timeindex,
            timeincrement=timeincrement,
        )

        # bare system to load pickled data
        if self.timeindex is not None:
            if investment_times is None:
                self.investment_times = [self.timeindex[0]]
            else:
                self.investment_times = sorted(
                    set([self.timeindex[0]] + investment_times)
                )

    @staticmethod
    def _init_timeincrement(timeincrement, timeindex, periods, tsa_parameters):
        """Check and initialize timeincrement"""

        # Timeincrement in TSAM mode
        if (
            timeincrement is not None
            and tsa_parameters is not None
            and any("segments" in params for params in tsa_parameters)
        ):
            msg = (
                "You must not specify timeincrement in TSAM mode. "
                "TSAM will define timeincrement itself."
            )
            raise AttributeError(msg)
        if (
            tsa_parameters is not None
            and any("segments" in params for params in tsa_parameters)
            and not all("segments" in params for params in tsa_parameters)
        ):
            msg = (
                "You have to set up segmentation in all periods, "
                "if you want to use segmentation in TSAM mode"
            )
            raise AttributeError(msg)
        if tsa_parameters is not None and all(
            "segments" in params for params in tsa_parameters
        ):
            # Concatenate segments from TSAM parameters to get timeincrement
            return list(
                itertools.chain(
                    *[params["segments"].values() for params in tsa_parameters]
                )
            )

        elif timeindex is not None and timeincrement is None:
            df = pd.DataFrame(timeindex)
            timedelta = df.diff()
            timeincrement = timedelta / np.timedelta64(1, "h")

            # we want a series (squeeze)
            # without the first item (no delta defined for first entry)
            # but starting with index 0 (reset)
            return timeincrement.squeeze()[1:].reset_index(drop=True)

        return timeincrement


1			# -- coding: utf-8 --
2
3			"""
4			solph version of oemof.network.energy_system
5
6			SPDX-FileCopyrightText: Uwe Krien <[email protected]>
7			SPDX-FileCopyrightText: Simon Hilpert
8			SPDX-FileCopyrightText: Cord Kaldemeyer
9			SPDX-FileCopyrightText: Stephan Günther
10			SPDX-FileCopyrightText: Birgit Schachler
11			SPDX-FileCopyrightText: Johannes Kochems
12
13			SPDX-License-Identifier: MIT
14
15			"""
16
17			import collections
18			import itertools
19			import warnings
20
21			import numpy as np
22			import pandas as pd
23			from oemof.network import energy_system as es
24			from oemof.tools import debugging
25
26
27			class EnergySystem(es.EnergySystem):
28			"""A variant of the class EnergySystem from
29			<oemof.network.network.energy_system.EnergySystem> specially tailored to
30			solph.
31
32			In order to work in tandem with solph, instances of this class always use
33			solph.GROUPINGS <oemof.solph.GROUPINGS>. If custom groupings are
34			supplied via the `groupings` keyword argument, solph.GROUPINGS
35			<oemof.solph.GROUPINGS> is prepended to those.
36
37			If you know what you are doing and want to use solph without
38			solph.GROUPINGS <oemof.solph.GROUPINGS>, you can just use
39			EnergySystem <oemof.network.network.energy_system.EnergySystem>` of
40			oemof.network directly.
41
42			Parameters
43			----------
44			timeindex : sequence of ascending numeric values
45			Typically a pandas.DatetimeIndex is used,
46			but for example also a list of floats works.
47
48			infer_last_interval : bool
49			Add an interval to the last time point. The end time of this interval
50			is unknown so it does only work for an equidistant DatetimeIndex with
51			a 'freq' attribute that is not None. The parameter has no effect on the
52			timeincrement parameter.
53
54			investment_times : list or None
55			The points in time investments can be made. Defaults to timeindex[0].
56			If multiple times are specified, it leads to creating a multi-period
57			model.
58
59			tsa_parameters : list of dicts, dict or None
60			Parameter can be set in order to use aggregated timeseries from TSAM.
61			If multi-period model is used, one dict per period has to be set.
62			If no multi-period (aka single period) approach is selected, a single
63			dict can be provided.
64			If parameter is None, model is set up as usual.
65
66			Dict must contain keys `timesteps_per_period`
67			(from TSAMs `hoursPerPeriod`), `order` (from TSAMs `clusterOrder`) and
68			`occurrences` (from TSAMs `clusterPeriodNoOccur`).
69			When activated, storage equations and flow rules for full_load_time
70			will be adapted. Note that timeseries for components have to
71			be set up as already aggregated timeseries.
72
73			kwargs
74			"""
75
76			def __init__(
77			self,
78			timeindex=None,
79			timeincrement=None,
80			infer_last_interval=False,
81			investment_times=None,
82			tsa_parameters=None,
83			groupings=None,
84			):
85			# Doing imports at runtime is generally frowned upon, but should work
86			# for now. See the TODO in :func:`constraint_grouping
87			# <oemof.solph.groupings.constraint_grouping>` for more information.
88			from oemof.solph import GROUPINGS
89
90			if groupings is None:
91			groupings = []
92			groupings = GROUPINGS + groupings
93
94			if infer_last_interval is True and timeindex is not None:
95			# Add one time interval to the timeindex by adding one time point.
96			if timeindex.freq is None:
97			msg = (
98			"You cannot infer the last interval if the 'freq' "
99			"attribute of your DatetimeIndex is None. Set "
100			" 'infer_last_interval=False' or specify a DatetimeIndex "
101			"with a valid frequency."
102			)
103			raise AttributeError(msg)
104
105			timeindex = timeindex.union(
106			pd.date_range(
107			timeindex[-1] + timeindex.freq,
108			periods=1,
109			freq=timeindex.freq,
110			)
111			)
112
113			# catch wrong combinations and infer timeincrement from timeindex.
114			if timeincrement is not None:
115			if timeindex is None:
116			msg = (
117			"Initialising an EnergySystem using a timeincrement"
118			" is deprecated. Please give a timeindex instead."
119			)
120			warnings.warn(msg, FutureWarning)
121			else:
122			msg = (
123			"The timeincrement is infered from the given timeindex."
124			" As both parameters might be conflicting to each other,"
125			" you cannot define both at the same time."
126			" Please give only a timeindex."
127			)
128			raise AttributeError(msg)
129
130			elif timeindex is None and timeincrement is not None:
131			timeindex = pd.Index(
132			np.append(np.array([0]), np.cumsum(timeincrement))
133			)
134			elif timeindex is not None and timeincrement is None:
135			if tsa_parameters is not None:
136			pass
137			else:
138			try:
139			df = pd.DataFrame(timeindex)
140			except ValueError:
141			raise ValueError("Invalid timeindex.")
142			timedelta = df.diff()
143			if isinstance(timeindex, pd.DatetimeIndex):
144			timeincrement = timedelta / np.timedelta64(1, "h")
145			else:
146			timeincrement = timedelta
147			# we want a series (squeeze)
148			# without the first item (no delta defined for first entry)
149			# but starting with index 0 (reset)
150			timeincrement = timeincrement.squeeze()[1:].reset_index(
151			drop=True
152			)
153
154			if timeincrement is not None and (pd.Series(timeincrement) <= 0).any():
155			msg = (
156			"The time increment is inconsistent. Negative values and zero "
157			"are not allowed.\nThis is caused by a inconsistent "
158			"timeincrement parameter or an incorrect timeindex."
159			)
160			raise TypeError(msg)
161			if tsa_parameters is not None:
162			msg = (
163			"CAUTION! You specified the 'tsa_parameters' attribute for "
164			"your energy system.\n This will lead to setting up "
165			"energysystem with aggregated timeseries. "
166			"Storages and flows will be adapted accordingly.\n"
167			"Please be aware that the feature is experimental as of "
168			"now. If you find anything suspicious or any bugs, "
169			"please report them."
170			)
171			warnings.warn(msg, debugging.SuspiciousUsageWarning)
172
173			if isinstance(tsa_parameters, dict):
174			# Set up tsa_parameters for single period:
175			tsa_parameters = [tsa_parameters]
176
177			# Construct occurrences of typical periods
178			if investment_times is not None:
179			for p in range(len(investment_times)):
180			tsa_parameters[p]["occurrences"] = collections.Counter(
181			tsa_parameters[p]["order"]
182			)
183			else:
184			tsa_parameters[0]["occurrences"] = collections.Counter(
185			tsa_parameters[0]["order"]
186			)
187
188			# If segmentation is used, timesteps is set to number of
189			# segmentations per period.
190			# Otherwise, default timesteps_per_period is used.
191			for params in tsa_parameters:
192			if "segments" in params:
193			params["timesteps"] = int(
194			len(params["segments"]) / len(params["occurrences"])
195			)
196			else:
197			params["timesteps"] = params["timesteps_per_period"]
198			self.tsa_parameters = tsa_parameters
199
200			timeincrement = self._init_timeincrement(
201			timeincrement, timeindex, investment_times, tsa_parameters
202			)
203			super().__init__(
204			groupings=groupings,
205			timeindex=timeindex,
206			timeincrement=timeincrement,
207			)
208
209			# bare system to load pickled data
210			if self.timeindex is not None:
211			if investment_times is None:
212			self.investment_times = [self.timeindex[0]]
213			else:
214			self.investment_times = sorted(
215			set([self.timeindex[0]] + investment_times)
216			)
217
218			@staticmethod
219			def _init_timeincrement(timeincrement, timeindex, periods, tsa_parameters):
220			"""Check and initialize timeincrement"""
221
222			# Timeincrement in TSAM mode
223			if (
224			timeincrement is not None
225			and tsa_parameters is not None
226			and any("segments" in params for params in tsa_parameters)
227			):
228			msg = (
229			"You must not specify timeincrement in TSAM mode. "
230			"TSAM will define timeincrement itself."
231			)
232			raise AttributeError(msg)
233			if (
234			tsa_parameters is not None
235			and any("segments" in params for params in tsa_parameters)
236			and not all("segments" in params for params in tsa_parameters)
237			):
238			msg = (
239			"You have to set up segmentation in all periods, "
240			"if you want to use segmentation in TSAM mode"
241			)
242			raise AttributeError(msg)
243			if tsa_parameters is not None and all(
244			"segments" in params for params in tsa_parameters
245			):
246			# Concatenate segments from TSAM parameters to get timeincrement
247			return list(
248			itertools.chain(
249			*[params["segments"].values() for params in tsa_parameters]
250			)
251			)
252
253			elif timeindex is not None and timeincrement is None:
254			df = pd.DataFrame(timeindex)
255			timedelta = df.diff()
256			timeincrement = timedelta / np.timedelta64(1, "h")
257
258			# we want a series (squeeze)
259			# without the first item (no delta defined for first entry)
260			# but starting with index 0 (reset)
261			return timeincrement.squeeze()[1:].reset_index(drop=True)
262
263			return timeincrement
264

oemof / oemof-solph

Pull Request — dev (#1183)

solph._energy_system.EnergySystem.get_period_duration() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like