Toggle Light / Dark / Auto color theme
Toggle table of contents sidebar
Source code for immuneML.caching.CacheHandler import hashlib
import logging
import os
import pickle
from pathlib import Path
import dill
from immuneML.caching.CacheObjectType import CacheObjectType
from immuneML.environment.EnvironmentSettings import EnvironmentSettings
from immuneML.util.PathBuilder import PathBuilder
[docs]
class CacheHandler :
[docs]
@staticmethod
def get_file_path ( cache_type = None ):
file_path = EnvironmentSettings . get_cache_path ( cache_type ) / "files"
PathBuilder . build ( file_path )
return file_path
[docs]
@staticmethod
def get ( params : tuple , object_type , cache_type = None ):
h = CacheHandler . _hash ( params )
return CacheHandler . get_by_key ( h , object_type , cache_type )
[docs]
@staticmethod
def get_by_key ( cache_key : str , object_type , cache_type = None ):
filename = CacheHandler . _build_filename ( cache_key , object_type , cache_type )
obj = None
if filename . is_file ():
with filename . open ( "rb" ) as file :
obj = dill . load ( file )
return obj
@staticmethod
def _build_filename ( cache_key : str , object_type : CacheObjectType , cache_type = None ) -> Path :
path = EnvironmentSettings . get_cache_path ( cache_type ) / object_type . name . lower ()
PathBuilder . build ( path )
return path / f " { cache_key } .pickle"
[docs]
@staticmethod
def exists ( params : tuple , object_type : CacheObjectType = CacheObjectType . OTHER , cache_type = None ):
return CacheHandler . get_by_key ( CacheHandler . generate_cache_key ( params ), object_type , cache_type ) is not None
[docs]
@staticmethod
def add ( params : tuple , caching_object , object_type : CacheObjectType = CacheObjectType . OTHER , cache_type = None ):
PathBuilder . build ( EnvironmentSettings . get_cache_path ( cache_type ))
h = CacheHandler . generate_cache_key ( params )
filename = CacheHandler . _build_filename ( cache_key = h , object_type = object_type , cache_type = cache_type )
with filename . open ( "wb" ) as file :
dill . dump ( caching_object , file , protocol = pickle . HIGHEST_PROTOCOL )
[docs]
@staticmethod
def add_by_key ( cache_key : str , caching_object , object_type : CacheObjectType = CacheObjectType . OTHER , cache_type = None ):
PathBuilder . build ( EnvironmentSettings . get_cache_path ( cache_type ))
filename = CacheHandler . _build_filename ( cache_key = cache_key , object_type = object_type , cache_type = cache_type )
try :
with filename . open ( "wb" ) as file :
dill . dump ( caching_object , file , protocol = pickle . HIGHEST_PROTOCOL )
except AttributeError :
os . remove ( filename )
logging . warning ( f "CacheHandler: could not cache object of class { type ( caching_object ) . __name__ } with key { cache_key } . "
f "Object: { caching_object } \n "
f "Next time this object is needed, it will be recomputed which will take more time but should not influence results." )
[docs]
@staticmethod
def generate_cache_key ( params : tuple ):
return hashlib . sha256 ( str ( params ) . encode ( 'utf-8' )) . hexdigest ()
[docs]
@staticmethod
def memo ( cache_key : str , fn , object_type : CacheObjectType = CacheObjectType . OTHER , cache_type = None ):
result = CacheHandler . get_by_key ( cache_key , object_type , cache_type )
if result is None :
result = fn ()
CacheHandler . add_by_key ( cache_key , result , object_type , cache_type )
return result
[docs]
@staticmethod
def memo_by_params ( params : tuple , fn , object_type : CacheObjectType = CacheObjectType . OTHER , cache_type = None ):
cache_key = CacheHandler . generate_cache_key ( params )
return CacheHandler . memo ( cache_key , fn , object_type , cache_type )
@staticmethod
def _hash ( params : tuple ) -> str :
return hashlib . sha256 ( str ( params ) . encode ( 'utf-8' )) . hexdigest ()