Update timing defaults in forecasting schema

documentation/tut/forecasting_scheduling.rst

@@ -104,9 +104,8 @@ There are two ways to queue a forecasting job:
    .. code-block:: json
 
        {
-         "start_date": "2025-01-01T00:00:00+00:00",
-         "start_predict_date": "2025-01-04T00:00:00+00:00",
-         "end_date": "2025-01-04T04:00:00+00:00"
+         "start": "2025-01-04T00:00:00+00:00",
+         "duration": "PT4H"
        }
 
    Example response:

flexmeasures/api/common/schemas/utils.py

@@ -4,15 +4,18 @@
 from marshmallow import Schema, fields
 
 from flexmeasures.utils.doc_utils import rst_to_openapi
-from flexmeasures.data.schemas.forecasting.pipeline import ForecastingTriggerSchema
+from flexmeasures.data.schemas.forecasting.pipeline import (
+    ForecastingTriggerSchema,
+    TrainPredictPipelineConfigSchema,
+)
 from flexmeasures.data.schemas.sensors import (
     SensorReferenceSchema,
     VariableQuantityField,
     VariableQuantityOpenAPISchema,
 )
 
 
-def make_openapi_compatible(schema_cls: Type[Schema]) -> Type[Schema]:
+def make_openapi_compatible(schema_cls: Type[Schema]) -> Type[Schema]:  # noqa: C901
     """
     Create an OpenAPI-compatible version of a Marshmallow schema.
 
@@ -28,11 +31,24 @@ def make_openapi_compatible(schema_cls: Type[Schema]) -> Type[Schema]:
     new_fields = {}
     for name, field in schema_cls._declared_fields.items():
 
-        if schema_cls == ForecastingTriggerSchema:
+        if schema_cls in (ForecastingTriggerSchema, TrainPredictPipelineConfigSchema):
             if "cli" in field.metadata and field.metadata["cli"].get(
                 "cli-exclusive", False
             ):
                 continue
+            if isinstance(field, fields.Nested):
+                nested_schema_cls = type(field.schema)
+                if nested_schema_cls is TrainPredictPipelineConfigSchema:
+                    field_copy = fields.Nested(
+                        make_openapi_compatible(nested_schema_cls),
+                        metadata=field.metadata,
+                        data_key=field.data_key,
+                        many=field.many,
+                        required=field.required,
+                        allow_none=field.allow_none,
+                    )
+                    new_fields[name] = field_copy
+                    continue
 
         # Copy metadata, but sanitize description for OpenAPI
         metadata = dict(getattr(field, "metadata", {}))

flexmeasures/api/v3_0/__init__.py

@@ -140,7 +140,7 @@ def create_openapi_specs(app: Flask):
     # Explicitly register OpenAPI-compatible schemas
     schemas = [
         ("FlexContextOpenAPISchema", flex_context_schema_openAPI),
-        ("forecaster_parameters_schema_openAPI", forecasting_trigger_schema_openAPI),
+        ("forecasting_trigger_schema_openAPI", forecasting_trigger_schema_openAPI),
         ("UserAPIQuerySchema", UserAPIQuerySchema),
         ("AssetAPIQuerySchema", AssetAPIQuerySchema),
         ("AssetSchema", AssetSchema),

flexmeasures/api/v3_0/sensors.py

@@ -82,7 +82,17 @@
 partial_sensor_schema = SensorSchema(partial=True, exclude=["generic_asset_id"])
 
 # Create ForecasterParametersSchema OpenAPI compatible schema
-forecasting_trigger_schema_openAPI = make_openapi_compatible(ForecastingTriggerSchema)
+EXCLUDED_FORECASTING_FIELDS = [
+    # todo: hide these in the config schema instead
+    # "train_period",
+    # "max_training_period",
+    "sensor_to_save",
+]
+forecasting_trigger_schema_openAPI = make_openapi_compatible(ForecastingTriggerSchema)(
+    # partial=True,
+    exclude=EXCLUDED_FORECASTING_FIELDS
+    + ["sensor"],
+)
 
 
 class SensorKwargsSchema(Schema):
@@ -1524,7 +1534,10 @@ def get_status(self, id, sensor):
 
     @route("/<id>/forecasts/trigger", methods=["POST"])
     @use_args(
-        ForecastingTriggerSchema(),
+        ForecastingTriggerSchema(
+            # partial=True,
+            exclude=EXCLUDED_FORECASTING_FIELDS,
+        ),
         location="combined_sensor_data_description",
         as_kwargs=True,
     )
@@ -1558,11 +1571,10 @@ def trigger_forecast(self, id: int, **params):
             required: true
             content:
               application/json:
-                schema: forecaster_parameters_schema_openAPI
+                schema: forecasting_trigger_schema_openAPI
                 example:
-                  start-date: "2026-01-01T00:00:00+01:00"
-                  start-predict-date: "2026-01-15T00:00:00+01:00"
-                  end-date: "2026-01-17T00:00:00+01:00"
+                  start: "2026-01-15T00:00:00+01:00"
+                  duration: "P2D"
           responses:
             200:
               description: PROCESSED
@@ -1609,7 +1621,7 @@ def trigger_forecast(self, id: int, **params):
             forecaster = get_data_generator(
                 source=None,
                 model=model,
-                config={},
+                config=parameters.pop("config", {}),
                 save_config=True,
                 data_generator_type=Forecaster,
             )
@@ -1780,8 +1792,8 @@ def get_forecast(self, id: int, uuid: str, sensor: Sensor, job_id: str):
             data_source = get_data_source_for_job(job, type="forecasting")
 
             forecasts = sensor.search_beliefs(
-                event_starts_after=job.meta.get("start_predict_date"),
-                event_ends_before=job.meta.get("end_date"),
+                event_starts_after=job.meta.get("start"),
+                event_ends_before=job.meta.get("end"),
                 source=data_source,
                 most_recent_beliefs_only=True,
                 use_latest_version_per_event=True,

flexmeasures/api/v3_0/tests/test_forecasting_api.py

@@ -2,9 +2,7 @@
 import isodate
 import pytest
 from flask import url_for
-from flexmeasures.data.services.scheduling import (
-    get_data_source_for_job,
-)
+
 from rq.job import Job
 from flexmeasures.utils.job_utils import work_on_rq
 from flexmeasures.api.tests.utils import get_auth_token
@@ -35,18 +33,21 @@ def test_trigger_and_fetch_forecasts(
 
     # Trigger job
     payload = {
-        "start-date": "2025-01-01T00:00:00+00:00",
-        "start-predict-date": "2025-01-05T00:00:00+00:00",
-        "end-date": "2025-01-05T02:00:00+00:00",
+        "start": "2025-01-05T00:00:00+00:00",
+        "end": "2025-01-05T02:00:00+00:00",
         "max-forecast-horizon": "PT1H",
-        "retrain-frequency": "PT1H",
+        "forecast-frequency": "PT1H",
+        "config": {
+            "train-start": "2025-01-01T00:00:00+00:00",
+            "retrain-frequency": "PT1H",
+        },
     }
 
     trigger_url = url_for("SensorAPI:trigger_forecast", id=sensor_0.id)
     trigger_res = client.post(
         trigger_url, json=payload, headers={"Authorization": token}
     )
-    assert trigger_res.status_code == 200
+    assert trigger_res.status_code == 200, trigger_res.json
 
     trigger_json = trigger_res.get_json()
     wrap_up_job = app.queues["forecasting"].fetch_job(trigger_json["forecast"])
@@ -77,15 +78,15 @@ def test_trigger_and_fetch_forecasts(
     payload["sensor"] = sensor_1.id
 
     # Run pipeline manually to compute expected forecasts
-    pipeline = TrainPredictPipeline()
+    pipeline = TrainPredictPipeline(config=payload.pop("config", {}))
     pipeline.compute(parameters=payload)
 
     # Fetch forecasts for each job
     for job_id in job_ids:
 
         fetch_url = url_for("SensorAPI:get_forecast", id=sensor_0.id, uuid=job_id)
         res = client.get(fetch_url, headers={"Authorization": token})
-        assert res.status_code == 200
+        assert res.status_code == 200, res.json
 
         data = res.get_json()
 
@@ -103,14 +104,11 @@ def test_trigger_and_fetch_forecasts(
         assert isinstance(api_forecasts, list)
         assert len(api_forecasts) > 0
 
-        # Identify which data source wrote these beliefs
-        data_source = get_data_source_for_job(job, type="forecasting")
-
         # Load only the latest belief per event_start
         forecasts_df = sensor_1.search_beliefs(
-            event_starts_after=job.meta.get("start_predict_date"),
-            event_ends_before=job.meta.get("end_date") + sensor_1.event_resolution,
-            source=data_source,
+            event_starts_after=job.meta.get("start"),
+            event_ends_before=job.meta.get("end"),
+            source_types=["forecaster"],
             most_recent_beliefs_only=True,
             use_latest_version_per_event=True,
         ).reset_index()

flexmeasures/data/models/forecasting/__init__.py

@@ -141,7 +141,8 @@ def _clean_parameters(self, parameters: dict) -> dict:
             "output-path",
             "sensor-to-save",
             "as-job",
-            "n_cycles",  # Computed internally, still uses snake_case
+            "m_viewpoints",  # Computed internally, still uses snake_case
+            "sensor",
         ]
 
         for field in fields_to_remove: