Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
# Download pre-built CACM index; append a random value to avoid filename clashes.
r = randint(0, 10000000)
self.collection_url = 'https://github.com/castorini/anserini-data/raw/master/CACM/lucene-index.cacm.tar.gz'
self.tarball_name = 'lucene-index.cacm-{}.tar.gz'.format(r)
self.index_dir = 'index{}/'.format(r)
_, _ = urlretrieve(self.collection_url, self.tarball_name)
tarball = tarfile.open(self.tarball_name)
tarball.extractall(self.index_dir)
tarball.close()
self.index_path = os.path.join(self.index_dir, 'lucene-index.cacm')
self.searcher = search.SimpleSearcher(self.index_path)
self.index_reader = index.IndexReader(self.index_path)
def setUp(self):
# Download pre-built CACM index; append a random value to avoid filename clashes.
r = randint(0, 10000000)
self.collection_url = 'https://github.com/castorini/anserini-data/raw/master/CACM/lucene-index.cacm.tar.gz'
self.tarball_name = 'lucene-index.cacm-{}.tar.gz'.format(r)
self.index_dir = 'index{}/'.format(r)
filename, headers = urlretrieve(self.collection_url, self.tarball_name)
tarball = tarfile.open(self.tarball_name)
tarball.extractall(self.index_dir)
tarball.close()
self.searcher = search.SimpleSearcher(f'{self.index_dir}lucene-index.cacm')
def setUp(self):
# Download pre-built CACM index; append a random value to avoid filename clashes.
r = randint(0, 10000000)
self.collection_url = 'https://github.com/castorini/anserini-data/raw/master/CACM/lucene-index.cacm.tar.gz'
self.tarball_name = 'lucene-index.cacm-{}.tar.gz'.format(r)
self.index_dir = 'index{}/'.format(r)
_, _ = urlretrieve(self.collection_url, self.tarball_name)
tarball = tarfile.open(self.tarball_name)
tarball.extractall(self.index_dir)
tarball.close()
self.searcher = search.SimpleSearcher(f'{self.index_dir}lucene-index.cacm')
self.index_utils = index.IndexReader(f'{self.index_dir}lucene-index.cacm')
def test_robust04(self):
topics = search.get_topics('robust04')
self.assertEqual(len(topics), 250)
self.assertTrue(isinstance(next(iter(topics.keys())), int))
def test_query_doc_score_custom_similarity(self):
custom_bm25 = search.LuceneSimilarities.bm25(0.8, 0.2)
queries = ['information retrieval', 'databases']
self.searcher.set_bm25(0.8, 0.2)
for query in queries:
hits = self.searcher.search(query)
# We're going to verify that the score of each hit is about the same as the output of
# compute_query_document_score
for i in range(0, len(hits)):
self.assertAlmostEqual(hits[i].score,
self.index_reader.compute_query_document_score(
hits[i].docid, query, similarity=custom_bm25), places=4)
custom_qld = search.LuceneSimilarities.qld(500)
self.searcher.set_qld(500)
for query in queries:
hits = self.searcher.search(query)
# We're going to verify that the score of each hit is about the same as the output of
# compute_query_document_score
for i in range(0, len(hits)):
self.assertAlmostEqual(hits[i].score,
self.index_reader.compute_query_document_score(
hits[i].docid, query, similarity=custom_qld), places=4)
def test_covid_round1(self):
topics = search.get_topics('covid_round1')
self.assertEqual(len(topics), 30)
self.assertEqual('coronavirus origin', topics[1]['query'])
self.assertEqual('coronavirus remdesivir', topics[30]['query'])
self.assertTrue(isinstance(next(iter(topics.keys())), int))
def __init__(self, lucene_index_path: str, min_df: int = 1, verbose: bool = False):
self.min_df: int = min_df
self.verbose: bool = verbose
self.index_reader = index.IndexReader(lucene_index_path)
self.searcher = search.SimpleSearcher(lucene_index_path)
self.num_docs: int = self.searcher.num_docs
# build vocabulary
self.vocabulary_ = set()
for term in self.index_reader.terms():
if term.df > self.min_df:
self.vocabulary_.add(term.term)
# build term to index mapping
self.term_to_index = {}
for i, term in enumerate(self.vocabulary_):
self.term_to_index[term] = i
self.vocabulary_size = len(self.vocabulary_)
if self.verbose:
print(f'Found {self.vocabulary_size} terms with min_df={self.min_df}')