Source code for ccpn.core.lib.Cache

Cache object and cached / cached.clear decorators

Typical usage:


    @cached('_myFuncCache', maxItems=100)  # cache for 100 items gets created if it does not exist
    def myFunc(obj, arg1, arg2, kwd1=True)
        # some action here
        return result

On cleaning up:


    def cleaningUp(self)
        # action here

or alternatively in your code:


    if hasattr(obj, '_myFuncCache'):
        cache = getattr(obj, '_myFuncCache')


# Licence, Reference and Credits
__copyright__ = "Copyright (C) CCPN project ( 2014 - 2019"
__credits__ = ("Ed Brooksbank, Luca Mureddu, Timothy J Ragan & Geerten W Vuister")
__licence__ = ("CCPN licence. See")
__reference__ = ("Skinner, S.P., Fogh, R.H., Boucher, W., Ragan, T.J., Mureddu, L.G., & Vuister, G.W.",
                 "CcpNmr AnalysisAssign: a flexible platform for integrated NMR analysis",
                 "J.Biomol.Nmr (2016), 66, 111-124,")
# Last code modification
__modifiedBy__ = "$modifiedBy: geertenv $"
__dateModified__ = "$dateModified: 2017-07-07 16:32:36 +0100 (Fri, July 07, 2017) $"
__version__ = "$Revision: 3.0.0 $"
# Created
__author__ = "$Author: geertenv $"
__date__ = "$Date: 2018-05-14 10:28:41 +0000 (Fri, April 07, 2017) $"
# Start of code

import decorator
import inspect
import sys
from collections import deque

DEBUG = False  # global debug

[docs]class Cache(object): """ A cache object; - Retains (item, value) pairs; item must be a hash-able object (e.g. tuple) - Retains either maxItem or unlimited number of objects. - maxItem == 0 disables caching - Clearing cache is responsibility of the instantiating code; e.g. by decorating a cleanup function with cached.clear(attributeName); see above in description for example. """ def __init__(self, maxItems=None, name='', debug=False): """ Initialise the cache :param maxItems: maximum number of items to hold; unlimited for maxItems==0 or maxItems == None :param debug: enable debug for this cache """ self._maxItems = maxItems # maximum number of cached items self._name = name # name of the cache (mainly for debugging) self._debug = DEBUG or debug # debug flag for this cache instance self._items = deque([]) # deque (FIFO stack) of items in the cache self._cacheDict = {} # cached (item, value) dict
[docs] def add(self, item, value): """add item,value to the cache """ if self._maxItems == 0: return # cache is disabled if item in self._cacheDict: # not using hasItem() to save another call return # item is already cached if len(self._items) == self._maxItems: # need to remove one item first self.pop() if self._debug: sys.stderr.write('DEBUG> %s ... Adding "%s"\n' % (self, item)) self._cacheDict[item] = value self._items.append(item)
[docs] def pop(self): """Remove oldest item from stack """ if len(self._items) > 0: itm = self._items.popleft() if self._debug: sys.stderr.write('DEBUG> %s ... removing "%s"\n' % (self, itm)) del (self._cacheDict[itm])
[docs] def resize(self, maxItems): """Resize the cache to contain maxItems """ while len(self._items) > max(0, maxItems): self.pop() self._maxItems = maxItems
[docs] def get(self, item): """Get item from cache; return None if not present """ result = self._cacheDict.get(item) if self._debug and result is not None: sys.stderr.write('DEBUG> %s ... Got cached item "%s"\n' % (self, item)) return result
[docs] def hasItem(self, item): """Return True of item is in cache """ return item in self._cacheDict
[docs] def clear(self): """Clear all items from the cache """ if self._debug: sys.stderr.write('DEBUG> %s ... clearing\n' % self) self._cacheDict = {} self._items = deque([])
def __str__(self): return '<Cache %s; items:(%d,max:%d)>' % (self._name, len(self._items), self._maxItems)
[docs]def cached(attributeName, maxItems=0, debug=False, doSignatureExpansion=True): """ A decorator for initiating cached function call Works on functions that pass an object as the first argument; e.g. self attributeName defines the cache object doSignatureExpansion: flag to do a full signature expansion of the arguments if False: use (repr(args), repr(kwds)) as hash cached.clear (defined below) is a decorator to clear the cache """ @decorator.decorator def decoratedFunc(*args, **kwds): # def myFunc(obj, *args, **kwds): # to avoid potential conflicts with potential 'func' named keywords func = args[0] args = args[1:] obj = args[0] if doSignatureExpansion: # create an item hash from *args and **kwds, skipping the obj argument ba = inspect.signature(func).bind(*args, **kwds) ba.apply_defaults() allArgs = ba.arguments # ordered dict of (argument,value) pairs; first corresponds to object argumentNames = [k for k in allArgs.keys()][1:] # skip the first one which is the object # sort to maintain a consistent tuple of tuples item to cache argumentNames.sort() item = tuple([(k, allArgs[k]) for k in argumentNames]) # convert to a string item = repr(item) else: item = (repr(args), repr(kwds)) if not hasattr(obj, attributeName): name = '%s.%s' % (obj, attributeName) setattr(obj, attributeName, Cache(maxItems=maxItems, name=name, debug=debug)) cache = getattr(obj, attributeName) if not isinstance(cache, Cache): raise RuntimeError('%s, %s is not a Cache object' % (obj, attributeName)) # Check the cache if item exists result = cache.get(item) if result is None: # execute the function result = func(*args, **kwds) cache.add(item, result) return result return decoratedFunc
def _clear(attributeName): """ cached.clear decorator; clear cache of object if it existed """ @decorator.decorator def decoratedFunc(*args, **kwds): # def myFunc(obj, *args, **kwds): # to avoid potential conflicts with potential 'func' named keywords func = args[0] args = args[1:] obj = args[0] # check attributeName for a cache instance, and if present, clear it if hasattr(obj, attributeName): cache = getattr(obj, attributeName) if isinstance(cache, Cache): cache.clear() result = func(*args, **kwds) return result return decoratedFunc cached.clear = _clear if __name__ == "__main__": class myclass(object): CACHE = 'cache' @cached(CACHE, maxItems=2, debug=True) def add(self, values, test=True): result = str(tuple([v.upper() for v in values])) + '\n' return result @cached.clear('cache') def doClear(self): sys.stderr.write('>>> clearing\n') def __str__(self): return '<myclass>' a = myclass() sys.stderr.write(a.add('aap noot mies'.split())) sys.stderr.write(a.add('aap noot mies'.split())) sys.stderr.write(a.add('kees hallo'.split())) sys.stderr.write(a.add('dag week'.split())) v = 'fijn zo'.split() sys.stderr.write(a.add(v)) sys.stderr.write(a.add(v, False)) a.doClear() sys.stderr.write(a.add(v, False))