-
from sklearn.pipeline import Pipeline from sklearn.base import TransformerMixin import numpy as np state = 0 class Cacher(TransformerMixin): def __init__(self, internal=None, cache=None): self.internal = internal self.cache = cache or dict() assert internal def fit(self, data, y=None): """Returns `self` unless something different happens in train and test""" return self def save(self): '''Save to disk...''' return True def transform(self, X, y=None): def cacheado(i): hash = str(i) if hash not in self.cache: self.cache[hash] = self.internal.transform(i) return self.cache[hash] res = np.apply_along_axis(cacheado, 0, X) self.save() return res class Transformer(TransformerMixin): def fit(self, data, y=None): return self def transform(self, X, y=None): print('Original transformer') return X+state # DO IT ONCE ner_pipe = Pipeline([ ('trans', Cacher(internal=Transformer())) ]) ner_pipe.fit(None) for i in range(10): state = i print(ner_pipe.transform(np.array([[0,1], [1, 2]]))) if i % 4 == 0: print('Reentrenado') ner_pipe.fit([[0,1], [1, 2]])
Please register or sign in to comment