# Path: splearn/linear_model/base.py
# class SparkLinearRegression(LinearRegression, SparkLinearModelMixin):
# 
#     """Distributed implementation of sklearn's Linear Regression.
# 
#     Parameters
#     ----------
#     fit_intercept : boolean, optional
#         whether to calculate the intercept for this model. If set
#         to false, no intercept will be used in calculations
#         (e.g. data is expected to be already centered).
#     normalize : boolean, optional, default False
#         If True, the regressors X will be normalized before regression.
#     copy_X : boolean, optional, default True
#         If True, X will be copied; else, it may be overwritten.
#     n_jobs : The number of jobs to use for the computation.
#         If -1 all CPUs are used. This will only provide speedup for
#         n_targets > 1 and sufficient large problems.
# 
#     Attributes
#     ----------
#     coef_ : array, shape (n_features, ) or (n_targets, n_features)
#         Estimated coefficients for the linear regression problem.
#         If multiple targets are passed during the fit (y 2D), this
#         is a 2D array of shape (n_targets, n_features), while if only
#         one target is passed, this is a 1D array of length n_features.
#     intercept_ : array
#         Independent term in the linear model.
# 
#     """
# 
#     def fit(self, Z):
#         """
#         Fit linear model.
# 
#         Parameters
#         ----------
#         Z : DictRDD with (X, y) values
#             X containing numpy array or sparse matrix - The training data
#             y containing the target values
# 
#         Returns
#         -------
#         self : returns an instance of self.
#         """
#         check_rdd(Z, {'X': (sp.spmatrix, np.ndarray)})
#         return self._spark_fit(SparkLinearRegression, Z)
# 
#     def predict(self, X):
#         """Distributed method to predict class labels for samples in X.
# 
#         Parameters
#         ----------
#         X : ArrayRDD containing {array-like, sparse matrix}
#             Samples.
# 
#         Returns
#         -------
#         C : ArrayRDD
#             Predicted class label per sample.
#         """
#         check_rdd(X, (sp.spmatrix, np.ndarray))
#         return self._spark_predict(SparkLinearRegression, X)
# 
#     def to_scikit(self):
#         return self._to_scikit(LinearRegression)
# 
# Path: splearn/utils/testing.py
# def assert_tuple_equal(tpl1, tpl2):
# def assert_multiple_tuples_equal(tpls1, tpls2):
#     def setUp(self):
#     def tearDown(self):
#     def make_blobs(self, centers, n_samples, blocks=-1):
#     def make_regression(self, n_targets, n_samples, blocks=-1):
#     def make_classification(self, n_classes, n_samples, blocks=-1,
#                             nonnegative=False):
#     def make_text_rdd(self, blocks=-1):
#     def make_dense_rdd(self, shape=(1e3, 10), block_size=-1):
#     def make_dense_randint_rdd(self, low, high=None, shape=(1e3, 10),
#                                block_size=-1):
#     def make_dense_range_rdd(self, shape=(1e3, 10), block_size=-1):
#     def make_sparse_rdd(self, shape=(1e3, 10), block_size=-1):
# class SplearnTestCase(unittest.TestCase):
#         Z = DictRDD([X_rdd, y_rdd], columns=('X', 'y'), bsize=blocks)
#             X = np.abs(X)
#         Z = DictRDD([X_rdd, y_rdd], columns=('X', 'y'), bsize=blocks)
#         X = ALL_FOOD_DOCS
#         X = rng.randn(*shape)
#         X = np.random.randint(low, high, size=shape)
#         X = np.arange(np.prod(shape)).reshape(shape)
#         X = sp.rand(shape[0], shape[1], random_state=42, density=0.3)
# 
# Path: splearn/utils/validation.py
# def check_rdd_dtype(rdd, expected_dtype):
#     """Checks if the blocks in the RDD matches the expected types.
# 
#     Parameters:
#     -----------
#     rdd: splearn.BlockRDD
#         The RDD to check
#     expected_dtype: {type, list of types, tuple of types, dict of types}
#         Expected type(s). If the RDD is a DictRDD the parameter type is
#         restricted to dict.
# 
#     Returns:
#     --------
#     accept: bool
#         Returns if the types are matched.
#     """
#     if not isinstance(rdd, BlockRDD):
#         raise TypeError("Expected {0} for parameter rdd, got {1}."
#                         .format(BlockRDD, type(rdd)))
#     if isinstance(rdd, DictRDD):
#         if not isinstance(expected_dtype, dict):
#             raise TypeError('Expected {0} for parameter '
#                             'expected_dtype, got {1}.'
#                             .format(dict, type(expected_dtype)))
#         accept = True
#         types = dict(list(zip(rdd.columns, rdd.dtype)))
#         for key, values in expected_dtype.items():
#             if not isinstance(values, (tuple, list)):
#                 values = [values]
#             accept = accept and types[key] in values
#         return accept
# 
#     if not isinstance(expected_dtype, (tuple, list)):
#         expected_dtype = [expected_dtype]
# 
#     return rdd.dtype in expected_dtype


