diff --git a/.github/workflows/python-build.yaml b/.github/workflows/python-build.yaml index 805c73194..30642441f 100644 --- a/.github/workflows/python-build.yaml +++ b/.github/workflows/python-build.yaml @@ -56,7 +56,7 @@ jobs: # Test - name: Unit test with pytest run: | - pip install pytest pytest-cov + pip install -r test-requirements.txt pytest --cov-report=xml --cov=openstef/ test/ --junitxml=pytest-report.xml # Fix relative paths in coverage file # Known bug: https://community.sonarsource.com/t/sonar-on-github-actions-with-python-coverage-source-issue/36057 diff --git a/openstef/validation/validation.py b/openstef/validation/validation.py index b0c2c4fb6..4207e4f69 100644 --- a/openstef/validation/validation.py +++ b/openstef/validation/validation.py @@ -1,7 +1,7 @@ # SPDX-FileCopyrightText: 2017-2023 Contributors to the OpenSTEF project # noqa E501> # # SPDX-License-Identifier: MPL-2.0 -from datetime import timedelta +from datetime import datetime, timedelta from typing import Union import math @@ -222,6 +222,8 @@ def detect_ongoing_zero_flatliner( bool: Indicating wether or not there is a zero flatliner ongoing for the given load. """ + # remove all timestamps in the future + load = load[load.index.tz_localize(None) <= datetime.utcnow()] latest_measurement_time = load.index.max() latest_measurements = load[ latest_measurement_time - timedelta(minutes=duration_threshold_minutes) : diff --git a/test-requirements.txt b/test-requirements.txt index af4dcebb4..4a610226a 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -9,6 +9,7 @@ autoflake==1.7.5 bandit==1.7.4 black==23.9.1 docformatter==1.5.0 +freezegun~=1.2.2 isort==5.10.1 pydocstyle==6.1.1 pylint==2.15.4 diff --git a/test/unit/validation/test_validation_detect_ongoing_zero_flatliner.py b/test/unit/validation/test_validation_detect_ongoing_zero_flatliner.py index b4c44a7e4..42f9a5dae 100644 --- a/test/unit/validation/test_validation_detect_ongoing_zero_flatliner.py +++ b/test/unit/validation/test_validation_detect_ongoing_zero_flatliner.py @@ -3,22 +3,31 @@ # SPDX-License-Identifier: MPL-2.0 from datetime import datetime, timedelta +from freezegun import freeze_time from test.unit.utils.base import BaseTestCase import numpy as np import pandas as pd from openstef.validation.validation import detect_ongoing_zero_flatliner -now = datetime.utcnow() -three_hour_range = pd.date_range( - start=now - timedelta(minutes=180), end=now, freq="0.25H" -) - +@freeze_time("2023-10-30 12:01:02") class TestDetectOngoingZeroFlatliners(BaseTestCase): + def setUp(self) -> None: + super().setUp() + now = datetime.utcnow() + self.three_hour_range = pd.date_range( + start=now - timedelta(minutes=180), end=now, freq="0.25H" + ) + self.four_hour_range_predict_setting = pd.date_range( + start=now - timedelta(minutes=180), + end=now + timedelta(minutes=60), + freq="0.25H", + ) + def test_all_zero(self): # Arrange - load = pd.Series(index=three_hour_range, data=[0 for i in range(13)]) + load = pd.Series(index=self.three_hour_range, data=[0 for i in range(13)]) duration_threshold = 120 # Act @@ -29,7 +38,7 @@ def test_all_zero(self): def test_all_nonzero(self): # Arrange - load = pd.Series(index=three_hour_range, data=[i for i in range(1, 14)]) + load = pd.Series(index=self.three_hour_range, data=[i for i in range(1, 14)]) duration_threshold = 120 # Act @@ -43,7 +52,9 @@ def test_only_last_nonzero(self): # now the pattern has ended since the last measurement is not zero anymore. # Arrange - load = pd.Series(index=three_hour_range, data=[0 for i in range(1, 13)] + [1]) + load = pd.Series( + index=self.three_hour_range, data=[0 for i in range(1, 13)] + [1] + ) duration_threshold = 120 # Act @@ -58,7 +69,7 @@ def test_zero_flatliner_pattern_below_threshold(self): # Arrange load = pd.Series( - index=three_hour_range, data=[i for i in range(1, 10)] + [0, 0, 0, 0] + index=self.three_hour_range, data=[i for i in range(1, 10)] + [0, 0, 0, 0] ) duration_threshold = 120 @@ -71,7 +82,7 @@ def test_zero_flatliner_pattern_below_threshold(self): def test_zero_flatliner_pattern_just_above_threshold(self): # Arrange load = pd.Series( - index=three_hour_range, data=[1, 2, 3, 4] + [0 for i in range(9)] + index=self.three_hour_range, data=[1, 2, 3, 4] + [0 for i in range(9)] ) duration_threshold = 120 @@ -84,7 +95,7 @@ def test_zero_flatliner_pattern_just_above_threshold(self): def test_zero_flatliner_and_missing_values(self): # Arrange load = pd.Series( - index=three_hour_range, + index=self.three_hour_range, data=[1, 2, 3, 4] + [0, 0, 0, 0, np.nan, np.nan, np.nan, np.nan, 0], ) duration_threshold = 120 @@ -97,7 +108,7 @@ def test_zero_flatliner_and_missing_values(self): def test_all_missing_values(self): # Arrange - load = pd.Series(index=three_hour_range, data=[np.nan for i in range(13)]) + load = pd.Series(index=self.three_hour_range, data=[np.nan for i in range(13)]) duration_threshold = 120 # Act @@ -105,3 +116,23 @@ def test_all_missing_values(self): # Assert assert zero_flatliner_ongoing == False + + def test_zero_flatliner_predict_future(self): + # Scenario: A forecast is made on a zero flatliner, which contains timestamps in the + # future with NaN values that need to be predicted. + # In this case: Time in future > duration_threshold. + + # Arrange + load = pd.Series( + index=self.four_hour_range_predict_setting, + data=[1, 2, 3] + + [0, 0, 0, 0, 0, 0, 0, 0, 0] + + [np.nan, np.nan, np.nan, np.nan, np.nan], + ) + duration_threshold = 60 + + # Act + zero_flatliner_ongoing = detect_ongoing_zero_flatliner(load, duration_threshold) + + # Assert + assert zero_flatliner_ongoing == True