wehub-resource-sync
122 lines
3.7 KiB
Python
122 lines
3.7 KiB
Python
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
import ray
|
|
from ray.data.preprocessors import Normalizer
|
|
|
|
|
|
def test_normalizer():
|
|
"""Tests basic Normalizer functionality."""
|
|
|
|
col_a = [10, 10, 10]
|
|
col_b = [1, 3, 3]
|
|
col_c = [2, 4, -4]
|
|
in_df = pd.DataFrame.from_dict({"A": col_a, "B": col_b, "C": col_c})
|
|
ds = ray.data.from_pandas(in_df)
|
|
|
|
# l2 norm
|
|
normalizer = Normalizer(["B", "C"])
|
|
transformed = normalizer.transform(ds)
|
|
out_df = transformed.to_pandas()
|
|
|
|
processed_col_a = col_a
|
|
processed_col_b = [1 / np.sqrt(5), 0.6, 0.6]
|
|
processed_col_c = [2 / np.sqrt(5), 0.8, -0.8]
|
|
expected_df = pd.DataFrame.from_dict(
|
|
{"A": processed_col_a, "B": processed_col_b, "C": processed_col_c}
|
|
).astype(out_df.dtypes.to_dict())
|
|
|
|
pd.testing.assert_frame_equal(out_df, expected_df, check_like=True)
|
|
|
|
# l1 norm
|
|
normalizer = Normalizer(["B", "C"], norm="l1")
|
|
|
|
transformed = normalizer.transform(ds)
|
|
out_df = transformed.to_pandas()
|
|
|
|
processed_col_a = col_a
|
|
processed_col_b = [1 / 3, 3 / 7, 3 / 7]
|
|
processed_col_c = [2 / 3, 4 / 7, -4 / 7]
|
|
expected_df = pd.DataFrame.from_dict(
|
|
{"A": processed_col_a, "B": processed_col_b, "C": processed_col_c}
|
|
).astype(out_df.dtypes.to_dict())
|
|
|
|
pd.testing.assert_frame_equal(out_df, expected_df, check_like=True)
|
|
|
|
# max norm
|
|
normalizer = Normalizer(["B", "C"], norm="max")
|
|
|
|
transformed = normalizer.transform(ds)
|
|
out_df = transformed.to_pandas()
|
|
|
|
processed_col_a = col_a
|
|
processed_col_b = [0.5, 0.75, 0.75]
|
|
processed_col_c = [1.0, 1.0, -1.0]
|
|
expected_df = pd.DataFrame.from_dict(
|
|
{"A": processed_col_a, "B": processed_col_b, "C": processed_col_c}
|
|
).astype(out_df.dtypes.to_dict())
|
|
|
|
pd.testing.assert_frame_equal(out_df, expected_df, check_like=True)
|
|
|
|
# append mode
|
|
with pytest.raises(ValueError):
|
|
Normalizer(columns=["B", "C"], output_columns=["B_encoded"])
|
|
|
|
normalizer = Normalizer(["B", "C"], output_columns=["B_normalized", "C_normalized"])
|
|
transformed = normalizer.transform(ds)
|
|
out_df = transformed.to_pandas()
|
|
|
|
processed_col_a = col_a
|
|
processed_col_b = [1 / np.sqrt(5), 0.6, 0.6]
|
|
processed_col_c = [2 / np.sqrt(5), 0.8, -0.8]
|
|
expected_df = pd.DataFrame.from_dict(
|
|
{
|
|
"A": col_a,
|
|
"B": col_b,
|
|
"C": col_c,
|
|
"B_normalized": processed_col_b,
|
|
"C_normalized": processed_col_c,
|
|
}
|
|
).astype(out_df.dtypes.to_dict())
|
|
|
|
pd.testing.assert_frame_equal(out_df, expected_df, check_like=True)
|
|
|
|
|
|
def test_normalizer_serialization():
|
|
"""Test Normalizer serialization and deserialization functionality."""
|
|
from ray.data.preprocessor import SerializablePreprocessorBase
|
|
|
|
# Create normalizer with test data
|
|
normalizer = Normalizer(columns=["A", "B"], norm="l1")
|
|
|
|
# Serialize using CloudPickle
|
|
serialized = normalizer.serialize()
|
|
|
|
# Verify it's binary CloudPickle format
|
|
assert isinstance(serialized, bytes)
|
|
assert serialized.startswith(SerializablePreprocessorBase.MAGIC_CLOUDPICKLE)
|
|
|
|
# Deserialize
|
|
deserialized = Normalizer.deserialize(serialized)
|
|
|
|
# Verify type and field values
|
|
assert isinstance(deserialized, Normalizer)
|
|
assert deserialized.columns == ["A", "B"]
|
|
assert deserialized.norm == "l1"
|
|
assert deserialized.output_columns == ["A", "B"]
|
|
|
|
# Verify it works correctly
|
|
df = pd.DataFrame({"A": [3.0, 4.0], "B": [4.0, 3.0]})
|
|
result = deserialized.transform_batch(df)
|
|
|
|
# For l1 norm, values should sum to 1 for each row
|
|
assert abs(result["A"][0] + result["B"][0] - 1.0) < 1e-10
|
|
assert abs(result["A"][1] + result["B"][1] - 1.0) < 1e-10
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
sys.exit(pytest.main(["-sv", __file__]))
|