Encode data as sound, and retrieve it using a FFT

data2sound.py

class GlobalFormat(object):

	def __init__(self):
		self.id = 0x46464952 # RIFF
		self.size = 0
		self.type = 0x45564157 # WAVE
		
	def as_bin(self):
		return struct.pack("III", self.id, self.size, self.type)
		
class FormatChunk(object):

	def __init__(self):
		self.id = 0x20746d66
		self.size = 0
		self.formatTag = 0
		self.channels = 0
		self.bitsPerSample = 0
		self.samplesPerSec = 0
		self.avgBytesPerSec = 0
		self.blockAlign = 4
		
	def as_bin(self):
		return struct.pack("IIhHIIHH", self.id, self.size, self.formatTag, self.channels, \
			self.samplesPerSec, self.avgBytesPerSec, self.blockAlign, self.bitsPerSample)

import struct
import math
import os.path

infile = "sieve1.py"
filename = "out.wav"
sampleFreq = 8000
denominator = 8
frequencies = (262, 294, 330, 350, 392, 440, 494, 523, 587, 659, 698, 784, 880, 988, 1047, 1175)

size = os.path.getsize(infile)

fmt = FormatChunk()
fmt.size = 0x10
fmt.formatTag = 1
fmt.channels = 1
fmt.bitsPerSample = 16
fmt.samplesPerSec = sampleFreq
fmt.avgBytesPerSec = fmt.samplesPerSec * fmt.bitsPerSample / 8
fmt.blockAlign = 4

print "channels: %d" % (fmt.channels,)
print "samplesPerSec: %d" % (fmt.samplesPerSec,)
print "bytesPerSec: %d" % (fmt.avgBytesPerSec,)
print "bitsPerSample: %d" % (fmt.bitsPerSample,)

amplitude = 3000
length = (3*denominator + size) * fmt.samplesPerSec  / denominator * fmt.channels * fmt.bitsPerSample / 8

infh = open(infile, "rb")
fh = open(filename, "wb")

format = GlobalFormat()
format.size = length

fh.write(format.as_bin())
fh.write(fmt.as_bin())

dataId = 0x61746164
fh.write(struct.pack("I", dataId))

dataLength = length - 12 - 16 - 12
fh.write(struct.pack("I", dataLength))

print "sync length: %d" % (3*fmt.avgBytesPerSec,)

frequency = 440
period = 2*math.pi/(fmt.samplesPerSec/float(frequency))
samples = []
for i in xrange(0, 3*fmt.samplesPerSec):
	sample = amplitude * (1 + math.sin(period*float(i)))
	samples.append(int(sample))
	fh.write(struct.pack("h", int(sample)))
	fh.flush()
	
halfByteSampleLength = fmt.samplesPerSec / denominator
print "half byte sample length: %d" % (halfByteSampleLength,)

j = 0
while True:
	data = infh.read(1)
	if data == "":
		break
		
	(m,) = struct.unpack("B", data)
	
	a = m & 0xF
	b = m >> 4

	for halfbyte in (a,b):
		frequency = frequencies[halfbyte]
		#print frequency
		period = 2*math.pi/(fmt.samplesPerSec/float(frequency))
		for i in xrange(0, halfByteSampleLength):
			sample = amplitude * (1 + math.sin(period*float(i)))
			fh.write(struct.pack("h", int(sample)))
			fh.flush()
		j += 1
		
#print "i: %d" % (j,)

infh.close()
fh.close()

sound2data.py

def read_and_unpack(fh, format):
	size = struct.calcsize(format)
	return struct.unpack(format, fh.read(size))
	
def get_maxfrequency(samples, norm, maxAcceptable = 1200):
	samplesArr = numpy.numarray.array(samples)
	transformedSamples = numpy.fft.fft(samplesArr)
	
	maxFreq = 0
	maxI = 0
	for i, freq in enumerate(transformedSamples):
		j = int(norm * i)
		a = math.sqrt(freq.real**2 + freq.imag**2)
		if a > maxFreq and i > 0 and j < maxAcceptable:
			#print "%d: %10.10f" % (i, a)
			maxFreq = a
			maxI = j
			
	return maxI
	
def get_closest_index(frequencies, freq):
	minDist = ()
	minI = -1
	for i, cmp in enumerate(frequencies):
		dist = abs(freq - cmp)
		if dist < minDist:
			minDist = dist
			minI = i
	return minI

import struct
import math
import os.path
import numpy.numarray
import numpy.fft

filename = "out.wav"
sourceSampleFreq = 8000
denominator = 8
frequencies = (262, 294, 330, 350, 392, 440, 494, 523, 587, 659, 698, 784, 880, 988, 1047, 1175)

fh = open(filename, "rb")

(id, size, type) = read_and_unpack(fh, "III")

(id, size, formatTag, channels, samplesPerSec, \
bytesPerSec, blockAlign, bitsPerSample) = read_and_unpack(fh, "IIhHIIHH")

print "channels: %d" % (channels,)
print "samplesPerSec: %d" % (samplesPerSec,)
print "bytesPerSec: %d" % (bytesPerSec,)
print "bitsPerSample: %d" % (bitsPerSample,)

# skip data chunk header
fh.read(4)

# read length
(length,) = read_and_unpack(fh, "I")
print "length: %d" % (length,)

syncToneLength = 3*samplesPerSec
print "sync length: %d" % (syncToneLength,)
syncSamples = []
for i in xrange(0, syncToneLength):
	(sample,) = read_and_unpack(fh, "h")
	syncSamples.append(sample)

syncFreq = get_maxfrequency(syncSamples, 1.0/3.0)
if syncFreq == 440:
	print "Found valid 3 second syncbeep of 440hz"
else:
	print "Invalid sync beep: %d" % (syncFreq,)
	exit(0)

halfByteSampleLength = samplesPerSec / denominator
print "half byte sample length: %d" % (halfByteSampleLength,)
j = 0
buffer = ""
while True:
	try:
		halfByte = []
		for w in xrange(0,2):
			samples = []
			for i in xrange(0, halfByteSampleLength):
				(sample,) = read_and_unpack(fh, "h")
				samples.append(sample)
				
			freq = get_maxfrequency(samples, samplesPerSec / float(halfByteSampleLength))
			v = get_closest_index(frequencies, freq)
			halfByte.append(v)
			
		fullbyte = halfByte[0] | (halfByte[1] << 4)
		buffer += chr(fullbyte)
		
		j += 1
	except:
		break

print
print "decoded data:"
print buffer
#print "i: %d" % (j,)

fh.close()