mirror/btstack
1
0
Fork 0
mirror of https://github.com/bluekitchen/btstack.git synced 2025-03-28 19:20:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

sbc: stereo, joint stereo working

Browse Source
This commit is contained in:
Milanka Ringwald 2016-05-19 16:00:28 +02:00
parent ba345fb8f4
commit ef8a7a12f4
6 changed files with 85 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
test/sbc/Makefile
View File

@ -4,32 +4,32 @@ test-sine:
./sbc_decoder_test.py data/sine-4sb-stereo.sbc data/sine-4sb-decoded-stereo.wav
./sbc_decoder_test.py data/sine-8sb-stereo.sbc data/sine-8sb-decoded-stereo.wav
./sbc_encoder_test.py data/sine-mono.wav 16 4 31 data/sine-4sb-mono.sbc
./sbc_encoder_test.py data/sine-mono.wav 16 8 31 data/sine-8sb-mono.sbc
./sbc_encoder_test.py data/sine-stereo.wav 16 4 31 data/sine-4sb-stereo.sbc
./sbc_encoder_test.py data/sine-stereo.wav 16 8 31 data/sine-8sb-stereo.sbc
./sbc_encoder_test.py data/sine-mono.wav 16 4 31 0 data/sine-4sb-mono.sbc
./sbc_encoder_test.py data/sine-mono.wav 16 8 64 0 data/sine-8sb-mono.sbc
./sbc_encoder_test.py data/sine-stereo.wav 16 4 30 1 2 data/sine-4sb-stereo.sbc
./sbc_encoder_test.py data/sine-stereo.wav 16 8 64 2 data/sine-8sb-stereo.sbc
test-short:
./sbc_decoder_test.py data/fanfare-short-4sb-mono.sbc data/fanfare-short-4sb-decoded-mono.wav
./sbc_decoder_test.py data/fanfare-short-8sb-mono.sbc data/fanfare-short-8sb-decoded-mono.wav
./sbc_decoder_test.py data/fanfare-short-4sb-stereo.sbc data/fanfare-short-4sb-decoded-stereo.wav
./sbc_decoder_test.py data/fanfare-short-8sb-stereo.sbc data/fanfare-short-8sb-decoded-stereo.wav
./sbc_encoder_test.py data/fanfare-short-mono.wav 16 4 31 data/fanfare-short-4sb-mono.sbc
./sbc_encoder_test.py data/fanfare-short-mono.wav 16 8 31 data/fanfare-short-8sb-mono.sbc
./sbc_encoder_test.py data/fanfare-short-stereo.wav 16 4 31 data/fanfare-short-4sb-stereo.sbc
./sbc_encoder_test.py data/fanfare-short-stereo.wav 16 8 31 data/fanfare-short-8sb-stereo.sbc
./sbc_encoder_test.py data/fanfare-short-mono.wav 16 4 31 1 0 data/fanfare-short-4sb-mono.sbc
./sbc_encoder_test.py data/fanfare-short-mono.wav 16 8 64 1 0 data/fanfare-short-8sb-mono.sbc
./sbc_encoder_test.py data/fanfare-short-stereo.wav 16 4 30 1 2 data/fanfare-short-4sb-stereo.sbc
./sbc_encoder_test.py data/fanfare-short-stereo.wav 16 8 62 1 2 data/fanfare-short-8sb-stereo.sbc
test:
./sbc_decoder_test.py data/fanfare-4sb-mono.sbc data/fanfare-4sb-decoded-mono.wav
./sbc_decoder_test.py data/fanfare-8sb-mono.sbc data/fanfare-8sb-decoded-mono.wav
./sbc_decoder_test.py data/fanfare-4sb-stereo.sbc data/fanfare-4sb-decoded-stereo.wav
./sbc_decoder_test.py data/fanfare-8sb-stereo.sbc data/fanfare-8sb-decoded-stereo.wav
./sbc_encoder_test.py data/fanfare-mono.wav 16 4 31 data/fanfare-4sb-mono.sbc
./sbc_encoder_test.py data/fanfare-mono.wav 16 8 31 data/fanfare-8sb-mono.sbc
./sbc_encoder_test.py data/fanfare-stereo.wav 16 4 31 data/fanfare-4sb-stereo.sbc
./sbc_encoder_test.py data/fanfare-stereo.wav 16 8 31 data/fanfare-8sb-stereo.sbc
./sbc_encoder_test.py data/fanfare-mono.wav 16 4 31 0 data/fanfare-4sb-mono.sbc
./sbc_encoder_test.py data/fanfare-mono.wav 16 8 64 0 data/fanfare-8sb-mono.sbc
./sbc_encoder_test.py data/fanfare-stereo.wav 16 4 31 2 data/fanfare-4sb-stereo.sbc
./sbc_encoder_test.py data/fanfare-stereo.wav 16 8 64 2 data/fanfare-8sb-stereo.sbc
clean:
rm -f *.pyc *.wav *.sbc
rm -f *.pyc *.wav *.sbc data/*-decoded.wav data/*-encoded.sbc

12
test/sbc/sbc.py
View File

@ -127,7 +127,7 @@ offset8 = np.array([[ -2, 0, 0, 0, 0, 0, 0, 1 ],
def calculate_scalefactor(max_subbandsample):
x = 0
while True:
y = (1 << x) + 1
y = 1 << (x + 1)
if y > max_subbandsample:
break
x += 1
@ -155,16 +155,19 @@ def calculate_scalefactors(nr_blocks, nr_channels, nr_subbands, sb_sample):
(scale_factor[ch][sb], scalefactor[ch][sb]) = calculate_scalefactor(max_subbandsample[ch][sb])
return scale_factor, scalefactor
def calculate_channel_mode_and_scale_factors(frame):
def calculate_channel_mode_and_scale_factors(frame, force_channel_mode):
frame.scale_factor, frame.scalefactor = calculate_scalefactors(frame.nr_blocks, frame.nr_channels, frame.nr_subbands, frame.sb_sample)
if frame.nr_channels == 1:
frame.channel_mode = MONO
return
frame.channel_mode = STEREO
frame.join = np.zeros(frame.nr_subbands, dtype = np.uint8)
if force_channel_mode == STEREO:
frame.channel_mode = STEREO
return
sb_sample = np.zeros(shape = (frame.nr_blocks,2,frame.nr_subbands), dtype = np.int32)
for blk in range(frame.nr_blocks):
for sb in range(frame.nr_subbands):
@ -177,7 +180,7 @@ def calculate_channel_mode_and_scale_factors(frame):
suma = frame.scale_factor[0][sb] + frame.scale_factor[1][sb]
sumb = scale_factor[0][sb] + scale_factor[1][sb]
if suma > sumb:
if suma > sumb or force_channel_mode == JOINT_STEREO:
frame.channel_mode = JOINT_STEREO
frame.join[sb] = 1
@ -361,6 +364,7 @@ def sbc_bit_allocation_stereo_joint(frame):
else:
ch = 1
if bits.sum() != frame.bitpool:
print "bit allocation failed, bitpool %d, allocated %d" % (bits.sum() , frame.bitpool)
exit(1)

7
test/sbc/sbc_decoder.py
View File

@ -97,8 +97,8 @@ def sbc_reconstruct_subband_samples(frame):
if frame.join[sb]==1:
ch_a = frame.sb_sample[blk][0][sb] + frame.sb_sample[blk][1][sb]
ch_b = frame.sb_sample[blk][0][sb] - frame.sb_sample[blk][1][sb]
frame.sb_sample[blk][0][sb] = ch_a/2
frame.sb_sample[blk][1][sb] = ch_b/2
frame.sb_sample[blk][0][sb] = ch_a
frame.sb_sample[blk][1][sb] = ch_b
return 0
@ -210,7 +210,6 @@ if __name__ == "__main__":
if frame_count % 200 == 0:
print "== Frame %d == %d" % (frame_count, fin.tell())
err = sbc_unpack_frame(fin, file_size - fin.tell(), sbc_decoder_frame)
if frame_count == 0:
print sbc_decoder_frame
@ -221,7 +220,7 @@ if __name__ == "__main__":
sbc_decode(sbc_decoder_frame)
if frame_count == 0:
fout = wave.open(wavfile, 'w')
fout.setnchannels(sbc_decoder_frame.nr_channels)

14
test/sbc/sbc_decoder_test.py
View File

@ -17,8 +17,8 @@ def sbc_compare_pcm(frame_count, actual_frame, expected_frame):
if M > max_error:
max_error = M
if M > error:
print "pcm error (%d, %f ) " % (frame_count, M)
if max_error > error:
print "pcm error (%d, %f ) " % (frame_count, max_error)
return -1
return 0
@ -66,7 +66,7 @@ def get_actual_frame(fin):
def get_expected_frame(fin_expected, nr_blocks, nr_subbands, nr_channels, sampling_frequency, bitpool, allocation_method):
expected_frame = SBCFrame(nr_blocks, nr_subbands, nr_channels, sampling_frequency, bitpool, allocation_method)
fetch_samples_for_next_sbc_frame(fin_expected, expected_frame)
calculate_scalefactors_and_channel_mode(expected_frame)
calculate_channel_mode_and_scale_factors(expected_frame)
return expected_frame
usage = '''
@ -113,18 +113,15 @@ try:
actual_frame.allocation_method)
err = sbc_compare_headers(subband_frame_count, actual_frame, expected_frame)
print ("%03d : %s %s"%( subband_frame_count,
channel_mode_to_str(actual_frame.channel_mode),
channel_mode_to_str(expected_frame.channel_mode)))
if err < 0:
print ("Headers differ \n%s\n%s" % (actual_frame, expected_frame))
exit(1)
sys.exit(1)
err = sbc_compare_pcm(subband_frame_count, actual_frame, expected_frame)
if err < 0:
print ("PCMs differ \n%s\n%s" % (actual_frame.pcm, expected_frame.pcm))
exit(1)
sys.exit(1)
if subband_frame_count == 0:
print actual_frame, expected_frame
@ -135,7 +132,6 @@ try:
fin_expected.close()
fin.close()
print ("DONE, max MSE PCM error %f" % max_error)
exit(0)
except IOError as e:
print(usage)

37
test/sbc/sbc_encoder.py
View File

@ -7,17 +7,15 @@ from sbc import *
X = np.zeros(shape=(2,80), dtype = np.int16)
def fetch_samples_for_next_sbc_frame(fin, frame):
nr_samples = frame.nr_blocks * frame.nr_subbands
raw_data = fin.readframes(nr_samples) # Returns byte data
raw_data = fin.readframes(frame.nr_blocks * frame.nr_subbands)
fmt = "%ih" % (len(raw_data) / 2)
data = struct.unpack(fmt, raw_data)
if frame.nr_channels == 2:
for i in range(len(data)/2):
frame.pcm[0][i] = data[2*i]
frame.pcm[1][i] = data[2*i+1]
frame.pcm[1][i] = data[2*i+1]
else:
for i in range(len(data)):
frame.pcm[0][i] = data[i]
@ -73,14 +71,14 @@ def sbc_analysis(frame):
sbc_frame_analysis(frame, ch, blk, C)
return 0
def sbc_encode(frame):
def sbc_encode(frame, force_channel_mode):
err = sbc_analysis(frame)
if err >= 0:
err = sbc_quantization(frame)
err = sbc_quantization(frame, force_channel_mode)
return err
def sbc_quantization(frame):
calculate_channel_mode_and_scale_factors(frame)
def sbc_quantization(frame, force_channel_mode):
calculate_channel_mode_and_scale_factors(frame, force_channel_mode)
frame.bits = sbc_bit_allocation(frame)
# Reconstruct the Audio Samples
@ -99,11 +97,7 @@ def sbc_quantization(frame):
if frame.levels[ch][sb] > 0:
SB = frame.sb_sample[blk][ch][sb]
L = frame.levels[ch][sb]
SF = frame.scalefactor[ch][sb]
# if frame.channel_mode == JOINT_STEREO and frame.join[sb]:
# SB = SB * 2
SF = frame.scalefactor[ch][sb]
frame.audio_sample[blk][ch][sb] = np.uint16(((SB * L / SF + L) - 1.0)/2.0)
else:
frame.audio_sample[blk][ch][sb] = 0
@ -117,7 +111,7 @@ def sbc_write_frame(fout, sbc_encoder_frame):
if __name__ == "__main__":
usage = '''
Usage: ./sbc_encoder.py input.wav blocks subbands bitpool allocation_method[0-LOUDNESS,1-SNR]
Usage: ./sbc_encoder.py input.wav blocks subbands bitpool allocation_method[0-LOUDNESS,1-SNR] force_channel_mode[2-STEREO,3-JOINT_STEREO]
Example: ./sbc_encoder.py fanfare.wav 16 4 31 0
'''
nr_blocks = 0
@ -137,7 +131,13 @@ if __name__ == "__main__":
nr_blocks = int(sys.argv[2])
nr_subbands = int(sys.argv[3])
bitpool = int(sys.argv[4])
allocation_method = int(sys.argv[5])
allocation_method = int(sys.argv[5])
force_channel_mode = 0
if len(sys.argv) == 6:
force_channel_mode = int(sys.argv[6])
if force_channel_mode != 2 or force_channel_mode != 3:
print(usage)
sys.exit(1)
fin = wave.open(infile, 'rb')
nr_channels = fin.getnchannels()
@ -155,17 +155,12 @@ if __name__ == "__main__":
sbc_encoder_frame = SBCFrame(nr_blocks, nr_subbands, nr_channels, bitpool, sampling_frequency, allocation_method)
fetch_samples_for_next_sbc_frame(fin, sbc_encoder_frame)
sbc_encode(sbc_encoder_frame)
sbc_encode(sbc_encoder_frame, force_channel_mode)
sbc_write_frame(fout, sbc_encoder_frame)
if subband_frame_count == 0:
print sbc_encoder_frame.channel_mode
print sbc_encoder_frame
audio_frame_count += nr_samples
subband_frame_count += 1
fin.close()
fout.close()
print("DONE, WAV file %s encoded into SBC file %s " % (infile, sbcfile))

53
test/sbc/sbc_encoder_test.py
View File

@ -10,15 +10,36 @@ from sbc_decoder import *
error = 0.99
max_error = -1
def sbc_compare_subband_samples(frame_count, actual_frame, expected_frame):
global error, max_error
for blk in range(actual_frame.nr_blocks):
for ch in range(actual_frame.nr_channels):
M = mse(actual_frame.sb_sample[blk][ch], expected_frame.sb_sample[blk][ch])
if M > max_error:
max_error = M
if max_error > error:
print "Frame %d: sb_sample error %f (ch %d, blk %d)" % (frame_count, max_error, ch, blk)
print actual_frame.sb_sample[blk]
print expected_frame.sb_sample[blk]
return -1
return 0
def sbc_compare_audio_frames(frame_count, actual_frame, expected_frame):
global error, max_error
for ch in range(actual_frame.nr_channels):
M = mse(actual_frame.audio_sample, expected_frame.audio_sample)
if M > max_error:
max_error = M
if M > error:
print "audio_sample error (%d, %f ) " % (frame_count, M)
for blk in range(actual_frame.nr_blocks):
for ch in range(actual_frame.nr_channels):
M = mse(actual_frame.audio_sample[blk][ch], expected_frame.audio_sample[blk][ch])
if M > max_error:
max_error = M
if max_error > error:
print "audio_sample error (%d, %f ) " % (frame_count, max_error)
print actual_frame.audio_sample[blk]
print expected_frame.audio_sample[blk]
return -1
return 0
@ -79,10 +100,10 @@ def sbc_compare_headers(frame_count, actual_frame, expected_frame):
return 0
def get_actual_frame(fin, nr_blocks, nr_subbands, nr_channels, sampling_frequency, bitpool, allocation_method):
def get_actual_frame(fin, nr_blocks, nr_subbands, nr_channels, sampling_frequency, bitpool, allocation_method, force_channel_mode):
actual_frame = SBCFrame(nr_blocks, nr_subbands, nr_channels, sampling_frequency, bitpool, allocation_method)
fetch_samples_for_next_sbc_frame(fin, actual_frame)
sbc_encode(actual_frame)
sbc_encode(actual_frame, force_channel_mode)
return actual_frame
file_size = 0
@ -90,14 +111,15 @@ def get_expected_frame(fin_expected):
global file_size
expected_frame = SBCFrame()
sbc_unpack_frame(fin_expected, file_size - fin_expected.tell(), expected_frame)
sbc_reconstruct_subband_samples(expected_frame)
return expected_frame
usage = '''
Usage: ./sbc_encoder_test.py encoder_input.wav blocks subbands bitpool allocation_method encoder_expected_output.sbc
Example: ./sbc_encoder_test.py fanfare.wav 16 4 31 0 fanfare-4sb.sbc
Usage: ./sbc_encoder_test.py encoder_input.wav blocks subbands bitpool allocation_method force_channel_mode encoder_expected_output.sbc
Example: ./sbc_encoder_test.py fanfare.wav 16 4 31 0 2 fanfare-4sb.sbc
'''
if (len(sys.argv) < 7):
if (len(sys.argv) < 8):
print(usage)
sys.exit(1)
try:
@ -107,6 +129,7 @@ try:
bitpool = int(sys.argv[4])
allocation_method = int(sys.argv[5])
encoder_expected_sbc = sys.argv[6]
force_channel_mode = sys.argv[7]
sampling_frequency = 44100
if not encoder_input_wav.endswith('.wav'):
@ -130,12 +153,12 @@ try:
subband_frame_count = 0
audio_frame_count = 0
nr_samples = nr_blocks * nr_subbands
while audio_frame_count < nr_audio_frames:
if subband_frame_count % 200 == 0:
print("== Frame %d ==" % (subband_frame_count))
actual_frame = get_actual_frame(fin, nr_blocks, nr_subbands, nr_channels, bitpool, sampling_frequency, allocation_method)
actual_frame = get_actual_frame(fin, nr_blocks, nr_subbands, nr_channels, bitpool, sampling_frequency, allocation_method, force_channel_mode)
expected_frame = get_expected_frame(fin_expected)
err = sbc_compare_headers(subband_frame_count, actual_frame, expected_frame)
@ -149,12 +172,12 @@ try:
audio_frame_count += nr_samples
subband_frame_count += 1
print ("DONE, max MSE audio sample error %f" % max_error)
print ("Max MSE audio sample error %f" % max_error)
fin.close()
fin_expected.close()
except TypeError:
print ("DONE, max MSE audio sample error %f" % max_error)
print ("Max MSE audio sample error %f" % max_error)
fin.close()
fin_expected.close()