#574 Avoid Recomputing Schedules

documentation/changelog.rst

@@ -13,6 +13,7 @@ New features
 * Keyboard control over replay [see `PR #562 <https://www.github.com/FlexMeasures/flexmeasures/pull/562>`_]
 * The ``FLEXMEASURES_MAX_PLANNING_HORIZON`` config setting can also be set as an integer number of planning steps rather than just as a fixed duration, which makes it possible to schedule further ahead in coarser time steps [see `PR #583 <https://www.github.com/FlexMeasures/flexmeasures/pull/583>`_]
 * Different text styles for CLI output for errors, warnings or success messages. [see `PR #609 <https://www.github.com/FlexMeasures/flexmeasures/pull/609>`_]
+* Avoid Recomputing Schedules[see `PR #616 <https://www.github.com/FlexMeasures/flexmeasures/pull/616>`_]
 
 Bugfixes
 -----------

flexmeasures/data/models/time_series.py

@@ -524,6 +524,17 @@ def find_closest(
         else:
             return query.limit(n).all()
 
+    def make_hashable(self) -> tuple:
+        """Returns a tuple with the properties subject to change
+        In principle all properties (except ID) of a given sensor could be changed, but not all changes are relevant to warrant reanalysis (e.g. scheduling or forecasting).
+        """
+
+        generic_asset_attributes = getattr(self.generic_asset, "attribute", None)
+
+        parameters = [self.id, self.attributes, generic_asset_attributes]
+
+        return tuple(parameters)
+
 
 class TimedBelief(db.Model, tb.TimedBeliefDBMixin):
     """A timed belief holds a precisely timed record of a belief about an event.

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 job_cache
 
 
+@job_cache("scheduling")
 def create_scheduling_job(
     sensor: Sensor,
     job_id: str | None = None,

flexmeasures/data/services/utils.py

@@ -0,0 +1,110 @@
+import functools
+from flask import current_app
+from rq.job import Job
+
+import hashlib
+import base64
+
+
+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 callable(
+        getattr(o, "make_hashable", None)
+    ):  # checks if the object o has the method make_hashable
+        return o.make_hashable()
+
+    return o
+
+
+def hash_function_arguments(args, kwags):
+    """Combines the hashes of the args and kargs
+
+    The way to go to do h(x,y) = hash(hash(x) || hash(y)) because it avoid the following:
+
+    1) h(x,y) = hash(x || y), might create a collision if we delete the last n characters of x and we append them in front of y. e.g h("abc", "d") = h("ab", "cd")
+    2) we don't want to sort x and y, because we need the function h(x,y) != h(y,x)
+    3) extra hashing just avoid that we can't decompose the input arguments and track if the same args or kwarg are called several times. More of a security measure I think.
+
+    source: https://crypto.stackexchange.com/questions/55162/best-way-to-hash-two-values-into-one
+    """
+    return make_hash_sha256(
+        make_hash_sha256(args) + make_hash_sha256(kwags)
+    )  # concat two hashes
+
+
+def job_cache(queue):
+    """
+    To avoid recomputing the same task multiple times, this decortator checks if the function was already been called with the
+    same arguments. Input arguments are hashed and stored as Redis keys with values the job ids (`input_arguments_hash:job_id`).
+
+    The benefits of using redis to store the input arguments over a local cache, such as LRU Cache, are:
+    1) It will work in distributed environments (in computing clusters), where multiple workers will avoid repeating
+      work as the cache will be shared across them.
+    2) Cached calls are logged, which means that we can easily debug.
+    3) Cache will still be there on restarts.
+    """
+
+    def decorator(func):
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+
+            # checking if force is an input argument of `func`
+            force_new_job_creation = kwargs.pop("force_new_job_creation", False)
+
+            # creating a hash from args and kwargs
+            args_hash = hash_function_arguments(args, kwargs)
+
+            # check if the key hash exists in the redis equeue
+            if (
+                current_app.queues[queue].connection.exists(args_hash)
+                and not force_new_job_creation
+            ):
+                current_app.logger.info(
+                    f"The function {func.__name__} has been called already with the same arguments. Skipping..."
+                )
+
+                # get job id
+                job_id = current_app.queues[queue].connection.get(args_hash).decode()
+
+                job = Job.fetch(
+                    job_id, connection=current_app.queues[queue].connection
+                )  # get job object from job id
+
+                return job  # returning the same job regardless of the status (SUCCESS, FAILED, ...)
+            else:
+                # if the job hasn't been called before or the job has failed -> create job
+                job = func(*args, **kwargs)  # create a new job
+
+                # 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
+
+        return wrapper
+
+    return decorator

flexmeasures/data/tests/test_scheduling_jobs.py

@@ -1,4 +1,3 @@
-# flake8: noqa: E402
 from datetime import datetime, timedelta
 import os