python类interpolate()的实例源码

def __init__(self,cqt,Ls):
        from scipy.interpolate import interp1d
        self.intp = interp1d(np.linspace(0,Ls,len(cqt)),cqt)

def moveout_correction(self):
   ####################################################################################
      '''
      Moveout correction relative to a reference ray parameter of 6.4 s/deg.  This stretches
      the time axis for smaller ray parameters (larger epicentral distances), and 
      shrinks the time axis for larger ray parameters (smaller epicentral distances).

      #NOTE 3-16-16, Moveout correction doesn't work properly... The time axis seems
      to be stretching in the opposite way that it should.
      '''
      p = self.slowness_table[:,0]
      s = self.slowness_table[:,1]

      #interpolate with np.interp. make sure values in the first vector are increasing
      scale = np.interp(self.ray_param,p[::-1],s[::-1])
      #print "ray parameter, scale = ",self.ray_param,scale

      #scale the receiver function and interpolate new time axis
      new_time = self.time * scale

      f        = interp1d(new_time,self.rf_st[0].data,bounds_error=False,fill_value=0)
      self.rf_st[0].data = f(self.time)

      f        = interp1d(new_time,self.rf_st[1].data,bounds_error=False,fill_value=0)
      self.rf_st[1].data = f(self.time)

      f        = interp1d(new_time,self.rf_st[2].data,bounds_error=False,fill_value=0)
      self.rf_st[2].data = f(self.time)

   ####################################################################################

def rf_moveout_correction(rf_trace,table='None'):
####################################################################################
   '''
   takes a receiver function trace in the rfh5 format. if table = 'None', the slowness
   lookup table will be read.  alternatively if calling this function repeatedly, pass 
   the table as an argument to avoid repetative i/o.
   '''

   if table == 'None':
      slowness_table = np.loadtxt('/geo/work10/romaguir/seismology/seis_tools/seispy/slowness_table.dat')
   else:
      slowness_table = table

   p = slowness_table[:,0]
   s = slowness_table[:,1]

   #interpolate with np.interp. make sure values in the first vector are increasing
   scale = np.interp(rf_trace.stats.ray_param,p[::-1],s[::-1])

   #scale the receiver function and interpolate new time axis
   time = np.linspace(0,rf_trace.stats.delta*rf_trace.stats.npts,rf_trace.stats.npts)
   new_time = time * scale

   f        = interp1d(new_time,rf_trace.data,bounds_error=False,fill_value=0)
   rf_mvc   = f(time)
   rf_trace.data = rf_mvc

   return rf_trace

####################################################################################

def _skip_if_no_pchip():
    try:
        from scipy.interpolate import pchip_interpolate  # noqa
    except ImportError:
        raise nose.SkipTest('scipy.interpolate.pchip missing')

# ----------------------------------------------------------------------
# Generic types test cases

def test_interp_regression(self):
        tm._skip_if_no_scipy()
        _skip_if_no_pchip()

        ser = Series(np.sort(np.random.uniform(size=100)))

        # interpolate at new_index
        new_index = ser.index.union(Index([49.25, 49.5, 49.75, 50.25, 50.5,
                                           50.75]))
        interp_s = ser.reindex(new_index).interpolate(method='pchip')
        # does not blow up, GH5977
        interp_s[49:51]

def test_interpolate_corners(self):
        s = Series([np.nan, np.nan])
        assert_series_equal(s.interpolate(), s)

        s = Series([]).interpolate()
        assert_series_equal(s.interpolate(), s)

        tm._skip_if_no_scipy()
        s = Series([np.nan, np.nan])
        assert_series_equal(s.interpolate(method='polynomial', order=1), s)

        s = Series([]).interpolate()
        assert_series_equal(s.interpolate(method='polynomial', order=1), s)

def test_nan_interpolate(self):
        s = Series([0, 1, np.nan, 3])
        result = s.interpolate()
        expected = Series([0., 1., 2., 3.])
        assert_series_equal(result, expected)

        tm._skip_if_no_scipy()
        result = s.interpolate(method='polynomial', order=1)
        assert_series_equal(result, expected)

def test_nan_irregular_index(self):
        s = Series([1, 2, np.nan, 4], index=[1, 3, 5, 9])
        result = s.interpolate()
        expected = Series([1., 2., 3., 4.], index=[1, 3, 5, 9])
        assert_series_equal(result, expected)

def test_nan_str_index(self):
        s = Series([0, 1, 2, np.nan], index=list('abcd'))
        result = s.interpolate()
        expected = Series([0., 1., 2., 2.], index=list('abcd'))
        assert_series_equal(result, expected)

def test_interp_quad(self):
        tm._skip_if_no_scipy()
        sq = Series([1, 4, np.nan, 16], index=[1, 2, 3, 4])
        result = sq.interpolate(method='quadratic')
        expected = Series([1., 4., 9., 16.], index=[1, 2, 3, 4])
        assert_series_equal(result, expected)

