Commit b5e10fff authored by Simon Sebastian Humpohl's avatar Simon Sebastian Humpohl
Browse files

Add caching module

parent 18122e99
......@@ -28,3 +28,7 @@ In this module there are some quantities and functions related to quantum inform
## qutil.itertools
This module contains a everything from `itertools`, `more_itertools` and custom functions.
## qutil.caching
Here you find decorators, functions and classes that help you implement caching like `file_cache` and `lru_cache`. This is helpful if you need to call computationally expensive functions with the same arguments repeatedly.
"""Programming tools"""
import shelve
from typing import Callable, Generator, Any
import inspect
import tempfile
import os.path
import itertools
import functools
import numpy as np
from qutil.itertools import separate_iterator
__all__ = ["file_cache", "lru_cache"]
def to_key(obj: Any) -> Generator[str]:
"""Convert to a string representation that is unique except for
- lists and ndarrays are treated the same
:param obj:
:return:
"""
if isinstance(obj, tuple):
element_key_generator = itertools.chain.from_iterable(map(to_key, obj))
yield '('
yield from separate_iterator(element_key_generator, ',')
yield ',)'
elif isinstance(obj, (list, np.ndarray)):
element_key_generator = itertools.chain.from_iterable(map(to_key, obj))
yield '['
yield from separate_iterator(element_key_generator, ',')
yield ']'
elif isinstance(obj, dict):
# we need to sort the representations in the case the objects are not sortable
sorted_items = sorted(''.join(itertools.chain(to_key(key), (':',), to_key(value)))
for key, value in obj.items())
yield '{'
yield from separate_iterator(sorted_items, ',')
yield '}'
elif isinstance(obj, (set, frozenset)):
sorted_elements = sorted(map(''.join, map(to_key, obj)))
yield '{'
yield from separate_iterator(sorted_elements, ',')
yield '}'
elif isinstance(obj, (int, float, complex, str, bytes)) or obj is None:
yield repr(obj)
else:
raise TypeError('not handled: ', type(obj))
class CachingWrapper:
"""This object wraps a callable and caches the results in a dbm database on the file system (by default in the temp
folder). The key is generated via to_key which means that large arguments need a large time to process."""
def __init__(self, func, storage_path=None):
self._func = func
if storage_path is None:
storage_path = os.path.join(tempfile.gettempdir(), inspect.getmodule(func).__name__ + func.__name__)
self.storage_path = storage_path
def __call__(self, *args, **kwargs):
key = ''.join(to_key((args, kwargs)))
with shelve.open(self.storage_path) as db:
if key in db:
result = db[key]
else:
result = self._func(*args, **kwargs)
db[key] = result
return result
def clear(self):
with shelve.open(self.storage_path) as db:
db.clear()
def file_cache(func: Callable) -> Callable:
return CachingWrapper(func)
lru_cache = functools.lru_cache
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment