7745340509

Committed 01 Feb 2024 05:36PM UTC coverage: 66.519% (-0.1%) from 66.619%

Build # 7745340509

Build Type

Pull #969

github

Committed by

Flix6x

Commit Message

style: clarify use of dummy config setting

Signed-off-by: F.N. Claessen <felix@seita.nl>

Pull Request Pull Request #969: Fix documentation issue after flask env deprecation

Run Details

3815 of 6400 branches covered (0.0%)

Branch coverage included in aggregate %.

18 of 44 new or added lines in 3 files covered. (40.91%)

11 existing lines in 3 files now uncovered.

6814 of 9579 relevant lines covered (71.13%)

2.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

57.3

/flexmeasures/utils/time_utils.py

"""
Utils for dealing with time
"""

from __future__ import annotations

import re
from datetime import datetime, timedelta

from flask import current_app
from flask_security.core import current_user
from humanize import naturaldate, naturaltime
import pandas as pd
from pandas.tseries.frequencies import to_offset
import pytz
from dateutil import tz


def server_now() -> datetime:
    """The current time (timezone aware), converted to the timezone of the FlexMeasures platform."""
    return get_timezone().fromutc(datetime.utcnow())


def ensure_local_timezone(
    dt: pd.Timestamp | datetime, tz_name: str = "Europe/Amsterdam"
) -> pd.Timestamp | datetime:
    """If no timezone is given, assume the datetime is in the given timezone and make it explicit.
    Otherwise, if a timezone is given, convert to that timezone."""
    if isinstance(dt, datetime):
        if dt.tzinfo is not None:
            return dt.astimezone(tz.gettz(tz_name))
        else:
            return dt.replace(tzinfo=tz.gettz(tz_name))
    if dt.tzinfo is not None:
        return dt.tz_convert(tz_name)
    else:
        return dt.tz_localize(tz_name)


def as_server_time(dt: datetime) -> datetime:
    """The datetime represented in the timezone of the FlexMeasures platform.
    If dt is naive, we assume it is UTC time.
    """
    return naive_utc_from(dt).replace(tzinfo=pytz.utc).astimezone(get_timezone())


def localized_datetime(dt: datetime) -> datetime:
    """
    Localise a datetime to the timezone of the FlexMeasures platform.
    Note: this will change nothing but the tzinfo field.
    """
    return get_timezone().localize(naive_utc_from(dt))


def naive_utc_from(dt: datetime) -> datetime:
    """
    Return a naive datetime, that is localised to UTC if it has a timezone.
    If dt is naive, we assume it is already in UTC time.
    """
    if not hasattr(dt, "tzinfo") or dt.tzinfo is None:
        # let's hope this is the UTC time you expect
        return dt
    else:
        return dt.astimezone(pytz.utc).replace(tzinfo=None)


def tz_index_naively(
    data: pd.DataFrame | pd.Series | pd.DatetimeIndex,
) -> pd.DataFrame | pd.Series | pd.DatetimeIndex:
    """Turn any DatetimeIndex into a tz-naive one, then return. Useful for bokeh, for instance."""
    if isinstance(data, pd.DatetimeIndex):
        return data.tz_localize(tz=None)
    if hasattr(data, "index") and isinstance(data.index, pd.DatetimeIndex):
        # TODO: if index is already naive, don't
        data.index = data.index.tz_localize(tz=None)
    return data


def localized_datetime_str(dt: datetime, dt_format: str = "%Y-%m-%d %I:%M %p") -> str:
    """
    Localise a datetime to the timezone of the FlexMeasures platform.
    If no datetime is passed in, use server_now() as basis.

    Hint: This can be set as a jinja filter, so we can display local time in the app, e.g.:
    app.jinja_env.filters['localized_datetime'] = localized_datetime_str
    """
    if dt is None:
        dt = server_now()
    local_tz = get_timezone()
    local_dt = naive_utc_from(dt).astimezone(local_tz)
    return local_dt.strftime(dt_format)


