add option to load forecast horizon

Author: peterdudfield

nowcasting_datamodel/read/read.py

@@ -12,6 +12,7 @@
 from sqlalchemy.orm import contains_eager, joinedload
 from sqlalchemy.orm.session import Session
 
+
 from nowcasting_datamodel import N_GSP
 from nowcasting_datamodel.models import (
     ForecastSQL,
@@ -212,6 +213,7 @@ def get_all_gsp_ids_latest_forecast(
     start_target_time: Optional[datetime] = None,
     preload_children: Optional[bool] = False,
     historic: bool = False,
+    forecast_horizon_hours: Optional[int] = None,
 ) -> List[ForecastSQL]:
     """
     Read forecasts
@@ -235,6 +237,7 @@ def get_all_gsp_ids_latest_forecast(
         preload_children=preload_children,
         historic=historic,
         gsp_ids=list(range(0, N_GSP + 1)),
+        forecast_horizon_hours=forecast_horizon_hours,
     )
 
 
@@ -244,6 +247,7 @@ def get_latest_forecast_for_gsps(
     start_target_time: Optional[datetime] = None,
     preload_children: Optional[bool] = False,
     historic: bool = False,
+    forecast_horizon_hours: Optional[int] = None,
     gsp_ids: List[int] = None,
 ):
     """
@@ -256,6 +260,9 @@ def get_latest_forecast_for_gsps(
     :param preload_children: Option to preload children. This is a speed up, if we need them.
     :param historic: Option to load historic values or not
     :param gsp_ids: Option to filter on gsps. If None, then only the lastest forecast is loaded.
+    :param forecast_horizon_hours: Optional filter on forecast horizon. For example
+        forecast_horizon_hours=2, means load the forecast than was made 2 hours before the target time.
+        Note this only works for non-historic data.
 
     return: List of forecasts objects from database
 
@@ -293,6 +300,16 @@ def get_latest_forecast_for_gsps(
             query=query, start_target_time=start_target_time, historic=historic
         )
 
+    from sqlalchemy import text
+    if forecast_horizon_hours is not None:
+        assert historic is False, Exception(
+            "Loading a forecast horizon only works on non latest data."
+        )
+        query = query.join(ForecastValueSQL).filter(
+            ForecastValueSQL.target_time - ForecastValueSQL.created_utc
+            >= text(f"interval '{forecast_horizon_hours} hour'")
+        )
+
     query = query.join(LocationSQL)
 
     # option to preload values, makes querying quicker

tests/read/test_read_forecast.py

@@ -0,0 +1,63 @@
+import logging
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from freezegun import freeze_time
+
+from nowcasting_datamodel.fake import (
+    make_fake_forecast,
+    make_fake_forecasts,
+    make_fake_national_forecast,
+    make_fake_pv_system,
+)
+from nowcasting_datamodel.models import (
+    Forecast,
+    ForecastValue,
+    ForecastValueLatestSQL,
+    ForecastValueSQL,
+    InputDataLastUpdatedSQL,
+    LocationSQL,
+    MLModel,
+    PVSystem,
+    Status,
+    national_gb_label,
+)
+from nowcasting_datamodel.read.read import (
+    get_all_gsp_ids_latest_forecast,
+    get_latest_forecast_for_gsps,
+    get_all_locations,
+    get_forecast_values,
+    get_latest_forecast,
+    get_latest_forecast_created_utc,
+    get_latest_input_data_last_updated,
+    get_latest_national_forecast,
+    get_latest_status,
+    get_location,
+    get_model,
+    get_pv_system,
+    update_latest_input_data_last_updated,
+)
+from nowcasting_datamodel.save import save_pv_system
+
+logger = logging.getLogger(__name__)
+
+
+def test_get_latest_forecast_created_utc_gsp(db_session):
+    t0_datetime_utc = datetime(2022, 1, 1, 12, tzinfo=timezone.utc)
+
+    f1 = make_fake_forecast(gsp_id=1, session=db_session, t0_datetime_utc=t0_datetime_utc)
+    f2 = make_fake_forecast(gsp_id=1, session=db_session, t0_datetime_utc=t0_datetime_utc)
+
+    f1.forecast_values[0].created_utc = datetime(2022, 1, 1, 10, tzinfo=timezone.utc)
+    f1.forecast_values[1].created_utc = datetime(2022, 1, 1, 10, tzinfo=timezone.utc)
+
+    f2.forecast_values[0].created_utc = datetime(2022, 1, 1, 12, tzinfo=timezone.utc)
+    f2.forecast_values[1].created_utc = datetime(2022, 1, 1, 12, tzinfo=timezone.utc)
+
+    db_session.add_all([f1, f2])
+    db_session.commit()
+
+    f = get_latest_forecast_for_gsps(session=db_session, forecast_horizon_hours=2, gsp_ids=[1])
+    assert len(f) == 1
+    assert f[0].forecast_values[0].created_utc == datetime(2022, 1, 1, 10, tzinfo=timezone.utc)
+    assert f[0].forecast_values[0].target_time == datetime(2022, 1, 1, 12, tzinfo=timezone.utc)