Source code for pygimli.utils.cache

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Caching manager with function decorator.

Input supports python base types and all pg.core objects with .hash() method.
Output supports DataContainerERT, ...


  *   Output types:
        numpy.ndarray, pg.Mesh. pg.Vector, pg.Matrix

To use just add the decorator.

def myLongRunningStuff(*args, **kwargs):
    return results
import sys
import os
import inspect
import hashlib
import json
import time

import numpy as np

import pygimli as pg

__NO_CACHE__ = False

[docs] def noCache(c): global __NO_CACHE__ __NO_CACHE__ = c
[docs] def strHash(string): return int(hashlib.sha224(string.encode()).hexdigest()[:16], 16)
def valHash(a): if isinstance(a, str): return strHash(a) elif isinstance(a, int): return a elif isinstance(a, list): hsh = 0 for item in a: hsh = hsh ^ valHash(item) return hsh elif isinstance(a, np.ndarray): if a.ndim == 1: return hash(pg.Vector(a)) elif a.ndim == 2: # convert to RVector to use memcopy return hash(pg.Vector(a.reshape((1,a.shape[0]*a.shape[1]))[0])) else: print(a) pg.error('no hash for numpy array') return hash(a) class Cache(object): def __init__(self, hashValue): self._value = None self._hash = hashValue self._name = CacheManager().cachingPath(str(self._hash)) self._info = None self.restore() @property def info(self): if self._info is None: self._info = {'type': '', 'file': '', 'date': 0, 'dur': 0.0, 'restored': 0, 'codeinfo': '', 'version': '', 'args': '', 'kwargs': {}, } return self._info @info.setter def info(self, i): self._info = i @property def value(self): return self._value @value.setter def value(self, v):['type'] = str(type(v).__name__) # if len(['type']) != 1: # pg.error('only single return caches supported for now.') # return['file'] = self._name self.updateCacheInfo() if['type'] == 'Mesh':'Save Mesh binary v2') v.saveBinaryV2(self._name) elif['type'] == 'RVector':'Save RVector binary'), format=pg.core.Binary) elif['type'] == 'ndarray':'Save ndarray'), v, allow_pickle=True) elif hasattr(v, 'save') and hasattr(v, 'load'): else:, v, allow_pickle=True) # pg.warn('ascii save of type',['type'], 'might by dangerous') # self._value = v'Cache stored:', self._name) def updateCacheInfo(self): with open(self._name + '.json', 'w') as of: json.dump(, of, sort_keys=False, indent=4, separators=(',', ': ')) def restore(self): """Read data from json infos""" if os.path.exists(self._name + '.json'): # Fricking mpl kills locale setting to system default .. this went # horrible wrong for german 'decimal_point': ',' pg.checkAndFixLocaleDecimal_point(verbose=False) try: with open(self._name + '.json') as file: = json.load(file) # if len(['type']) != 1: # pg.error('only single return caches supported for now.') #pg._y( if['type'] == 'DataContainerERT': self._value = pg.DataContainerERT(['file'], removeInvalid=False) # print(self._value) elif['type'] == 'RVector': self._value = pg.Vector() self._value.load(['file'], format=pg.core.Binary) elif['type'] == 'Mesh': pg.tic() self._value = pg.Mesh() self._value.loadBinaryV2(['file'] + '.bms') pg.debug("Restoring cache took:", pg.dur(), "s") elif['type'] == 'ndarray': self._value = np.load(['file'] + '.npy', allow_pickle=True) elif['type'] == 'Cm05Matrix': self._value = pg.matrix.Cm05Matrix(['file']) elif['type'] == 'GeostatisticConstraintsMatrix': self._value = pg.matrix.GeostatisticConstraintsMatrix(['file']) else: self._value = np.load(['file'] + '.npy', allow_pickle=True) if self.value is not None:['restored'] =['restored'] + 1 self.updateCacheInfo()'Cache {3} restored ({1}s x {0}): {2}'.\ format(['restored'], round(['dur'], 1), self._name,['codeinfo'])) else: # default try numpy pg.warn('Could not restore cache of type {0}.'.format(['type'])) pg.debug("Restoring cache took:", pg.dur(), "s") except Exception as e: import traceback traceback.print_exc(file=sys.stdout) print( pg.error('Cache restoring failed.') #@pg.singleton class CacheManager(object): __instance = None __has_init = False def __new__(cls): if cls.__instance is None: cls.__instance = object.__new__(cls) return cls.__instance def __init__(self): if not self.__has_init: self._caches = {} self.__has_init = True @staticmethod def instance(cls): return cls.__instance__ def cachingPath(self, fName): """Create a path name for the cache""" if pg.rc["globalCache"]: path = pg.getCachePath() else: path = ".cache" if not os.path.exists(path): os.mkdir(path) return os.path.join(path, fName) def functInfo(self, funct): """Return unique info string about the called function.""" return funct.__code__.co_filename + ":" + funct.__qualname__ def hash(self, funct, *args, **kwargs): """"Create a hash value""" pg.tic() functInfo = self.functInfo(funct) funcHash = strHash(functInfo) versionHash = strHash(pg.versionStr()) codeHash = strHash(inspect.getsource(funct)) argHash = 0 for i, a in enumerate(args): if pg.isScalar(a): argHash = argHash ^ valHash(str(i) + str(a)) else: argHash = argHash ^ (valHash(i) ^ valHash(a)) for k, v in kwargs.items(): if pg.isScalar(v): argHash = argHash ^ (valHash(k + str(v))) else: argHash = argHash ^ valHash(k) ^ valHash(v) pg.debug("Hashing took:", pg.dur(), "s") return funcHash ^ versionHash ^ codeHash ^ argHash def cache(self, funct, *args, **kwargs): """ Create a unique cache """ hashVal = self.hash(funct, *args, **kwargs) cached = Cache(hashVal)['codeinfo'] = self.functInfo(funct)['version'] = pg.versionStr()['args'] = str(args)['kwargs'] = str(kwargs) return cached
[docs] def cache(funct): """Cache decorator.""" def wrapper(*args, **kwargs): nc = kwargs.pop('skipCache', False) if any(('--noCache' in sys.argv, '-N' in sys.argv, nc is True, __NO_CACHE__)): return funct(*args, **kwargs) cache = CacheManager().cache(funct, *args, **kwargs) if cache.value is not None: return cache.value else: # pg.tic will not work because there is only one global __swatch__ sw = pg.Stopwatch(True) rv = funct(*args, **kwargs)['date'] = time.time()['dur'] = sw.duration() try: cache.value = rv except Exception as e: print(e) pg.warn("Can't cache:", rv) return rv return wrapper