flexmeasures-main/flexmeasures/utils/tests/test_time_utils.py

from __future__ import annotations

from datetime import datetime, timedelta, timezone

from isodate import duration_isoformat as original_duration_isoformat
import pandas as pd
import pytz
import pytest

from flexmeasures.utils.time_utils import (
    duration_isoformat,
    server_now,
    naturalized_datetime_str,
    get_most_recent_clocktime_window,
    apply_offset_chain,
    to_utc_timestamp,
)


@pytest.mark.parametrize(
    "td, iso",
    [
        (timedelta(hours=1), "PT1H"),
        (timedelta(hours=14), "PT14H"),
        (timedelta(hours=24), "PT24H"),
        (timedelta(days=1), "PT24H"),
        (timedelta(days=1, seconds=22), "PT24H22S"),
        (timedelta(days=1, seconds=122), "PT24H2M2S"),
    ],
)
def test_duration_isoformat(td: timedelta, iso: str):
    assert duration_isoformat(td) == iso


@pytest.mark.parametrize(
    "td, iso",
    [
        (timedelta(hours=1), "PT1H"),
        (timedelta(hours=14), "PT14H"),
        # todo: if the following test cases fail, we can start using isodate.duration_isoformat again (see #459)
        (timedelta(hours=24), "P1D"),
        (timedelta(days=1), "P1D"),
        (timedelta(days=1, seconds=22), "P1DT22S"),
        (timedelta(days=1, seconds=122), "P1DT2M2S"),
    ],
)
def test_original_duration_isoformat(td: timedelta, iso: str):
    assert original_duration_isoformat(td) == iso


@pytest.mark.parametrize(
    "dt_tz, now, server_tz, delta_in_h, exp_result",
    # there can be two results depending of today's date, due to humanize.
    # Monekypatching was too hard.
    [
        (None, pd.Timestamp.utcnow(), "UTC", 3, "3 hours ago"),
        (None, pd.Timestamp.utcnow().tz_convert("Asia/Seoul"), "UTC", 3, "3 hours ago"),
        (None, datetime.now(timezone.utc), "UTC", 3, "3 hours ago"),
        (
            None,
            datetime.now(timezone.utc).replace(tzinfo=None),
            "UTC",
            3,
            "3 hours ago",
        ),
        (
            None,
            datetime(2021, 5, 17, 3),
            "Europe/Amsterdam",
            48,
            ("May 15", "May 15 2021"),
        ),
        ("Asia/Seoul", "server_now", "Europe/Amsterdam", 1, "an hour ago"),
        (
            "UTC",
            datetime(2021, 5, 17, 3),
            "Asia/Seoul",
            24 * 7,
            ("May 10", "May 10 2021"),
        ),
        ("UTC", datetime(2021, 5, 17, 3), "Asia/Seoul", None, "never"),
    ],
)
def test_naturalized_datetime_str(
    app,
    monkeypatch,
    dt_tz,
    now,
    server_tz,
    delta_in_h,
    exp_result,
):
    monkeypatch.setitem(app.config, "FLEXMEASURES_TIMEZONE", server_tz)
    if now == "server_now":
        now = server_now()  # done this way as it needs (patched) app context
    if now.tzinfo is None:
        now = now.replace(tzinfo=pytz.utc)  # assuming UTC
    if delta_in_h is not None:
        h_ago = now - timedelta(hours=delta_in_h)
        if dt_tz is not None:
            h_ago = h_ago.astimezone(pytz.timezone(dt_tz))
    else:
        h_ago = None
    if isinstance(exp_result, tuple):
        assert naturalized_datetime_str(h_ago, now=now) in exp_result
    else:
        assert naturalized_datetime_str(h_ago, now=now) == exp_result