def naturalized_datetime_str(dt: datetime | None, now: datetime | None = None) -> str:
    """
    Naturalise a datetime object (into a human-friendly string).
    The dt parameter (as well as the now parameter if you use it)
    can be either naive or tz-aware. We assume UTC in the naive case.

    We use the `humanize` library to generate a human-friendly string.
    If dt is not longer ago than 24 hours, we use humanize.naturaltime (e.g. "3 hours ago"),
    otherwise humanize.naturaldate (e.g. "one week ago")

    Hint: This can be set as a jinja filter, so we can display local time in the app, e.g.:
    app.jinja_env.filters['naturalized_datetime'] = naturalized_datetime_str
    """
    if dt is None:
        return "never"
    if now is None:
        now = datetime.utcnow()
    naive_utc_now = naive_utc_from(now)

    # Convert or localize to utc
    if dt.tzinfo is None:
        utc_dt = pd.Timestamp(dt).tz_localize("utc")
    else:
        utc_dt = pd.Timestamp(dt).tz_convert("utc")

    # decide which humanize call to use for naturalization
    if naive_utc_from(utc_dt) >= naive_utc_now - timedelta(hours=24):
        # return natural time (naive utc dt with respect to naive utc now)
        return naturaltime(
            utc_dt.replace(tzinfo=None),
            when=naive_utc_now,
        )
    else:
        # return natural date in the user's timezone
        local_dt = utc_dt.tz_convert(get_timezone(of_user=True))
        return naturaldate(local_dt)


def resolution_to_hour_factor(resolution: str | timedelta) -> float:
    """Return the factor with which a value needs to be multiplied in order to get the value per hour,
    e.g. 10 MW at a resolution of 15min are 2.5 MWh per time step.

    :param resolution: timedelta or pandas offset such as "15T" or "1H"
    """
    if isinstance(resolution, timedelta):
        return resolution / timedelta(hours=1)
    return pd.Timedelta(resolution).to_pytimedelta() / timedelta(hours=1)


def decide_resolution(start: datetime | None, end: datetime | None) -> str:
    """
    Decide on a practical resolution given the length of the selected time period.
    Useful for querying or plotting.
    """
    if start is None or end is None:
        return "15T"  # default if we cannot tell period
    period_length = end - start
    if period_length > timedelta(weeks=16):
        resolution = "168h"  # So upon switching from days to weeks, you get at least 16 data points
    elif period_length > timedelta(days=14):
        resolution = "24h"  # So upon switching from hours to days, you get at least 14 data points
    elif period_length > timedelta(hours=48):
        resolution = "1h"  # So upon switching from 15min to hours, you get at least 48 data points
    elif period_length > timedelta(hours=8):
        resolution = "15T"
    else:
        resolution = "5T"  # we are (currently) not going lower than 5 minutes
    return resolution


def get_timezone(of_user=False) -> pytz.BaseTzInfo:
    """Return the FlexMeasures timezone, or if desired try to return the timezone of the current user."""
    default_timezone = pytz.timezone(
        current_app.config.get("FLEXMEASURES_TIMEZONE", "")
    )
    if not of_user:
        return default_timezone
    if current_user.is_anonymous:
        return default_timezone
    if current_user.timezone not in pytz.common_timezones:
        return default_timezone
    return pytz.timezone(current_user.timezone)


def round_to_closest_quarter(dt: datetime) -> datetime:
    new_hour = dt.hour
    new_minute = 15 * round((float(dt.minute) + float(dt.second) / 60) / 15)
    if new_minute == 60:
        new_hour += 1
        new_minute = 0
    return datetime(dt.year, dt.month, dt.day, new_hour, new_minute, tzinfo=dt.tzinfo)


def round_to_closest_hour(dt: datetime) -> datetime:
    if dt.minute >= 30:
        return datetime(dt.year, dt.month, dt.day, dt.hour + 1, tzinfo=dt.tzinfo)
    else:
        return datetime(dt.year, dt.month, dt.day, dt.hour, tzinfo=dt.tzinfo)


def get_most_recent_quarter() -> datetime:
    now = server_now()
    return now.replace(minute=now.minute - (now.minute % 15), second=0, microsecond=0)


def get_most_recent_hour() -> datetime:
    now = server_now()
    return now.replace(minute=now.minute - (now.minute % 60), second=0, microsecond=0)


