solph._energy_system.EnergySystem._init_timeincrement() - Code Metrics - Inspection of "Merge pull request #980 from oemof/feature/integra..." - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev ( 077813...629ebf )

by Patrik

created 2025-03-10 11:26 UTC

EnergySystem._init_timeincrement() C

↳ Parent: solph._energy_system

Complexity

Conditions

Size

Total Lines	46
Code Lines	27

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	27
dl	0
loc	46
rs	5.4
c	0
b	0
f	0
cc	11
nop	4

How to fix Complexity

# -*- 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 : pandas.DatetimeIndex

    timeincrement : iterable

    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=None,
        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 not (
            isinstance(timeindex, pd.DatetimeIndex)
            or isinstance(timeindex, type(None))
        ):
            msg = (
                "Parameter 'timeindex' has to be of type "
                "pandas.DatetimeIndex or NoneType and not of type {0}"
            )
            raise TypeError(msg.format(type(timeindex)))

        if infer_last_interval is None and timeindex is not None:
            msg = (
                "The default behaviour will change in future versions.\n"
                "At the moment the last interval of an equidistant time "
                "index is added implicitly by default. Set "
                "'infer_last_interval' explicitly 'True' or 'False' to avoid "
                "this warning. In future versions 'False' will be the default"
                "behaviour"
            )
            warnings.warn(msg, FutureWarning)
            infer_last_interval = True

        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 and timeindex is not None:
            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:
                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)
                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 : pandas.DatetimeIndex
