def get_timepixel_g2( oned_count ):
n = len( oned_count )
norm = ( np.arange( n, 0, -1) *
np.array( [np.average(oned_count[i:]) for i in range( n )] ) *
np.array( [np.average(oned_count[0:n-i]) for i in range( n )] ) )
return np.correlate(oned_count, oned_count, mode = 'full')[-n:]/norm
#########################################
python类correlate()的实例源码
def get_timepixel_g2( oned_count ):
n = len( oned_count )
norm = ( np.arange( n, 0, -1) *
np.array( [np.average(oned_count[i:]) for i in range( n )] ) *
np.array( [np.average(oned_count[0:n-i]) for i in range( n )] ) )
return np.correlate(oned_count, oned_count, mode = 'full')[-n:]/norm
#########################################
def get_timepixel_g2( oned_count ):
n = len( oned_count )
norm = ( np.arange( n, 0, -1) *
np.array( [np.average(oned_count[i:]) for i in range( n )] ) *
np.array( [np.average(oned_count[0:n-i]) for i in range( n )] ) )
return np.correlate(oned_count, oned_count, mode = 'full')[-n:]/norm
#########################################
def gi_correlation(self, p_gi_len, p_symbol_len, signal):
''' Mode Erkennung '''
corr_delay = p_symbol_len - p_gi_len
# Verzögertes Signal / Delay
corr_A = signal[0:p_gi_len-1];
corr_B = signal[corr_delay:len(signal)-1]
# Normierung
corr_A = corr_A / np.sqrt( np.sum(np.square(np.abs(corr_A))) )
corr_B = corr_B / np.sqrt( np.sum(np.square(np.abs(corr_B))) )
# Korrelation
erg_corr = np.correlate(corr_A,corr_B)
return erg_corr[0]
def crosscorr(data, fb, fs, pairs=None):
"""
Parameters
----------
Returns
-------
"""
n_channels, n_samples = np.shape(data)
filtered, _, _ = analytic_signal(data, fb, fs)
r = np.zeros([n_channels, n_channels], dtype=np.float32)
for i in range(n_channels):
for ii in range(n_channels):
r[i, ii] = np.correlate(filtered[i, ], filtered[ii, ], mode='valid')
return r
def estimate_range(tx,rx,fs,quiet=False):
"""
tx: the known, noise-free, undelayed transmit signal (bistatic radars agree beforehand on the psuedorandom sequence)
rx: the noisy, corrupted, interference, jammed signal to estimate distance from
fs: baseband sample frequency
"""
Rxy = np.correlate(tx, rx, 'full')
lags = np.arange(Rxy.size) - Rxy.size // 2
pklag = lags[Rxy.argmax()]
distest_m = -pklag / fs / 2 * c
mR = abs(Rxy) # magnitude of complex cross-correlation
if not quiet and figure is not None:
ax = figure().gca()
ax.plot(lags,mR)
ax.plot(pklag,mR[mR.argmax()], color='red', marker='*')
ax.set_title('cross-correlation of receive waveform with transmit waveform')
ax.set_ylabel('$|R_{xy}|$')
ax.set_xlabel('lags')
ax.set_xlim(pklag-100,pklag+100)
return distest_m
def procchunk(rx, tx, P:dict):
if P['rxfn'] is not None:
rx = scipy.signal.resample_poly(rx,
P['resample'].numerator,
P['resample'].denominator)
fs = P['txfs']
# %% resamples parameters
NrxPRI = int(fs * P['pri']) # Number of RX samples per PRI (resampled)
assert NrxPRI >= tx.size,'PRI must be longer than chirp length!'
NrxChirp = rx.size // NrxPRI # number of complete PRIs received in this data
assert NrxChirp == P['Nchirp']
Rxy = 0.
for i in range(P['Nchirp']):
r = rx[i*NrxPRI:(i+1)*NrxPRI]
Rxy += np.correlate(tx, r,'same')
if P['verbose']:
plotxcor(Rxy, fs)
draw()
pause(0.1)
return Rxy
def find_audio_period(aclip, t_min=.1, t_max=2, t_res=.01):
""" Finds the period, in seconds of an audioclip.
The beat is then given by bpm = 60/T
t_min and _tmax are bounds for the returned value, t_res
is the numerical precision
"""
chunksize = int(t_res*aclip.fps)
chunk_duration = 1.0*chunksize/aclip.fps
# v denotes the list of volumes
v = np.array([(c**2).sum() for c in
aclip.iter_chunks(chunksize)])
v = v-v.mean()
corrs = np.correlate(v, v, mode = 'full')[-len(v):]
corrs[:int(t_min/chunk_duration)]=0
corrs[int(t_max/chunk_duration):]=0
return chunk_duration*np.argmax(corrs)
def extractMseq(cover, stego, secret_length, m, tau=1):
u"""Extract secret informations by spread spectrum using m-sequence.
@param cover : cover data (2 dimensional np.ndarray)
@param stego : stego data (2 dimension np.ndarray)
@param secret_length : length of secret information
@param m : M-Sequence
@param tau : embed shift interval
@return secret : extracted secret information
"""
cover = _image2vrctor(cover)
stego = _image2vrctor(stego)
m_length = len(m)
data = stego - cover
data = data[:m_length:tau]
secret_data = correlate(m, data, cycle=CYCLE)
center = ((m_length-1)*2+1)//2
secret_data = secret_data[center:center+secret_length]
secret_data = list(map(_checkData, secret_data))
return secret_data
def _convolve_or_correlate(f, a, v, mode, propagate_mask):
"""
Helper function for ma.correlate and ma.convolve
"""
if propagate_mask:
# results which are contributed to by either item in any pair being invalid
mask = (
f(getmaskarray(a), np.ones(np.shape(v), dtype=np.bool), mode=mode)
| f(np.ones(np.shape(a), dtype=np.bool), getmaskarray(v), mode=mode)
)
data = f(getdata(a), getdata(v), mode=mode)
else:
# results which are not contributed to by any pair of valid elements
mask = ~f(~getmaskarray(a), ~getmaskarray(v))
data = f(filled(a, 0), filled(v, 0), mode=mode)
return masked_array(data, mask=mask)
def plot_correlate():
"""Plots the autocorrelation function computed by numpy.
"""
wave = thinkdsp.read_wave('28042__bcjordan__voicedownbew.wav')
wave.normalize()
segment = wave.segment(start=0.2, duration=0.01)
lags, corrs = autocorr(segment)
corrs2 = numpy.correlate(segment.ys, segment.ys, mode='same')
thinkplot.plot(corrs2)
thinkplot.config(xlabel='lag',
ylabel='correlation',
xlim=[0, len(corrs2)])
thinkplot.save(root='autocorr9')
N = len(corrs2)
half = corrs2[N//2:]
lengths = range(N, N//2, -1)
half /= lengths
half /= half[0]
def find_audio_period(aclip, t_min=.1, t_max=2, t_res=.01):
""" Finds the period, in seconds of an audioclip.
The beat is then given by bpm = 60/T
t_min and _tmax are bounds for the returned value, t_res
is the numerical precision
"""
chunksize = int(t_res*aclip.fps)
chunk_duration = 1.0*chunksize/aclip.fps
# v denotes the list of volumes
v = np.array([(c**2).sum() for c in
aclip.iter_chunks(chunksize)])
v = v-v.mean()
corrs = np.correlate(v, v, mode = 'full')[-len(v):]
corrs[:int(t_min/chunk_duration)]=0
corrs[int(t_max/chunk_duration):]=0
return chunk_duration*np.argmax(corrs)
def find_audio_period(aclip, t_min=.1, t_max=2, t_res=.01):
""" Finds the period, in seconds of an audioclip.
The beat is then given by bpm = 60/T
t_min and _tmax are bounds for the returned value, t_res
is the numerical precision
"""
chunksize = int(t_res*aclip.fps)
chunk_duration = 1.0*chunksize/aclip.fps
# v denotes the list of volumes
v = np.array([(c**2).sum() for c in
aclip.iter_chunks(chunksize)])
v = v-v.mean()
corrs = np.correlate(v, v, mode = 'full')[-len(v):]
corrs[:int(t_min/chunk_duration)]=0
corrs[int(t_max/chunk_duration):]=0
return chunk_duration*np.argmax(corrs)
def estimated_autocorrelation(x):
n = len(x)
variance = x.var()
x = x - x.mean()
r = np.correlate(x, x, mode='full')[-n:]
assert np.allclose(r, np.array([(x[:n - k] * x[-(n - k):]).sum() for k in range(n)]))
result = r / (variance * (np.arange(n, 0, -1)))
return result
def _delta(x, window):
return np.correlate(x, window, mode="same")
def test_float(self):
self._setup(np.float)
z = np.correlate(self.x, self.y, 'full')
assert_array_almost_equal(z, self.z1)
z = np.correlate(self.x, self.y[:-1], 'full')
assert_array_almost_equal(z, self.z1_4)
z = np.correlate(self.y, self.x, 'full')
assert_array_almost_equal(z, self.z2)
z = np.correlate(self.x[::-1], self.y, 'full')
assert_array_almost_equal(z, self.z1r)
z = np.correlate(self.y, self.x[::-1], 'full')
assert_array_almost_equal(z, self.z2r)
z = np.correlate(self.xs, self.y, 'full')
assert_array_almost_equal(z, self.zs)
def test_object(self):
self._setup(Decimal)
z = np.correlate(self.x, self.y, 'full')
assert_array_almost_equal(z, self.z1)
z = np.correlate(self.y, self.x, 'full')
assert_array_almost_equal(z, self.z2)
def test_no_overwrite(self):
d = np.ones(100)
k = np.ones(3)
np.correlate(d, k)
assert_array_equal(d, np.ones(100))
assert_array_equal(k, np.ones(3))
def test_complex(self):
x = np.array([1, 2, 3, 4+1j], dtype=np.complex)
y = np.array([-1, -2j, 3+1j], dtype=np.complex)
r_z = np.array([3-1j, 6, 8+1j, 11+5j, -5+8j, -4-1j], dtype=np.complex)
r_z = r_z[::-1].conjugate()
z = np.correlate(y, x, mode='full')
assert_array_almost_equal(z, r_z)
def align(l1, l2, axis):
if axis == 1: #horizontal alignment, we do not care about the right line end
#cw = min(l2.shape[1],l1.shape[1])
#l1 = l1[:,:cw]
#l2 = l2[:,:cw]
#compute correlation
sc1 = np.sum(l1, axis=1-axis)
sc2 = np.sum(l2, axis=1-axis)
cor = np.correlate(sc1,sc2,"same")
posErr = np.argmax(cor)-sc1.shape[0]/2
#place at right position
if posErr > 0:
l2c = l2.copy()
l2c[:]=0
l2c[:,posErr:] = l2[:,:-posErr]
l2 = l2c
elif posErr < 0:
l1c = l1.copy()
l1c[:]=0
l1c[:,-posErr:] = l1[:,:posErr]
l1=l1c
else: #vertical alignment, we cate about both ends
#compute correlation
sc1 = np.sum(l1, axis=1-axis)
sc2 = np.sum(l2, axis=1-axis)
cor = np.correlate(sc1,sc2,"same")
posErr = np.argmax(cor)-sc1.shape[0]/2
#place at right position
if posErr > 0:
l2c=l2.copy()
l2c[:]=0
l2c[posErr:,:] = l2[:-posErr,:]
l2 = l2c
elif posErr < 0:
l1c=l1.copy()
l1c[:]=0
l1c[-posErr:,:]=l1[:posErr,:]
l1 = l1c
return posErr, l1, l2
def autocorr(x):
x=x-np.mean(x)
y=np.correlate(x,x,mode='full')
return y[y.size/2:]/y[y.size/2]
def time_delay_func_paralel(start, end, outs, multi):
for idx in range(start, end):
print 'locating ...', getpid()
c = numpy.correlate(multi[0, ][:, 0], multi[idx, ][:, 0], "full")
C, I = c.max(0), c.argmax(0)
outs[idx] = ((float(len(c))+1.0)/2.0 - I)/44100.0
def time_delay_func(x, y):
print 'locating ...'
c = numpy.correlate(x[:, 0], y[:, 0], "full")
C, I = c.max(0), c.argmax(0)
out = ((float(len(c))+1.0)/2.0 - I)/44100.0
return out
def fst_delay_snd(fst, snd, samp_rate, max_delay):
# Verify argument shape.
s1, s2 = fst.shape, snd.shape
if len(s1) != 1 or len(s2) != 1 or s1[0] != s2[0]:
raise Exception("Argument shape invalid, in 'fst_delay_snd' function")
half_len = int(s1[0]/2)
a = numpy.array(fst, dtype=numpy.double)
b = numpy.array(snd, dtype=numpy.double)
corr = numpy.correlate(a, b, 'same')
max_pos = numpy.argmax(corr)
# plot(s1[0], samp_rate, a, b, corr)
return corr, (max_pos - half_len) / samp_rate
def fst_delay_snd(fst, snd, samp_rate):
# Verify argument shape.
s1, s2 = fst.shape, snd.shape
if len(s1) != 1 or len(s2) != 1 or s1[0] != s2[0]:
raise Exception("Argument shape invalid, in 'fst_delay_snd' function")
half_len = int(s1[0]/2)
corre = numpy.correlate(fst, snd, 'same')
max_pos = numpy.argmax(corre)
return (max_pos - half_len)/samp_rate
def rolling_mean(X, window_size):
w = 1.0 / window_size * np.ones((window_size))
return np.correlate(X, w, 'valid')
def rolling_mean(X, window_size):
w = 1.0 / window_size * np.ones((window_size))
return np.correlate(X, w, 'valid')
def test_float(self):
self._setup(np.float)
z = np.correlate(self.x, self.y, 'full')
assert_array_almost_equal(z, self.z1)
z = np.correlate(self.x, self.y[:-1], 'full')
assert_array_almost_equal(z, self.z1_4)
z = np.correlate(self.y, self.x, 'full')
assert_array_almost_equal(z, self.z2)
z = np.correlate(self.x[::-1], self.y, 'full')
assert_array_almost_equal(z, self.z1r)
z = np.correlate(self.y, self.x[::-1], 'full')
assert_array_almost_equal(z, self.z2r)
z = np.correlate(self.xs, self.y, 'full')
assert_array_almost_equal(z, self.zs)
def test_object(self):
self._setup(Decimal)
z = np.correlate(self.x, self.y, 'full')
assert_array_almost_equal(z, self.z1)
z = np.correlate(self.y, self.x, 'full')
assert_array_almost_equal(z, self.z2)
def test_no_overwrite(self):
d = np.ones(100)
k = np.ones(3)
np.correlate(d, k)
assert_array_equal(d, np.ones(100))
assert_array_equal(k, np.ones(3))
