path: root/snips_inference_agl/intent_classifier/featurizer.py

from __future__ import division, unicode_literals

import json
from builtins import str, zip
from copy import deepcopy
from pathlib import Path

from future.utils import iteritems

from snips_inference_agl.common.utils import (
    fitted_required, replace_entities_with_placeholders)
from snips_inference_agl.constants import (
    DATA, ENTITY, ENTITY_KIND, NGRAM, TEXT)
from snips_inference_agl.dataset import get_text_from_chunks
from snips_inference_agl.entity_parser.builtin_entity_parser import (
    is_builtin_entity)
from snips_inference_agl.exceptions import (LoadingError)
from snips_inference_agl.languages import get_default_sep
from snips_inference_agl.pipeline.configs import FeaturizerConfig
from snips_inference_agl.pipeline.configs.intent_classifier import (
    CooccurrenceVectorizerConfig, TfidfVectorizerConfig)
from snips_inference_agl.pipeline.processing_unit import ProcessingUnit
from snips_inference_agl.preprocessing import stem, tokenize_light
from snips_inference_agl.resources import get_stop_words, get_word_cluster
from snips_inference_agl.slot_filler.features_utils import get_all_ngrams


@ProcessingUnit.register("featurizer")
class Featurizer(ProcessingUnit):
    """Feature extractor for text classification relying on ngrams tfidf and
    optionally word cooccurrences features"""

    config_type = FeaturizerConfig

    def __init__(self, config=None, **shared):
        super(Featurizer, self).__init__(config, **shared)
        self.language = None
        self.tfidf_vectorizer = None
        self.cooccurrence_vectorizer = None

    @property
    def fitted(self):
        if not self.tfidf_vectorizer or not self.tfidf_vectorizer.vocabulary:
            return False
        return True
    
    def transform(self, utterances):
        import scipy.sparse as sp

        x = self.tfidf_vectorizer.transform(utterances)
        if self.cooccurrence_vectorizer:
            x_cooccurrence = self.cooccurrence_vectorizer.transform(utterances)
            x = sp.hstack((x, x_cooccurrence))
        return x

    @classmethod
    def from_path(cls, path, **shared):
        path = Path(path)

        model_path = path / "featurizer.json"
        if not model_path.exists():
            raise LoadingError("Missing featurizer model file: %s"
                               % model_path.name)
        with model_path.open("r", encoding="utf-8") as f:
            featurizer_dict = json.load(f)

        featurizer_config = featurizer_dict["config"]
        featurizer = cls(featurizer_config, **shared)

        featurizer.language = featurizer_dict["language_code"]

        tfidf_vectorizer = featurizer_dict["tfidf_vectorizer"]
        if tfidf_vectorizer:
            vectorizer_path = path / featurizer_dict["tfidf_vectorizer"]
            tfidf_vectorizer = TfidfVectorizer.from_path(
                vectorizer_path, **shared)
        featurizer.tfidf_vectorizer = tfidf_vectorizer

        cooccurrence_vectorizer = featurizer_dict["cooccurrence_vectorizer"]
        if cooccurrence_vectorizer:
            vectorizer_path = path / featurizer_dict["cooccurrence_vectorizer"]
            cooccurrence_vectorizer = CooccurrenceVectorizer.from_path(
                vectorizer_path, **shared)
        featurizer.cooccurrence_vectorizer = cooccurrence_vectorizer

        return featurizer


@ProcessingUnit.register("tfidf_vectorizer")
class TfidfVectorizer(ProcessingUnit):
    """Wrapper of the scikit-learn TfidfVectorizer"""

    config_type = TfidfVectorizerConfig

    def __init__(self, config=None, **shared):
        super(TfidfVectorizer, self).__init__(config, **shared)
        self._tfidf_vectorizer = None
        self._language = None
        self.builtin_entity_scope = None

    @property
    def fitted(self):
        return self._tfidf_vectorizer is not None and hasattr(
            self._tfidf_vectorizer, "vocabulary_")
    
    @fitted_required
    def transform(self, x):
        """Featurizes the given utterances after enriching them with builtin
        entities matches, custom entities matches and the potential word
        clusters matches

        Args:
            x (list of dict): list of utterances

        Returns:
            :class:`.scipy.sparse.csr_matrix`: A sparse matrix X of shape
            (len(x), len(self.vocabulary)) where X[i, j] contains tfdif of
            the ngram of index j of the vocabulary in the utterance i

        Raises:
            NotTrained: when the vectorizer is not fitted:
        """
        utterances = [self._enrich_utterance(*data)
                      for data in zip(*self._preprocess(x))]
        return self._tfidf_vectorizer.transform(utterances)

    def _preprocess(self, utterances):
        normalized_utterances = deepcopy(utterances)
        for u in normalized_utterances:
            nb_chunks = len(u[DATA])
            for i, chunk in enumerate(u[DATA]):
                chunk[TEXT] = _normalize_stem(
                    chunk[TEXT], self.language, self.resources,
                    self.config.use_stemming)
                if i < nb_chunks - 1:
                    chunk[TEXT] += " "

        # Extract builtin entities on unormalized utterances
        builtin_ents = [
            self.builtin_entity_parser.parse(
                get_text_from_chunks(u[DATA]),
                self.builtin_entity_scope, use_cache=True)
            for u in utterances
        ]
        # Extract builtin entities on normalized utterances
        custom_ents = [
            self.custom_entity_parser.parse(
                get_text_from_chunks(u[DATA]), use_cache=True)
            for u in normalized_utterances
        ]
        if self.config.word_clusters_name:
            # Extract world clusters on unormalized utterances
            original_utterances_text = [get_text_from_chunks(u[DATA])
                                        for u in utterances]
            w_clusters = [
                _get_word_cluster_features(
                    tokenize_light(u.lower(), self.language),
                    self.config.word_clusters_name,
                    self.resources)
                for u in original_utterances_text
            ]
        else:
            w_clusters = [None for _ in normalized_utterances]

        return normalized_utterances, builtin_ents, custom_ents, w_clusters

    def _enrich_utterance(self, utterance, builtin_entities, custom_entities,
                          word_clusters):
        custom_entities_features = [
            _entity_name_to_feature(e[ENTITY_KIND], self.language)
            for e in custom_entities]

        builtin_entities_features = [
            _builtin_entity_to_feature(ent[ENTITY_KIND], self.language)
            for ent in builtin_entities
        ]

        # We remove values of builtin slots from the utterance to avoid
        # learning specific samples such as '42' or 'tomorrow'
        filtered_tokens = [
            chunk[TEXT] for chunk in utterance[DATA]
            if ENTITY not in chunk or not is_builtin_entity(chunk[ENTITY])
        ]

        features = get_default_sep(self.language).join(filtered_tokens)

        if builtin_entities_features:
            features += " " + " ".join(sorted(builtin_entities_features))
        if custom_entities_features:
            features += " " + " ".join(sorted(custom_entities_features))
        if word_clusters:
            features += " " + " ".join(sorted(word_clusters))

        return features

    @property
    def language(self):
        # Create this getter to prevent the language from being set elsewhere
        # than in the fit
        return self._language

    @property
    def vocabulary(self):
        if self._tfidf_vectorizer and hasattr(
                self._tfidf_vectorizer, "vocabulary_"):
            return self._tfidf_vectorizer.vocabulary_
        return None

    @property
    def idf_diag(self):
        if self._tfidf_vectorizer and hasattr(
                self._tfidf_vectorizer, "vocabulary_"):
            return self._tfidf_vectorizer.idf_
        return None

    @classmethod
    # pylint: disable=W0212
    def from_path(cls, path, **shared):
        import numpy as np
        import scipy.sparse as sp
        from sklearn.feature_extraction.text import (
            TfidfTransformer, TfidfVectorizer as SklearnTfidfVectorizer)

        path = Path(path)

        model_path = path / "vectorizer.json"
        if not model_path.exists():
            raise LoadingError("Missing vectorizer model file: %s"
                               % model_path.name)
        with model_path.open("r", encoding="utf-8") as f:
            vectorizer_dict = json.load(f)

        vectorizer = cls(vectorizer_dict["config"], **shared)
        vectorizer._language = vectorizer_dict["language_code"]

        builtin_entity_scope = vectorizer_dict["builtin_entity_scope"]
        if builtin_entity_scope is not None:
            builtin_entity_scope = set(builtin_entity_scope)
        vectorizer.builtin_entity_scope = builtin_entity_scope

        vectorizer_ = vectorizer_dict["vectorizer"]
        if vectorizer_:
            vocab = vectorizer_["vocab"]
            idf_diag_data = vectorizer_["idf_diag"]
            idf_diag_data = np.array(idf_diag_data)

            idf_diag_shape = (len(idf_diag_data), len(idf_diag_data))
            row = list(range(idf_diag_shape[0]))
            col = list(range(idf_diag_shape[0]))
            idf_diag = sp.csr_matrix(
                (idf_diag_data, (row, col)), shape=idf_diag_shape)

            tfidf_transformer = TfidfTransformer()
            tfidf_transformer._idf_diag = idf_diag

            vectorizer_ = SklearnTfidfVectorizer(
                tokenizer=lambda x: tokenize_light(x, vectorizer._language))
            vectorizer_.vocabulary_ = vocab

            vectorizer_._tfidf = tfidf_transformer

        vectorizer._tfidf_vectorizer = vectorizer_
        return vectorizer