import numpy as np
from sklearn.linear_model import LinearRegression
from splearn.linear_model import SparkLinearRegression
from splearn.utils.testing import (SplearnTestCase, assert_array_almost_equal,
                                   assert_true)
from splearn.utils.validation import check_rdd_dtype


class TestLinearRegression(SplearnTestCase):

    def test_same_coefs(self):
        X, y, Z = self.make_regression(1, 100000)

        local = LinearRegression()
        dist = SparkLinearRegression()

        local.fit(X, y)
        dist.fit(Z)
        converted = dist.to_scikit()

        assert_array_almost_equal(local.coef_, dist.coef_)
        assert_array_almost_equal(local.intercept_, dist.intercept_)
        assert_array_almost_equal(local.coef_, converted.coef_)
        assert_array_almost_equal(local.intercept_, converted.intercept_)


# Path: splearn/linear_model/base.py
# class SparkLinearRegression(LinearRegression, SparkLinearModelMixin):
# 
#     """Distributed implementation of sklearn's Linear Regression.
# 
#     Parameters
#     ----------
#     fit_intercept : boolean, optional
#         whether to calculate the intercept for this model. If set
#         to false, no intercept will be used in calculations
#         (e.g. data is expected to be already centered).
#     normalize : boolean, optional, default False
#         If True, the regressors X will be normalized before regression.
#     copy_X : boolean, optional, default True
#         If True, X will be copied; else, it may be overwritten.
#     n_jobs : The number of jobs to use for the computation.
#         If -1 all CPUs are used. This will only provide speedup for
#         n_targets > 1 and sufficient large problems.
# 
#     Attributes
#     ----------
#     coef_ : array, shape (n_features, ) or (n_targets, n_features)
#         Estimated coefficients for the linear regression problem.
#         If multiple targets are passed during the fit (y 2D), this
#         is a 2D array of shape (n_targets, n_features), while if only
#         one target is passed, this is a 1D array of length n_features.
#     intercept_ : array
#         Independent term in the linear model.
# 
#     """
# 
#     def fit(self, Z):
#         """
#         Fit linear model.
# 
#         Parameters
#         ----------
#         Z : DictRDD with (X, y) values
#             X containing numpy array or sparse matrix - The training data
#             y containing the target values
# 
#         Returns
#         -------
#         self : returns an instance of self.
#         """
#         check_rdd(Z, {'X': (sp.spmatrix, np.ndarray)})
#         return self._spark_fit(SparkLinearRegression, Z)
# 
#     def predict(self, X):
#         """Distributed method to predict class labels for samples in X.
# 
#         Parameters
#         ----------
#         X : ArrayRDD containing {array-like, sparse matrix}
#             Samples.
# 
#         Returns
#         -------
#         C : ArrayRDD
#             Predicted class label per sample.
#         """
#         check_rdd(X, (sp.spmatrix, np.ndarray))
#         return self._spark_predict(SparkLinearRegression, X)
# 
#     def to_scikit(self):
#         return self._to_scikit(LinearRegression)
# 
# Path: splearn/utils/testing.py
# def assert_tuple_equal(tpl1, tpl2):
# def assert_multiple_tuples_equal(tpls1, tpls2):
#     def setUp(self):
#     def tearDown(self):
#     def make_blobs(self, centers, n_samples, blocks=-1):
#     def make_regression(self, n_targets, n_samples, blocks=-1):
#     def make_classification(self, n_classes, n_samples, blocks=-1,
#                             nonnegative=False):
#     def make_text_rdd(self, blocks=-1):
#     def make_dense_rdd(self, shape=(1e3, 10), block_size=-1):
#     def make_dense_randint_rdd(self, low, high=None, shape=(1e3, 10),
#                                block_size=-1):
#     def make_dense_range_rdd(self, shape=(1e3, 10), block_size=-1):
#     def make_sparse_rdd(self, shape=(1e3, 10), block_size=-1):
# class SplearnTestCase(unittest.TestCase):
#         Z = DictRDD([X_rdd, y_rdd], columns=('X', 'y'), bsize=blocks)
#             X = np.abs(X)
#         Z = DictRDD([X_rdd, y_rdd], columns=('X', 'y'), bsize=blocks)
#         X = ALL_FOOD_DOCS
#         X = rng.randn(*shape)
#         X = np.random.randint(low, high, size=shape)
#         X = np.arange(np.prod(shape)).reshape(shape)
#         X = sp.rand(shape[0], shape[1], random_state=42, density=0.3)
# 
# Path: splearn/utils/validation.py
# def check_rdd_dtype(rdd, expected_dtype):
#     """Checks if the blocks in the RDD matches the expected types.
# 
#     Parameters:
#     -----------
#     rdd: splearn.BlockRDD
#         The RDD to check
#     expected_dtype: {type, list of types, tuple of types, dict of types}
#         Expected type(s). If the RDD is a DictRDD the parameter type is
#         restricted to dict.
# 
#     Returns:
#     --------
#     accept: bool
#         Returns if the types are matched.
#     """
#     if not isinstance(rdd, BlockRDD):
#         raise TypeError("Expected {0} for parameter rdd, got {1}."
#                         .format(BlockRDD, type(rdd)))
#     if isinstance(rdd, DictRDD):
#         if not isinstance(expected_dtype, dict):
#             raise TypeError('Expected {0} for parameter '
#                             'expected_dtype, got {1}.'
#                             .format(dict, type(expected_dtype)))
#         accept = True
#         types = dict(list(zip(rdd.columns, rdd.dtype)))
#         for key, values in expected_dtype.items():
#             if not isinstance(values, (tuple, list)):
#                 values = [values]
#             accept = accept and types[key] in values
#         return accept
# 
#     if not isinstance(expected_dtype, (tuple, list)):
#         expected_dtype = [expected_dtype]
# 
#     return rdd.dtype in expected_dtype


# Path: splearn/linear_model/tests/test_base.py
import numpy as np
from sklearn.linear_model import LinearRegression
from splearn.linear_model import SparkLinearRegression
from splearn.utils.testing import (SplearnTestCase, assert_array_almost_equal,
                                   assert_true)
from splearn.utils.validation import check_rdd_dtype



class TestLinearRegression(SplearnTestCase):

    def test_same_coefs(self):
        X, y, Z = self.make_regression(1, 100000)

        local = LinearRegression()
        dist = SparkLinearRegression()

        local.fit(X, y)
        dist.fit(Z)
        converted = dist.to_scikit()

        assert_array_almost_equal(local.coef_, dist.coef_)
        assert_array_almost_equal(local.intercept_, dist.intercept_)
        assert_array_almost_equal(local.coef_, converted.coef_)
        assert_array_almost_equal(local.intercept_, converted.intercept_)


    def test_same_prediction(self):