"""
In this module lives the type tree.
"""
from collections import defaultdict
from typing import Any, DefaultDict, Dict, List, Optional, Tuple, Type, Union
import dask.dataframe as dd
import numpy as np
import pandas as pd
from ..clean import validate_country, validate_lat_long
from ..errors import UnreachableError
CATEGORICAL_NUMPY_DTYPES = [np.bool, np.object]
CATEGORICAL_PANDAS_DTYPES = [pd.CategoricalDtype, pd.PeriodDtype]
CATEGORICAL_DTYPES = CATEGORICAL_NUMPY_DTYPES + CATEGORICAL_PANDAS_DTYPES
STRING_PANDAS_DTYPES = [pd.StringDtype]
STRING_DTYPES = STRING_PANDAS_DTYPES
NUMERICAL_NUMPY_DTYPES = [np.number]
NUMERICAL_DTYPES = NUMERICAL_NUMPY_DTYPES
DATETIME_NUMPY_DTYPES = [np.datetime64]
DATETIME_PANDAS_DTYPES = [pd.DatetimeTZDtype]
DATETIME_DTYPES = DATETIME_NUMPY_DTYPES + DATETIME_PANDAS_DTYPES
NULL_VALUES = {
float("NaN"),
"#N/A",
"#N/A N/A",
"#NA",
"-1.#IND",
"-1.#QNAN",
"-NaN",
"-nan",
"1.#IND",
"1.#QNAN",
"<NA>",
"N/A",
"NA",
"NULL",
"NaN",
"n/a",
"nan",
"null",
"",
}
[docs]class DType:
"""
Root of Type Tree
"""
############## Syntactic DTypes ##############
[docs]class Categorical(DType):
"""
Type Categorical
"""
[docs]class Nominal(Categorical):
"""
Type Nominal, Subtype of Categorical
"""
[docs]class Ordinal(Categorical):
"""
Type Ordinal, Subtype of Categorical
"""
[docs]class Numerical(DType):
"""
Type Numerical
"""
[docs]class Continuous(Numerical):
"""
Type Continuous, Subtype of Numerical
"""
[docs]class Discrete(Numerical):
"""
Type Discrete, Subtype of Numerical
"""
############## Semantic DTypes ##############
[docs]class DateTime(Numerical):
"""
Type DateTime, Subtype of Numerical
"""
[docs]class Text(Nominal):
"""
Type Text, Subtype of Nominal
"""
[docs]class GeoGraphy(Categorical):
"""
Type GeoGraphy, Subtype of Categorical
"""
[docs]class GeoPoint(DType):
"""
Type GeoPoint
"""
[docs]class LatLong(GeoPoint):
"""
Type LatLong, Tuple
"""
def __init__(self, lat_col: str, long_col: str) -> None:
self.lat = lat_col
self.long = long_col
############## End of the Type Tree ##############
DTypeOrStr = Union[DType, Type[DType], str, None]
DTypeDict = Union[Dict[str, Union[DType, Type[DType], str]], None]
DTypeDef = Union[Dict[str, Union[DType, Type[DType], str]], DType, Type[DType], None]
[docs]def detect_dtype(
col: dd.Series,
known_dtype: Optional[DTypeDef] = None,
detect_small_distinct: bool = True,
) -> DType:
"""
Given a column, detect its type or transform its type according to users' specification
Parameters
----------
col: dask.datafram.Series
A dataframe column
known_dtype: Optional[Union[Dict[str, Union[DType, str]], DType]], default None
A dictionary or single DType given by users to specify the types for designated columns or
all columns. E.g. known_dtype = {"a": Continuous, "b": "Nominal"} or
known_dtype = {"a": Continuous(), "b": "nominal"} or
known_dtype = Continuous() or known_dtype = "Continuous" or known_dtype = Continuous()
detect_small_distinct: bool, default True
Whether to detect numerical columns with small distinct values as categorical column.
"""
if not known_dtype:
return detect_without_known(col, detect_small_distinct=detect_small_distinct)
if isinstance(known_dtype, dict):
if col.name in known_dtype:
dtype = normalize_dtype(known_dtype[col.name])
return map_dtype(dtype)
elif isinstance(normalize_dtype(known_dtype), DType):
return map_dtype(normalize_dtype(known_dtype))
return detect_without_known(col, detect_small_distinct=detect_small_distinct)
[docs]def map_dtype(dtype: DType) -> DType:
"""
Currently, we want to keep our Type System flattened.
We will map Categorical() to Nominal() and Numerical() to Continuous()
"""
if (
isinstance(dtype, Categorical) is True
and isinstance(dtype, Ordinal) is False
and isinstance(dtype, Nominal) is False
):
return Nominal()
elif (
isinstance(dtype, Numerical) is True
and isinstance(dtype, Continuous) is False
and isinstance(dtype, Discrete) is False
):
return Continuous()
else:
return dtype
[docs]def detect_without_known(col: dd.Series, detect_small_distinct: bool) -> DType:
# pylint: disable=too-many-return-statements
"""
This function detects dtypes of column when users didn't specify.
"""
if is_nominal(col.dtype):
if is_geography(col):
return GeoGraphy()
if is_geopoint(col):
return GeoPoint()
else:
return Nominal()
elif is_continuous(col.dtype):
if detect_small_distinct:
# detect as categorical if distinct value is small
nuniques = col.nunique_approx().compute()
if nuniques < 10:
return Nominal()
else:
return Continuous()
else:
return Continuous()
elif is_datetime(col.dtype):
return DateTime()
else:
raise UnreachableError
[docs]def is_dtype(dtype1: Any, dtype2: DType) -> bool:
"""
This function detects if dtype2 is dtype1.
"""
return isinstance(dtype1, dtype2.__class__)
[docs]def normalize_dtype(dtype_repr: Any) -> DType:
"""
This function normalizes a dtype repr.
"""
normalized: DType
str_dic = {
"Categorical": Categorical,
"Ordinal": Ordinal,
"Nominal": Nominal,
"Numerical": Numerical,
"Continuous": Continuous,
"Discrete": Discrete,
"DateTime": DateTime,
"Text": Text,
}
for str_dtype, dtype in str_dic.items():
if isinstance(dtype_repr, str):
if dtype_repr.lower() == str_dtype.lower():
normalized = dtype()
break
elif isinstance(dtype_repr, dtype):
normalized = dtype_repr
break
elif dtype_repr == dtype:
normalized = dtype()
break
return normalized
[docs]def is_nominal(dtype: Any) -> bool:
"""
Given a type, return if that type is a nominal type
"""
if is_continuous(dtype) or is_datetime(dtype):
return False
if isinstance(dtype, np.dtype):
dtype = dtype.type
return any(issubclass(dtype, c) for c in CATEGORICAL_NUMPY_DTYPES)
else:
return any(isinstance(dtype, c) for c in CATEGORICAL_PANDAS_DTYPES)
[docs]def is_geography(col: dd.Series) -> bool:
"""
Given a column, return if its type is a geography type
"""
geo = col.compute()[:100]
geo_ratio: float = np.sum(validate_country(geo)) / geo.shape[0]
return geo_ratio > 0.8
[docs]def is_geopoint(col: dd.Series) -> bool:
"""
Given a column, return if its type is a geopoint type
"""
lat_long = pd.Series(col.compute()[:100], dtype="string")
lat_long_ratio: float = np.sum(validate_lat_long(lat_long)) / lat_long.shape[0]
return lat_long_ratio > 0.8
[docs]def is_continuous(dtype: Any) -> bool:
"""
Given a type, return if that type is a continuous type
"""
dtype = dtype.type
return any(issubclass(dtype, c) for c in NUMERICAL_NUMPY_DTYPES)
[docs]def is_datetime(dtype: Any) -> bool:
"""
Given a type, return if that type is a datetime type
"""
if isinstance(dtype, np.dtype):
dtype = dtype.type
return any(issubclass(dtype, c) for c in DATETIME_NUMPY_DTYPES)
else:
return any(isinstance(dtype, c) for c in DATETIME_PANDAS_DTYPES)
[docs]def is_pandas_categorical(dtype: Any) -> bool:
"""
Detect if a dtype is categorical and from pandas.
"""
return any(isinstance(dtype, c) for c in CATEGORICAL_PANDAS_DTYPES)
[docs]def drop_null(
var: Union[dd.Series, pd.DataFrame, dd.DataFrame]
) -> Union[pd.Series, dd.Series, pd.DataFrame, dd.DataFrame]:
"""
Drop the null values (specified in NULL_VALUES) from a series or DataFrame
"""
if isinstance(var, (pd.Series, dd.Series)):
if is_datetime(var.dtype):
return var.dropna()
return var[~var.isin(NULL_VALUES)]
elif isinstance(var, (pd.DataFrame, dd.DataFrame)):
df = var
for values in df.columns:
if is_datetime(df[values].dtype):
df = df.dropna(subset=[values])
else:
df = df[~df[values].isin(NULL_VALUES)]
return df
raise ValueError("Input should be a Pandas/Dask Dataframe or Series")
[docs]def get_dtype_cnts_and_num_cols(
df: dd.DataFrame,
dtype: Union[Dict[str, Union[DType, Type[DType], str]], DType, Type[DType], None],
) -> Tuple[Dict[str, int], List[str]]:
"""
Get the count of each dtype in a dataframe
"""
dtype_cnts: DefaultDict[str, int] = defaultdict(int)
num_cols: List[str] = []
for col in df.columns:
col_dtype = detect_dtype(df[col], dtype)
if is_dtype(col_dtype, Nominal()):
dtype_cnts["Categorical"] += 1
elif is_dtype(col_dtype, Continuous()):
dtype_cnts["Numerical"] += 1
num_cols.append(col)
elif is_dtype(col_dtype, DateTime()):
dtype_cnts["DateTime"] += 1
elif is_dtype(col_dtype, GeoGraphy()):
dtype_cnts["GeoGraphy"] += 1
elif is_dtype(col_dtype, GeoPoint()):
dtype_cnts["GeoPoint"] += 1
else:
raise NotImplementedError
return dtype_cnts, num_cols