diff --git a/tests/test_models.py b/tests/test_models.py index 2f2ce06..d23dcc4 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -2,13 +2,15 @@ import datetime as dt import json import logging +import re from pathlib import Path import pandas as pd import pytest +from pydantic import ValidationError from tests.conftest import TEST_CONFIG_DIR -from wind_up.models import PrePost, WindUpConfig +from wind_up.models import DEFAULT_TIMEBASE_S, PrePost, WindUpConfig def test_lsa_asset_name(test_lsa_t13_config: WindUpConfig) -> None: @@ -214,3 +216,94 @@ def test_with_pre_last_dt_utc_start(self) -> None: assert cfg.upgrade_first_dt_utc_start == pd.Timestamp("2021-09-30 00:00:00+0000", tz="UTC") assert cfg.prepost.post_first_dt_utc_start == pd.Timestamp("2021-09-30 00:00:00+0000", tz="UTC") assert cfg.prepost.post_last_dt_utc_start == pd.Timestamp("2022-07-20 23:50:00+0000", tz="UTC") + + +def test_windupconfig_with_extended_post_period_length() -> None: + """Check that the pre-period does not extend over the upgrade date. + + Check that if the `analysis_last_dt_utc_start` is >1 year post upgrade that if the `years_offset_for_pre_period` is + 1 year, then the maximum end date of the pre-period is one timebase before upgrade date. + """ + # Modify yaml file and then load it to ensure the override works as expected + yaml_path = TEST_CONFIG_DIR / "test_LSA_T13.yaml" + with yaml_path.open() as f: + yaml_str = f.read() + + # Replace the existing line containing "pre_last_dt_utc_start" + analysis_end = "2026-01-01 23:50:00+0000" + yaml_str = re.sub(r"analysis_last_dt_utc_start:.*", f"analysis_last_dt_utc_start: {analysis_end}", yaml_str) + + modified_yaml_path = TEST_CONFIG_DIR / "modified_test_LSA_T13.yaml" + with modified_yaml_path.open("w") as mf: + mf.write(yaml_str) + + cfg = WindUpConfig.from_yaml(modified_yaml_path) + + # delete the modified yaml file after loading the config + modified_yaml_path.unlink() + + assert cfg.prepost.pre_first_dt_utc_start == pd.Timestamp("2020-09-30 00:00:00+0000", tz="UTC") + assert cfg.prepost.pre_last_dt_utc_start == ( + cfg.upgrade_first_dt_utc_start - pd.Timedelta(seconds=DEFAULT_TIMEBASE_S) + ) # key check + assert cfg.upgrade_first_dt_utc_start == pd.Timestamp("2021-09-30 00:00:00+0000", tz="UTC") + assert cfg.prepost.post_first_dt_utc_start == pd.Timestamp("2021-09-30 00:00:00+0000", tz="UTC") + assert cfg.prepost.post_last_dt_utc_start == pd.Timestamp(analysis_end, tz="UTC") + + +class TestPrePostValidation: + @pytest.fixture + def valid_dates(self) -> dict[str, dt.datetime]: + return { + "pre_first_dt_utc_start": dt.datetime(2000, 1, 1, tzinfo=dt.timezone.utc), + "pre_last_dt_utc_start": dt.datetime(2000, 1, 15, tzinfo=dt.timezone.utc), + "post_first_dt_utc_start": dt.datetime(2000, 1, 16, tzinfo=dt.timezone.utc), + "post_last_dt_utc_start": dt.datetime(2000, 1, 29, tzinfo=dt.timezone.utc), + } + + def test_valid_prepost(self, valid_dates: dict[str, dt.datetime]) -> None: + PrePost(**valid_dates) + + def test_pre_period_start_after_end_raises(self, valid_dates: dict[str, dt.datetime]) -> None: + valid_dates["pre_first_dt_utc_start"] = dt.datetime(2000, 1, 20, tzinfo=dt.timezone.utc) + with pytest.raises( + ValidationError, match=re.escape("Start date of pre-period must be before the end date of pre-period.") + ): + PrePost(**valid_dates) + + def test_pre_period_equal_start_and_end_is_invalid(self, valid_dates: dict[str, dt.datetime]) -> None: + valid_dates["pre_first_dt_utc_start"] = valid_dates["pre_last_dt_utc_start"] + with pytest.raises( + ValidationError, match=re.escape("Start date of pre-period must be before the end date of pre-period.") + ): + PrePost(**valid_dates) + + def test_post_period_start_after_end_raises(self, valid_dates: dict[str, dt.datetime]) -> None: + valid_dates["post_first_dt_utc_start"] = dt.datetime(2000, 1, 30, tzinfo=dt.timezone.utc) + with pytest.raises( + ValidationError, match=re.escape("Start date of post-period must be before the end date of post-period.") + ): + PrePost(**valid_dates) + + def test_post_period_equal_start_and_end_is_invalid(self, valid_dates: dict[str, dt.datetime]) -> None: + valid_dates["post_first_dt_utc_start"] = valid_dates["post_last_dt_utc_start"] + with pytest.raises( + ValidationError, match=re.escape("Start date of post-period must be before the end date of post-period.") + ): + PrePost(**valid_dates) + + def test_pre_last_after_post_first_raises(self, valid_dates: dict[str, dt.datetime]) -> None: + valid_dates["pre_last_dt_utc_start"] = dt.datetime(2000, 1, 20, tzinfo=dt.timezone.utc) + with pytest.raises( + ValidationError, match=re.escape("End date of pre-period must be before the Start date of post-period.") + ): + PrePost(**valid_dates) + + def test_pre_last_equal_post_first_raises(self, valid_dates: dict[str, dt.datetime]) -> None: + same_dt = dt.datetime(2000, 1, 16, tzinfo=dt.timezone.utc) + valid_dates["pre_last_dt_utc_start"] = same_dt + valid_dates["post_first_dt_utc_start"] = same_dt + with pytest.raises( + ValidationError, match=re.escape("End date of pre-period must be before the Start date of post-period.") + ): + PrePost(**valid_dates) diff --git a/wind_up/models.py b/wind_up/models.py index 23ed55a..b78d132 100644 --- a/wind_up/models.py +++ b/wind_up/models.py @@ -19,6 +19,8 @@ logger = logging.getLogger(__name__) +DEFAULT_TIMEBASE_S = 10 * 60 + class PlotConfig(BaseModel): """Plot configuration model.""" @@ -159,6 +161,27 @@ class PrePost(BaseModel): description="Last time to use in post-upgrade analysis, UTC Start format", ) + @model_validator(mode="after") + def _validate_pre_period_dates(self) -> PrePost: + if self.pre_first_dt_utc_start >= self.pre_last_dt_utc_start: + msg = "Start date of pre-period must be before the end date of pre-period." + raise ValueError(msg) + return self + + @model_validator(mode="after") + def _validate_post_period_dates(self) -> PrePost: + if self.post_first_dt_utc_start >= self.post_last_dt_utc_start: + msg = "Start date of post-period must be before the end date of post-period." + raise ValueError(msg) + return self + + @model_validator(mode="after") + def _validate_pre_is_prior_to_post(self) -> PrePost: + if self.pre_last_dt_utc_start >= self.post_first_dt_utc_start: + msg = "End date of pre-period must be before the Start date of post-period." + raise ValueError(msg) + return self + class WindUpConfig(BaseModel): """WindUpConfig model. @@ -171,7 +194,7 @@ class WindUpConfig(BaseModel): description="Name used for assessment output folder", ) timebase_s: int = Field( - default=10 * 60, + default=DEFAULT_TIMEBASE_S, description="Timebase in seconds for SCADA data, other data is converted to this timebase", ) ignore_turbine_anemometer_data: bool = Field( @@ -319,6 +342,17 @@ class WindUpConfig(BaseModel): ), ) + @model_validator(mode="after") + def _validate_pre_period_is_before_upgrade_date(self: WindUpConfig) -> WindUpConfig: + if ( + (self.toggle is None) + and (self.prepost is not None) + and (self.prepost.pre_last_dt_utc_start >= self.upgrade_first_dt_utc_start) + ): + msg = "pre_last_dt_utc_start must be before upgrade_first_dt_utc_start" + raise ValueError(msg) + return self + @model_validator(mode="after") def _check_years_offset_for_pre_period(self: WindUpConfig) -> WindUpConfig: if self.toggle is None and self.years_offset_for_pre_period is None: @@ -417,6 +451,11 @@ def from_yaml(cls, file_path: Path) -> "WindUpConfig": # noqa ANN102 pre_last_dt_utc_start = pd.to_datetime( cfg_dct["analysis_last_dt_utc_start"] - pd.DateOffset(years=cfg_dct["years_offset_for_pre_period"]) ) + if pre_last_dt_utc_start > ( + pre_max_threshold := pd.to_datetime(cfg_dct["upgrade_first_dt_utc_start"]) + - pd.Timedelta(seconds=cfg_dct.get("timebase_s", DEFAULT_TIMEBASE_S)) + ): + pre_last_dt_utc_start = pre_max_threshold if pre_last_dt_utc_start.tzinfo is None: pre_last_dt_utc_start = pre_last_dt_utc_start.tz_localize("UTC") pre_post_dict = {