@pytest.mark.parametrize(
    "window_size, now, exp_start, exp_end, grace_period",
    [
        (
            5,
            datetime(2021, 4, 30, 15, 1),
            datetime(2021, 4, 30, 14, 55),
            datetime(2021, 4, 30, 15),
            0,
        ),
        (
            15,
            datetime(2021, 4, 30, 3, 36),
            datetime(2021, 4, 30, 3, 15),
            datetime(2021, 4, 30, 3, 30),
            0,
        ),
        (
            5,
            datetime(2021, 4, 30, 9, 15, 16),
            datetime(2021, 4, 30, 9, 10),
            datetime(2021, 4, 30, 9, 15),
            0,
        ),
        (
            10,
            datetime(2021, 4, 30, 0, 5),
            datetime(2021, 4, 29, 23, 50),
            datetime(2021, 4, 30, 0, 0),
            0,
        ),
        (
            5,
            datetime(
                2021, 5, 20, 10, 5, 34
            ),  # less than grace period into the new window
            datetime(2021, 5, 20, 9, 55),
            datetime(2021, 5, 20, 10, 0),
            60,
        ),
        (
            5,
            datetime(2021, 5, 20, 10, 7, 4),
            datetime(2021, 5, 20, 9, 55),
            datetime(2021, 5, 20, 10, 0),
            130,  # grace period > 2 minutes
        ),
        (
            60,
            datetime(2021, 1, 1, 0, 4),  # new year
            datetime(2020, 12, 31, 23, 00),
            datetime(2021, 1, 1, 0, 0),
            0,
        ),
        (
            60,
            datetime(2021, 3, 28, 3, 10),  # DST transition
            datetime(2021, 3, 28, 2),
            datetime(2021, 3, 28, 3),
            0,
        ),
    ],
)
def test_recent_clocktime_window(window_size, now, exp_start, exp_end, grace_period):
    start, end = get_most_recent_clocktime_window(
        window_size, now=now, grace_period_in_seconds=grace_period
    )
    assert start == exp_start
    assert end == exp_end


def test_recent_clocktime_window_invalid_window():
    with pytest.raises(AssertionError):
        get_most_recent_clocktime_window(25, now=datetime(2021, 4, 30, 3, 36))
        get_most_recent_clocktime_window(120, now=datetime(2021, 4, 30, 3, 36))
        get_most_recent_clocktime_window(0, now=datetime(2021, 4, 30, 3, 36))


@pytest.mark.parametrize(
    "input_date, offset_chain, output_date",
    [
        (datetime(2023, 5, 17, 10, 15), ",,, , , ", datetime(2023, 5, 17, 10, 15)),
        (datetime(2023, 5, 17, 10, 15), "", datetime(2023, 5, 17, 10, 15)),
        (datetime(2023, 5, 17, 10, 15), ",,hb, , , ", datetime(2023, 5, 17, 10)),
        (datetime(2023, 5, 17, 10, 15), "DB", datetime(2023, 5, 17)),
        (datetime(2023, 5, 17, 10, 15), "2D,DB", datetime(2023, 5, 19)),
        (datetime(2023, 5, 17, 10, 15), "-2D,DB", datetime(2023, 5, 15)),
        # The day of this moment started on May 17th in Amsterdam (but started on May 16th in UTC).
        (
            pytz.timezone("Europe/Amsterdam").localize(datetime(2023, 5, 17, 0, 15)),
            "DB",
            pytz.timezone("Europe/Amsterdam").localize(datetime(2023, 5, 17)),
        ),
        # Check Pandas structure, too.
        (
            pd.Timestamp("2023-05-17T00:15", tz="Europe/Amsterdam"),
            "DB",
            pd.Timestamp("2023-05-17", tz="Europe/Amsterdam"),
        ),
    ],
)
def test_apply_offset_chain(
    input_date: pd.Timestamp | datetime,
    offset_chain: str,
    output_date: pd.Timestamp | datetime,
):
    assert apply_offset_chain(input_date, offset_chain) == output_date


@pytest.mark.parametrize(
    "input_value, expected_output",
    [
        (datetime(2024, 4, 28, 8, 55, 58), 1714294558.0),
        (pd.Timestamp("2024-04-28 08:55:58", tz="UTC").to_pydatetime(), 1714294558.0),
        ("Sun, 28 Apr 2024 08:55:58 GMT", 1714294558.0),
        ("Invalid date string", None),
        (None, None),
    ],
)
def test_to_utc_timestamp(input_value, expected_output):
    result = to_utc_timestamp(input_value)
    if expected_output is None:
        assert result is None
    else:
        assert result == pytest.approx(expected_output)