| import copy |
| import pytest |
| import numpy as np |
| import pandas as pd |
| from sklearn.feature_extraction.text import CountVectorizer |
|
|
|
|
| @pytest.mark.parametrize('model', [('kmeans_pca_topic_model'), |
| ('base_topic_model'), |
| ('custom_topic_model'), |
| ('merged_topic_model'), |
| ('reduced_topic_model')]) |
| def test_update_topics(model, documents, request): |
| topic_model = copy.deepcopy(request.getfixturevalue(model)) |
| old_ctfidf = topic_model.c_tf_idf_ |
| old_topics = topic_model.topics_ |
|
|
| topic_model.update_topics(documents, n_gram_range=(1, 3)) |
|
|
| assert old_ctfidf.shape[1] < topic_model.c_tf_idf_.shape[1] |
| assert old_topics == topic_model.topics_ |
|
|
| updated_topics = [topic if topic != 1 else 0 for topic in old_topics] |
| topic_model.update_topics(documents, topics=updated_topics, n_gram_range=(1, 3)) |
|
|
| assert len(set(old_topics)) - 1 == len(set(topic_model.topics_)) |
|
|
| old_topics = topic_model.topics_ |
| updated_topics = [topic if topic != 2 else 0 for topic in old_topics] |
| topic_model.update_topics(documents, topics=updated_topics, n_gram_range=(1, 3)) |
|
|
| assert len(set(old_topics)) - 1 == len(set(topic_model.topics_)) |
|
|
|
|
| @pytest.mark.parametrize('model', [('kmeans_pca_topic_model'), |
| ('base_topic_model'), |
| ('custom_topic_model'), |
| ('merged_topic_model'), |
| ('reduced_topic_model'), |
| ('online_topic_model')]) |
| def test_extract_topics(model, documents, request): |
| topic_model = copy.deepcopy(request.getfixturevalue(model)) |
| nr_topics = 5 |
| documents = pd.DataFrame({"Document": documents, |
| "ID": range(len(documents)), |
| "Topic": np.random.randint(-1, nr_topics-1, len(documents))}) |
| topic_model._update_topic_size(documents) |
| topic_model._extract_topics(documents) |
| freq = topic_model.get_topic_freq() |
|
|
| assert topic_model.c_tf_idf_.shape[0] == 5 |
| assert topic_model.c_tf_idf_.shape[1] > 100 |
| assert isinstance(freq, pd.DataFrame) |
| assert nr_topics == len(freq.Topic.unique()) |
| assert freq.Count.sum() == len(documents) |
| assert len(freq.Topic.unique()) == len(freq) |
|
|
|
|
| @pytest.mark.parametrize('model', [('kmeans_pca_topic_model'), |
| ('base_topic_model'), |
| ('custom_topic_model'), |
| ('merged_topic_model'), |
| ('reduced_topic_model'), |
| ('online_topic_model')]) |
| def test_extract_topics_custom_cv(model, documents, request): |
| topic_model = copy.deepcopy(request.getfixturevalue(model)) |
| nr_topics = 5 |
| documents = pd.DataFrame({"Document": documents, |
| "ID": range(len(documents)), |
| "Topic": np.random.randint(-1, nr_topics-1, len(documents))}) |
|
|
| cv = CountVectorizer(ngram_range=(1, 2)) |
| topic_model.vectorizer_model = cv |
| topic_model._update_topic_size(documents) |
| topic_model._extract_topics(documents) |
| freq = topic_model.get_topic_freq() |
|
|
| assert topic_model.c_tf_idf_.shape[0] == 5 |
| assert topic_model.c_tf_idf_.shape[1] > 100 |
| assert isinstance(freq, pd.DataFrame) |
| assert nr_topics == len(freq.Topic.unique()) |
| assert freq.Count.sum() == len(documents) |
| assert len(freq.Topic.unique()) == len(freq) |
|
|
|
|
| @pytest.mark.parametrize('model', [('kmeans_pca_topic_model'), |
| ('base_topic_model'), |
| ('custom_topic_model'), |
| ('merged_topic_model'), |
| ('reduced_topic_model'), |
| ('online_topic_model')]) |
| @pytest.mark.parametrize("reduced_topics", [2, 4, 10]) |
| def test_topic_reduction(model, reduced_topics, documents, request): |
| topic_model = copy.deepcopy(request.getfixturevalue(model)) |
| old_topics = copy.deepcopy(topic_model.topics_) |
| old_freq = topic_model.get_topic_freq() |
|
|
| topic_model.reduce_topics(documents, nr_topics=reduced_topics) |
|
|
| new_freq = topic_model.get_topic_freq() |
|
|
| if model != "online_topic_model": |
| assert old_freq.Count.sum() == new_freq.Count.sum() |
| assert len(old_freq.Topic.unique()) == len(old_freq) |
| assert len(new_freq.Topic.unique()) == len(new_freq) |
| assert len(topic_model.topics_) == len(old_topics) |
| assert topic_model.topics_ != old_topics |
|
|
|
|
| @pytest.mark.parametrize('model', [('kmeans_pca_topic_model'), |
| ('base_topic_model'), |
| ('custom_topic_model'), |
| ('merged_topic_model'), |
| ('reduced_topic_model'), |
| ('online_topic_model')]) |
| def test_topic_reduction_edge_cases(model, documents, request): |
| topic_model = copy.deepcopy(request.getfixturevalue(model)) |
| topic_model.nr_topics = 100 |
| nr_topics = 5 |
| topics = np.random.randint(-1, nr_topics - 1, len(documents)) |
| old_documents = pd.DataFrame({"Document": documents, |
| "ID": range(len(documents)), |
| "Topic": topics}) |
| topic_model._update_topic_size(old_documents) |
| topic_model._extract_topics(old_documents) |
| old_freq = topic_model.get_topic_freq() |
|
|
| new_documents = topic_model._reduce_topics(old_documents) |
| new_freq = topic_model.get_topic_freq() |
|
|
| assert not set(old_documents.Topic).difference(set(new_documents.Topic)) |
| pd.testing.assert_frame_equal(old_documents, new_documents) |
| pd.testing.assert_frame_equal(old_freq, new_freq) |
|
|
|
|
| @pytest.mark.parametrize('model', [('kmeans_pca_topic_model'), |
| ('base_topic_model'), |
| ('custom_topic_model'), |
| ('merged_topic_model'), |
| ('reduced_topic_model'), |
| ('online_topic_model')]) |
| def test_find_topics(model, request): |
| topic_model = copy.deepcopy(request.getfixturevalue(model)) |
| similar_topics, similarity = topic_model.find_topics("car") |
|
|
| assert np.mean(similarity) > 0.1 |
| assert len(similar_topics) > 0 |
|
|
