Add MutableCachableProperty with tests

stonesoup/base.py

@@ -53,13 +53,15 @@ def __init__(self, foo, bar=bar.default, *args, **kwargs):
 
 
 """
+import functools
 import inspect
 import textwrap
 from reprlib import Repr
 from abc import ABCMeta
 from collections import OrderedDict
 from copy import copy
 from types import MappingProxyType
+from typing import Callable, Sequence
 
 from ._util import cached_property
 
@@ -457,3 +459,95 @@ def __repr__(self):
         rep = Base._repr.whitespace_remove(max_len_whitespace, value)
         truncate = '\n...\n...  (truncated due to length)\n...'
         return ''.join([rep[:max_out], truncate]) if len(rep) > max_out else rep
+
+
+class MutableCacheableProperty:
+    """ A value of ``None`` cannot be cached.
+    Only works if all properties of the object are immutable
+    """
+    def __call__(self, cls):
+        cls._cached_properties = set()
+        cls.__setattr__ = MutableCacheableProperty._cache_set_attr(cls.__setattr__)
+        return cls
+
+    @staticmethod
+    def _cache_set_attr(orig_setattr):
+        def wrapper(self, name, value):
+            result = orig_setattr(self, name, value)
+            if not name.endswith('_cache'):
+                for cp in self._cached_properties:
+                    setattr(self, MutableCacheableProperty._get_cache_name(cp), None)
+            return result
+        return wrapper
+
+    @staticmethod
+    def cached_property(depends_on: Sequence = None) -> Callable:
+        if depends_on is None:
+            depends_on = []
+
+        def rcp_decorator(func: Callable) -> Callable:
+            @functools.wraps(func)
+            def rcp_wrapper(obj):
+                name = func.__name__
+                cache_name = MutableCacheableProperty._get_cache_name(name)
+                dependents = []
+                for d in depends_on:
+                    try:
+                        dependent = getattr(obj, d._property_name)
+                    except AttributeError:
+                        dependent = d.fget(obj)
+                    dependents.append(dependent)
+                MutableCacheableProperty._set_up_cache_variables(obj, name, dependents)
+                cache_value = MutableCacheableProperty._cache_value_with_dependent_check(
+                    obj, name, dependents)
+                if cache_value is None:
+                    result = func(obj)
+                    setattr(obj, cache_name, result)
+                    qualified_name = MutableCacheableProperty._get_qualified_name(obj, name)
+                    for dependent in dependents:
+                        setattr(dependent,
+                                MutableCacheableProperty._get_cache_name(qualified_name),
+                                True)
+                else:
+                    result = cache_value
+                return result
+            return rcp_wrapper
+        return rcp_decorator
+
+    @staticmethod
+    def _get_cache_name(name):
+        cache_name = '_{}_cache'.format(name)
+        return cache_name
+
+    @staticmethod
+    def _get_qualified_name(obj, name):
+        return '{}_{}'.format(obj.__class__.__name__, name)
+
+    @staticmethod
+    def _set_up_cache_variables(obj, name, dependents):
+        full_cache_name = MutableCacheableProperty._get_qualified_name(obj, name)
+        for dependent in dependents:
+            dependent._cached_properties.add(full_cache_name)
+        # noinspection PyProtectedMember
+        obj._cached_properties.add(name)
+
+    @staticmethod
+    def _cache_value_with_dependent_check(obj, name, dependents):
+        # first check for dependents having invalidated the cache
+        dependent_cache_name = MutableCacheableProperty._get_cache_name(
+            MutableCacheableProperty._get_qualified_name(obj, name))
+        for dependent in dependents:
+            try:
+                value = getattr(dependent, dependent_cache_name)
+                if value is None:
+                    return None
+            except AttributeError:
+                return None
+
+        # if still good, then check cache value
+        cache_name = MutableCacheableProperty._get_cache_name(name)
+        try:
+            cache_value = getattr(obj, cache_name)
+        except AttributeError:
+            cache_value = None
+        return cache_value

stonesoup/tests/test_caching.py

@@ -0,0 +1,155 @@
+from stonesoup.base import MutableCacheableProperty, Base, Property
+
+
+def test_cache_creation():
+    """Test that :class:`MutableCacheableProperty` correctly caches a property value and then
+    clears it when a property is updated."""
+    @MutableCacheableProperty()
+    class TestClass:
+        def __init__(self):
+            self.a = 3
+            self.b = 4
+            self.calls = 0
+
+        @property
+        @MutableCacheableProperty.cached_property()
+        def checksum(self):
+            self.calls += 1
+            return self.a * 1 + self.b * 2
+
+    obj = TestClass()
+    assert obj.a == 3
+    assert hasattr(obj, '_cached_properties')
+    assert hasattr(TestClass, '_cached_properties')
+    assert obj._cached_properties == set()
+    assert TestClass._cached_properties == set()
+
+    assert obj.checksum == 11
+    assert obj.calls == 1
+    assert 'checksum' in obj._cached_properties
+    assert len(obj._cached_properties) == 1
+
+    # test that another call doesn't change things
+    assert obj.checksum == 11
+    assert obj.calls == 1
+    assert 'checksum' in obj._cached_properties
+    assert len(obj._cached_properties) == 1
+
+    # try changing the value
+    obj.b = 2
+    # check cache is cleared
+    assert obj._checksum_cache is None
+    # check value recalculates
+    assert obj.checksum == 7
+    assert obj.calls == 2
+
+    # Another call should use the cached value
+    assert obj.checksum == 7
+    assert obj.calls == 2
+
+
+def test_cache_dependency():
+    """Test that :class:`MutableCacheableProperty` correctly caches a property value and then
+    clears it when a another object on which the property depends is updated."""
+    @MutableCacheableProperty()
+    class ChildClass:
+        def __init__(self):
+            self.c = 1
+            self.d = 4
+
+    @MutableCacheableProperty()
+    class TestClass(Base):
+        a: int = Property()
+        child: ChildClass = Property()
+
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            self.b = 4
+            self.calls = 0
+
+        @property
+        @MutableCacheableProperty.cached_property(depends_on=[child])
+        def checksum(self):
+            self.calls += 1
+            return self.a * 1 + self.b * 2 + self.child.c * 3 + self.child.d * 4
+
+    obj = TestClass(a=3, child=ChildClass())
+    assert obj.a == 3
+    child = obj.child
+    assert child.d == 4
+
+    assert obj.checksum == 30
+    assert obj.calls == 1
+
+    # test that another call doesn't change things
+    assert obj.checksum == 30
+    assert obj.calls == 1
+
+    assert child._TestClass_checksum_cache is True
+    # try changing the value
+    child.d = 2
+    # check cache is cleared
+    assert child._TestClass_checksum_cache is None
+    # check value recalculates
+    assert obj.checksum == 22
+    assert obj.calls == 2
+
+    # Another call should use the cached value
+    assert obj.checksum == 22
+    assert obj.calls == 2
+
+
+def test_cache_dependency_python_property():
+    """Test that :class:`MutableCacheableProperty` correctly caches a property value and then
+    clears it when a another object on which the property depends is updated if the two
+    objects are linked as Python properties rather than StoneSoup properties."""
+    @MutableCacheableProperty()
+    class ChildClass:
+        def __init__(self):
+            self.c = 1
+            self.d = 4
+
+    @MutableCacheableProperty()
+    class TestClass(Base):
+        a: int = Property()
+        child: ChildClass = Property()
+
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            self.b = 4
+            self.calls = 0
+
+        @property
+        def child_prop(self):
+            return self.child
+
+        @property
+        @MutableCacheableProperty.cached_property(depends_on=[child_prop])
+        def checksum(self):
+            self.calls += 1
+            return self.a * 1 + self.b * 2 + self.child.c * 3 + self.child.d * 4
+
+    obj = TestClass(a=3, child=ChildClass())
+    assert obj.a == 3
+    child = obj.child
+    assert child.d == 4
+
+    assert obj.checksum == 30
+    assert obj.calls == 1
+
+    # test that another call doesn't change things
+    assert obj.checksum == 30
+    assert obj.calls == 1
+
+    assert child._TestClass_checksum_cache is True
+    # try changing the value
+    child.d = 2
+    # check cache is cleared
+    assert child._TestClass_checksum_cache is None
+    # check value recalculates
+    assert obj.checksum == 22
+    assert obj.calls == 2
+
+    # Another call should use the cached value
+    assert obj.checksum == 22
+    assert obj.calls == 2