<commit_before>import os
import numpy as np
import pandas as pd
import logging

from anacode.api import writers
from anacode.api.writers import CSV_FILES

import random
import matplotlib.pyplot as plt
from wordcloud import WordCloud, STOPWORDS


class ApiCallDataset:
    """Base class for specific call data sets."""
    pass


class NoRelevantData(Exception):
    """Thrown when :class:`anacode.agg.aggregations.ApiCallDataset` does not
    have data needed to finish aggregation.
    """
    pass


def generate_color_func(colormap_name):
    def color_func(word, font_size, position, orientation, random_state=None,
                   **kwargs):
        color = plt.get_cmap(colormap_name)(random.random())
        color = map(lambda val: int(round(val * 255)), color)
        return tuple(color)
    return color_func


class ConceptsDataset(ApiCallDataset):
    """Concept data sets container that provides easy aggregation and
    plotting capabilities.

    """
    def __init__(self, concepts: pd.DataFrame, expressions: pd.DataFrame):
        """Initialize instance by providing two concept relevant data frames.

        :param concepts: List of found concepts with metadata
        :type concepts: pandas.DataFrame
        :param expressions: List of expressions realizing found concepts
        :type expressions: pandas.DataFrame
        """
        self._concepts = concepts
        self._expressions = expressions

    def concept_frequency(self, concept: str):
        """Return count of concept occurrences in this dataset. It's case
        insensitive.

        :param concept: name of concept to count it's occurrences
        :type concept: str
        :return: int -- Number of concept occurences in dataset
        """
        if self._concepts is None:
            raise NoRelevantData('Relevant concept data is not available!')

        concept = concept.lower()
        con = self._concepts
        return con[con.concept.str.lower() == concept].freq.sum()

    def most_common_concepts(self, n=15, concept_type='') -> pd.Series:
        """Counts concepts and returns n most occurring ones sorted by their
        count descending. Counted concepts can be filtered by their type.

        :param n: Maximum number of most common concepts to return
        :type n: int
        :param concept_type: Limit concept counts only to concepts whose type
         starts with this string
        :type concept_type: str
        :return: pandas.Series -- Concept names as index and their counts as
         values sorted descending
        """
        if self._concepts is None:
            raise NoRelevantData('Relevant concept data is not available!')

        con = self._concepts
        con = con[con.concept_type.str.startswith(concept_type)]
        con_counts = con.groupby('concept').agg({'freq': 'sum'}).freq
        return con_counts.rename('Count').sort_values(ascending=False)[:n]

    def least_common_concepts(self, n=15, concept_type='') -> pd.Series:
        """Counts concepts and returns n least occurring ones sorted by their
        count ascending. Counted concepts can be filtered by their type.

        :param n: Maximum number of least common concepts to return
        :type n: int
        :param concept_type: Limit concept counts only to concepts whose type
         starts with this string
        :type concept_type: str
        :return: pandas.Series -- Concept names as index and their counts as
         values sorted ascending
        """
        if self._concepts is None:
            raise NoRelevantData('Relevant concept data is not available!')

        con = self._concepts
        con = con[con.concept_type.str.startswith(concept_type)]
        con_counts = con.groupby('concept').agg({'freq': 'sum'}).freq
        return con_counts.rename('Count').sort_values()[:n]

    def co_occurring_concepts(self, concept: str, n=15,
                              concept_type='') -> pd.Series:
        """Find n concepts co-occurring frequently in texts of this dataset with
        given concept, sorted descending. Co-occurring concepts can be
        filtered by their type.

        :param concept: Concept to inspect for co-occurring concepts
        :type concept: str
        :param n: Maximum count of returned concepts
        :type n: int
        :param concept_type: Limit co-occurring concept counts only to this type
         of concepts.
        :type concept_type: str
        :return: pandas.Series -- Co-occurring concept names as index and their
         counts as values sorted descending
        """
        if self._concepts is None:
            raise NoRelevantData('Relevant concept data is not available!')

        con = self._concepts

        identity_filter = con.concept.str.lower() == concept.lower()
        relevant_texts = con[identity_filter][['doc_id', 'text_order']]
        relevant_texts = relevant_texts.set_index(['doc_id', 'text_order'])

        type_filter = con.concept_type.str.startswith(concept_type)
        con = con[type_filter & (identity_filter == False)]
        con = relevant_texts.join(con.set_index(['doc_id', 'text_order']))

        con_counts = con.groupby('concept').agg({'freq': 'sum'}).freq
        con_counts = con_counts.rename('Count').sort_values(ascending=False)
        return con_counts[:n].astype(int)

    def word_cloud(self, path, size=(600, 350), background='white',
                   colormap_name='Accent', max_concepts=200, stopwords=None):
        """Saves word cloud image to *path*. If *path* is not returns image as
        np.ndarray. On way to view np.ndarray resulting image is to use
        matplotlib's imshow method.

        :param path: Save plot to this file. Set to None if you want IPython to
         show this plot instead
        :type path: str
        :param size: Size of plot in pixels
        :type size: tuple - pair - of ints
        :param background: Name of background color
        :type background: str
        :param colormap_name: Name of matplotlib colormap that will be used to
         sample random colors for concepts in plot
        :type colormap_name: str
        :param max_concepts: Maximum number of concepts that will be plotted
        :type max_concepts: int
        :param stopwords: Optionally set stopwords to use for the plot
        :type stopwords: set
        """
        if self._concepts is None:
            raise NoRelevantData('Relevant concept data is not available!')

        data = self._concepts.groupby('concept')['freq'].sum()
        data = data.sort_values().tail(max_concepts).reset_index()
        frequencies = [tuple(row.tolist()) for _, row in data.iterrows()]

        if stopwords is None:
            stopwords = STOPWORDS
        word_cloud = WordCloud(
            width=size[0], height=size[1], stopwords=stopwords,
            background_color=background, prefer_horizontal=0.8,
            color_func=generate_color_func(colormap_name),
        ).fit_words(frequencies)

        if path is not None:
            word_cloud.to_file(path)
        else:
            return np.asarray(word_cloud.to_image())


