def test_ind_to_sub_rdd(self):
        dataLocal = map(lambda x: (x, array([1.0])), range(1, 13))

        data = Series(self.sc.parallelize(dataLocal))
        subs = data.indToSub(dims=[2, 3, 2]).keys().collect()
        assert(allclose(subs, array([(1, 1, 1), (2, 1, 1), (1, 2, 1), (2, 2, 1), (1, 3, 1), (2, 3, 1),
                                     (1, 1, 2), (2, 1, 2), (1, 2, 2), (2, 2, 2), (1, 3, 2), (2, 3, 2)])))

 def test_to_row_matrix(self):
     from thunder.rdds.matrices import RowMatrix
     rdd = self.sc.parallelize([(0, array([4, 5, 6, 7])), (1, array([8, 9, 10, 11]))])
     data = Series(rdd)
     mat = data.toRowMatrix()
     assert(isinstance(mat, RowMatrix))
     assert(mat.nrows == 2)
     assert(mat.ncols == 4)

    def test_sub_to_ind_rdd(self):
        subs = [(1, 1, 1), (2, 1, 1), (1, 2, 1), (2, 2, 1), (1, 3, 1), (2, 3, 1),
                (1, 1, 2), (2, 1, 2), (1, 2, 2), (2, 2, 2), (1, 3, 2), (2, 3, 2)]
        dataLocal = map(lambda x: (x, array([1.0])), subs)

        data = Series(self.sc.parallelize(dataLocal))
        inds = array(data.subToInd().keys().collect())
        assert(allclose(inds, array(range(1, 13))))

    def test_round_trip_rdd(self):
        subs = [(1, 1, 1), (2, 1, 1), (1, 2, 1), (2, 2, 1), (1, 3, 1), (2, 3, 1),
                (1, 1, 2), (2, 1, 2), (1, 2, 2), (2, 2, 2), (1, 3, 2), (2, 3, 2)]
        dataLocal = map(lambda x: (x, array([1.0])), subs)

        data = Series(self.sc.parallelize(dataLocal))
        start = data.keys().collect()
        stop = data.subToInd().indToSub().keys().collect()
        assert(allclose(array(start), array(stop)))

 def test_select(self):
     rdd = self.sc.parallelize([(0, array([4, 5, 6, 7])), (1, array([8, 9, 10, 11]))])
     data = Series(rdd, index=['label1', 'label2', 'label3', 'label4'])
     selection1 = data.select(['label1'])
     assert(allclose(selection1.first()[1], 4))
     selection1 = data.select('label1')
     assert(allclose(selection1.first()[1], 4))
     selection2 = data.select(['label1', 'label2'])
     assert(allclose(selection2.first()[1], array([4, 5])))

 def test_standardization_axis1(self):
     rdd = self.sc.parallelize([(0, array([1, 2], dtype='float16')), (0, array([3, 4], dtype='float16'))])
     data = Series(rdd, dtype='float16')
     centered = data.center(1)
     standardized = data.standardize(1)
     zscored = data.zscore(1)
     assert(allclose(centered.first()[1], array([-1, -1]), atol=1e-3))
     assert(allclose(standardized.first()[1], array([1, 2]), atol=1e-3))
     assert(allclose(zscored.first()[1], array([-1, -1]), atol=1e-3))

 def test_squelch(self):
     rdd = self.sc.parallelize([(0, array([1, 2])), (0, array([3, 4]))])
     data = Series(rdd)
     squelched = data.squelch(5)
     assert(allclose(squelched.collectValuesAsArray(), [[0, 0], [0, 0]]))
     squelched = data.squelch(3)
     assert(allclose(squelched.collectValuesAsArray(), [[0, 0], [3, 4]]))
     squelched = data.squelch(1)
     assert(allclose(squelched.collectValuesAsArray(), [[1, 2], [3, 4]]))

 def test_correlate(self):
     rdd = self.sc.parallelize([(0, array([1, 2, 3, 4, 5]))])
     data = Series(rdd)
     sig1 = [4, 5, 6, 7, 8]
     corr = data.correlate(sig1).values().collect()
     assert(allclose(corr[0], 1))
     sig12 = [[4, 5, 6, 7, 8], [8, 7, 6, 5, 4]]
     corrs = data.correlate(sig12).values().collect()
     assert(allclose(corrs[0], [1, -1]))

 def test_standardization_axis0(self):
     rdd = self.sc.parallelize([(0, array([1, 2, 3, 4, 5], dtype='float16'))])
     data = Series(rdd, dtype='float16')
     centered = data.center(0)
     standardized = data.standardize(0)
     zscored = data.zscore(0)
     assert(allclose(centered.first()[1], array([-2, -1, 0, 1, 2]), atol=1e-3))
     assert(allclose(standardized.first()[1], array([0.70710,  1.41421,  2.12132,  2.82842,  3.53553]), atol=1e-3))
     assert(allclose(zscored.first()[1], array([-1.41421, -0.70710,  0,  0.70710,  1.41421]), atol=1e-3))

 def test_normalization(self):
     rdd = self.sc.parallelize([(0, array([1, 2, 3, 4, 5], dtype='float16'))])
     data = Series(rdd, dtype='float16')
     out = data.normalize('percentile')
     # check that _dtype has been set properly *before* calling first(), b/c first() will update this
     # value even if it hasn't been correctly set
     assert_equals('float16', str(out._dtype))
     vals = out.first()[1]
     assert_equals('float16', str(vals.dtype))
     assert(allclose(vals, array([-0.42105,  0.10526,  0.63157,  1.15789,  1.68421]), atol=1e-3))

    def test_toImages(self):
        from thunder.rdds.images import Images
        rdd = self.sc.parallelize([((0, 0), array([1])), ((0, 1), array([2])),
                                   ((1, 0), array([3])), ((1, 1), array([4]))])
        data = Series(rdd)
        imgs = data.toImages()
        assert(isinstance(imgs, Images))

        im = imgs.values().first()
        assert(allclose(im, [[1, 2], [3, 4]]))

 def test_subset(self):
     rdd = self.sc.parallelize([(0, array([1, 5], dtype='float16')),
                                (0, array([1, 10], dtype='float16')),
                                (0, array([1, 15], dtype='float16'))])
     data = Series(rdd)
     assert_equal(len(data.subset(3, stat='min', thresh=0)), 3)
     assert_array_equal(data.subset(1, stat='max', thresh=10), [[1, 15]])
     assert_array_equal(data.subset(1, stat='mean', thresh=6), [[1, 15]])
     assert_array_equal(data.subset(1, stat='std', thresh=6), [[1, 15]])
     assert_array_equal(data.subset(1, thresh=6), [[1, 15]])

 def test_correlate(self):
     rdd = self.sc.parallelize([(0, array([1, 2, 3, 4, 5], dtype='float16'))])
     data = Series(rdd, dtype='float16')
     sig1 = [4, 5, 6, 7, 8]
     corrData = data.correlate(sig1)
     assert_equals('float64', corrData._dtype)
     corr = corrData.values().collect()
     assert(allclose(corr[0], 1))
     sig12 = [[4, 5, 6, 7, 8], [8, 7, 6, 5, 4]]
     corrs = data.correlate(sig12).values().collect()
     assert(allclose(corrs[0], [1, -1]))

 def test_normalization_bymean(self):
     rdd = self.sc.parallelize([(0, array([1, 2, 3, 4, 5], dtype='float16'))])
     data = Series(rdd, dtype='float16')
     out = data.normalize('mean')
     # check that _dtype has been set properly *before* calling first(), b/c first() will update this
     # value even if it hasn't been correctly set
     assert_equals('float16', str(out._dtype))
     vals = out.first()[1]
     assert_equals('float16', str(vals.dtype))
     assert(allclose(out.first()[1],
                     array([-0.64516,  -0.32258,  0.0,  0.32258,  0.64516]), atol=1e-3))

    def test_query_linear_singleton(self):
        data_local = [
            ((1,), array([1.0, 2.0, 3.0])),
            ((2,), array([2.0, 2.0, 4.0])),
            ((3,), array([4.0, 2.0, 1.0]))
        ]

        data = Series(self.sc.parallelize(data_local))

        inds = array([array([1, 2])])
        keys, values = data.query(inds)
        assert(allclose(values[0, :], array([1.5, 2., 3.5])))

 def test_seriesStats(self):
     rdd = self.sc.parallelize([(0, array([1, 2, 3, 4, 5]))])
     data = Series(rdd)
     assert(allclose(data.seriesMean().first()[1], 3.0))
     assert(allclose(data.seriesSum().first()[1], 15.0))
     assert(allclose(data.seriesMedian().first()[1], 3.0))
     assert(allclose(data.seriesStdev().first()[1], 1.4142135))
     assert(allclose(data.seriesStat('mean').first()[1], 3.0))
     assert(allclose(data.seriesStats().select('mean').first()[1], 3.0))
     assert(allclose(data.seriesStats().select('count').first()[1], 5))
     assert(allclose(data.seriesPercentile(25).first()[1], 2.0))
     assert(allclose(data.seriesPercentile((25, 75)).first()[1], array([2.0, 4.0])))

    def test_meanByFixedLength(self):
        rdd = self.sc.parallelize([((0,), array([0, 1, 2, 3, 4, 5, 6, 7], dtype='float16'))])
        data = Series(rdd)

        test1 = data.meanByFixedLength(4)
        assert(test1.keys().collect() == [(0,)])
        assert(allclose(test1.index, array([0, 1, 2, 3])))
        assert(allclose(test1.values().collect(), [[2, 3, 4, 5]]))

        test2 = data.meanByFixedLength(2)
        assert(test2.keys().collect() == [(0,)])
        assert(allclose(test2.index, array([0, 1])))
        assert(allclose(test2.values().collect(), [[3, 4]]))

    def test_query_subscripts(self):
        dataLocal = [
            ((1, 1), array([1.0, 2.0, 3.0])),
            ((2, 1), array([2.0, 2.0, 4.0])),
            ((1, 2), array([4.0, 2.0, 1.0]))
        ]

        data = Series(self.sc.parallelize(dataLocal))

        inds = array([array([1, 2]), array([3])])
        keys, values = data.query(inds)
        assert(allclose(values[0, :], array([1.5, 2., 3.5])))
        assert(allclose(values[1, :], array([4.0, 2.0, 1.0])))

    def test_toRowMatrix(self):
        from thunder.rdds.matrices import RowMatrix
        rdd = self.sc.parallelize([(0, array([4, 5, 6, 7])), (1, array([8, 9, 10, 11]))])
        data = Series(rdd)
        mat = data.toRowMatrix()
        assert(isinstance(mat, RowMatrix))
        assert(mat.nrows == 2)
        assert(mat.ncols == 4)

        # check a basic operation from superclass
        newmat = mat.applyValues(lambda x: x + 1)
        out = newmat.collectValuesAsArray()
        assert(array_equal(out, array([[5, 6, 7, 8], [9, 10, 11, 12]])))

    def test_index_setter_getter(self):
        dataLocal = [
            ((1,), array([1.0, 2.0, 3.0])),
            ((2,), array([2.0, 2.0, 4.0])),
            ((3,), array([4.0, 2.0, 1.0]))
        ]
        data = Series(self.sc.parallelize(dataLocal))

        assert_true(array_equal(data.index, array([0, 1, 2])))
        data.index = [3, 2, 1]
        assert_true(data.index == [3, 2, 1])

        def setIndex(data, idx):
            data.index = idx

        assert_raises(ValueError, setIndex, data, 5)
        assert_raises(ValueError, setIndex, data, [1, 2])