As part of creating a software defined receiver and decoder, I want to find sync pulses in weather satellite data. The data is first digitized to a 0-1 array.
The code below is working. The convolution method is sufficiently fast. The slow method is not.
I was wondering if there was a more efficient or "Pythonic" way to do it.
def findPulseSlow(data, pulse):
# find indexes of occurrences of pulse in data
# data and pulse are numpy arrays of 0-1 elements
# tolerate up to 3 incorrect values
# takes 2 minutes per Mb on AMD FX8150
maxerr=3
return np.fromiter( \
( i for i in range(0,len(data)-len(pulse)) \
if sum(np.logical_xor(data[i:(i+len(pulse))], pulse))<=maxerr ),\
np.int)
def findPulseConvolve2(data,pulse):
maxerr=3
required = len(pulse)-maxerr
doubleConvolution = np.convolve(data,pulse[::-1])+\
np.convolve(1-data,1-pulse[::-1])
return 1-len(pulse)+np.where(doubleConvolution>=required)[0]
Here is a complete test framework:
testPulse.py
import numpy as np
NOAAsyncA=np.concatenate( ([4*[0],7*[1,1,0,0],8*[0]]) )
def getFixedTestData():
data = np.zeros(1000)
syncL = len(NOAAsyncA)
data[12:(12+syncL)]=NOAAsyncA
data[257:(257+syncL)]=NOAAsyncA
data[900:(900+syncL)]=NOAAsyncA
return ([12,257,900], data)
def getRandomTestData():
# create random 1 Megabit data, insert 10 NOAA sync pulses in data
# return correct pulse start locations, data
global NOAAsyncA
data = np.random.randint(2, size=1000000)
# make an ascending list of random spots to insert the pulse
insertAidx = np.sort(np.random.randint(1000000, size=10))
# insert the pulses
for l in insertAidx:
data[l:(l+len(NOAAsyncA))]=NOAAsyncA
# return the pulse locations and the data
return (insertAidx, data)
def findPulseSlow(data, pulse):
# find indexes of occurrences of pulse in data
# data and pulse are numpy arrays of 0-1 elements
# tolerate up to 3 incorrect values
# takes 2 minutes per Mb on AMD FX8150
maxerr=3
return np.fromiter( \
( i for i in range(0,len(data)-len(pulse)) \
if sum(np.logical_xor(data[i:(i+len(pulse))], pulse))<=maxerr ),\
np.int)
def findPulseConvolve1(data, pulse):
# tests for locations of 1's only. Bad idea.
# max error count
maxerr=0
required = sum(pulse)-maxerr
return 1-len(pulse)+np.where(np.convolve(data,pulse[::-1])>=required)[0]
def findPulseConvolve2(data,pulse):
maxerr=3
required = len(pulse)-maxerr
doubleConvolution = np.convolve(data,pulse[::-1])+\
np.convolve(1-data,1-pulse[::-1])
return 1-len(pulse)+np.where(doubleConvolution>=required)[0]
def Test(name,generator,finder):
global NOAAsyncA
(correctLocations, data) = generator()
syncAidx = finder(data, NOAAsyncA)
passedTest = False
if len(syncAidx)==len(correctLocations):
if (syncAidx==correctLocations).all():
passedTest = True
print "Test: "+name
print "Test Result: "+str(passedTest)
print "found syncA at: "+str(syncAidx)
print "should be at: "+str(correctLocations)
return passedTest
Test("fixed findPulseConvolve1", getFixedTestData, findPulseConvolve1)
Test("fixed findPulseConvolve2", getFixedTestData, findPulseConvolve2)
Test("random findPulseConvolve1", getRandomTestData, findPulseConvolve1)
Test("random findPulseConvolve2", getRandomTestData, findPulseConvolve2)