def get_most_recent_clocktime_window(
    window_size_in_minutes: int,
    now: datetime | None = None,
    grace_period_in_seconds: int | None = 0,
) -> tuple[datetime, datetime]:
    """
    Calculate a recent time window, returning a start and end minute so that
    a full hour can be filled with such windows, e.g.:

    Calling this function at 15:01:xx with window size 5 -> (14:55:00, 15:00:00)
    Calling this function at 03:36:xx with window size 15 -> (03:15:00, 03:30:00)

    We can demand a grace period (of x seconds) to have passed before we are ready to accept that we're in a new window:
    Calling this function at 15:00:16 with window size 5 and grace period of 30 seconds -> (14:50:00, 14:55:00)

    window_size_in_minutes is assumed to > 0 and < = 60, and a divisor of 60 (1, 2, ..., 30, 60).

    If now is not given, the current server time is used.
    if now / the current time lies within a boundary minute (e.g. 15 when window_size_in_minutes=5),
    then the window is not deemed over and the previous one is returned (in this case, [5, 10])

    Returns two datetime objects. They'll be in the timezone (if given) of the now parameter,
    or in the server timezone (see FLEXMEASURES_TIMEZONE setting).
    """
    assert window_size_in_minutes > 0
    assert 60 % window_size_in_minutes == 0
    if now is None:
        now = server_now()
    last_full_minute = now.replace(second=0, microsecond=0)
    if (
        grace_period_in_seconds is not None
        and grace_period_in_seconds > 0
        and (now - last_full_minute).seconds < grace_period_in_seconds
    ):
        last_full_minute -= timedelta(minutes=1 + grace_period_in_seconds // 60)
    last_round_minute = last_full_minute.minute - (
        last_full_minute.minute % window_size_in_minutes
    )
    begin_time = last_full_minute.replace(minute=last_round_minute) - timedelta(
        minutes=window_size_in_minutes
    )
    end_time = begin_time + timedelta(minutes=window_size_in_minutes)
    return begin_time, end_time


def get_first_day_of_next_month() -> datetime:
    return (datetime.now().replace(day=1) + timedelta(days=32)).replace(day=1)


def freq_label_to_human_readable_label(freq_label: str) -> str:
    """Translate pandas frequency labels to human-readable labels."""
    f2h_map = {
        "5T": "5 minutes",
        "15T": "15 minutes",
        "1h": "1 hour",
        "24h": "1 day",
        "168h": "1 week",
    }
    return f2h_map.get(freq_label, freq_label)


def forecast_horizons_for(resolution: str | timedelta) -> list[str] | list[timedelta]:
    """Return a list of horizons that are supported per resolution.
    Return values or of the same type as the input."""
    if isinstance(resolution, timedelta):
        resolution_str = timedelta_to_pandas_freq_str(resolution)
    else:
        resolution_str = resolution
    horizons = []
    if resolution_str in ("5T", "10T"):
        horizons = ["1h", "6h", "24h"]
    elif resolution_str in ("15T", "1h", "H"):
        horizons = ["1h", "6h", "24h", "48h"]
    elif resolution_str in ("24h", "D"):
        horizons = ["24h", "48h"]
    elif resolution_str in ("168h", "7D"):
        horizons = ["168h"]
    if isinstance(resolution, timedelta):
        return [pd.to_timedelta(to_offset(h)) for h in horizons]
    else:
        return horizons


def supported_horizons() -> list[timedelta]:
    return [
        timedelta(hours=1),
        timedelta(hours=6),
        timedelta(hours=24),
        timedelta(hours=48),
    ]


def timedelta_to_pandas_freq_str(resolution: timedelta) -> str:
    return to_offset(resolution).freqstr


def duration_isoformat(duration: timedelta):
    """Adapted version of isodate.duration_isoformat for formatting a datetime.timedelta.

    The difference is that absolute days are not formatted as nominal days.
    Workaround for https://github.com/gweis/isodate/issues/74.
    """
    ret = []
    usecs = abs(
        (duration.days * 24 * 60 * 60 + duration.seconds) * 1000000
        + duration.microseconds
    )
    seconds, usecs = divmod(usecs, 1000000)
    minutes, seconds = divmod(seconds, 60)
    hours, minutes = divmod(minutes, 60)
    if hours or minutes or seconds or usecs:
        ret.append("T")
        if hours:
            ret.append("%sH" % hours)
        if minutes:
            ret.append("%sM" % minutes)
        if seconds or usecs:
            if usecs:
                ret.append(("%d.%06d" % (seconds, usecs)).rstrip("0"))
            else:
                ret.append("%d" % seconds)
            ret.append("S")
    # at least one component has to be there.
    repl = ret and "".join(ret) or "T0H"
    return re.sub("%P", repl, "P%P")


def determine_minimum_resampling_resolution(
    event_resolutions: list[timedelta],
) -> timedelta:
    """Return minimum non-zero event resolution, or zero resolution if none of the event resolutions is non-zero."""
    condition = list(
        event_resolution
        for event_resolution in event_resolutions
        if event_resolution > timedelta(0)
    )
    return min(condition) if any(condition) else timedelta(0)


def to_http_time(dt: pd.Timestamp | datetime) -> str:
    """Formats datetime using the Internet Message Format fixdate.

    >>> to_http_time(pd.Timestamp("2022-12-13 14:06:23Z"))
    Tue, 13 Dec 2022 14:06:23 GMT

    References
    ----------
    IMF-fixdate: https://www.rfc-editor.org/rfc/rfc7231#section-7.1.1.1
    """
    return dt.strftime("%a, %d %b %Y %H:%M:%S GMT")


def get_max_planning_horizon(resolution: timedelta) -> timedelta | None:
    """Determine the maximum planning horizon for the given sensor resolution."""
    max_planning_horizon = current_app.config.get("FLEXMEASURES_MAX_PLANNING_HORIZON")
    if isinstance(max_planning_horizon, int):
        # Config setting specifies maximum number of planning steps
        max_planning_horizon *= resolution
    return max_planning_horizon


def apply_offset_chain(
    dt: pd.Timestamp | datetime, offset_chain: str
) -> pd.Timestamp | datetime:
    """Apply an offset chain to a date.

    An offset chain consist of multiple (pandas) offset strings separated by commas. Moreover,
    this function implements the offset string "DB", which stands for Day Begin, to
    get a date from a datetime, i.e. removing time details finer than a day.

    Args:
        dt (pd.Timestamp | datetime)
        offset_chain (str)

    Returns:
        pd.Timestamp | datetime (same type as given dt)
    """

    if len(offset_chain) == 0:
        return dt

    # We need a Pandas structure to apply the offsets
    if isinstance(dt, datetime):
        _dt = pd.Timestamp(dt)
    elif isinstance(dt, pd.Timestamp):
        _dt = dt
    else:
        raise TypeError()

    # Apply the offsets
    for offset in offset_chain.split(","):
        try:
            _dt += to_offset(offset.strip())
        except ValueError:
            if offset.strip().lower() == "db":  # db = day begin
                _dt = _dt.floor("D")
            elif offset.strip().lower() == "hb":  # hb = hour begin
                _dt = _dt.floor("H")

    # Return output in the same type as the input
    if isinstance(dt, datetime):
        return _dt.to_pydatetime()
    return _dt

1	"""	4✔
2	Utils for dealing with time
3	"""
4
5	from __future__ import annotations	4✔
6
7	import re	4✔
8	from datetime import datetime, timedelta	4✔
9
10	from flask import current_app	4✔
11	from flask_security.core import current_user	4✔
12	from humanize import naturaldate, naturaltime	4✔
13	import pandas as pd	4✔
14	from pandas.tseries.frequencies import to_offset	4✔
15	import pytz	4✔
16	from dateutil import tz	4✔
17
18
19	def server_now() -> datetime:	4✔
20	"""The current time (timezone aware), converted to the timezone of the FlexMeasures platform."""
21	return get_timezone().fromutc(datetime.utcnow())	4✔
22
23
24	def ensure_local_timezone(	4✔
25	dt: pd.Timestamp \| datetime, tz_name: str = "Europe/Amsterdam"
26	) -> pd.Timestamp \| datetime:
27	"""If no timezone is given, assume the datetime is in the given timezone and make it explicit.
28	Otherwise, if a timezone is given, convert to that timezone."""
29	if isinstance(dt, datetime):	×
30	if dt.tzinfo is not None:	×
31	return dt.astimezone(tz.gettz(tz_name))	×
32	else:
33	return dt.replace(tzinfo=tz.gettz(tz_name))	×
34	if dt.tzinfo is not None:	×
35	return dt.tz_convert(tz_name)	×
36	else:
37	return dt.tz_localize(tz_name)	×
38
39
40	def as_server_time(dt: datetime) -> datetime:	4✔
41	"""The datetime represented in the timezone of the FlexMeasures platform.
42	If dt is naive, we assume it is UTC time.
43	"""
44	return naive_utc_from(dt).replace(tzinfo=pytz.utc).astimezone(get_timezone())	4✔
45
46
47	def localized_datetime(dt: datetime) -> datetime:	4✔
48	"""
49	Localise a datetime to the timezone of the FlexMeasures platform.
50	Note: this will change nothing but the tzinfo field.
51	"""
52	return get_timezone().localize(naive_utc_from(dt))	×
53
54
55	def naive_utc_from(dt: datetime) -> datetime:	4✔
56	"""
57	Return a naive datetime, that is localised to UTC if it has a timezone.
58	If dt is naive, we assume it is already in UTC time.
59	"""
60	if not hasattr(dt, "tzinfo") or dt.tzinfo is None:	4✔
61	# let's hope this is the UTC time you expect
62	return dt	4✔
63	else:
64	return dt.astimezone(pytz.utc).replace(tzinfo=None)	4✔
65
66
67	def tz_index_naively(	4✔
68	data: pd.DataFrame \| pd.Series \| pd.DatetimeIndex,
69	) -> pd.DataFrame \| pd.Series \| pd.DatetimeIndex:
70	"""Turn any DatetimeIndex into a tz-naive one, then return. Useful for bokeh, for instance."""
71	if isinstance(data, pd.DatetimeIndex):	×
72	return data.tz_localize(tz=None)	×
73	if hasattr(data, "index") and isinstance(data.index, pd.DatetimeIndex):	×
74	# TODO: if index is already naive, don't
75	data.index = data.index.tz_localize(tz=None)	×
76	return data	×
77
78
79	def localized_datetime_str(dt: datetime, dt_format: str = "%Y-%m-%d %I:%M %p") -> str:	4✔
80	"""
81	Localise a datetime to the timezone of the FlexMeasures platform.
82	If no datetime is passed in, use server_now() as basis.
83
84	Hint: This can be set as a jinja filter, so we can display local time in the app, e.g.:
85	app.jinja_env.filters['localized_datetime'] = localized_datetime_str
86	"""
87	if dt is None:	4✔
88	dt = server_now()	4✔
89	local_tz = get_timezone()	4✔
90	local_dt = naive_utc_from(dt).astimezone(local_tz)	4✔
91	return local_dt.strftime(dt_format)	4✔
92
93
94	def naturalized_datetime_str(dt: datetime \| None, now: datetime \| None = None) -> str:	4✔
95	"""
96	Naturalise a datetime object (into a human-friendly string).
97	The dt parameter (as well as the now parameter if you use it)
98	can be either naive or tz-aware. We assume UTC in the naive case.
99
100	We use the `humanize` library to generate a human-friendly string.
101	If dt is not longer ago than 24 hours, we use humanize.naturaltime (e.g. "3 hours ago"),
102	otherwise humanize.naturaldate (e.g. "one week ago")
103
104	Hint: This can be set as a jinja filter, so we can display local time in the app, e.g.:
105	app.jinja_env.filters['naturalized_datetime'] = naturalized_datetime_str
106	"""
107	if dt is None:	4✔
108	return "never"	4✔
109	if now is None:	4✔
110	now = datetime.utcnow()	4✔
111	naive_utc_now = naive_utc_from(now)	4✔
112
113	# Convert or localize to utc
114	if dt.tzinfo is None:	4✔
UNCOV 115	utc_dt = pd.Timestamp(dt).tz_localize("utc")	×
116	else:
117	utc_dt = pd.Timestamp(dt).tz_convert("utc")	4✔
118
119	# decide which humanize call to use for naturalization
120	if naive_utc_from(utc_dt) >= naive_utc_now - timedelta(hours=24):	4✔
121	# return natural time (naive utc dt with respect to naive utc now)
122	return naturaltime(	4✔
123	utc_dt.replace(tzinfo=None),
124	when=naive_utc_now,
125	)
126	else:
127	# return natural date in the user's timezone
128	local_dt = utc_dt.tz_convert(get_timezone(of_user=True))	4✔
129	return naturaldate(local_dt)	4✔
130
131
132	def resolution_to_hour_factor(resolution: str \| timedelta) -> float:	4✔
133	"""Return the factor with which a value needs to be multiplied in order to get the value per hour,
134	e.g. 10 MW at a resolution of 15min are 2.5 MWh per time step.
135
136	:param resolution: timedelta or pandas offset such as "15T" or "1H"
137	"""
138	if isinstance(resolution, timedelta):	×
139	return resolution / timedelta(hours=1)	×
140	return pd.Timedelta(resolution).to_pytimedelta() / timedelta(hours=1)	×
141
142
143	def decide_resolution(start: datetime \| None, end: datetime \| None) -> str:	4✔
144	"""
145	Decide on a practical resolution given the length of the selected time period.
146	Useful for querying or plotting.
147	"""
148	if start is None or end is None:	×
149	return "15T" # default if we cannot tell period	×
150	period_length = end - start	×
151	if period_length > timedelta(weeks=16):	×
152	resolution = "168h" # So upon switching from days to weeks, you get at least 16 data points	×
153	elif period_length > timedelta(days=14):	×
154	resolution = "24h" # So upon switching from hours to days, you get at least 14 data points	×
155	elif period_length > timedelta(hours=48):	×
156	resolution = "1h" # So upon switching from 15min to hours, you get at least 48 data points	×
157	elif period_length > timedelta(hours=8):	×
158	resolution = "15T"	×
159	else:
160	resolution = "5T" # we are (currently) not going lower than 5 minutes	×
161	return resolution	×
162
163
164	def get_timezone(of_user=False) -> pytz.BaseTzInfo:	4✔
165	"""Return the FlexMeasures timezone, or if desired try to return the timezone of the current user."""
166	default_timezone = pytz.timezone(	4✔
167	current_app.config.get("FLEXMEASURES_TIMEZONE", "")
168	)
169	if not of_user:	4✔
170	return default_timezone	4✔
171	if current_user.is_anonymous:	4✔
172	return default_timezone	4✔
173	if current_user.timezone not in pytz.common_timezones:	4✔
174	return default_timezone	×
175	return pytz.timezone(current_user.timezone)	4✔
176
177
178	def round_to_closest_quarter(dt: datetime) -> datetime:	4✔
179	new_hour = dt.hour	×
180	new_minute = 15 * round((float(dt.minute) + float(dt.second) / 60) / 15)	×
181	if new_minute == 60:	×
182	new_hour += 1	×
183	new_minute = 0	×
184	return datetime(dt.year, dt.month, dt.day, new_hour, new_minute, tzinfo=dt.tzinfo)	×
185
186
187	def round_to_closest_hour(dt: datetime) -> datetime:	4✔
188	if dt.minute >= 30:	×
189	return datetime(dt.year, dt.month, dt.day, dt.hour + 1, tzinfo=dt.tzinfo)	×
190	else:
191	return datetime(dt.year, dt.month, dt.day, dt.hour, tzinfo=dt.tzinfo)	×
192
193
194	def get_most_recent_quarter() -> datetime:	4✔
195	now = server_now()	×
196	return now.replace(minute=now.minute - (now.minute % 15), second=0, microsecond=0)	×
197
198
199	def get_most_recent_hour() -> datetime:	4✔
200	now = server_now()	4✔
201	return now.replace(minute=now.minute - (now.minute % 60), second=0, microsecond=0)	4✔
202
203
204	def get_most_recent_clocktime_window(	4✔
205	window_size_in_minutes: int,
206	now: datetime \| None = None,
207	grace_period_in_seconds: int \| None = 0,
208	) -> tuple[datetime, datetime]:
209	"""
210	Calculate a recent time window, returning a start and end minute so that
211	a full hour can be filled with such windows, e.g.:
212
213	Calling this function at 15:01:xx with window size 5 -> (14:55:00, 15:00:00)
214	Calling this function at 03:36:xx with window size 15 -> (03:15:00, 03:30:00)
215
216	We can demand a grace period (of x seconds) to have passed before we are ready to accept that we're in a new window:
217	Calling this function at 15:00:16 with window size 5 and grace period of 30 seconds -> (14:50:00, 14:55:00)
218
219	window_size_in_minutes is assumed to > 0 and < = 60, and a divisor of 60 (1, 2, ..., 30, 60).
220
221	If now is not given, the current server time is used.
222	if now / the current time lies within a boundary minute (e.g. 15 when window_size_in_minutes=5),
223	then the window is not deemed over and the previous one is returned (in this case, [5, 10])
224
225	Returns two datetime objects. They'll be in the timezone (if given) of the now parameter,
226	or in the server timezone (see FLEXMEASURES_TIMEZONE setting).
227	"""
228	assert window_size_in_minutes > 0	4✔
229	assert 60 % window_size_in_minutes == 0	4✔
230	if now is None:	4✔
231	now = server_now()	×
232	last_full_minute = now.replace(second=0, microsecond=0)	4✔
233	if (	4✔
234	grace_period_in_seconds is not None
235	and grace_period_in_seconds > 0
236	and (now - last_full_minute).seconds < grace_period_in_seconds
237	):
238	last_full_minute -= timedelta(minutes=1 + grace_period_in_seconds // 60)	4✔
239	last_round_minute = last_full_minute.minute - (	4✔
240	last_full_minute.minute % window_size_in_minutes
241	)
242	begin_time = last_full_minute.replace(minute=last_round_minute) - timedelta(	4✔
243	minutes=window_size_in_minutes
244	)
245	end_time = begin_time + timedelta(minutes=window_size_in_minutes)	4✔
246	return begin_time, end_time	4✔
247
248
249	def get_first_day_of_next_month() -> datetime:	4✔
250	return (datetime.now().replace(day=1) + timedelta(days=32)).replace(day=1)	4✔
251
252
253	def freq_label_to_human_readable_label(freq_label: str) -> str:	4✔
254	"""Translate pandas frequency labels to human-readable labels."""
255	f2h_map = {	4✔
256	"5T": "5 minutes",
257	"15T": "15 minutes",
258	"1h": "1 hour",
259	"24h": "1 day",
260	"168h": "1 week",
261	}
262	return f2h_map.get(freq_label, freq_label)	4✔
263
264
265	def forecast_horizons_for(resolution: str \| timedelta) -> list[str] \| list[timedelta]:	4✔
266	"""Return a list of horizons that are supported per resolution.
267	Return values or of the same type as the input."""
268	if isinstance(resolution, timedelta):	×
269	resolution_str = timedelta_to_pandas_freq_str(resolution)	×
270	else:
271	resolution_str = resolution	×
272	horizons = []	×
273	if resolution_str in ("5T", "10T"):	×
274	horizons = ["1h", "6h", "24h"]	×
275	elif resolution_str in ("15T", "1h", "H"):	×
276	horizons = ["1h", "6h", "24h", "48h"]	×
277	elif resolution_str in ("24h", "D"):	×
278	horizons = ["24h", "48h"]	×
279	elif resolution_str in ("168h", "7D"):	×
280	horizons = ["168h"]	×
281	if isinstance(resolution, timedelta):	×
282	return [pd.to_timedelta(to_offset(h)) for h in horizons]	×
283	else:
284	return horizons	×
285
286
287	def supported_horizons() -> list[timedelta]:	4✔
288	return [	4✔
289	timedelta(hours=1),
290	timedelta(hours=6),
291	timedelta(hours=24),
292	timedelta(hours=48),
293	]
294
295
296	def timedelta_to_pandas_freq_str(resolution: timedelta) -> str:	4✔
297	return to_offset(resolution).freqstr	×
298
299
300	def duration_isoformat(duration: timedelta):	4✔
301	"""Adapted version of isodate.duration_isoformat for formatting a datetime.timedelta.
302
303	The difference is that absolute days are not formatted as nominal days.
304	Workaround for https://github.com/gweis/isodate/issues/74.
305	"""
306	ret = []	4✔
307	usecs = abs(	4✔
308	(duration.days * 24 * 60 * 60 + duration.seconds) * 1000000
309	+ duration.microseconds
310	)
311	seconds, usecs = divmod(usecs, 1000000)	4✔
312	minutes, seconds = divmod(seconds, 60)	4✔
313	hours, minutes = divmod(minutes, 60)	4✔
314	if hours or minutes or seconds or usecs:	4✔
315	ret.append("T")	4✔
316	if hours:	4✔
317	ret.append("%sH" % hours)	4✔
318	if minutes:	4✔
319	ret.append("%sM" % minutes)	4✔
320	if seconds or usecs:	4✔
321	if usecs:	4✔
322	ret.append(("%d.%06d" % (seconds, usecs)).rstrip("0"))	×
323	else:
324	ret.append("%d" % seconds)	4✔
325	ret.append("S")	4✔
326	# at least one component has to be there.
327	repl = ret and "".join(ret) or "T0H"	4✔
328	return re.sub("%P", repl, "P%P")	4✔
329
330
331	def determine_minimum_resampling_resolution(	4✔
332	event_resolutions: list[timedelta],
333	) -> timedelta:
334	"""Return minimum non-zero event resolution, or zero resolution if none of the event resolutions is non-zero."""
335	condition = list(	×
336	event_resolution
337	for event_resolution in event_resolutions
338	if event_resolution > timedelta(0)
339	)
340	return min(condition) if any(condition) else timedelta(0)	×
341
342
343	def to_http_time(dt: pd.Timestamp \| datetime) -> str:	4✔
344	"""Formats datetime using the Internet Message Format fixdate.
345
346	>>> to_http_time(pd.Timestamp("2022-12-13 14:06:23Z"))
347	Tue, 13 Dec 2022 14:06:23 GMT
348
349	References
350	----------
351	IMF-fixdate: https://www.rfc-editor.org/rfc/rfc7231#section-7.1.1.1
352	"""
353	return dt.strftime("%a, %d %b %Y %H:%M:%S GMT")	4✔
354
355
356	def get_max_planning_horizon(resolution: timedelta) -> timedelta \| None:	4✔
357	"""Determine the maximum planning horizon for the given sensor resolution."""
358	max_planning_horizon = current_app.config.get("FLEXMEASURES_MAX_PLANNING_HORIZON")	4✔
359	if isinstance(max_planning_horizon, int):	4✔
360	# Config setting specifies maximum number of planning steps
361	max_planning_horizon *= resolution	4✔
362	return max_planning_horizon	4!
363
364
365	def apply_offset_chain(	4✔
366	dt: pd.Timestamp \| datetime, offset_chain: str
367	) -> pd.Timestamp \| datetime:
368	"""Apply an offset chain to a date.
369
370	An offset chain consist of multiple (pandas) offset strings separated by commas. Moreover,
371	this function implements the offset string "DB", which stands for Day Begin, to
372	get a date from a datetime, i.e. removing time details finer than a day.
373
374	Args:
375	dt (pd.Timestamp \| datetime)
376	offset_chain (str)
377
378	Returns:
379	pd.Timestamp \| datetime (same type as given dt)
380	"""
381
382	if len(offset_chain) == 0:	4✔
383	return dt	4✔
384
385	# We need a Pandas structure to apply the offsets
386	if isinstance(dt, datetime):	4✔
387	_dt = pd.Timestamp(dt)	4✔
388	elif isinstance(dt, pd.Timestamp):	×
389	_dt = dt	×
390	else:
391	raise TypeError()	×
392
393	# Apply the offsets
394	for offset in offset_chain.split(","):	4✔
395	try:	4✔
396	_dt += to_offset(offset.strip())	4✔
397	except ValueError:	4✔
398	if offset.strip().lower() == "db": # db = day begin	4✔
399	_dt = _dt.floor("D")	4✔
400	elif offset.strip().lower() == "hb": # hb = hour begin	4✔
401	_dt = _dt.floor("H")	4✔
402
403	# Return output in the same type as the input
404	if isinstance(dt, datetime):	4✔
405	return _dt.to_pydatetime()	4✔
406	return _dt	×

FlexMeasures / flexmeasures / 7745340509

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous