(
        self,
        tokenized_docs: Iterable[Iterable[str]],
        grps: Iterable[str],
        fixed_vocab_terms: bool,
        fixed_vocab_grps: bool,
    )
return grp_term_matrix, vocabulary_terms, vocabulary_grps
Count terms and build up a vocabulary based on the terms found in the
        ``tokenized_docs`` and the groups found in ``grps``.

        Args:
            tokenized_docs
            grps
            fixed_vocab_terms
            fixed_vocab_grps

        Returns:
            Group-term matrix, terms vocabulary, groups vocabulary.
Count terms and build up a vocabulary based on the terms found in the
        ``tokenized_docs`` and the groups found in ``grps``.
Count terms and build up a vocabulary based on the terms found in the tokenized_docs and the groups found in grps .
def _count_terms(
        self,
        tokenized_docs: Iterable[Iterable[str]],
        grps: Iterable[str],
        fixed_vocab_terms: bool,
        fixed_vocab_grps: bool,
    ) -> Tuple[sp.csr_matrix, Dict[str, int], Dict[str, int]]:
        """
        Count terms and build up a vocabulary based on the terms found in the
        ``tokenized_docs`` and the groups found in ``grps``.

        Args:
            tokenized_docs
            grps
            fixed_vocab_terms
            fixed_vocab_grps

        Returns:
            Group-term matrix, terms vocabulary, groups vocabulary.
        """
        # TODO: can we adapt the optimization from `Vectorizer._count_terms()` here?
        if fixed_vocab_terms is False:
            # add a new value when a new term is seen
            vocabulary_terms: Union[Dict, DefaultDict] = collections.defaultdict()
            vocabulary_terms.default_factory = vocabulary_terms.__len__
        else:
            vocabulary_terms = self.vocabulary_terms

        if fixed_vocab_grps is False:
            # add a new value when a new group is seen
            vocabulary_grps = collections.defaultdict()
            vocabulary_grps.default_factory = vocabulary_grps.__len__
        else:
            vocabulary_grps = self.vocabulary_grps

        data = array(str("i"))
        cols = array(str("i"))
        rows = array(str("i"))
        for grp, terms in zip(grps, tokenized_docs):

            try:
                grp_idx = vocabulary_grps[grp]
            except KeyError:
                # ignore out-of-vocabulary groups when fixed_grps=True
                continue

            term_counter = collections.defaultdict(int)
            for term in terms:
                try:
                    term_idx = vocabulary_terms[term]
                    term_counter[term_idx] += 1
                except KeyError:
                    # ignore out-of-vocabulary terms when fixed_terms=True
                    continue

            data.extend(term_counter.values())
            cols.extend(term_counter.keys())
            rows.extend(grp_idx for _ in range(len(term_counter)))

        # do we still want defaultdict behaviour?
        if fixed_vocab_terms is False:
            vocabulary_terms = dict(vocabulary_terms)
        if fixed_vocab_grps is False:
            vocabulary_grps = dict(vocabulary_grps)

        data = np.frombuffer(data, dtype=np.intc)
        rows = np.frombuffer(rows, dtype=np.intc)
        cols = np.frombuffer(cols, dtype=np.intc)

        grp_term_matrix = sp.csr_matrix(
            (data, (rows, cols)),
            shape=(len(vocabulary_grps), len(vocabulary_terms)),
            dtype=np.int32,
        )
        grp_term_matrix.sort_indices()

        return grp_term_matrix, vocabulary_terms, vocabulary_grps
def _count_terms ( self , tokenized_docs : Iterable [ Iterable [ str ] ] , grps : Iterable [ str ] , fixed_vocab_terms : bool , fixed_vocab_grps : bool , ) -> Tuple [ sp . csr_matrix , Dict [ str , int ] , Dict [ str , int ] ] : # TODO: can we adapt the optimization from `Vectorizer._count_terms()` here? if fixed_vocab_terms is False : # add a new value when a new term is seen vocabulary_terms : Union [ Dict , DefaultDict ] = collections . defaultdict ( ) vocabulary_terms . default_factory = vocabulary_terms . __len__ else : vocabulary_terms = self . vocabulary_terms if fixed_vocab_grps is False : # add a new value when a new group is seen vocabulary_grps = collections . defaultdict ( ) vocabulary_grps . default_factory = vocabulary_grps . __len__ else : vocabulary_grps = self . vocabulary_grps data = array ( str ( "i" ) ) cols = array ( str ( "i" ) ) rows = array ( str ( "i" ) ) for grp , terms in zip ( grps , tokenized_docs ) : try : grp_idx = vocabulary_grps [ grp ] except KeyError : # ignore out-of-vocabulary groups when fixed_grps=True continue term_counter = collections . defaultdict ( int ) for term in terms : try : term_idx = vocabulary_terms [ term ] term_counter [ term_idx ] += 1 except KeyError : # ignore out-of-vocabulary terms when fixed_terms=True continue data . extend ( term_counter . values ( ) ) cols . extend ( term_counter . keys ( ) ) rows . extend ( grp_idx for _ in range ( len ( term_counter ) ) ) # do we still want defaultdict behaviour? if fixed_vocab_terms is False : vocabulary_terms = dict ( vocabulary_terms ) if fixed_vocab_grps is False : vocabulary_grps = dict ( vocabulary_grps ) data = np . frombuffer ( data , dtype = np . intc ) rows = np . frombuffer ( rows , dtype = np . intc ) cols = np . frombuffer ( cols , dtype = np . intc ) grp_term_matrix = sp . csr_matrix ( ( data , ( rows , cols ) ) , shape = ( len ( vocabulary_grps ) , len ( vocabulary_terms ) ) , dtype = np . int32 , ) grp_term_matrix . sort_indices ( ) return grp_term_matrix , vocabulary_terms , vocabulary_grps