@ProcessingUnit.register("cooccurrence_vectorizer")
class CooccurrenceVectorizer(ProcessingUnit):
    """Featurizer that takes utterances and extracts ordered word cooccurrence
     features matrix from them"""

    config_type = CooccurrenceVectorizerConfig

    def __init__(self, config=None, **shared):
        super(CooccurrenceVectorizer, self).__init__(config, **shared)
        self._word_pairs = None
        self._language = None
        self.builtin_entity_scope = None

    @property
    def language(self):
        # Create this getter to prevent the language from being set elsewhere
        # than in the fit
        return self._language

    @property
    def word_pairs(self):
        return self._word_pairs

    @property
    def fitted(self):
        """Whether or not the vectorizer is fitted"""
        return self.word_pairs is not None
    
    @fitted_required
    def transform(self, x):
        """Computes the cooccurrence feature matrix.

        Args:
            x (list of dict): list of utterances

        Returns:
            :class:`.scipy.sparse.csr_matrix`: A sparse matrix X of shape
            (len(x), len(self.word_pairs)) where X[i, j] = 1.0 if
            x[i][0] contains the words cooccurrence (w1, w2) and if
            self.word_pairs[(w1, w2)] = j

        Raises:
            NotTrained: when the vectorizer is not fitted
        """
        import numpy as np
        import scipy.sparse as sp

        preprocessed = self._preprocess(x)
        utterances = [
            self._enrich_utterance(utterance, builtin_ents, custom_ent)
            for utterance, builtin_ents, custom_ent in zip(*preprocessed)]

        x_coo = sp.dok_matrix((len(x), len(self.word_pairs)), dtype=np.int32)
        for i, u in enumerate(utterances):
            for p in self._extract_word_pairs(u):
                if p in self.word_pairs:
                    x_coo[i, self.word_pairs[p]] = 1

        return x_coo.tocsr()

    def _preprocess(self, x):
        # Extract all entities on unnormalized data
        builtin_ents = [
            self.builtin_entity_parser.parse(
                get_text_from_chunks(u[DATA]),
                self.builtin_entity_scope,
                use_cache=True
            ) for u in x
        ]
        custom_ents = [
            self.custom_entity_parser.parse(
                get_text_from_chunks(u[DATA]), use_cache=True)
            for u in x
        ]
        return x, builtin_ents, custom_ents

    def _extract_word_pairs(self, utterance):
        if self.config.filter_stop_words:
            stop_words = get_stop_words(self.resources)
            utterance = [t for t in utterance if t not in stop_words]
        pairs = set()
        for j, w1 in enumerate(utterance):
            max_index = None
            if self.config.window_size is not None:
                max_index = j + self.config.window_size + 1
            for w2 in utterance[j + 1:max_index]:
                key = (w1, w2)
                if not self.config.keep_order:
                    key = tuple(sorted(key))
                pairs.add(key)
        return pairs
    
    def _enrich_utterance(self, x, builtin_ents, custom_ents):
        utterance = get_text_from_chunks(x[DATA])
        all_entities = builtin_ents + custom_ents
        placeholder_fn = self._placeholder_fn
        # Replace entities with placeholders
        enriched_utterance = replace_entities_with_placeholders(
            utterance, all_entities, placeholder_fn)[1]
        # Tokenize
        enriched_utterance = tokenize_light(enriched_utterance, self.language)
        # Remove the unknownword strings if needed
        if self.config.unknown_words_replacement_string:
            enriched_utterance = [
                t for t in enriched_utterance
                if t != self.config.unknown_words_replacement_string
            ]
        return enriched_utterance

    def _extract_word_pairs(self, utterance):
        if self.config.filter_stop_words:
            stop_words = get_stop_words(self.resources)
            utterance = [t for t in utterance if t not in stop_words]
        pairs = set()
        for j, w1 in enumerate(utterance):
            max_index = None
            if self.config.window_size is not None:
                max_index = j + self.config.window_size + 1
            for w2 in utterance[j + 1:max_index]:
                key = (w1, w2)
                if not self.config.keep_order:
                    key = tuple(sorted(key))
                pairs.add(key)
        return pairs

    def _placeholder_fn(self, entity_name):
        return "".join(
            tokenize_light(str(entity_name), str(self.language))).upper()

    @classmethod
    # pylint: disable=protected-access
    def from_path(cls, path, **shared):
        path = Path(path)
        model_path = path / "vectorizer.json"
        if not model_path.exists():
            raise LoadingError("Missing vectorizer model file: %s"
                               % model_path.name)

        with model_path.open(encoding="utf8") as f:
            vectorizer_dict = json.load(f)
        config = vectorizer_dict.pop("config")

        self = cls(config, **shared)
        self._language = vectorizer_dict["language_code"]
        self._word_pairs = None

        builtin_entity_scope = vectorizer_dict["builtin_entity_scope"]
        if builtin_entity_scope is not None:
            builtin_entity_scope = set(builtin_entity_scope)
        self.builtin_entity_scope = builtin_entity_scope

        if vectorizer_dict["word_pairs"]:
            self._word_pairs = {
                tuple(p): int(i)
                for i, p in iteritems(vectorizer_dict["word_pairs"])
            }
        return self

def _entity_name_to_feature(entity_name, language):
    return "entityfeature%s" % "".join(tokenize_light(
        entity_name.lower(), language))


def _builtin_entity_to_feature(builtin_entity_label, language):
    return "builtinentityfeature%s" % "".join(tokenize_light(
        builtin_entity_label.lower(), language))


def _normalize_stem(text, language, resources, use_stemming):
    from snips_nlu_utils import normalize

    if use_stemming:
        return stem(text, language, resources)
    return normalize(text)


def _get_word_cluster_features(query_tokens, clusters_name, resources):
    if not clusters_name:
        return []
    ngrams = get_all_ngrams(query_tokens)
    cluster_features = []
    for ngram in ngrams:
        cluster = get_word_cluster(resources, clusters_name).get(
            ngram[NGRAM].lower(), None)
        if cluster is not None:
            cluster_features.append(cluster)
    return cluster_features