solph._energy_system.EnergySystem.get_period_duration() - Code Metrics - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

EnergySystem.get_period_duration() A
last analyzed 2025-08-20 07:05 UTC

↳ Parent: solph._energy_system

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.

    periods : list or None
        The periods of a multi-period model.
        If this is explicitly specified, it leads to creating a multi-period
        model, providing a respective user warning as a feedback.

        list of pd.date_range objects carrying the timeindex for the
        respective period;

        For a standard model, periods are not (to be) declared, i.e. None.
        A list with one entry is derived, i.e. [0].

    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.

    use_remaining_value : bool
        If True, compare the remaining value of an investment to the
        original value (only applicable for multi-period models)

    kwargs
    """

    def __init__(
        self,
        timeindex=None,
        timeincrement=None,
        infer_last_interval=False,
        periods=None,
        tsa_parameters=None,
        use_remaining_value=False,
        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:
                if periods is None:
                    msg = (
                        "Specifying the timeincrement and the timeindex"
                        " parameter at the same time is not allowed since"
                        " these might be conflicting to each other."
                    )
                    raise AttributeError(msg)
                else:
                    msg = (
                        "Ensure that your timeindex and timeincrement are "
                        "consistent."
                    )
                    warnings.warn(msg, debugging.ExperimentalFeatureWarning)

        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 periods is not None:
                for p in range(len(periods)):
                    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, periods, tsa_parameters
        )
        super().__init__(
            groupings=groupings,
            timeindex=timeindex,
            timeincrement=timeincrement,
        )

        self.periods = periods
        if self.periods is not None:
            msg = (
                "CAUTION! You specified the 'periods' attribute for your "
                "energy system.\n This will lead to creating "
                "a multi-period optimization modeling which can be "
                "used e.g. for long-term investment modeling.\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.ExperimentalFeatureWarning)
            self._extract_periods_years()
            self._extract_periods_matrix()
            self._extract_end_year_of_optimization()
            self.use_remaining_value = use_remaining_value
        else:
            self.end_year_of_optimization = 1

    def _extract_periods_years(self):
        """Map years in optimization to respective period based on time indices

        Attribute `periods_years` of type list is set. It contains
        the year of the start of each period, relative to the
        start of the optimization run and starting with 0.
        """
        periods_years = [0]
        start_year = self.periods[0].min().year
        for k, v in enumerate(self.periods):
            if k >= 1:
                periods_years.append(v.min().year - start_year)

        self.periods_years = periods_years

    def _extract_periods_matrix(self):
        """Determines a matrix describing the temporal distance to each period.

        Attribute `periods_matrix` of type list np.array is set.
        Rows represent investment/commissioning periods, columns represent
        decommissioning periods. The values describe the temporal distance
        between each investment period to each decommissioning period.
        """
        periods_matrix = []
        period_years = np.array(self.periods_years)
        for v in period_years:
            row = period_years - v
            row = np.where(row < 0, 0, row)
            periods_matrix.append(row)
        self.periods_matrix = np.array(periods_matrix)

    def _extract_end_year_of_optimization(self):
        """Extract the end of the optimization in years

        Attribute `end_year_of_optimization` of int is set.
        """
        duration_last_period = self.get_period_duration(-1)
        self.end_year_of_optimization = (
            self.periods_years[-1] + duration_last_period
        )

    def get_period_duration(self, period):
        """Get duration of a period in full years

        Parameters
        ----------
        period : int
            Period for which the duration in years shall be obtained

        Returns
        -------
        int
            Duration of the period
        """
        return (
            self.periods[period].max().year
            - self.periods[period].min().year
            + 1
        )

    @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			periods : list or None
55			The periods of a multi-period model.
56			If this is explicitly specified, it leads to creating a multi-period
57			model, providing a respective user warning as a feedback.
58
59			list of pd.date_range objects carrying the timeindex for the
60			respective period;
61
62			For a standard model, periods are not (to be) declared, i.e. None.
63			A list with one entry is derived, i.e. [0].
64
65			tsa_parameters : list of dicts, dict or None
66			Parameter can be set in order to use aggregated timeseries from TSAM.
67			If multi-period model is used, one dict per period has to be set.
68			If no multi-period (aka single period) approach is selected, a single
69			dict can be provided.
70			If parameter is None, model is set up as usual.
71
72			Dict must contain keys `timesteps_per_period`
73			(from TSAMs `hoursPerPeriod`), `order` (from TSAMs `clusterOrder`) and
74			`occurrences` (from TSAMs `clusterPeriodNoOccur`).
75			When activated, storage equations and flow rules for full_load_time
76			will be adapted. Note that timeseries for components have to
77			be set up as already aggregated timeseries.
78
79			use_remaining_value : bool
80			If True, compare the remaining value of an investment to the
81			original value (only applicable for multi-period models)
82
83			kwargs
84			"""
85
86			def __init__(
87			self,
88			timeindex=None,
89			timeincrement=None,
90			infer_last_interval=False,
91			periods=None,
92			tsa_parameters=None,
93			use_remaining_value=False,
94			groupings=None,
95			):
96			# Doing imports at runtime is generally frowned upon, but should work
97			# for now. See the TODO in :func:`constraint_grouping
98			# <oemof.solph.groupings.constraint_grouping>` for more information.
99			from oemof.solph import GROUPINGS
100
101			if groupings is None:
102			groupings = []
103			groupings = GROUPINGS + groupings
104
105			if infer_last_interval is True and timeindex is not None:
106			# Add one time interval to the timeindex by adding one time point.
107			if timeindex.freq is None:
108			msg = (
109			"You cannot infer the last interval if the 'freq' "
110			"attribute of your DatetimeIndex is None. Set "
111			" 'infer_last_interval=False' or specify a DatetimeIndex "
112			"with a valid frequency."
113			)
114			raise AttributeError(msg)
115
116			timeindex = timeindex.union(
117			pd.date_range(
118			timeindex[-1] + timeindex.freq,
119			periods=1,
120			freq=timeindex.freq,
121			)
122			)
123
124			# catch wrong combinations and infer timeincrement from timeindex.
125			if timeincrement is not None:
126			if timeindex is None:
127			msg = (
128			"Initialising an EnergySystem using a timeincrement"
129			" is deprecated. Please give a timeindex instead."
130			)
131			warnings.warn(msg, FutureWarning)
132			else:
133			if periods is None:
134			msg = (
135			"Specifying the timeincrement and the timeindex"
136			" parameter at the same time is not allowed since"
137			" these might be conflicting to each other."
138			)
139			raise AttributeError(msg)
140			else:
141			msg = (
142			"Ensure that your timeindex and timeincrement are "
143			"consistent."
144			)
145			warnings.warn(msg, debugging.ExperimentalFeatureWarning)
146
147			elif timeindex is not None and timeincrement is None:
148			if tsa_parameters is not None:
149			pass
150			else:
151			try:
152			df = pd.DataFrame(timeindex)
153			except ValueError:
154			raise ValueError("Invalid timeindex.")
155			timedelta = df.diff()
156			if isinstance(timeindex, pd.DatetimeIndex):
157			timeincrement = timedelta / np.timedelta64(1, "h")
158			else:
159			timeincrement = timedelta
160			# we want a series (squeeze)
161			# without the first item (no delta defined for first entry)
162			# but starting with index 0 (reset)
163			timeincrement = timeincrement.squeeze()[1:].reset_index(
164			drop=True
165			)
166
167			if timeincrement is not None and (pd.Series(timeincrement) <= 0).any():
168			msg = (
169			"The time increment is inconsistent. Negative values and zero "
170			"are not allowed.\nThis is caused by a inconsistent "
171			"timeincrement parameter or an incorrect timeindex."
172			)
173			raise TypeError(msg)
174			if tsa_parameters is not None:
175			msg = (
176			"CAUTION! You specified the 'tsa_parameters' attribute for "
177			"your energy system.\n This will lead to setting up "
178			"energysystem with aggregated timeseries. "
179			"Storages and flows will be adapted accordingly.\n"
180			"Please be aware that the feature is experimental as of "
181			"now. If you find anything suspicious or any bugs, "
182			"please report them."
183			)
184			warnings.warn(msg, debugging.SuspiciousUsageWarning)
185
186			if isinstance(tsa_parameters, dict):
187			# Set up tsa_parameters for single period:
188			tsa_parameters = [tsa_parameters]
189
190			# Construct occurrences of typical periods
191			if periods is not None:
192			for p in range(len(periods)):
193			tsa_parameters[p]["occurrences"] = collections.Counter(
194			tsa_parameters[p]["order"]
195			)
196			else:
197			tsa_parameters[0]["occurrences"] = collections.Counter(
198			tsa_parameters[0]["order"]
199			)
200
201			# If segmentation is used, timesteps is set to number of
202			# segmentations per period.
203			# Otherwise, default timesteps_per_period is used.
204			for params in tsa_parameters:
205			if "segments" in params:
206			params["timesteps"] = int(
207			len(params["segments"]) / len(params["occurrences"])
208			)
209			else:
210			params["timesteps"] = params["timesteps_per_period"]
211			self.tsa_parameters = tsa_parameters
212
213			timeincrement = self._init_timeincrement(
214			timeincrement, timeindex, periods, tsa_parameters
215			)
216			super().__init__(
217			groupings=groupings,
218			timeindex=timeindex,
219			timeincrement=timeincrement,
220			)
221
222			self.periods = periods
223			if self.periods is not None:
224			msg = (
225			"CAUTION! You specified the 'periods' attribute for your "
226			"energy system.\n This will lead to creating "
227			"a multi-period optimization modeling which can be "
228			"used e.g. for long-term investment modeling.\n"
229			"Please be aware that the feature is experimental as of "
230			"now. If you find anything suspicious or any bugs, "
231			"please report them."
232			)
233			warnings.warn(msg, debugging.ExperimentalFeatureWarning)
234			self._extract_periods_years()
235			self._extract_periods_matrix()
236			self._extract_end_year_of_optimization()
237			self.use_remaining_value = use_remaining_value
238			else:
239			self.end_year_of_optimization = 1
240
241			def _extract_periods_years(self):
242			"""Map years in optimization to respective period based on time indices
243
244			Attribute `periods_years` of type list is set. It contains
245			the year of the start of each period, relative to the
246			start of the optimization run and starting with 0.
247			"""
248			periods_years = [0]
249			start_year = self.periods[0].min().year
250			for k, v in enumerate(self.periods):
251			if k >= 1:
252			periods_years.append(v.min().year - start_year)
253
254			self.periods_years = periods_years
255
256			def _extract_periods_matrix(self):
257			"""Determines a matrix describing the temporal distance to each period.
258
259			Attribute `periods_matrix` of type list np.array is set.
260			Rows represent investment/commissioning periods, columns represent
261			decommissioning periods. The values describe the temporal distance
262			between each investment period to each decommissioning period.
263			"""
264			periods_matrix = []
265			period_years = np.array(self.periods_years)
266			for v in period_years:
267			row = period_years - v
268			row = np.where(row < 0, 0, row)
269			periods_matrix.append(row)
270			self.periods_matrix = np.array(periods_matrix)
271
272			def _extract_end_year_of_optimization(self):
273			"""Extract the end of the optimization in years
274
275			Attribute `end_year_of_optimization` of int is set.
276			"""
277			duration_last_period = self.get_period_duration(-1)
278			self.end_year_of_optimization = (
279			self.periods_years[-1] + duration_last_period
280			)
281
282			def get_period_duration(self, period):
283			"""Get duration of a period in full years
284
285			Parameters
286			----------
287			period : int
288			Period for which the duration in years shall be obtained
289
290			Returns
291			-------
292			int
293			Duration of the period
294			"""
295			return (
296			self.periods[period].max().year
297			- self.periods[period].min().year
298			+ 1
299			)
300
301			@staticmethod
302			def _init_timeincrement(timeincrement, timeindex, periods, tsa_parameters):
303			"""Check and initialize timeincrement"""
304
305			# Timeincrement in TSAM mode
306			if (
307			timeincrement is not None
308			and tsa_parameters is not None
309			and any("segments" in params for params in tsa_parameters)
310			):
311			msg = (
312			"You must not specify timeincrement in TSAM mode. "
313			"TSAM will define timeincrement itself."
314			)
315			raise AttributeError(msg)
316			if (
317			tsa_parameters is not None
318			and any("segments" in params for params in tsa_parameters)
319			and not all("segments" in params for params in tsa_parameters)
320			):
321			msg = (
322			"You have to set up segmentation in all periods, "
323			"if you want to use segmentation in TSAM mode"
324			)
325			raise AttributeError(msg)
326			if tsa_parameters is not None and all(
327			"segments" in params for params in tsa_parameters
328			):
329			# Concatenate segments from TSAM parameters to get timeincrement
330			return list(
331			itertools.chain(
332			*[params["segments"].values() for params in tsa_parameters]
333			)
334			)
335
336			elif timeindex is not None and timeincrement is None:
337			df = pd.DataFrame(timeindex)
338			timedelta = df.diff()
339			timeincrement = timedelta / np.timedelta64(1, "h")
340
341			# we want a series (squeeze)
342			# without the first item (no delta defined for first entry)
343			# but starting with index 0 (reset)
344			return timeincrement.squeeze()[1:].reset_index(drop=True)
345
346			return timeincrement
347

oemof / oemof-solph

EnergySystem.get_period_duration() A last analyzed 2025-08-20 07:05 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

EnergySystem.get_period_duration() A
last analyzed 2025-08-20 07:05 UTC