#574 Avoid Recomputing Schedules

flexmeasures/data/services/scheduling.py

@@ -19,8 +19,10 @@
 from flexmeasures.data.models.data_sources import DataSource
 from flexmeasures.data.utils import get_data_source, save_to_db
 from flexmeasures.utils.time_utils import server_now
+from flexmeasures.data.services.utils import redis_cache
 
 
+@redis_cache("scheduling")
 def create_scheduling_job(
     sensor: Sensor,
     job_id: str | None = None,

flexmeasures/data/services/utils.py

@@ -0,0 +1,90 @@
+import functools
+from flask import current_app
+from rq.job import Job
+
+import hashlib
+import base64
+
+from flexmeasures.data.models.time_series import Sensor
+
+
+def make_hash_sha256(o):
+    """
+    SHA256 instead of Python's hash function because apparently, python native hashing function
+    yields different results on restarts.
+    Source: https://stackoverflow.com/a/42151923
+    """
+    hasher = hashlib.sha256()
+    hasher.update(repr(make_hashable(o)).encode())
+    return base64.b64encode(hasher.digest()).decode()
+
+
+def make_hashable(o):
+    """
+    Function to create hashes for dictionaries with nested objects
+    Source: https://stackoverflow.com/a/42151923
+    """
+    if isinstance(o, (tuple, list)):
+        return tuple((make_hashable(e) for e in o))
+
+    if isinstance(o, dict):
+        return tuple(sorted((k, make_hashable(v)) for k, v in o.items()))
+
+    if isinstance(o, (set, frozenset)):
+        return tuple(sorted(make_hashable(e) for e in o))
+
+    if isinstance(o, Sensor):
+        return tuple(
+            (
+                make_hashable(getattr(o, attr, None))
+                for attr in ["attributes", "annotations", "id"]
+            )
+        )
+
+    return o
+
+
+def hash_function_arguments(args, kwags):
+    return make_hash_sha256(args) + make_hash_sha256(kwags)  # concat two hashes
+
+
+def redis_cache(queue):
+    """
+    Decorator that checks if a function has alread been called with the same arguments and
+    fetches the job using the job id, which is stored in Redis.
+    """
+
+    def decorator(func):
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            # Creating a hash from args and kwargs
+            args_hash = hash_function_arguments(args, kwargs)
+
+            # Checking if there exist the key hash in the redis queue
+            if current_app.queues[queue].connection.exists(args_hash):
+                current_app.logger.info(
+                    f"The function {func.__name__} has been called alread with the same arguments. Skipping..."
+                )
+
+                # get job id
+                job_id = current_app.queues[queue].connection.get(args_hash).decode()
+
+                # get job object from job id
+                return Job.fetch(
+                    job_id, connection=current_app.queues[queue].connection
+                )
+            else:  # new call
+
+                # call function
+                job = func(*args, **kwargs)
+
+                # store function call in redis
+                current_app.queues[queue].connection.set(
+                    args_hash, job.id
+                )  # setting return value of function call to the hash of its inputs
+
+                return job  # todo: try catch, if it fails, don't hash it or impose a max retry number
+
+        return wrapper
+
+    return decorator

flexmeasures/data/tests/test_scheduling_repeated_jobs.py

@@ -0,0 +1,215 @@
+# flake8: noqa: E402
+from datetime import datetime, timedelta
+import os
+import copy
+import logging
+
+import pytz
+import pytest
+from rq.job import Job
+
+from flexmeasures.data.models.data_sources import DataSource
+from flexmeasures.data.models.time_series import Sensor, TimedBelief
+from flexmeasures.data.tests.utils import work_on_rq, exception_reporter
+from flexmeasures.data.services.scheduling import (
+    create_scheduling_job,
+    load_custom_scheduler,
+)
+from flexmeasures.data.services.utils import hash_function_arguments
+
+
+@pytest.mark.parametrize(
+    "args_modified,kwargs_modified,equal",
+    [
+        (
+            [1, 2, "1"],
+            {
+                "key1": "value1",
+                "key2": "value2",
+                "key3": 3,
+                "key4": {"key1_nested": 1, "key2_nested": 2},
+            },
+            True,
+        ),
+        (
+            [1, 2, 1],
+            {
+                "key1": "value1",
+                "key2": "value2",
+                "key3": 3,
+                "key4": {"key1_nested": 1, "key2_nested": 2},
+            },
+            False,
+        ),
+        (
+            [1, 2, "1"],
+            {
+                "key1": "different",
+                "key2": "value2",
+                "key3": 3,
+                "key4": {"key1_nested": 1, "key2_nested": 2},
+            },
+            False,
+        ),
+        (
+            [1, 2, "1"],
+            {
+                "different": "value1",
+                "key2": "value2",
+                "key3": 3,
+                "key4": {"key1_nested": 1, "key2_nested": 2},
+            },
+            False,
+        ),
+        (
+            [1, 2, "1"],
+            {
+                "key1": "value1",
+                "key2": "value2",
+                "key3": 3,
+                "key4": {"key1_nested": "different", "key2_nested": 2},
+            },
+            False,
+        ),
+    ],
+)
+def test_hashing_simple(args_modified: list, kwargs_modified: dict, equal: bool):
+    args = [1, 2, "1"]
+    kwargs = {
+        "key1": "value1",
+        "key2": "value2",
+        "key3": 3,
+        "key4": {"key1_nested": 1, "key2_nested": 2},
+    }
+
+    hash_original = hash_function_arguments(args, kwargs)
+    hash_modified = hash_function_arguments(args_modified, kwargs_modified)
+
+    if equal:
+        assert hash_original == hash_modified
+    else:
+        assert hash_original != hash_modified
+
+
+def test_hashing(db, app, add_charging_station_assets, setup_test_data):
+    soc_at_start = 1
+    target_soc = 5
+    duration_until_target = timedelta(hours=2)
+
+    charging_station = Sensor.query.filter(
+        Sensor.name == "Test charging station"
+    ).one_or_none()
+    tz = pytz.timezone("Europe/Amsterdam")
+    start = tz.localize(datetime(2015, 1, 2))
+    end = tz.localize(datetime(2015, 1, 3))
+    target_datetime = start + duration_until_target
+    resolution = timedelta(minutes=15)
+    soc_targets = [dict(datetime=target_datetime.isoformat(), value=target_soc)]
+
+    kwargs = dict(
+        sensor=charging_station,
+        start=start,
+        end=end,
+        belief_time=start,
+        resolution=resolution,
+        flex_model={"soc-at-start": soc_at_start, "soc-targets": soc_targets},
+    )
+    args = []
+
+    hash = hash_function_arguments(args, kwargs)
+
+    # checks that hashes are consistent between calls
+    assert (
+        hash
+        == "LjjneyLDFKRJ6R+v7ZKkOCasaqQDrmqKy2z1gjn7r10=Q3FfKNrQYBgY/qWrw+lEjVEI/GWm9sThyBbOd+sGIKk="
+    )
+
+    # checks that different arguments yield different hashes
+    kwargs2 = copy.deepcopy(kwargs)
+    kwargs2["resolution"] = timedelta(minutes=12)
+
+    hash2 = hash_function_arguments(args, kwargs2)
+
+    assert hash != hash2
+
+
+def test_scheduling_multiple_triggers(
+    caplog, db, app, add_charging_station_assets, setup_test_data
+):
+    caplog.set_level(
+        logging.INFO
+    )  # setting the logging level of the log capture fixture
+
+    soc_at_start = 1
+    target_soc = 5
+    duration_until_target = timedelta(hours=2)
+
+    charging_station = Sensor.query.filter(
+        Sensor.name == "Test charging station"
+    ).one_or_none()
+    tz = pytz.timezone("Europe/Amsterdam")
+    start = tz.localize(datetime(2015, 1, 2))
+    end = tz.localize(datetime(2015, 1, 3))
+    target_datetime = start + duration_until_target
+    resolution = timedelta(minutes=15)
+    soc_targets = [dict(datetime=target_datetime.isoformat(), value=target_soc)]
+
+    assert (
+        DataSource.query.filter_by(name="FlexMeasures", type="scheduling script")
+        .where()
+        .one_or_none()
+        is None
+    )  # Make sure the scheduler data source isn't there
+
+    # schedule 1 job
+    job1 = create_scheduling_job(
+        sensor=charging_station,
+        start=start,
+        end=end,
+        belief_time=start,
+        resolution=resolution,
+        flex_model={"soc-at-start": soc_at_start, "soc-targets": soc_targets},
+    )
+    work_on_rq(app.queues["scheduling"], exc_handler=exception_reporter)
+
+    caplog.clear()
+
+    # Schedule same job
+    job2 = create_scheduling_job(
+        sensor=charging_station,
+        start=start,
+        end=end,
+        belief_time=start,
+        resolution=resolution,
+        flex_model={"soc-at-start": soc_at_start, "soc-targets": soc_targets},
+    )
+    work_on_rq(app.queues["scheduling"], exc_handler=exception_reporter)
+
+    # checking that the decorator is detecting that the job is repeated
+    assert (
+        sum(
+            [
+                "The function create_scheduling_job has been called alread with the same arguments. Skipping..."
+                in rec.message
+                for rec in caplog.records
+            ]
+        )
+        == 1
+    )
+
+    # checking that they have the same job id
+    assert job1.id == job2.id
+
+    # checking that a different schedule trigger is actually computed
+    soc_at_start = 2
+    job3 = create_scheduling_job(
+        sensor=charging_station,
+        start=start,
+        end=end,
+        belief_time=start,
+        resolution=resolution,
+        flex_model={"soc-at-start": soc_at_start, "soc-targets": soc_targets},
+    )
+    work_on_rq(app.queues["scheduling"], exc_handler=exception_reporter)
+
+    assert job3.id != job1.id