Choose better weather forecasting horizons (#131)

- Let create_forecasting_jobs decide which horizons to use for weather data
- add forecast resolutions for 5T & 10T resolutions

documentation/changelog.rst

@@ -15,7 +15,7 @@ New features
 
 Bugfixes
 -----------
-
+* Choose better forecasting horizons when weather data is posted [see `PR #131 <http://www.github.com/SeitaBV/flexmeasures/pull/131>`_]
 
 Infrastructure / Support
 ----------------------

flexmeasures/api/v1_1/implementations.py

@@ -210,7 +210,6 @@ def post_weather_data_response(  # noqa: C901
                         start,
                         start + duration,
                         resolution=duration / len(value_group),
-                        horizons=[horizon],
                         enqueue=False,  # will enqueue later, only if we successfully saved weather measurements
                     )
                 )

flexmeasures/api/v1_1/tests/conftest.py

@@ -20,7 +20,6 @@ def setup_api_test_data(db, setup_roles_users, add_market_prices):
 
     from flexmeasures.data.models.user import User, Role
     from flexmeasures.data.models.assets import Asset, AssetType
-    from flexmeasures.data.models.weather import WeatherSensor, WeatherSensorType
 
     user_datastore = SQLAlchemySessionUserDatastore(db.session, User, Role)
 
@@ -98,7 +97,23 @@ def setup_api_test_data(db, setup_roles_users, add_market_prices):
         power_forecasts.append(p_3)
     db.session.bulk_save_objects(power_forecasts)
 
-    # Create 2 weather sensors
+    add_weather_sensors(db)
+
+    print("Done setting up data for API v1.1 tests")
+
+
+@pytest.fixture(scope="function")
+def setup_fresh_api_v1_1_test_data(
+    fresh_db, setup_roles_users_fresh_db, setup_markets_fresh_db
+):
+    return fresh_db
+
+
+def add_weather_sensors(db):
+    """ Create 2 weather sensors """
+
+    from flexmeasures.data.models.weather import WeatherSensor, WeatherSensorType
+
     test_sensor_type = WeatherSensorType(name="wind_speed")
     db.session.add(test_sensor_type)
     sensor = WeatherSensor(
@@ -122,12 +137,3 @@ def setup_api_test_data(db, setup_roles_users, add_market_prices):
         unit="°C",
     )
     db.session.add(sensor)
-
-    print("Done setting up data for API v1.1 tests")
-
-
-@pytest.fixture(scope="function")
-def setup_fresh_api_v1_1_test_data(
-    fresh_db, setup_roles_users_fresh_db, setup_markets_fresh_db
-):
-    return fresh_db

flexmeasures/api/v1_1/tests/test_api_v1_1.py

@@ -19,6 +19,7 @@
     message_for_post_price_data,
     message_for_post_weather_data,
     verify_prices_in_db,
+    get_forecasting_jobs,
 )
 from flexmeasures.data.auth_setup import UNAUTH_ERROR_STATUS
 
@@ -189,10 +190,12 @@ def test_post_price_data_invalid_unit(setup_api_test_data, client, post_message)
     "post_message",
     [message_for_post_weather_data(), message_for_post_weather_data(temperature=True)],
 )
-def test_post_weather_data(setup_api_test_data, client, post_message):
+def test_post_weather_forecasts(setup_api_test_data, app, client, post_message):
     """
-    Try to post wind speed data as a logged-in test user with the Supplier role, which should succeed.
+    Try to post wind speed and temperature forecasts as a logged-in test user with the Supplier role, which should succeed.
+    As only forecasts are sent, no forecasting jobs are expected.
     """
+    assert len(get_forecasting_jobs("Weather")) == 0
 
     # post weather data
     auth_token = get_auth_token(client, "test_supplier@seita.nl", "testtest")
@@ -205,11 +208,13 @@ def test_post_weather_data(setup_api_test_data, client, post_message):
     assert post_weather_data_response.status_code == 200
     assert post_weather_data_response.json["type"] == "PostWeatherDataResponse"
 
+    assert len(get_forecasting_jobs("Weather")) == 0
+
 
 @pytest.mark.parametrize(
     "post_message", [message_for_post_weather_data(invalid_unit=True)]
 )
