fork/cellxgene
1
0
Fork 0
mirror of https://github.com/chanzuckerberg/cellxgene.git synced 2026-10-04 02:28:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

clean up type inferencing (#2332)

Browse Source 
* unit tests for 64 bit conversion

* clean up type handling

* type inference tests

* more type inference fixes

* use schema to determine user intent for data typing

* stop using deprecated API

* fbs type encoding test

* add missing test

* add more tests

* correctly infer X type for CXG adaptor

* lint

* fix typo

* ts migration

* cleanup from PR review

* lint

* PR review changes
This commit is contained in:
Bruce Martin
2021-07-28 15:10:12 -07:00
committed by GitHub
parent 1140676106
commit 32f60a1547
15 changed files with 730 additions and 346 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/test/test_common/unit/fbs/test_matrix.py
+138 -4
View File
@@ -2,16 +2,20 @@ import unittest
import pandas as pd
import numpy as np
from scipy import sparse
from parameterized import parameterized_class
import json
import backend.test.decode_fbs as decode_fbs
from backend.test import decode_fbs
from backend.common.fbs.matrix import encode_matrix_fbs, decode_matrix_fbs
from backend.common.utils.type_conversion_utils import get_dtypes_and_schemas_of_dataframe
import backend.common.fbs as fbs
class FbsTests(unittest.TestCase):
"""Test Case for Matrix FBS data encode/decode """
"""Test Case for Matrix FBS data encode/decode"""
def test_encode_boundary(self):
""" test various boundary checks """
"""test various boundary checks"""
# row indexing is unsupported
with self.assertRaises(ValueError):
@@ -46,7 +50,7 @@ class FbsTests(unittest.TestCase):
"d": pd.Series(["x", "y", "z", "x", "y", "z", "a", "x", "y", "z"], dtype="category"),
}
)
expected_types = ((np.ndarray, np.float32), (np.ndarray, np.int32), (np.ndarray, np.uint32), (list, None))
expected_types = ((np.ndarray, np.float32), (np.ndarray, np.int32), (np.ndarray, np.int32), (list, None))
fbs = encode_matrix_fbs(matrix=df, row_idx=None, col_idx=df.columns)
self.fbs_checks(fbs, (10, 4), expected_types, ["a", "b", "c", "d"])
@@ -80,3 +84,133 @@ class FbsTests(unittest.TestCase):
self.assertTrue(np.all(dfSrc[c] == dfDst[c]))
else:
self.assertEqual(dfSrc[c], dfDst[c])
"""
Test type consistency between FBS encoding and the underlying schema hint.
Basic assertion: the FBS type returned by encode_matrix_fbs() will be consistent
with the schema hint returned by type_conversion_utils (which is in turn used
to create the client schema).
The following test cases are all dicts which contain the following keys:
- dataframe - the dataframe used as input for encode_matrix_fbs
- expected_fbs_types - upon success, dict of FBS column types expected (eg, Float32Array)
- expected_schema_hints - upon success, dict of schema hint
All are keyed by column name.
"""
# simple tests that we convert all ints to int32
int_dtypes = [np.dtype(d) for d in [np.int8, np.uint8, np.int16, np.uint16, np.int32, np.uint32, np.int64, np.uint64]]
int_test_cases = [
{
"dataframe": pd.DataFrame({dtype.name: np.zeros((10,), dtype=dtype) for dtype in int_dtypes}),
"expected_fbs_types": dict(
[(dtype.name, fbs.NetEncoding.TypedArray.TypedArray.Int32Array) for dtype in int_dtypes]
),
"expected_schema_hints": dict([(dtype.name, {"type": "int32"}) for dtype in int_dtypes]),
}
]
# simple tests that we convert all floats to float32
float_dtypes = [np.dtype(d) for d in [np.float16, np.float32, np.float64]]
float_test_cases = [
{
"dataframe": pd.DataFrame({dtype.name: np.zeros((10,), dtype=dtype) for dtype in float_dtypes}),
"expected_fbs_types": dict(
[(dtype.name, fbs.NetEncoding.TypedArray.TypedArray.Float32Array) for dtype in float_dtypes]
),
"expected_schema_hints": dict([(dtype.name, {"type": "float32"}) for dtype in float_dtypes]),
}
]
# boolean - should be encoded as an uint32
bool_dtypes = [np.dtype(d) for d in [np.bool_, bool]]
bool_test_cases = [
{
"dataframe": pd.DataFrame({dtype.name: np.ones((10,), dtype=dtype) for dtype in bool_dtypes}),
"expected_fbs_types": dict(
[(dtype.name, fbs.NetEncoding.TypedArray.TypedArray.Uint32Array) for dtype in bool_dtypes]
),
"expected_schema_hints": dict([(dtype.name, {"type": "boolean"}) for dtype in bool_dtypes]),
}
]
cat_test_cases = [
{
"dataframe": pd.DataFrame({"a": pd.Series(["a", "b", "c", "a", "b", "c"], dtype="category")}),
"expected_fbs_types": {"a": fbs.NetEncoding.TypedArray.TypedArray.JSONEncodedArray},
"expected_schema_hints": {"a": {"type": "categorical", "categories": ["a", "b", "c"]}},
},
{
"dataframe": pd.DataFrame(
{"a": pd.Series(["a", "b", "c", "a", "b", "c"], dtype="category").cat.remove_categories("b")}
),
"expected_fbs_types": {"a": fbs.NetEncoding.TypedArray.TypedArray.JSONEncodedArray},
"expected_schema_hints": {"a": {"type": "categorical", "categories": ["a", "c"]}},
},
{
"dataframe": pd.DataFrame({"a": pd.Series(np.arange(0, 10, dtype=np.int64), dtype="category")}),
"expected_fbs_types": {"a": fbs.NetEncoding.TypedArray.TypedArray.Int32Array},
"expected_schema_hints": {"a": {"type": "categorical"}},
},
{
"dataframe": pd.DataFrame(
{"a": pd.Series(np.arange(0, 10, dtype=np.int64), dtype="category").cat.remove_categories(2)}
),
"expected_fbs_types": {"a": fbs.NetEncoding.TypedArray.TypedArray.Float32Array},
"expected_schema_hints": {"a": {"type": "categorical"}},
},
{
"dataframe": pd.DataFrame({"a": pd.Series(np.arange(0, 10, dtype=np.float64), dtype="category")}),
"expected_fbs_types": {"a": fbs.NetEncoding.TypedArray.TypedArray.Float32Array},
"expected_schema_hints": {"a": {"type": "categorical"}},
},
{
"dataframe": pd.DataFrame(
{"a": pd.Series(np.arange(0, 10, dtype=np.float64), dtype="category").cat.remove_categories(2)}
),
"expected_fbs_types": {"a": fbs.NetEncoding.TypedArray.TypedArray.Float32Array},
"expected_schema_hints": {"a": {"type": "categorical"}},
},
]
test_cases = [
*int_test_cases,
*float_test_cases,
*bool_test_cases,
*cat_test_cases,
]
@parameterized_class(test_cases)
class TestTypeConversionConsistency(unittest.TestCase):
def test_type_conversion_consistency(self):
self.assertEqual(self.dataframe.shape[1], len(self.expected_fbs_types))
self.assertEqual(self.dataframe.shape[1], len(self.expected_schema_hints))
buf = encode_matrix_fbs(matrix=self.dataframe, col_idx=self.dataframe.columns)
encoding_dtypes, schema_hints = get_dtypes_and_schemas_of_dataframe(self.dataframe)
# check schema hints
# print(schema_hints)
# print(self.expected_schema_hints)
self.assertEqual(schema_hints, self.expected_schema_hints)
# inspect the FBS types
matrix = fbs.NetEncoding.Matrix.Matrix.GetRootAsMatrix(buf, 0)
columns_length = matrix.ColumnsLength()
self.assertEqual(columns_length, self.dataframe.shape[1])
self.assertEqual(matrix.ColIndexType(), fbs.NetEncoding.TypedArray.TypedArray.JSONEncodedArray)
col_labels_arr = fbs.NetEncoding.JSONEncodedArray.JSONEncodedArray()
col_labels_arr.Init(matrix.ColIndex().Bytes, matrix.ColIndex().Pos)
col_index_labels = json.loads(col_labels_arr.DataAsNumpy().tobytes().decode("utf-8"))
self.assertEqual(len(col_index_labels), self.dataframe.shape[1])
for col_idx in range(0, columns_length):
col_label = col_index_labels[col_idx]
col = matrix.Columns(col_idx)
col_type = col.UType()
self.assertEqual(self.expected_fbs_types[col_label], col_type)
backend/test/test_common/unit/utils/test_type_conversion_utils.py
+288 -181
View File
@@ -1,179 +1,22 @@
import unittest
from time import time
from unittest.mock import patch
import logging
from parameterized import parameterized_class
import numpy as np
import pandas as pd
from pandas import Series, DataFrame
from scipy import sparse
from backend.common.utils.type_conversion_utils import (
can_cast_to_float32,
can_cast_to_int32,
get_dtype_of_array,
get_encoding_dtype_of_array,
get_schema_type_hint_of_array,
get_dtypes_and_schemas_of_dataframe,
convert_pandas_series_to_numpy,
get_dtype_and_schema_of_array,
get_schema_type_hint_from_dtype,
)
class TestTypeConversionUtils(unittest.TestCase):
def test__can_cast_to_float32__string_is_false(self):
array_to_convert = Series(data=["1", "2", "3"], dtype=str)
can_cast = can_cast_to_float32(array_to_convert.dtype, array_to_convert)
self.assertFalse(can_cast)
def test__can_cast_to_float32__float64_is_true_warning_outputted(self):
array_to_convert = Series(data=[1, 2, 3], dtype=np.dtype(np.float64))
with self.assertLogs(level="WARN") as logger:
can_cast = can_cast_to_float32(array_to_convert.dtype, array_to_convert)
self.assertIn("may lose precision", logger.output[0])
self.assertTrue(can_cast)
@patch("logging.warning")
def test__can_cast_to_float32__float32_is_false(self, mock_log_warning):
array_to_convert = Series(data=[1, 2, 3], dtype=np.dtype(np.float32))
can_cast = can_cast_to_float32(array_to_convert.dtype, array_to_convert)
self.assertTrue(can_cast)
assert not mock_log_warning.called
def test__can_cast_to_float32__categorical_float64_is_false(self):
array_to_convert = Series(data=[1.1, 2.2, 3.3], dtype="category")
can_cast = can_cast_to_float32(array_to_convert.dtype, array_to_convert)
self.assertFalse(can_cast)
def test__can_cast_to_float32__categorical_int64_with_nans_is_true(self):
array_to_convert = Series(data=[1, 2, np.NaN], dtype="category")
can_cast = can_cast_to_float32(array_to_convert.dtype, array_to_convert)
self.assertTrue(can_cast)
def test__can_cast_to_float_32__float_32_with_nans_is_true(self):
array_to_convert = Series(data=[1, 2, np.NaN], dtype=np.dtype(np.float32))
can_cast = can_cast_to_float32(array_to_convert.dtype, array_to_convert)
self.assertTrue(can_cast)
def test__can_cast_to_int32__string_is_false(self):
array_to_convert = Series(data=["1", "2", "3"], dtype=str)
can_cast = can_cast_to_int32(array_to_convert.dtype, array_to_convert)
self.assertFalse(can_cast)
def test__can_cast_to_int32__int64_is_true(self):
array_to_convert = Series(data=[1, 2, 3], dtype=np.dtype(np.int64))
can_cast = can_cast_to_int32(array_to_convert.dtype, array_to_convert)
self.assertTrue(can_cast)
def test__can_cast_to_int32__int16_is_true(self):
array_to_convert = Series(data=[1, 2, 3], dtype=np.dtype(np.int16))
can_cast = can_cast_to_int32(array_to_convert.dtype, array_to_convert)
self.assertTrue(can_cast)
def test__can_cast_to_int32__int64_with_large_value_is_false(self):
array_to_convert = Series(data=[3000000000, 2, 3], dtype=np.dtype(np.int64))
can_cast = can_cast_to_int32(array_to_convert.dtype, array_to_convert)
self.assertFalse(can_cast)
def test__can_cast_to_int32__int64_with_nans_is_false(self):
array_to_convert = Series(data=[np.NaN, "2", "3"], dtype="category")
can_cast = can_cast_to_int32(array_to_convert.dtype, array_to_convert)
self.assertFalse(can_cast)
def test__get_dtype_of_array__supported_dtypes_return_as_expected(self):
types = [np.float32, np.int32, np.bool_, str]
expected_dtypes = [np.float32, np.int32, np.uint8, str]
for test_type_index in range(len(types)):
with self.subTest(
f"Testing get_dtype_of_array with type {types[test_type_index].__name__}", i=test_type_index
):
array = Series(data=[], dtype=types[test_type_index])
self.assertEqual(get_dtype_of_array(array), expected_dtypes[test_type_index])
def test__get_dtype_of_array__categories_return_as_expected(self):
array = Series(data=["a", "b", "c"], dtype="category")
expected_dtype = str
actual_dtype = get_dtype_of_array(array)
self.assertEqual(expected_dtype, actual_dtype)
def test__get_dtype_of_array__unordered_integer_categories_return_as_expected(self):
array = Series(data=[2, 3, 1, 3, 1, 2], dtype="category")
expected_dtype = np.int32
actual_dtype = get_dtype_of_array(array)
self.assertEqual(expected_dtype, actual_dtype)
def test__get_dtype_of_array__castable_dtypes_return_as_expected(self):
types = [np.float64, np.int64]
expected_dtypes = [np.float32, np.int32]
for test_type_index in range(len(types)):
with self.subTest(
f"Testing get_dtype_of_array with castable type {types[test_type_index].__name__}", i=test_type_index
):
array = Series(data=[], dtype=types[test_type_index])
self.assertEqual(get_dtype_of_array(array), expected_dtypes[test_type_index])
def test__get_dtype_of_array__unsupported_type_raises_exception(self):
unsupported_array = Series(list([time() for _ in range(2)]), dtype="datetime64[ns]")
with self.assertRaises(TypeError) as exception_context:
get_dtype_of_array(unsupported_array)
self.assertIn("unsupported", str(exception_context.exception))
def test__get_schema_type_hint_of_array__supported_dtypes_return_as_expected(self):
types = [np.float32, np.int32, np.bool_, str]
expected_schema_hints = [{"type": "float32"}, {"type": "int32"}, {"type": "boolean"}, {"type": "string"}]
for test_type_index in range(len(types)):
with self.subTest(
f"Testing get_schema_type_hint_of_array with type {types[test_type_index].__name__}", i=test_type_index
):
array = Series(data=[], dtype=types[test_type_index])
self.assertEqual(get_schema_type_hint_of_array(array), expected_schema_hints[test_type_index])
def test__get_schema_type_hint_of_array__categories_return_as_expected(self):
array = Series(data=["a", "b", "b"], dtype="category")
expected_schema_hint = {"type": "categorical", "categories": ["a", "b"]}
actual_schema_hint = get_schema_type_hint_of_array(array)
self.assertEqual(expected_schema_hint, actual_schema_hint)
def test__get_schema_type_hint_of_array__castable_dtypes_return_as_expected(self):
types = [np.float64, np.int64]
expected_schema_hints = [{"type": "float32"}, {"type": "int32"}]
for test_type_index in range(len(types)):
with self.subTest(
f"Testing get_schema_type_hint_of_array with castable type {types[test_type_index].__name__}",
i=test_type_index,
):
array = Series(data=[], dtype=types[test_type_index])
self.assertEqual(get_schema_type_hint_of_array(array), expected_schema_hints[test_type_index])
def test__get_dtypes_and_schemas_of_dataframe__dtype_and_schema_returns_as_expected(self):
float_array = Series(data=[1, 2, 3], dtype=np.dtype(np.float64))
category_array = Series(data=["a", "b", "b"], dtype="category")
@@ -190,28 +33,292 @@ class TestTypeConversionUtils(unittest.TestCase):
self.assertEqual(expected_data_types_dict, actual_dataframe_data_types)
self.assertEqual(expected_schema_type_hints_dict, actual_dataframe_schema_type_hints)
def test__convert_pandas_series_to_numpy__categorical_float64_to_float64_with_nans(self):
expected_float_array = np.array([1.1, 2.2, np.NaN], dtype=np.float64)
float_series = Series(data=[1.1, 2.2, np.NaN], dtype="category")
def test__get_schema_type_hint_from_dtype(self):
self.assertEqual(get_schema_type_hint_from_dtype(np.dtype(np.bool_)), {"type": "boolean"})
actual_float_array = convert_pandas_series_to_numpy(float_series, np.float64)
for dtype in [np.int8, np.int8, np.int16, np.uint16, np.int32]:
self.assertEqual(get_schema_type_hint_from_dtype(np.dtype(dtype)), {"type": "int32"})
for dtype in [np.uint32, np.int64, np.uint64]:
with self.assertRaises(TypeError):
get_schema_type_hint_from_dtype(np.dtype(dtype))
np.testing.assert_equal(expected_float_array, actual_float_array)
for dtype in [np.float16, np.float32, np.float64]:
self.assertEqual(get_schema_type_hint_from_dtype(np.dtype(dtype)), {"type": "float32"})
def test__convert_pandas_series_to_numpy__float64_to_float64(self):
expected_float_array = np.array([1.1, 2.2], dtype=np.float64)
float_series = Series(data=[1.1, 2.2], dtype=np.dtype(np.float64))
for dtype in [np.dtype(object), np.dtype(str)]:
self.assertEqual(get_schema_type_hint_from_dtype(dtype), {"type": "string"})
actual_float_array = convert_pandas_series_to_numpy(float_series, np.float64)
np.testing.assert_equal(expected_float_array, actual_float_array)
# Credit: https://stackoverflow.com/questions/35871815/python-3-unit-testing-assert-logger-not-called/64774103#64774103
class AssertNoLog:
def assertNoLogs(self, logger, level):
"""functions as a context manager. To be introduced in python 3.10"""
def test__convert_pandas_series_to_numpy__int64_to_int32_with_nans_throws_error(self):
int_series = Series(data=[1, 2, np.NaN], dtype="category")
class AssertNoLogsContext(unittest.TestCase):
def __init__(self, logger, level):
self.logger = logger
self.level = level
self.context = self.assertLogs(logger, level)
with self.assertLogs(level="ERROR") as logger:
convert_pandas_series_to_numpy(int_series, np.int32)
def __enter__(self):
"""enter self.assertLogs as context manager, and log something"""
self.initial_logmsg = "sole message"
self.cm = self.context.__enter__()
self.logger.log(self.level, self.initial_logmsg)
return self.cm
self.assertIn(
"Cannot convert a pandas Series object to an integer dtype if it contains NaNs", logger.output[0]
)
def __exit__(self, exc_type, exc_val, exc_tb):
"""cleanup logs, and then check nothing extra was logged"""
# assertLogs.__exit__ should never fail because of initial msg
self.context.__exit__(exc_type, exc_val, exc_tb)
if len(self.cm.output) > 1:
"""override any exception passed to __exit__"""
self.context._raiseFailure(
"logs of level {} or higher triggered on {} : {}".format(
logging.getLevelName(self.level), self.logger.name, self.cm.output[1:]
)
)
return AssertNoLogsContext(logger, level)
"""
See table of expected cases in type_conversion_utils.py.
This probes all edge cases. Each case is a dict containing keys:
- data - the array to be introspected
- throws - if not None, the expected Error (eg, TypeError)
- expected_encoding_dtype - upon success
- expected_schema_hint - upon success
- logs - if not None, specify expected log output
"""
bool_OK_cases = [
{
"data": data,
"expected_encoding_dtype": np.uint8,
"expected_schema_hint": {"type": "boolean"},
}
for data in [
np.array([0, 1, 0, 1], dtype=np.bool_),
pd.Series(np.array([0, 1, 0, 1], dtype=np.bool_)),
# pd.Index with bools doesn't really make any sense...and becomes dtype=object
]
]
int_OK_cases = [
{
"data": data,
"expected_encoding_dtype": np.int32,
"expected_schema_hint": {"type": "int32"},
}
for dtype in [np.int8, np.uint8, np.int16, np.uint16, np.int32, np.uint32, np.int64, np.uint64]
for data in [
np.arange(0, 1000, dtype=dtype),
pd.Series(np.arange(0, 1000, dtype=dtype)),
pd.Index(np.arange(0, 1000, dtype=dtype)),
sparse.csr_matrix((10, 100), dtype=dtype),
]
]
float_OK_cases = [
{
"data": data,
"expected_encoding_dtype": np.float32,
"expected_schema_hint": {"type": "float32"},
"logs": None if data.dtype != np.float64 else {"level": logging.WARNING, "output": "may lose precision"},
}
for dtype in [np.float16, np.float32, np.float64]
for data in [
np.arange(-128, 1000, dtype=dtype),
pd.Series(np.arange(-128, 1000, dtype=dtype)),
pd.Index(np.arange(-129, 1000, dtype=dtype)),
np.array([-np.nan, np.NINF, -1, np.NZERO, 0, np.PZERO, 1, np.PINF, np.nan], dtype=dtype),
np.array([np.finfo(dtype).min, 0, np.finfo(dtype).max], dtype=dtype),
sparse.csr_matrix((10, 100), dtype=dtype),
]
]
numeric_ERR_cases = [
{
"data": data,
"throws": TypeError,
}
for data in [
np.array([np.iinfo(np.int64).min, np.iinfo(np.int64).max], dtype=np.int64),
np.array([np.iinfo(np.uint64).min, np.iinfo(np.uint64).max], dtype=np.uint64),
np.array([np.iinfo(np.uint32).min, np.iinfo(np.uint32).max], dtype=np.uint32),
]
]
string_OK_cases = [
{
"data": data,
"expected_encoding_dtype": np.dtype(str),
"expected_schema_hint": {"type": "string"},
}
for data in [
np.array(["a", "b", "c"]),
np.array(["a", "b", "c"], dtype="object"),
pd.Series(["a", "b", "c"]),
pd.Index(["a", "b", "c"]),
np.array(["a", [], {}, None, True, False, 383.2], dtype="object"),
]
]
category_nonnumeric_OK_cases = [
{
"data": data,
"expected_encoding_dtype": np.dtype(str),
"expected_schema_hint": {"type": "categorical", "categories": data.dtype.categories.to_list()},
}
for data in [
pd.Series(["a", "b", "c"], dtype="category"),
pd.Series(["a", "b", "c", 0, 1, 2], dtype="category"),
pd.Series(["a", "b", "c"], dtype="category").cat.remove_categories(["b"]),
pd.Series(["a", "b", "c", 0, 1, 2], dtype="category").cat.remove_categories(["b", 0]),
]
]
category_numeric_OK_cases = [
# numeric, no NA/NaN, int
*[
{
"data": data,
"expected_encoding_dtype": np.int32,
"expected_schema_hint": {"type": "categorical"},
}
for dtype in [np.int8, np.uint8, np.int16, np.uint16, np.int32, np.uint32, np.int64, np.uint64]
for data in [
pd.Series(np.array([0, 1, 2], dtype=dtype), dtype="category"),
]
],
# numeric, no NA/NaN, float
*[
{
"data": data,
"expected_encoding_dtype": np.float32,
"expected_schema_hint": {"type": "categorical"},
"logs": {"level": logging.WARNING, "output": "may lose precision"},
}
for dtype in [np.float16, np.float32, np.float64]
for data in [
pd.Series(np.array([0, 1, 2], dtype=dtype), dtype="category"),
pd.Series(np.array([0, 1, 2], dtype=dtype), dtype="category").cat.remove_categories([1]),
pd.Categorical(np.array([0, 1, 2], dtype=dtype)),
]
],
# numeric, has NA-induced cast to float32
*[
{
"data": data,
"expected_encoding_dtype": np.float32,
"expected_schema_hint": {"type": "categorical"},
"logs": {"level": logging.WARNING, "output": "may lose precision"},
}
for dtype in [
np.int8,
np.uint8,
np.int16,
np.uint16,
np.int32,
np.uint32,
np.int64,
np.uint64,
np.float16,
np.float32,
np.float64,
]
for data in [
pd.Series(np.array([0, 1, 2], dtype=dtype), dtype="category").cat.remove_categories([1]),
pd.Categorical(np.array([0, 1, 2], dtype=dtype), categories=np.array([0, 1], dtype=dtype)),
]
],
]
category_ERR_cases = [
# catch expected categorical exceptions for Int64(etc) that have large values
{
"data": data,
"throws": TypeError,
}
for data in [
pd.Categorical(np.array([np.iinfo(np.int64).min, np.iinfo(np.int64).max], dtype=np.int64)),
pd.Categorical(np.array([np.iinfo(np.uint64).min, np.iinfo(np.uint64).max], dtype=np.uint64)),
pd.Categorical(np.array([np.iinfo(np.uint32).min, np.iinfo(np.uint32).max], dtype=np.uint32)),
]
]
object_OK_cases = [
{
"data": data,
"expected_encoding_dtype": np.dtype(str),
"expected_schema_hint": {"type": "string"},
}
for data in [
np.array(["a", True, 1, [], {}], dtype="object"),
pd.Series(["a", True, 1, [], {}], dtype="object"),
pd.Index(["a", True, 1, [], {}], dtype="object"),
]
]
err_cases = [
{"data": np.array, "throws": TypeError}
for data in [
np.ones((10,), dtype=np.complex64),
np.ones((10,), dtype=np.complex128),
np.array([b"foobar"], dtype=np.bytes_),
np.ones((10,), dtype=np.void),
np.arange("2005-02", "2005-03", dtype="datetime64[D]"),
np.arange("2005-02", "2005-03", dtype="datetime64[D]") - np.datetime64("2008-01-01"),
[],
{},
]
]
test_cases = [
*bool_OK_cases,
*int_OK_cases,
*float_OK_cases,
*numeric_ERR_cases,
*string_OK_cases,
*category_nonnumeric_OK_cases,
*category_numeric_OK_cases,
*category_ERR_cases,
*object_OK_cases,
*err_cases,
]
@parameterized_class(test_cases)
class TestTypeInference(unittest.TestCase, AssertNoLog):
def test_type_inference(self):
throws = getattr(self, "throws", None)
if throws:
with self.assertRaises(throws):
get_dtype_and_schema_of_array(self.data)
with self.assertRaises(throws):
get_encoding_dtype_of_array(self.data)
with self.assertRaises(throws):
get_schema_type_hint_of_array(self.data)
else:
logs = getattr(self, "logs", None)
if logs is not None:
with self.assertLogs(level=logs["level"]) as logger:
encoding_dtype, schema_hint = get_dtype_and_schema_of_array(self.data)
self.assertEqual(encoding_dtype, self.expected_encoding_dtype)
self.assertEqual(schema_hint, self.expected_schema_hint)
self.assertIn(logs["output"], logger.output[0])
else:
with self.assertNoLogs(logging.getLogger(), logging.WARNING):
encoding_dtype, schema_hint = get_dtype_and_schema_of_array(self.data)
self.assertEqual(encoding_dtype, self.expected_encoding_dtype)
self.assertEqual(schema_hint, self.expected_schema_hint)
# also test the other public API
self.assertEqual(get_encoding_dtype_of_array(self.data), self.expected_encoding_dtype)
self.assertEqual(get_schema_type_hint_of_array(self.data), self.expected_schema_hint)