<s> import json <EOL> import numpy as np <EOL> from nearpy . hashes import RandomBinaryProjections <EOL> from nearpy . hashes import PCABinaryProjections <EOL> from nearpy . hashes import RandomBinaryProjectionTree <EOL> from nearpy . filters import NearestFilter , UniqueFilter <EOL> from nearpy . distances import EuclideanDistance <EOL> from nearpy . distances import CosineDistance <EOL> from nearpy . storage import MemoryStorage <EOL> from nearpy . utils . utils import unitvec <EOL> class Engine ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , dim , lshashes = None , <EOL> distance = None , <EOL> fetch_vector_filters = None , <EOL> vector_filters = None , <EOL> storage = None ) : <EOL> """<STR_LIT>""" <EOL> if lshashes is None : lshashes = [ RandomBinaryProjections ( '<STR_LIT:default>' , <NUM_LIT:10> ) ] <EOL> self . lshashes = lshashes <EOL> if distance is None : distance = CosineDistance ( ) <EOL> self . distance = distance <EOL> if vector_filters is None : vector_filters = [ NearestFilter ( <NUM_LIT:10> ) ] <EOL> self . vector_filters = vector_filters <EOL> if fetch_vector_filters is None : fetch_vector_filters = [ UniqueFilter ( ) ] <EOL> self . fetch_vector_filters = fetch_vector_filters <EOL> if storage is None : storage = MemoryStorage ( ) <EOL> self . storage = storage <EOL> for lshash in self . lshashes : <EOL> lshash . reset ( dim ) <EOL> def store_vector ( self , v , data = None ) : <EOL> """<STR_LIT>""" <EOL> nv = unitvec ( v ) <EOL> for lshash in self . lshashes : <EOL> for bucket_key in lshash . hash_vector ( v ) : <EOL> self . storage . store_vector ( lshash . hash_name , bucket_key , <EOL> nv , data ) <EOL> def candidate_count ( self , v ) : <EOL> """<STR_LIT>""" <EOL> candidates = self . _get_candidates ( ) <EOL> return len ( candidates ) <EOL> def neighbours ( self , v ) : <EOL> """<STR_LIT>""" <EOL> candidates = self . _get_candidates ( v ) <EOL> if self . fetch_vector_filters : <EOL> candidates = self . _apply_filter ( self . fetch_vector_filters , candidates ) <EOL> if self . distance : <EOL> candidates = self . _append_distances ( v , self . distance , candidates ) <EOL> if self . vector_filters : <EOL> candidates = self . _apply_filter ( self . vector_filters , candidates ) <EOL> return candidates <EOL> def _get_candidates ( self , v ) : <EOL> """<STR_LIT>""" <EOL> candidates = [ ] <EOL> for lshash in self . lshashes : <EOL> for bucket_key in lshash . hash_vector ( v , querying = True ) : <EOL> bucket_content = self . storage . get_bucket ( lshash . hash_name , <EOL> bucket_key ) <EOL> candidates . extend ( bucket_content ) <EOL> return candidates <EOL> def _apply_filter ( self , filters , candidates ) : <EOL> """<STR_LIT>""" <EOL> if filters : <EOL> filter_input = candidates <EOL> for fetch_vector_filter in filters : <EOL> filter_input = fetch_vector_filter . filter_vectors ( filter_input ) <EOL> return filter_input <EOL> else : <EOL> return candidates <EOL> def _append_distances ( self , v , distance , candidates ) : <EOL> """<STR_LIT>""" <EOL> if distance : <EOL> nv = unitvec ( v ) <EOL> candidates = [ ( x [ <NUM_LIT:0> ] , x [ <NUM_LIT:1> ] , self . distance . distance ( x [ <NUM_LIT:0> ] , nv ) ) for x <EOL> in candidates ] <EOL> return candidates <EOL> def clean_all_buckets ( self ) : <EOL> """<STR_LIT>""" <EOL> self . storage . clean_all_buckets ( ) <EOL> def clean_buckets ( self , hash_name ) : <EOL> """<STR_LIT>""" <EOL> self . storage . clean_buckets ( hash_name ) </s>
