scikit-learn-intelex/sklearnex/spmd/basic_statistics/tests/test_basic_statistics_spmd.py at 676fead0e1ea2bab918a3bd60db0a30d54800286 · uxlfoundation/scikit-learn-intelex · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
# ==============================================================================
# Copyright 2024 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================

import numpy as np
import pytest
from numpy.testing import assert_allclose

from onedal.basic_statistics.tests.utils import options_and_tests
from onedal.tests.utils._dataframes_support import (
    _as_numpy,
    _convert_to_dataframe,
    get_dataframes_and_queues,
)
from sklearnex import config_context
from sklearnex.tests.utils.spmd import (
    _generate_statistic_data,
    _get_local_tensor,
    _mpi_libs_and_gpu_available,
)


@pytest.mark.skipif(
    not _mpi_libs_and_gpu_available,
    reason="GPU device and MPI libs required for test",
)
@pytest.mark.parametrize(
    "dataframe,queue",
    get_dataframes_and_queues(dataframe_filter_="dpnp,torch", device_filter_="gpu"),
)
@pytest.mark.mpi
def test_basic_stats_spmd_gold(dataframe, queue):
    # Import spmd and batch algo
    from onedal.basic_statistics import BasicStatistics as BasicStatistics_Batch
    from sklearnex.spmd.basic_statistics import BasicStatistics as BasicStatistics_SPMD

    # Create gold data and convert to dataframe
    data = np.array(
        [
            [0.0, 0.0, 0.0],
            [0.0, 1.0, 2.0],
            [0.0, 2.0, 4.0],
            [0.0, 3.0, 8.0],
            [0.0, 4.0, 16.0],
            [0.0, 5.0, 32.0],
            [0.0, 6.0, 64.0],
        ]
    )

    local_dpt_data = _convert_to_dataframe(
        _get_local_tensor(data), sycl_queue=queue, target_df=dataframe
    )

    # Ensure results of batch algo match spmd
    spmd = BasicStatistics_SPMD()
    spmd_result = spmd.fit(local_dpt_data)
    batch_result = BasicStatistics_Batch().fit(data)

    for option in options_and_tests:
        attr = option + "_"
        assert_allclose(getattr(spmd_result, attr), getattr(batch_result, attr))


@pytest.mark.skipif(
    not _mpi_libs_and_gpu_available,
    reason="GPU device and MPI libs required for test",
)
@pytest.mark.parametrize("n_samples", [100, 10000])
@pytest.mark.parametrize("n_features", [10, 100])
@pytest.mark.parametrize("dtype", [np.float32, np.float64])
@pytest.mark.parametrize(
    "dataframe,queue",
    get_dataframes_and_queues(dataframe_filter_="dpnp,torch", device_filter_="gpu"),
)
@pytest.mark.parametrize("array_api_dispatch", [True, False])
@pytest.mark.mpi
def test_basic_stats_spmd_synthetic(
    n_samples, n_features, dataframe, queue, dtype, array_api_dispatch
):
    # Import spmd and batch algo
    from onedal.basic_statistics import BasicStatistics as BasicStatistics_Batch
    from sklearnex.spmd.basic_statistics import BasicStatistics as BasicStatistics_SPMD

    # Generate data and convert to dataframe
    data = _generate_statistic_data(n_samples, n_features, dtype=dtype)

    local_dpt_data = _convert_to_dataframe(
        _get_local_tensor(data), sycl_queue=queue, target_df=dataframe
    )

    # Ensure results of batch algo match spmd
    # Configure array API dispatch status for spmd estimator
    with config_context(array_api_dispatch=array_api_dispatch):
        spmd_result = BasicStatistics_SPMD().fit(local_dpt_data)
    batch_result = BasicStatistics_Batch().fit(data)

    tol = 1e-5 if dtype == np.float32 else 1e-7
    for option in options_and_tests:
        attr = option + "_"
        assert_allclose(
            _as_numpy(getattr(spmd_result, attr)),
            _as_numpy(getattr(batch_result, attr)),
            atol=tol,
            rtol=tol,
        )