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

EnergySystem._extract_end_year_of_optimization() A
last analyzed 2025-11-04 19:41 UTC

↳ Parent: solph._energy_system

Complexity

Conditions

Size

Total Lines	8
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	8
rs	10
c	0
b	0
f	0
cc	1
nop	1

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

oemof / oemof-solph

EnergySystem._extract_end_year_of_optimization() A last analyzed 2025-11-04 19:41 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

EnergySystem._extract_end_year_of_optimization() A
last analyzed 2025-11-04 19:41 UTC