class CategoriesDataset(ApiCallDataset):
    """Categories data set container that will provides easy aggregation
    capabilities.

    """
    def __init__(self, categories):
        """Initialize instance by providing categories data set.

        :param categories: List of document topic probabilities
        :type categories: pandas.DateFrame
        """
        self._categories = categories
<commit_msg>Added better font support to word cloud generation<commit_after>import os
import numpy as np
import pandas as pd
import logging

from anacode.api import writers
from anacode.api.writers import CSV_FILES

import random
import matplotlib.pyplot as plt
import matplotlib.font_manager
from wordcloud import WordCloud, STOPWORDS


class ApiCallDataset:
    """Base class for specific call data sets."""
    pass


class NoRelevantData(Exception):
    """Thrown when :class:`anacode.agg.aggregations.ApiCallDataset` does not
    have data needed to finish aggregation.
    """
    pass


def generate_color_func(colormap_name):
    def color_func(word, font_size, position, orientation, random_state=None,
                   **kwargs):
        color = plt.get_cmap(colormap_name)(random.random())
        color = map(lambda val: int(round(val * 255)), color)
        return tuple(color)
    return color_func


class ConceptsDataset(ApiCallDataset):
    """Concept data sets container that provides easy aggregation and
    plotting capabilities.

    """
    def __init__(self, concepts: pd.DataFrame, expressions: pd.DataFrame):
        """Initialize instance by providing two concept relevant data frames.

        :param concepts: List of found concepts with metadata
        :type concepts: pandas.DataFrame
        :param expressions: List of expressions realizing found concepts
        :type expressions: pandas.DataFrame
        """
        self._concepts = concepts
        self._expressions = expressions

    def concept_frequency(self, concept: str):
        """Return count of concept occurrences in this dataset. It's case
        insensitive.

        :param concept: name of concept to count it's occurrences
        :type concept: str
        :return: int -- Number of concept occurences in dataset
        """
        if self._concepts is None:
            raise NoRelevantData('Relevant concept data is not available!')

        concept = concept.lower()
        con = self._concepts
        return con[con.concept.str.lower() == concept].freq.sum()

    def most_common_concepts(self, n=15, concept_type='') -> pd.Series:
        """Counts concepts and returns n most occurring ones sorted by their
        count descending. Counted concepts can be filtered by their type.

        :param n: Maximum number of most common concepts to return
        :type n: int
        :param concept_type: Limit concept counts only to concepts whose type
         starts with this string
        :type concept_type: str
        :return: pandas.Series -- Concept names as index and their counts as
         values sorted descending
        """
        if self._concepts is None:
            raise NoRelevantData('Relevant concept data is not available!')

        con = self._concepts
        con = con[con.concept_type.str.startswith(concept_type)]
        con_counts = con.groupby('concept').agg({'freq': 'sum'}).freq
        return con_counts.rename('Count').sort_values(ascending=False)[:n]

    def least_common_concepts(self, n=15, concept_type='') -> pd.Series:
        """Counts concepts and returns n least occurring ones sorted by their
        count ascending. Counted concepts can be filtered by their type.

        :param n: Maximum number of least common concepts to return
        :type n: int
        :param concept_type: Limit concept counts only to concepts whose type
         starts with this string
        :type concept_type: str
        :return: pandas.Series -- Concept names as index and their counts as
         values sorted ascending
        """
        if self._concepts is None:
            raise NoRelevantData('Relevant concept data is not available!')

        con = self._concepts
        con = con[con.concept_type.str.startswith(concept_type)]
        con_counts = con.groupby('concept').agg({'freq': 'sum'}).freq
        return con_counts.rename('Count').sort_values()[:n]

    def co_occurring_concepts(self, concept: str, n=15,
                              concept_type='') -> pd.Series:
        """Find n concepts co-occurring frequently in texts of this dataset with
        given concept, sorted descending. Co-occurring concepts can be
        filtered by their type.

        :param concept: Concept to inspect for co-occurring concepts
        :type concept: str
        :param n: Maximum count of returned concepts
        :type n: int
        :param concept_type: Limit co-occurring concept counts only to this type
         of concepts.
        :type concept_type: str
        :return: pandas.Series -- Co-occurring concept names as index and their
         counts as values sorted descending
        """
        if self._concepts is None:
            raise NoRelevantData('Relevant concept data is not available!')

        con = self._concepts

        identity_filter = con.concept.str.lower() == concept.lower()
        relevant_texts = con[identity_filter][['doc_id', 'text_order']]
        relevant_texts = relevant_texts.set_index(['doc_id', 'text_order'])

        type_filter = con.concept_type.str.startswith(concept_type)
        con = con[type_filter & (identity_filter == False)]
        con = relevant_texts.join(con.set_index(['doc_id', 'text_order']))

        con_counts = con.groupby('concept').agg({'freq': 'sum'}).freq
        con_counts = con_counts.rename('Count').sort_values(ascending=False)
        return con_counts[:n].astype(int)

    def word_cloud(self, path, size=(600, 350), background='white',
                   colormap_name='Accent', max_concepts=200, stopwords=None,
                   font=None):
        """Saves word cloud image to *path*. If *path* is not returns image as
        np.ndarray. On way to view np.ndarray resulting image is to use
        matplotlib's imshow method.

        :param path: Save plot to this file. Set to None if you want IPython to
         show this plot instead
        :type path: str
        :param size: Size of plot in pixels
        :type size: tuple - pair - of ints
        :param background: Name of background color
        :type background: str
        :param colormap_name: Name of matplotlib colormap that will be used to
         sample random colors for concepts in plot
        :type colormap_name: str
        :param max_concepts: Maximum number of concepts that will be plotted
        :type max_concepts: int
        :param stopwords: Optionally set stopwords to use for the plot
        :type stopwords: set
        :param font: Path to font that will be used
        :type font: str
        """
        if self._concepts is None:
            raise NoRelevantData('Relevant concept data is not available!')

        data = self._concepts.groupby('concept')['freq'].sum()
        data = data.sort_values().tail(max_concepts).reset_index()
        frequencies = [tuple(row.tolist()) for _, row in data.iterrows()]

        if font is None:
            font = matplotlib.font_manager.findfont('')
        elif not os.path.isfile(font):
            font = matplotlib.font_manager.findfont(font)

        if stopwords is None:
            stopwords = STOPWORDS
        word_cloud = WordCloud(
            width=size[0], height=size[1], stopwords=stopwords, font_path=font,
            background_color=background, prefer_horizontal=0.8,
            color_func=generate_color_func(colormap_name),
        ).fit_words(frequencies)

        if path is not None:
            word_cloud.to_file(path)
        else:
            return np.asarray(word_cloud.to_image())


class CategoriesDataset(ApiCallDataset):
    """Categories data set container that will provides easy aggregation
    capabilities.

    """
    def __init__(self, categories):
        """Initialize instance by providing categories data set.

        :param categories: List of document topic probabilities
        :type categories: pandas.DateFrame
        """
        self._categories = categories

{'fraction_non_alphanumeric': 0.08429987688718979, 'fraction_numerical': 0.002851033499643621, 'mean_word_length': 2.6695197337137424, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2855001', 'n_tokens_mistral': 4274, 'n_tokens_neox': 4049, 'n_words': 1404}