45
46			timeincrement : iterable
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=None,
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 not (
106			isinstance(timeindex, pd.DatetimeIndex)
107			or isinstance(timeindex, type(None))
108			):
109			msg = (
110			"Parameter 'timeindex' has to be of type "
111			"pandas.DatetimeIndex or NoneType and not of type {0}"
112			)
113			raise TypeError(msg.format(type(timeindex)))
114
115			if infer_last_interval is None and timeindex is not None:
116			msg = (
117			"The default behaviour will change in future versions.\n"
118			"At the moment the last interval of an equidistant time "
119			"index is added implicitly by default. Set "
120			"'infer_last_interval' explicitly 'True' or 'False' to avoid "
121			"this warning. In future versions 'False' will be the default"
122			"behaviour"
123			)
124			warnings.warn(msg, FutureWarning)
125			infer_last_interval = True
126
127			if infer_last_interval is True and timeindex is not None:
128			# Add one time interval to the timeindex by adding one time point.
129			if timeindex.freq is None:
130			msg = (
131			"You cannot infer the last interval if the 'freq' "
132			"attribute of your DatetimeIndex is None. Set "
133			" 'infer_last_interval=False' or specify a DatetimeIndex "
134			"with a valid frequency."
135			)
136			raise AttributeError(msg)
137
138			timeindex = timeindex.union(
139			pd.date_range(
140			timeindex[-1] + timeindex.freq,
141			periods=1,
142			freq=timeindex.freq,
143			)
144			)
145
146			# catch wrong combinations and infer timeincrement from timeindex.
147			if timeincrement is not None and timeindex is not None:
148			if periods is None:
149			msg = (
150			"Specifying the timeincrement and the timeindex parameter "
151			"at the same time is not allowed since these might be "
152			"conflicting to each other."
153			)
154			raise AttributeError(msg)
155			else:
156			msg = (
157			"Ensure that your timeindex and timeincrement are "
158			"consistent."
159			)
160			warnings.warn(msg, debugging.ExperimentalFeatureWarning)
161
162			elif timeindex is not None and timeincrement is None:
163			if tsa_parameters is not None:
164			pass
165			else:
166			df = pd.DataFrame(timeindex)
167			timedelta = df.diff()
168			timeincrement = timedelta / np.timedelta64(1, "h")
169
170			# we want a series (squeeze)
171			# without the first item (no delta defined for first entry)
172			# but starting with index 0 (reset)
173			timeincrement = timeincrement.squeeze()[1:].reset_index(
174			drop=True
175			)
176
177			if timeincrement is not None and (pd.Series(timeincrement) <= 0).any():
178			msg = (
179			"The time increment is inconsistent. Negative values and zero "
180			"are not allowed.\nThis is caused by a inconsistent "
181			"timeincrement parameter or an incorrect timeindex."
182			)
183			raise TypeError(msg)
184			if tsa_parameters is not None:
185			msg = (
186			"CAUTION! You specified the 'tsa_parameters' attribute for "
187			"your energy system.\n This will lead to setting up "
188			"energysystem with aggregated timeseries. "
189			"Storages and flows will be adapted accordingly.\n"
190			"Please be aware that the feature is experimental as of "
191			"now. If you find anything suspicious or any bugs, "
192			"please report them."
193			)
194			warnings.warn(msg, debugging.SuspiciousUsageWarning)
195
196			if isinstance(tsa_parameters, dict):
197			# Set up tsa_parameters for single period:
198			tsa_parameters = [tsa_parameters]
199
200			# Construct occurrences of typical periods
201			if periods is not None:
202			for p in range(len(periods)):
203			tsa_parameters[p]["occurrences"] = collections.Counter(
204			tsa_parameters[p]["order"]
205			)
206			else:
207			tsa_parameters[0]["occurrences"] = collections.Counter(
208			tsa_parameters[0]["order"]
209			)
210
211			# If segmentation is used, timesteps is set to number of
212			# segmentations per period.
213			# Otherwise, default timesteps_per_period is used.
214			for params in tsa_parameters:
215			if "segments" in params:
216			params["timesteps"] = int(
217			len(params["segments"]) / len(params["occurrences"])
218			)
219			else:
220			params["timesteps"] = params["timesteps_per_period"]
221			self.tsa_parameters = tsa_parameters
222
223			timeincrement = self._init_timeincrement(
224			timeincrement, timeindex, periods, tsa_parameters
225			)
226			super().__init__(
227			groupings=groupings,
228			timeindex=timeindex,
229			timeincrement=timeincrement,
230			)
231
232			self.periods = periods
233			if self.periods is not None:
234			msg = (
235			"CAUTION! You specified the 'periods' attribute for your "
236			"energy system.\n This will lead to creating "
237			"a multi-period optimization modeling which can be "
238			"used e.g. for long-term investment modeling.\n"
239			"Please be aware that the feature is experimental as of "
240			"now. If you find anything suspicious or any bugs, "
241			"please report them."
242			)
243			warnings.warn(msg, debugging.ExperimentalFeatureWarning)
244			self._extract_periods_years()
245			self._extract_periods_matrix()
246			self._extract_end_year_of_optimization()
247			self.use_remaining_value = use_remaining_value
248			else:
249			self.end_year_of_optimization = 1
250
251			def _extract_periods_years(self):
252			"""Map years in optimization to respective period based on time indices
253
254			Attribute `periods_years` of type list is set. It contains
255			the year of the start of each period, relative to the
256			start of the optimization run and starting with 0.
257			"""
258			periods_years = [0]
259			start_year = self.periods[0].min().year
260			for k, v in enumerate(self.periods):
261			if k >= 1:
262			periods_years.append(v.min().year - start_year)
263
264			self.periods_years = periods_years
265
266			def _extract_periods_matrix(self):
267			"""Determines a matrix describing the temporal distance to each period.
268
269			Attribute `periods_matrix` of type list np.array is set.
270			Rows represent investment/commissioning periods, columns represent
271			decommissioning periods. The values describe the temporal distance
272			between each investment period to each decommissioning period.
273			"""
274			periods_matrix = []
275			period_years = np.array(self.periods_years)
276			for v in period_years:
277			row = period_years - v
278			row = np.where(row < 0, 0, row)
279			periods_matrix.append(row)
280			self.periods_matrix = np.array(periods_matrix)
281
282			def _extract_end_year_of_optimization(self):
283			"""Extract the end of the optimization in years
284
285			Attribute `end_year_of_optimization` of int is set.
286			"""
287			duration_last_period = self.get_period_duration(-1)
288			self.end_year_of_optimization = (
289			self.periods_years[-1] + duration_last_period
290			)
291
292			def get_period_duration(self, period):
293			"""Get duration of a period in full years
294
295			Parameters
296			----------
297			period : int
298			Period for which the duration in years shall be obtained
299
300			Returns
301			-------
302			int
303			Duration of the period
304			"""
305			return (
306			self.periods[period].max().year
307			- self.periods[period].min().year
308			+ 1
309			)
310
311			@staticmethod
312			def _init_timeincrement(timeincrement, timeindex, periods, tsa_parameters):
313			"""Check and initialize timeincrement"""
314
315			# Timeincrement in TSAM mode
316			if (
317			timeincrement is not None
318			and tsa_parameters is not None
319			and any("segments" in params for params in tsa_parameters)
320			):
321			msg = (
322			"You must not specify timeincrement in TSAM mode. "
323			"TSAM will define timeincrement itself."
324			)
325			raise AttributeError(msg)
326			if (
327			tsa_parameters is not None
328			and any("segments" in params for params in tsa_parameters)
329			and not all("segments" in params for params in tsa_parameters)
330			):
331			msg = (
332			"You have to set up segmentation in all periods, "
333			"if you want to use segmentation in TSAM mode"
334			)
335			raise AttributeError(msg)
336			if tsa_parameters is not None and all(
337			"segments" in params for params in tsa_parameters
338			):
339			# Concatenate segments from TSAM parameters to get timeincrement
340			return list(
341			itertools.chain(
342			*[params["segments"].values() for params in tsa_parameters]
343			)
344			)
345
346			elif timeindex is not None and timeincrement is None:
347			df = pd.DataFrame(timeindex)
348			timedelta = df.diff()
349			timeincrement = timedelta / np.timedelta64(1, "h")
350
351			# we want a series (squeeze)
352			# without the first item (no delta defined for first entry)
353			# but starting with index 0 (reset)
354			return timeincrement.squeeze()[1:].reset_index(drop=True)
355
356			return timeincrement
357

oemof / oemof-solph

Push — dev ( 077813...629ebf )

EnergySystem._init_timeincrement() C

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like