-def test_post_weather_data_invalid_unit(setup_api_test_data, client, post_message):
+def test_post_weather_forecasts_invalid_unit(setup_api_test_data, client, post_message):
     """
     Try to post wind speed data as a logged-in test user with the Supplier role, but with a wrong unit for wind speed,
     which should fail.

flexmeasures/api/v1_1/tests/test_api_v1_1_fresh_db.py

@@ -1,14 +1,19 @@
 from datetime import timedelta
+from iso8601 import parse_date
 
 import pytest
 from flask import url_for
 from isodate import duration_isoformat
 
+from flexmeasures.utils.time_utils import forecast_horizons_for
 from flexmeasures.api.common.responses import unapplicable_resolution
 from flexmeasures.api.tests.utils import get_auth_token
+from flexmeasures.api.v1_1.tests.conftest import add_weather_sensors
 from flexmeasures.api.v1_1.tests.utils import (
     message_for_post_price_data,
+    message_for_post_weather_data,
     verify_prices_in_db,
+    get_forecasting_jobs,
 )
 
 
@@ -47,3 +52,36 @@ def test_post_price_data_unexpected_resolution(
         verify_prices_in_db(
             post_message, [v for v in post_message["values"] for i in range(4)], db
         )
+
+
+@pytest.mark.parametrize(
+    "post_message",
+    [message_for_post_weather_data(as_forecasts=False)],
+)
+def test_post_weather_data(setup_fresh_api_v1_1_test_data, app, client, post_message):
+    """
+    Try to post wind speed data as a logged-in test user, which should lead to forecasting jobs.
+    """
+    db = setup_fresh_api_v1_1_test_data
+    add_weather_sensors(db)
+
+    auth_token = get_auth_token(client, "test_supplier@seita.nl", "testtest")
+    post_weather_data_response = client.post(
+        url_for("flexmeasures_api_v1_1.post_weather_data"),
+        json=post_message,
+        headers={"Authorization": auth_token},
+    )
+    print("Server responded with:\n%s" % post_weather_data_response.json)
+    assert post_weather_data_response.status_code == 200
+    assert post_weather_data_response.json["type"] == "PostWeatherDataResponse"
+
+    forecast_horizons = forecast_horizons_for(timedelta(minutes=5))
+    jobs = get_forecasting_jobs("Weather")
+    for job, horizon in zip(
+        sorted(jobs, key=lambda x: x.kwargs["horizon"]), forecast_horizons
+    ):
+        # check if jobs have expected horizons
+        assert job.kwargs["horizon"] == horizon
+        # check if jobs' start time (the time to be forecasted)
+        # is the weather observation plus the horizon
+        assert job.kwargs["start"] == parse_date(post_message["start"]) + horizon

flexmeasures/api/v1_1/tests/utils.py

@@ -1,10 +1,12 @@
 """Useful test messages"""
-from typing import Optional, Dict, Any
+from typing import Optional, Dict, List, Any
 from datetime import timedelta
 from isodate import duration_isoformat, parse_duration, parse_datetime
 
 import pandas as pd
 from numpy import tile
+from rq.job import Job
+from flask import current_app
 
 from flexmeasures.api.common.utils.api_utils import get_generic_asset
 from flexmeasures.data.models.markets import Market, Price
@@ -113,7 +115,7 @@ def message_for_post_price_data(
 
 
 def message_for_post_weather_data(
-    invalid_unit: bool = False, temperature: bool = False
+    invalid_unit: bool = False, temperature: bool = False, as_forecasts: bool = True
 ) -> dict:
     message: Dict[str, Any] = {
         "type": "PostWeatherDataRequest",
@@ -136,6 +138,8 @@ def message_for_post_weather_data(
             message["unit"] = "°C"  # Right unit for temperature
     elif invalid_unit:
         message["unit"] = "°C"  # Wrong unit for wind speed
+    if not as_forecasts:
+        message["horizon"] = "PT0H"  # weather measurements
     return message
 
 
@@ -159,3 +163,11 @@ def verify_prices_in_db(post_message, values, db, swapped_sign: bool = False):
     if swapped_sign:
         df["value"] = -df["value"]
     assert df.value.tolist() == values
+
+
+def get_forecasting_jobs(timed_value_type: str) -> List[Job]:
+    return [
+        job
+        for job in current_app.queues["forecasting"].jobs
+        if job.kwargs["timed_value_type"] == timed_value_type
+    ]

flexmeasures/data/services/forecasting.py

@@ -78,7 +78,7 @@ def create_forecasting_jobs(
     current default model configuration will be used.
 
     It's possible to customize model parameters, but this feature is (currently) meant to only
-    be used by tests, so that model behavior can be adapted to test conditions. If used outside
+    be used by tests, so that model behaviour can be adapted to test conditions. If used outside
     of testing, an exception is raised.
 
     if enqueue is True (default), the jobs are put on the redis queue.

flexmeasures/utils/time_utils.py

@@ -195,7 +195,9 @@ def forecast_horizons_for(
     else:
         resolution_str = resolution
     horizons = []
-    if resolution_str in ("15T", "1h", "H"):
+    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"]