add fallback scheduler switch mechanism

Signed-off-by: Victor Garcia Reolid <victor@seita.nl>

documentation/configuration.rst

@@ -647,3 +647,11 @@ FLEXMEASURES_API_SUNSET_LINK
 Allow to override the default sunset link for your clients.
 
 Default: ``None`` (defaults are set internally for each sunset API version, e.g. ``"https://flexmeasures.readthedocs.io/en/v0.13.0/api/v2_0.html"`` for v2.0)
+
+
+FLEXMEASURES_FALLBACK_REDIRECT
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Allow control over the effect of redirecting to the fallback scheduler.
+
+Default : ``True``

flexmeasures/api/v3_0/sensors.py

@@ -445,11 +445,22 @@ def get_schedule(  # noqa: C901
 
         # Look up the scheduling job
         connection = current_app.queues["scheduling"].connection
+
         try:  # First try the scheduling queue
             job = Job.fetch(job_id, connection=connection)
         except NoSuchJobError:
             return unrecognized_event(job_id, "job")
 
+        if (
+            not current_app.config.get("FLEXMEASURES_FALLBACK_REDIRECT")
+            and job.is_failed
+            and ("fallback_job_id" in job.meta)
+        ):
+            try:  # First try the scheduling queue
+                job = Job.fetch(job.meta["fallback_job_id"], connection=connection)
+            except NoSuchJobError:
+                return unrecognized_event(job.meta["fallback_job_id"], "fallback-job")
+
         scheduler_info_msg = ""
         scheduler_info = job.meta.get("scheduler_info", dict(scheduler=""))
         scheduler_info_msg = f"{scheduler_info['scheduler']} was used."
@@ -470,20 +481,23 @@ def get_schedule(  # noqa: C901
 
             fallback_job_id = job.meta.get("fallback_job_id")
 
-            # redirect to the fallback schedule endpoint if the fallback_job_id
-            # is defined in the metadata of the original job
-            if fallback_job_id is not None:
-                return fallback_schedule_redirect(
-                    message,
-                    url_for(
-                        "SensorAPI:get_schedule",
-                        uuid=fallback_job_id,
-                        id=sensor.id,
-                        _external=True,
-                    ),
-                )
-            else:
-                return unknown_schedule(message)
+            if current_app.config.get("FLEXMEASURES_FALLBACK_REDIRECT"):
+                # redirect to the fallback schedule endpoint if the fallback_job_id
+                # is defined in the metadata of the original job
+                if fallback_job_id is not None:
+                    return fallback_schedule_redirect(
+                        message,
+                        url_for(
+                            "SensorAPI:get_schedule",
+                            uuid=fallback_job_id,
+                            id=sensor.id,
+                            _external=True,
+                        ),
+                    )
+                else:
+                    return unknown_schedule(message)
+            # case that the job failed or the fallback schedule job failed (FLEXMEASURES_FALLBACK_REDIRECT = False)
+            return unknown_schedule(message)
 
         elif job.is_started:
             return unknown_schedule(f"Scheduling job in progress. {scheduler_info_msg}")

flexmeasures/api/v3_0/tests/test_sensor_schedules.py

@@ -347,6 +347,7 @@ def test_get_schedule_fallback(
     Test if the fallback job is created after a failing StorageScheduler call. This test
     is based on flexmeasures/data/models/planning/tests/test_solver.py
     """
+    app.config["FLEXMEASURES_FALLBACK_REDIRECT"] = True
 
     target_soc = 9
     charging_station_name = "Test charging station"
@@ -471,3 +472,122 @@ def test_get_schedule_fallback(
             for source in charging_station.search_beliefs().sources.unique()
         ]
         assert "StorageFallbackScheduler" in models
+
+        app.config["FLEXMEASURES_FALLBACK_REDIRECT"] = False
+
+
+@pytest.mark.parametrize(
+    "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True
+)
+def test_get_schedule_fallback_not_redirect(
+    app,
+    add_battery_assets,
+    add_market_prices,
+    battery_soc_sensor,
+    add_charging_station_assets,
+    keep_scheduling_queue_empty,
+    requesting_user,
+):
+    """
+    Test if the fallback scheduler is returned directly after a failing StorageScheduler call. This test
+    is based on flexmeasures/data/models/planning/tests/test_solver.py
+    """
+    app.config["FLEXMEASURES_FALLBACK_REDIRECT"] = False
+
+    target_soc = 9
+    charging_station_name = "Test charging station"
+
+    start = "2015-01-02T00:00:00+01:00"
+    epex_da = Sensor.query.filter(Sensor.name == "epex_da").one_or_none()
+    charging_station = add_charging_station_assets[charging_station_name].sensors[0]
+
+    assert charging_station.get_attribute("capacity_in_mw") == 2
+    assert charging_station.get_attribute("market_id") == epex_da.id
+
+    # create a scenario that yields an infeasible problem (unreachable target SOC at 2am)
+    message = {
+        "start": start,
+        "duration": "PT24H",
+        "flex-model": {
+            "soc-at-start": 10,
+            "soc-min": charging_station.get_attribute("min_soc_in_mwh", 0),
+            "soc-max": charging_station.get_attribute("max-soc-in-mwh", target_soc),
+            "roundtrip-efficiency": charging_station.get_attribute(
+                "roundtrip-efficiency", 1
+            ),
+            "storage-efficiency": charging_station.get_attribute(
+                "storage-efficiency", 1
+            ),
+            "soc-targets": [
+                {"value": target_soc, "datetime": "2015-01-02T02:00:00+01:00"}
+            ],
+        },
+    }
+
+    with app.test_client() as client:
+        # trigger storage scheduler
+        trigger_schedule_response = client.post(
+            url_for("SensorAPI:trigger_schedule", id=charging_station.id),
+            json=message,
+        )
+
+        # check that the call is successful
+        assert trigger_schedule_response.status_code == 200
+        job_id = trigger_schedule_response.json["schedule"]
+
+        # look for scheduling jobs in queue
+        assert (
+            len(app.queues["scheduling"]) == 1
+        )  # only 1 schedule should be made for 1 asset
+        job = app.queues["scheduling"].jobs[0]
+        assert job.kwargs["sensor_id"] == charging_station.id
+        assert job.kwargs["start"] == parse_datetime(message["start"])
+        assert job.id == job_id
+
+        # process only the job that runs the storage scheduler (max_jobs=1)
+        work_on_rq(
+            app.queues["scheduling"],
+            exc_handler=handle_scheduling_exception,
+            max_jobs=1,
+        )
+
+        # check that the job is failing
+        assert Job.fetch(
+            job_id, connection=app.queues["scheduling"].connection
+        ).is_failed
+
+        # the callback creates the fallback job which is still pending
+        assert len(app.queues["scheduling"]) == 1
+
+        fallback_job_id = Job.fetch(
+            job_id, connection=app.queues["scheduling"].connection
+        ).meta.get("fallback_job_id")
+
+        # check that the fallback_job_id is stored on the metadata of the original job
+        assert app.queues["scheduling"].get_job_ids()[0] == fallback_job_id
+        assert fallback_job_id != job_id
+
+        get_schedule_response = client.get(
+            url_for("SensorAPI:get_schedule", id=charging_station.id, uuid=job_id),
+        )
+
+        work_on_rq(
+            app.queues["scheduling"],
+            exc_handler=handle_scheduling_exception,
+            max_jobs=1,
+        )
+
+        get_schedule_response = client.get(
+            url_for("SensorAPI:get_schedule", id=charging_station.id, uuid=job_id),
+        )
+
+        assert get_schedule_response.status_code == 200
+
+        schedule = get_schedule_response.json
+
+        # check that the fallback schedule has the right status and start dates
+        assert schedule["status"] == "PROCESSED"
+        assert parse_datetime(schedule["start"]) == parse_datetime(start)
+        assert schedule["scheduler_info"]["scheduler"] == "StorageFallbackScheduler"
+
+        app.config["FLEXMEASURES_FALLBACK_REDIRECT"] = False

flexmeasures/data/tests/test_scheduling_jobs.py

@@ -204,6 +204,7 @@ def test_fallback_chain(
 
         FailingScheduler1 -> FailingScheduler2 -> SuccessfulScheduler
     """
+    app.config["FLEXMEASURES_FALLBACK_REDIRECT"] = True
 
     battery = add_battery_assets["Test battery"].sensors[0]
     app.db.session.flush()
@@ -254,3 +255,4 @@ def test_fallback_chain(
     assert success_job.kwargs["scheduler_specs"]["class"] == "SuccessfulScheduler"
 
     assert len(app.queues["scheduling"]) == 0
+    app.config["FLEXMEASURES_FALLBACK_REDIRECT"] = False

flexmeasures/utils/config_defaults.py

@@ -128,6 +128,8 @@ class Config(object):
         # todo: expand with other js versions used in FlexMeasures
     )
 
+    FLEXMEASURES_FALLBACK_REDIRECT: bool = False
+
     # Custom sunset switches
     FLEXMEASURES_API_SUNSET_ACTIVE: bool = False  # if True, sunset endpoints return 410 (Gone) responses; if False, they return 404 (Not Found) responses or will work as before, depending on whether the current FlexMeasures version still contains the endpoint logic
     FLEXMEASURES_API_SUNSET_DATE: str | None = None  # e.g. 2023-05-01