Source code for mlinsights.mlbatch.cache_model

import numpy

_caches = {}


[docs] class MLCache: """ Implements a cache to reduce the number of trainings a grid search has to do. """ def __init__(self, name): """ @param name name of the cache """ self.name = name self.cached = {} self.count_ = {}
[docs] def cache(self, params, value): """ Caches one object. @param params dictionary of parameters @param value value to cache """ key = MLCache.as_key(params) assert key not in self.cached, f"Key {params} already exists" self.cached[key] = value self.count_[key] = 0
[docs] def get(self, params, default=None): """ Retrieves an element from the cache. @param params dictionary of parameters @param default if not found @return value or None if it does not exists """ key = MLCache.as_key(params) res = self.cached.get(key, default) if res != default: self.count_[key] += 1 return res
[docs] def count(self, params): """ Retrieves the number of times an elements was retrieved from the cache. @param params dictionary of parameters @return int """ key = MLCache.as_key(params) return self.count_.get(key, 0)
[docs] @staticmethod def as_key(params): """ Converts a list of parameters into a key. @param params dictionary @return key as a string """ if isinstance(params, str): return params els = [] for k, v in sorted(params.items()): if isinstance(v, (int, float, str)): sv = str(v) elif isinstance(v, tuple): assert all( map(lambda e: isinstance(e, (int, float, str)), v) ), f"Unable to create a key with value {k!r}:{v}" return str(v) elif isinstance(v, numpy.ndarray): # id(v) may have been better but # it does not play well with joblib. sv = hash(v.tobytes()) elif v is None: sv = "" else: raise TypeError(f"Unable to create a key with value '{k}':{v}") els.append((k, sv)) return str(els)
def __len__(self): """ Returns the number of cached items. """ return len(self.cached)
[docs] def items(self): """ Enumerates all cached items. """ for item in self.cached.items(): yield item
[docs] def keys(self): """ Enumerates all cached keys. """ for k in self.cached.keys(): yield k
[docs] @staticmethod def create_cache(name): """ Creates a new cache. @param name name @return created cache """ global _caches assert name not in _caches, f"cache {name!r} already exists." cache = MLCache(name) _caches[name] = cache return cache
[docs] @staticmethod def remove_cache(name): """ Removes a cache with a given name. @param name name """ global _caches del _caches[name]
[docs] @staticmethod def get_cache(name): """ Gets a cache with a given name. @param name name @return created cache """ global _caches return _caches[name]
[docs] @staticmethod def has_cache(name): """ Tells if cache *name* is present. @param name name @return boolean """ global _caches return name in _caches