test_carbonara.py 文件源码

def test_from_timeseries(self):
        sampling = numpy.timedelta64(5, 's')
        points = 100000
        ts = carbonara.TimeSerie.from_data(
            timestamps=list(map(datetime.datetime.utcfromtimestamp,
                                six.moves.range(points))),
            values=list(six.moves.range(points)))
        agg = self._resample(ts, sampling, 'mean')

        split = [t[1] for t in list(agg.split())]

        self.assertEqual(agg,
                         carbonara.AggregatedTimeSerie.from_timeseries(
                             split,
                             sampling=agg.sampling,
                             max_size=agg.max_size,
                             aggregation_method=agg.aggregation_method))