solph._energy_system.create_time_index() - Code Metrics - Inspection of "Replace constraint tests" - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#1080)

by Patrik

created 2024-08-20 13:53 UTC

solph._energy_system.create_time_index() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	66
Code Lines	19

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	19
dl	0
loc	66
rs	8.9833
c	0
b	0
f	0
cc	5
nop	4

How to fix Long Method

# -*- 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 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].

    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,
        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:
            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)

        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

    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
        )


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 warnings
18
19			import numpy as np
20			import pandas as pd
21			from oemof.network import energy_system as es
22			from oemof.tools import debugging
23
24
25			class EnergySystem(es.EnergySystem):
26			"""A variant of the class EnergySystem from
27			<oemof.network.network.energy_system.EnergySystem> specially tailored to
28			solph.
29
30			In order to work in tandem with solph, instances of this class always use
31			solph.GROUPINGS <oemof.solph.GROUPINGS>. If custom groupings are
32			supplied via the `groupings` keyword argument, solph.GROUPINGS
33			<oemof.solph.GROUPINGS> is prepended to those.
34
35			If you know what you are doing and want to use solph without
36			solph.GROUPINGS <oemof.solph.GROUPINGS>, you can just use
37			EnergySystem <oemof.network.network.energy_system.EnergySystem>` of
38			oemof.network directly.
39
40			Parameters
41			----------
42			timeindex : pandas.DatetimeIndex
43
44			timeincrement : iterable
45
46			infer_last_interval : bool
47			Add an interval to the last time point. The end time of this interval
48			is unknown so it does only work for an equidistant DatetimeIndex with
49			a 'freq' attribute that is not None. The parameter has no effect on the
50			timeincrement parameter.
51
52			periods : list or None
53			The periods of a multi-period model.
54			If this is explicitly specified, it leads to creating a multi-period
55			model, providing a respective user warning as a feedback.
56
57			list of pd.date_range objects carrying the timeindex for the
58			respective period;
59
60			For a standard model, periods are not (to be) declared, i.e. None.
61			A list with one entry is derived, i.e. [0].
62
63			use_remaining_value : bool
64			If True, compare the remaining value of an investment to the
65			original value (only applicable for multi-period models)
66
67			kwargs
68			"""
69
70			def __init__(
71			self,
72			timeindex=None,
73			timeincrement=None,
74			infer_last_interval=None,
75			periods=None,
76			use_remaining_value=False,
77			groupings=None,
78			):
79			# Doing imports at runtime is generally frowned upon, but should work
80			# for now. See the TODO in :func:`constraint_grouping
81			# <oemof.solph.groupings.constraint_grouping>` for more information.
82			from oemof.solph import GROUPINGS
83
84			if groupings is None:
85			groupings = []
86			groupings = GROUPINGS + groupings
87
88			if not (
89			isinstance(timeindex, pd.DatetimeIndex)
90			or isinstance(timeindex, type(None))
91			):
92			msg = (
93			"Parameter 'timeindex' has to be of type "
94			"pandas.DatetimeIndex or NoneType and not of type {0}"
95			)
96			raise TypeError(msg.format(type(timeindex)))
97
98			if infer_last_interval is None and timeindex is not None:
99			msg = (
100			"The default behaviour will change in future versions.\n"
101			"At the moment the last interval of an equidistant time "
102			"index is added implicitly by default. Set "
103			"'infer_last_interval' explicitly 'True' or 'False' to avoid "
104			"this warning. In future versions 'False' will be the default"
105			"behaviour"
106			)
107			warnings.warn(msg, FutureWarning)
108			infer_last_interval = True
109
110			if infer_last_interval is True and timeindex is not None:
111			# Add one time interval to the timeindex by adding one time point.
112			if timeindex.freq is None:
113			msg = (
114			"You cannot infer the last interval if the 'freq' "
115			"attribute of your DatetimeIndex is None. Set "
116			" 'infer_last_interval=False' or specify a DatetimeIndex "
117			"with a valid frequency."
118			)
119			raise AttributeError(msg)
120
121			timeindex = timeindex.union(
122			pd.date_range(
123			timeindex[-1] + timeindex.freq,
124			periods=1,
125			freq=timeindex.freq,
126			)
127			)
128
129			# catch wrong combinations and infer timeincrement from timeindex.
130			if timeincrement is not None and timeindex is not None:
131			if periods is None:
132			msg = (
133			"Specifying the timeincrement and the timeindex parameter "
134			"at the same time is not allowed since these might be "
135			"conflicting to each other."
136			)
137			raise AttributeError(msg)
138			else:
139			msg = (
140			"Ensure that your timeindex and timeincrement are "
141			"consistent."
142			)
143			warnings.warn(msg, debugging.ExperimentalFeatureWarning)
144
145			elif timeindex is not None and timeincrement is None:
146			df = pd.DataFrame(timeindex)
147			timedelta = df.diff()
148			timeincrement = timedelta / np.timedelta64(1, "h")
149
150			# we want a series (squeeze)
151			# without the first item (no delta defined for first entry)
152			# but starting with index 0 (reset)
153			timeincrement = timeincrement.squeeze()[1:].reset_index(drop=True)
154
155			if timeincrement is not None and (pd.Series(timeincrement) <= 0).any():
156			msg = (
157			"The time increment is inconsistent. Negative values and zero "
158			"are not allowed.\nThis is caused by a inconsistent "
159			"timeincrement parameter or an incorrect timeindex."
160			)
161			raise TypeError(msg)
162
163			super().__init__(
164			groupings=groupings,
165			timeindex=timeindex,
166			timeincrement=timeincrement,
167			)
168
169			self.periods = periods
170			if self.periods is not None:
171			msg = (
172			"CAUTION! You specified the 'periods' attribute for your "
173			"energy system.\n This will lead to creating "
174			"a multi-period optimization modeling which can be "
175			"used e.g. for long-term investment modeling.\n"
176			"Please be aware that the feature is experimental as of "
177			"now. If you find anything suspicious or any bugs, "
178			"please report them."
179			)
180			warnings.warn(msg, debugging.ExperimentalFeatureWarning)
181			self._extract_periods_years()
182			self._extract_periods_matrix()
183			self._extract_end_year_of_optimization()
184			self.use_remaining_value = use_remaining_value
185
186			def _extract_periods_years(self):
187			"""Map years in optimization to respective period based on time indices
188
189			Attribute `periods_years` of type list is set. It contains
190			the year of the start of each period, relative to the
191			start of the optimization run and starting with 0.
192			"""
193			periods_years = [0]
194			start_year = self.periods[0].min().year
195			for k, v in enumerate(self.periods):
196			if k >= 1:
197			periods_years.append(v.min().year - start_year)
198
199			self.periods_years = periods_years
200
201			def _extract_periods_matrix(self):
202			"""Determines a matrix describing the temporal distance to each period.
203
204			Attribute `periods_matrix` of type list np.array is set.
205			Rows represent investment/commissioning periods, columns represent
206			decommissioning periods. The values describe the temporal distance
207			between each investment period to each decommissioning period.
208			"""
209			periods_matrix = []
210			period_years = np.array(self.periods_years)
211			for v in period_years:
212			row = period_years - v
213			row = np.where(row < 0, 0, row)
214			periods_matrix.append(row)
215			self.periods_matrix = np.array(periods_matrix)
216
217			def _extract_end_year_of_optimization(self):
218			"""Extract the end of the optimization in years
219
220			Attribute `end_year_of_optimization` of int is set.
221			"""
222			duration_last_period = self.get_period_duration(-1)
223			self.end_year_of_optimization = (
224			self.periods_years[-1] + duration_last_period
225			)
226
227			def get_period_duration(self, period):
228			"""Get duration of a period in full years
229
230			Parameters
231			----------
232			period : int
233			Period for which the duration in years shall be obtained
234
235			Returns
236			-------
237			int
238			Duration of the period
239			"""
240			return (
241			self.periods[period].max().year
242			- self.periods[period].min().year
243			+ 1
244			)
245

oemof / oemof-solph

Pull Request — dev (#1080)

solph._energy_system.create_time_index() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like