def test_interp_scipy_basic(self):
        tm._skip_if_no_scipy()
        s = Series([1, 3, np.nan, 12, np.nan, 25])
        # slinear
        expected = Series([1., 3., 7.5, 12., 18.5, 25.])
        result = s.interpolate(method='slinear')
        assert_series_equal(result, expected)

        result = s.interpolate(method='slinear', downcast='infer')
        assert_series_equal(result, expected)
        # nearest
        expected = Series([1, 3, 3, 12, 12, 25])
        result = s.interpolate(method='nearest')
        assert_series_equal(result, expected.astype('float'))

        result = s.interpolate(method='nearest', downcast='infer')
        assert_series_equal(result, expected)
        # zero
        expected = Series([1, 3, 3, 12, 12, 25])
        result = s.interpolate(method='zero')
        assert_series_equal(result, expected.astype('float'))

        result = s.interpolate(method='zero', downcast='infer')
        assert_series_equal(result, expected)
        # quadratic
        expected = Series([1, 3., 6.769231, 12., 18.230769, 25.])
        result = s.interpolate(method='quadratic')
        assert_series_equal(result, expected)

        result = s.interpolate(method='quadratic', downcast='infer')
        assert_series_equal(result, expected)
        # cubic
        expected = Series([1., 3., 6.8, 12., 18.2, 25.])
        result = s.interpolate(method='cubic')
        assert_series_equal(result, expected)

def test_interp_limit_forward(self):
        s = Series([1, 3, np.nan, np.nan, np.nan, 11])

        # Provide 'forward' (the default) explicitly here.
        expected = Series([1., 3., 5., 7., np.nan, 11.])

        result = s.interpolate(method='linear', limit=2,
                               limit_direction='forward')
        assert_series_equal(result, expected)

        result = s.interpolate(method='linear', limit=2,
                               limit_direction='FORWARD')
        assert_series_equal(result, expected)

def test_interp_limit_bad_direction(self):
        s = Series([1, 3, np.nan, np.nan, np.nan, 11])

        self.assertRaises(ValueError, s.interpolate, method='linear', limit=2,
                          limit_direction='abc')

        # raises an error even if no limit is specified.
        self.assertRaises(ValueError, s.interpolate, method='linear',
                          limit_direction='abc')

def test_interp_limit_direction(self):
        # These tests are for issue #9218 -- fill NaNs in both directions.
        s = Series([1, 3, np.nan, np.nan, np.nan, 11])

        expected = Series([1., 3., np.nan, 7., 9., 11.])
        result = s.interpolate(method='linear', limit=2,
                               limit_direction='backward')
        assert_series_equal(result, expected)

        expected = Series([1., 3., 5., np.nan, 9., 11.])
        result = s.interpolate(method='linear', limit=1,
                               limit_direction='both')
        assert_series_equal(result, expected)

        # Check that this works on a longer series of nans.
        s = Series([1, 3, np.nan, np.nan, np.nan, 7, 9, np.nan, np.nan, 12,
                    np.nan])

        expected = Series([1., 3., 4., 5., 6., 7., 9., 10., 11., 12., 12.])
        result = s.interpolate(method='linear', limit=2,
                               limit_direction='both')
        assert_series_equal(result, expected)

        expected = Series([1., 3., 4., np.nan, 6., 7., 9., 10., 11., 12., 12.])
        result = s.interpolate(method='linear', limit=1,
                               limit_direction='both')
        assert_series_equal(result, expected)

def test_interp_limit_to_ends(self):
        # These test are for issue #10420 -- flow back to beginning.
        s = Series([np.nan, np.nan, 5, 7, 9, np.nan])

        expected = Series([5., 5., 5., 7., 9., np.nan])
        result = s.interpolate(method='linear', limit=2,
                               limit_direction='backward')
        assert_series_equal(result, expected)

        expected = Series([5., 5., 5., 7., 9., 9.])
        result = s.interpolate(method='linear', limit=2,
                               limit_direction='both')
        assert_series_equal(result, expected)

def test_interp_multiIndex(self):
        idx = MultiIndex.from_tuples([(0, 'a'), (1, 'b'), (2, 'c')])
        s = Series([1, 2, np.nan], index=idx)

        expected = s.copy()
        expected.loc[2] = 2
        result = s.interpolate()
        assert_series_equal(result, expected)

        tm._skip_if_no_scipy()
        with tm.assertRaises(ValueError):
            s.interpolate(method='polynomial', order=1)

def test_interp_nonmono_raise(self):
        tm._skip_if_no_scipy()
        s = Series([1, np.nan, 3], index=[0, 2, 1])
        with tm.assertRaises(ValueError):
            s.interpolate(method='krogh')

def test_interp_datetime64(self):
        tm._skip_if_no_scipy()
        df = Series([1, np.nan, 3], index=date_range('1/1/2000', periods=3))
        result = df.interpolate(method='nearest')
        expected = Series([1., 1., 3.],
                          index=date_range('1/1/2000', periods=3))
        assert_series_equal(result, expected)

def test_interp_limit_no_nans(self):
        # GH 7173
        s = pd.Series([1., 2., 3.])
        result = s.interpolate(limit=1)
        expected = s
        assert_series_equal(result, expected)

def test_interp_combo(self):
        df = DataFrame({'A': [1., 2., np.nan, 4.],
                        'B': [1, 4, 9, np.nan],
                        'C': [1, 2, 3, 5],
                        'D': list('abcd')})

        result = df['A'].interpolate()
        expected = Series([1., 2., 3., 4.], name='A')
        assert_series_equal(result, expected)

        result = df['A'].interpolate(downcast='infer')
        expected = Series([1, 2, 3, 4], name='A')
        assert_series_equal(result, expected)