<commit_before>from pandas import DataFrame
from geopandas import GeoDataFrame

from ..io.managers.context_manager import ContextManager
from ..utils.geom_utils import set_geometry, has_geometry
from ..utils.utils import encode_geodataframe, get_geodataframe_bounds, get_geodataframe_geom_type, \
                          get_datetime_column_names

RFC_2822_DATETIME_FORMAT = "%a, %d %b %Y %T %z"


class SourceType:
    QUERY = 'Query'
    GEOJSON = 'GeoJSON'


class Source:
    """Source

    Args:
        data (str, pandas.DataFrame, geopandas.GeoDataFrame): a table name,
            SQL query, DataFrame, GeoDataFrame instance.
        credentials (:py:class:`Credentials <cartoframes.auth.Credentials>`, optional):
            A Credentials instance. If not provided, the credentials will be automatically
            obtained from the default credentials if available.
        bounds (dict or list, optional): a dict with `west`, `south`, `east`, `north`
            keys, or an array of floats in the following structure: [[west,
            south], [east, north]]. If not provided the bounds will be automatically
            calculated to fit all features.
        geom_col (str, optional): string indicating the geometry column name in the source `DataFrame`.

    Example:

        Table name.

        >>> Source('table_name')

        SQL query.

        >>> Source('SELECT * FROM table_name')

        DataFrame object.

        >>> Source(df, geom_col='my_geom')

        GeoDataFrame object.

        >>> Source(gdf)

        Setting the credentials.

        >>> Source('table_name', credentials)

    """
    def __init__(self, source, credentials=None, geom_col=None):
        self.credentials = None
        self.datetime_column_names = None

        if isinstance(source, str):
            # Table, SQL query
            self.type = SourceType.QUERY
            self.manager = ContextManager(credentials)
            self.query = self.manager.compute_query(source)
            self.credentials = self.manager.credentials
        elif isinstance(source, DataFrame):
            # DataFrame, GeoDataFrame
            self.type = SourceType.GEOJSON
            self.gdf = GeoDataFrame(source, copy=True)
            self.set_datetime_columns()

            if geom_col in self.gdf:
                set_geometry(self.gdf, geom_col, inplace=True)
            elif has_geometry(source):
                self.gdf.set_geometry(source.geometry.name, inplace=True)
            else:
                raise ValueError('No valid geometry found. Please provide an input source with ' +
                                 'a valid geometry or specify the "geom_col" param with a geometry column.')

            # Checking the uniqueness of the geometry type
            geometry_types = self.gdf.geom_type.unique()
            if len(geometry_types) > 1:
                raise ValueError('No valid geometry column, it has more that one ' +
                                 'geometry type: {}.'.format(geometry_types))

        else:
            raise ValueError('Wrong source input. Valid values are str and DataFrame.')

    def get_credentials(self):
        if self.type == SourceType.QUERY:
            if self.credentials:
                return {
                    # CARTO VL requires a username but CARTOframes allows passing only the base_url.
                    # That's why 'user' is used by default if username is empty.
                    'username': self.credentials.username or 'user',
                    'api_key': self.credentials.api_key,
                    'base_url': self.credentials.base_url<commit_msg>More fine type check for viz<commit_after>from pandas import DataFrame
from geopandas import GeoDataFrame

from ..io.managers.context_manager import ContextManager
from ..utils.geom_utils import set_geometry, has_geometry
from ..utils.utils import encode_geodataframe, get_geodataframe_bounds, \
                          get_geodataframe_geom_type, get_datetime_column_names

RFC_2822_DATETIME_FORMAT = "%a, %d %b %Y %T %z"

VALID_GEOMETRY_TYPES = [
    {'Point'},
    {'MultiPoint'},
    {'LineString'},
    {'MultiLineString'},
    {'LineString', 'MultiLineString'},
    {'Polygon'},
    {'MultiPolygon'},
    {'Polygon', 'MultiPolygon'}
]


class SourceType:
    QUERY = 'Query'
    GEOJSON = 'GeoJSON'


class Source:
    """Source

    Args:
        data (str, pandas.DataFrame, geopandas.GeoDataFrame): a table name,
            SQL query, DataFrame, GeoDataFrame instance.
        credentials (:py:class:`Credentials <cartoframes.auth.Credentials>`, optional):
            A Credentials instance. If not provided, the credentials will be automatically
            obtained from the default credentials if available.
        bounds (dict or list, optional): a dict with `west`, `south`, `east`, `north`
            keys, or an array of floats in the following structure: [[west,
            south], [east, north]]. If not provided the bounds will be automatically
            calculated to fit all features.
        geom_col (str, optional): string indicating the geometry column name in the source `DataFrame`.

    Example:

        Table name.

        >>> Source('table_name')

        SQL query.

        >>> Source('SELECT * FROM table_name')

        DataFrame object.

        >>> Source(df, geom_col='my_geom')

        GeoDataFrame object.

        >>> Source(gdf)

        Setting the credentials.

        >>> Source('table_name', credentials)

    """
    def __init__(self, source, credentials=None, geom_col=None):
        self.credentials = None
        self.datetime_column_names = None

        if isinstance(source, str):
            # Table, SQL query
            self.type = SourceType.QUERY
            self.manager = ContextManager(credentials)
            self.query = self.manager.compute_query(source)
            self.credentials = self.manager.credentials
        elif isinstance(source, DataFrame):
            # DataFrame, GeoDataFrame
            self.type = SourceType.GEOJSON
            self.gdf = GeoDataFrame(source, copy=True)
            self.set_datetime_columns()

            if geom_col in self.gdf:
                set_geometry(self.gdf, geom_col, inplace=True)
            elif has_geometry(source):
                self.gdf.set_geometry(source.geometry.name, inplace=True)
            else:
                raise ValueError('No valid geometry found. Please provide an input source with ' +
                                 'a valid geometry or specify the "geom_col" param with a geometry column.')

            # Checking the uniqueness of the geometry type
            geometry_types = set(self.gdf.geom_type.unique())
            if geometry_types not in VALID_GEOMETRY_TYPES:
                raise ValueError('No valid geometry column types ({}), it has '.format(geometry_types) +
                                 'to be one of the next type sets: {}.'.format(VALID_GEOMETRY_TYPES))

        else:
            raise ValueError('Wrong source input. Valid values are str and DataFrame.')

    def get_credentials(self):
        if self.type == SourceType.QUERY:
            if self.credentials:
                return {
                    # CARTO VL requires a username but CARTOframes allows passing only the base_url.
                    # That's why 'user' is used by default if username is empty.
                    'username': self.credentials.username or 'user',
                    'api_key': self.credentials.api_key,
                    'base_url': self.credentials.base_url
{'fraction_non_alphanumeric': 0.08684175795169592, 'fraction_numerical': 0.0011878051999472086, 'mean_word_length': 2.200168918918919, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 36, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '630315', 'n_tokens_mistral': 2069, 'n_tokens_neox': 1983, 'n_words': 606}