mirror/btstack
1
0
Fork 0
mirror of https://github.com/bluekitchen/btstack.git synced 2025-04-16 08:42:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

sbc stereo fixed create joint stereo, read stereo

Browse Source
This commit is contained in:
Milanka Ringwald 2016-05-18 14:33:52 +02:00
parent f08a674b88
commit 5c9bef5ba2
5 changed files with 73 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
test/sbc/sbc.py
View File

@ -149,49 +149,47 @@ def calculate_scalefactors(nr_blocks, nr_channels, nr_subbands, sb_sample):
scale_factor = np.zeros(shape=(nr_channels, nr_subbands), dtype = np.int32) scale_factor = np.zeros(shape=(nr_channels, nr_subbands), dtype = np.int32)
scalefactor = np.zeros(shape=(nr_channels, nr_subbands), dtype = np.int32) scalefactor = np.zeros(shape=(nr_channels, nr_subbands), dtype = np.int32)
# max_subbandsample = calculate_max_subbandsample(nr_blocks, nr_channels, nr_subbands, sb_sample) max_subbandsample = calculate_max_subbandsample(nr_blocks, nr_channels, nr_subbands, sb_sample)
# for ch in range(nr_channels):
# for sb in range(nr_subbands):
# (scale_factor[ch][sb], scalefactor[ch][sb]) = calculate_scalefactor(max_subbandsample[ch][sb])
for ch in range(nr_channels): for ch in range(nr_channels):
for sb in range(nr_subbands): for sb in range(nr_subbands):
scale_factor[ch][sb] = 0 (scale_factor[ch][sb], scalefactor[ch][sb]) = calculate_scalefactor(max_subbandsample[ch][sb])
scalefactor[ch][sb] = 2
for blk in range(nr_blocks):
while scalefactor[ch][sb] < abs(sb_sample[blk][ch][sb]):
scale_factor[ch][sb]+=1
scalefactor[ch][sb] *= 2
return scale_factor, scalefactor return scale_factor, scalefactor
def calculate_scalefactors_and_channel_mode(frame): def calculate_channel_mode(frame):
frame.scale_factor, frame.scalefactor = calculate_scalefactors(frame.nr_blocks, frame.nr_channels, frame.nr_subbands, frame.sb_sample)
#print "calculate_scalefactors_and_channel_mode1 ", frame.scale_factor
if frame.nr_channels == 1: if frame.nr_channels == 1:
frame.channel_mode = MONO frame.channel_mode = MONO
else: else:
frame.channel_mode = STEREO
frame.join = np.zeros(frame.nr_subbands, dtype = np.uint8)
return
sb_sample1 = np.zeros(shape = (frame.nr_blocks,2,frame.nr_subbands), dtype = np.uint16) sb_sample1 = np.zeros(shape = (frame.nr_blocks,2,frame.nr_subbands), dtype = np.uint16)
for blk in range(frame.nr_blocks): for blk in range(frame.nr_blocks):
for sb in range(frame.nr_subbands): for sb in range(frame.nr_subbands):
sb_sample1[blk][0][sb] = frame.sb_sample[blk][0][sb] + frame.sb_sample[blk][1][sb] sb_sample1[blk][0][sb] = (frame.sb_sample[blk][0][sb] + frame.sb_sample[blk][1][sb])/2
sb_sample1[blk][1][sb] = frame.sb_sample[blk][0][sb] - frame.sb_sample[blk][1][sb] sb_sample1[blk][1][sb] = (frame.sb_sample[blk][0][sb] - frame.sb_sample[blk][1][sb])/2
scale_factor, scalefactor = calculate_scalefactors(frame.nr_blocks, frame.nr_channels, frame.nr_subbands, sb_sample1) scale_factor, scalefactor = calculate_scalefactors(frame.nr_blocks, frame.nr_channels, frame.nr_subbands, sb_sample1)
#print "calculate_scalefactors_and_channel_mode 2", scale_factor
sumb = 0
suma = 0
for sb in range(frame.nr_subbands): for sb in range(frame.nr_subbands):
suma += frame.scale_factor[0][sb] + frame.scale_factor[1][sb] suma = frame.scale_factor[0][sb] + frame.scale_factor[1][sb]
sumb += scale_factor[0][sb] + scale_factor[1][sb] sumb = scale_factor[0][sb] + scale_factor[1][sb]
#print "calculate_scalefactors_and_channel_mode 3", suma, sumb if suma > sumb:
if suma > sumb: frame.channel_mode = JOINT_STEREO
frame.channel_mode = JOINT_STEREO frame.join[sb] = 1
else:
frame.channel_mode = STEREO frame.scale_factor[0][sb] = scale_factor[0][sb]
frame.scale_factor[1][sb] = scale_factor[1][sb]
frame.scalefactor[0][sb] = scalefactor[0][sb]
frame.scalefactor[1][sb] = scalefactor[1][sb]
for blk in range(frame.nr_blocks):
frame.sb_sample[blk][0][sb] = sb_sample1[blk][0][sb]
frame.sb_sample[blk][1][sb] = sb_sample1[blk][1][sb]
print " channel_mode = JOINT_STEREO"
class SBCFrame: class SBCFrame:
@ -224,13 +222,19 @@ class SBCFrame:
self.sampling_frequency = sampling_frequency_index(sampling_frequency) self.sampling_frequency = sampling_frequency_index(sampling_frequency)
self.bitpool = bitpool self.bitpool = bitpool
self.allocation_method = allocation_method self.allocation_method = allocation_method
self.init(nr_blocks, nr_subbands, nr_channels)
return
def init(self, nr_blocks, nr_subbands, nr_channels):
self.scale_factor = np.zeros(shape=(nr_channels, nr_subbands), dtype = np.int32) self.scale_factor = np.zeros(shape=(nr_channels, nr_subbands), dtype = np.int32)
self.scalefactor = np.zeros(shape=(nr_channels, nr_subbands), dtype = np.int32) self.scalefactor = np.zeros(shape=(nr_channels, nr_subbands), dtype = np.int32)
self.audio_sample = np.zeros(shape=(nr_blocks, nr_channels, nr_subbands), dtype = np.uint16) self.audio_sample = np.zeros(shape=(nr_blocks, nr_channels, nr_subbands), dtype = np.uint16)
self.sb_sample = np.zeros(shape=(nr_blocks, nr_channels, nr_subbands), dtype = np.uint16) self.sb_sample = np.zeros(shape=(nr_blocks, nr_channels, nr_subbands), dtype = np.uint16)
self.levels = np.zeros(shape=(nr_channels, nr_subbands), dtype = np.int32) self.levels = np.zeros(shape=(nr_channels, nr_subbands), dtype = np.int32)
self.pcm = np.zeros(shape=(nr_channels, nr_subbands*nr_blocks), dtype = np.int16)
self.join = np.zeros(nr_subbands, dtype = np.uint8)
self.X = np.zeros(nr_subbands, dtype = np.int16)
self.EX = np.zeros(nr_subbands) self.EX = np.zeros(nr_subbands)
return
def dump_audio_samples(self, blk, ch): def dump_audio_samples(self, blk, ch):
print self.audio_sample[blk][ch] print self.audio_sample[blk][ch]
@ -545,6 +549,7 @@ def calculate_crc(frame):
add_bits(frame.bitpool, 8) add_bits(frame.bitpool, 8)
if frame.channel_mode == JOINT_STEREO: if frame.channel_mode == JOINT_STEREO:
#print ("Joint Stereo!")
for sb in range(frame.nr_subbands): for sb in range(frame.nr_subbands):
add_bits(frame.join[sb],1) add_bits(frame.join[sb],1)
@ -554,7 +559,8 @@ def calculate_crc(frame):
bitstream_len = (bitstream_index + 1) * 8 bitstream_len = (bitstream_index + 1) * 8
if bitstream_bits_available: if bitstream_bits_available:
bitstream_len += (8-bitstream_bits_available) bitstream_len -= bitstream_bits_available
return sbc_crc8(bitstream, bitstream_len) return sbc_crc8(bitstream, bitstream_len)

24
test/sbc/sbc_decoder.py
View File

@ -14,14 +14,6 @@ def sbc_unpack_frame(fin, available_bytes, frame):
frame.syncword = get_bits(fin,8) frame.syncword = get_bits(fin,8)
if frame.syncword != 156: if frame.syncword != 156:
# i = 0
# while available_bytes:
# if i%10 == 0:
# print
# bt = get_bits(fin,8)
# print "0x%0x "% bt,
# available_bytes -= 1
# i+=1
print ("out of sync %02x" % frame.syncword) print ("out of sync %02x" % frame.syncword)
return -1 return -1
frame.sampling_frequency = get_bits(fin,2) frame.sampling_frequency = get_bits(fin,2)
@ -32,13 +24,16 @@ def sbc_unpack_frame(fin, available_bytes, frame):
frame.nr_channels = 1 frame.nr_channels = 1
else: else:
frame.nr_channels = 2 frame.nr_channels = 2
frame.allocation_method = get_bits(fin,1) frame.allocation_method = get_bits(fin,1)
frame.nr_subbands = nr_subbands[get_bits(fin,1)] frame.nr_subbands = nr_subbands[get_bits(fin,1)]
frame.init(frame.nr_blocks, frame.nr_subbands, frame.nr_channels)
frame.bitpool = get_bits(fin,8) frame.bitpool = get_bits(fin,8)
frame.crc_check = get_bits(fin,8) frame.crc_check = get_bits(fin,8)
frame.join = np.zeros(frame.nr_subbands, dtype = np.uint8)
# frame.join = np.zeros(frame.nr_subbands, dtype = np.uint8)
if frame.channel_mode == JOINT_STEREO: if frame.channel_mode == JOINT_STEREO:
frame.join = np.zeros(frame.nr_subbands-1) frame.join = np.zeros(frame.nr_subbands-1)
@ -146,11 +141,9 @@ def sbc_frame_synthesis(frame, ch, blk, proto_table):
offset = blk*M offset = blk*M
for j in range(M): for j in range(M):
for i in range(10): for i in range(10):
frame.X[j] += W[j+M*i] frame.X[j] += W[j+M*i]
frame.pcm[ch][offset + j] = np.int16(frame.X[j]) frame.pcm[ch][offset + j] = np.int16(frame.X[j])
@ -161,7 +154,6 @@ def sbc_synthesis(frame):
proto_table = Proto_8_80 proto_table = Proto_8_80
else: else:
return -1 return -1
for ch in range(frame.nr_channels): for ch in range(frame.nr_channels):
for blk in range(frame.nr_blocks): for blk in range(frame.nr_blocks):
sbc_frame_synthesis(frame, ch, blk, proto_table) sbc_frame_synthesis(frame, ch, blk, proto_table)
@ -223,15 +215,17 @@ if __name__ == "__main__":
err = sbc_unpack_frame(fin, file_size - fin.tell(), sbc_decoder_frame) err = sbc_unpack_frame(fin, file_size - fin.tell(), sbc_decoder_frame)
if frame_count == 0:
print sbc_decoder_frame
if err: if err:
print "error, frame_count: ", frame_count print "error, frame_count: ", frame_count
break break
sbc_decode(sbc_decoder_frame) sbc_decode(sbc_decoder_frame)
if frame_count == 0: if frame_count == 0:
print sbc_decoder_frame
fout = wave.open(wavfile, 'w') fout = wave.open(wavfile, 'w')
fout.setnchannels(sbc_decoder_frame.nr_channels) fout.setnchannels(sbc_decoder_frame.nr_channels)
fout.setsampwidth(2) fout.setsampwidth(2)

2
test/sbc/sbc_decoder_test.py
View File

@ -18,7 +18,7 @@ def sbc_compare_pcm(frame_count, actual_frame, expected_frame):
max_error = M max_error = M
if M > error: if M > error:
print "pcm error (%d, %d ) " % (frame_count, M) print "pcm error (%d, %f ) " % (frame_count, M)
return -1 return -1
return 0 return 0

75
test/sbc/sbc_encoder.py
View File

@ -5,26 +5,22 @@ import struct
import sys import sys
from sbc import * from sbc import *
X = np.zeros(80, dtype = np.int16) X = np.zeros(shape=(2,80), dtype = np.int16)
def fetch_samples_for_next_sbc_frame(fin, frame): def fetch_samples_for_next_sbc_frame(fin, frame):
nr_samples = frame.nr_blocks * frame.nr_subbands * frame.nr_channels nr_samples = frame.nr_blocks * frame.nr_subbands
raw_data = fin.readframes(nr_samples) # Returns byte data raw_data = fin.readframes(nr_samples) # Returns byte data
len_raw_data = len(raw_data) / 2 fmt = "%ih" % (len(raw_data) / 2)
fmt = "%ih" % len_raw_data # read signed 2 byte shorts
data = struct.unpack(fmt, raw_data) data = struct.unpack(fmt, raw_data)
len_data = len(data)
for i in range(frame.nr_blocks * frame.nr_subbands): if frame.nr_channels == 2:
for ch in range(frame.nr_channels): for i in range(len(data)/2 - 1):
index = i*2 + ch frame.pcm[0][i] = data[2*i]
if index < len_data: frame.pcm[1][i] = data[2*i+1]
frame.pcm[ch][i] = data[i*2 + ch] else:
else: for i in range(len(data)):
frame.pcm[ch][i] = 0 frame.pcm[0][i] = data[i]
def sbc_frame_analysis(frame, ch, blk, C): def sbc_frame_analysis(frame, ch, blk, C):
@ -41,12 +37,12 @@ def sbc_frame_analysis(frame, ch, blk, C):
S = np.zeros(M) S = np.zeros(M)
for i in range(L-1, M-1, -1): for i in range(L-1, M-1, -1):
X[i] = X[i-M] X[ch][i] = X[ch][i-M]
for i in range(M-1, -1, -1): for i in range(M-1, -1, -1):
X[i] = frame.EX[M-1-i] X[ch][i] = frame.EX[M-1-i]
for i in range(L): for i in range(L):
Z[i] = X[i] * C[i] Z[i] = X[ch][i] * C[i]
for i in range(M2): for i in range(M2):
for k in range(5): for k in range(5):
@ -69,8 +65,8 @@ def sbc_analysis(frame):
return -1 return -1
frame.sb_sample = np.ndarray(shape=(frame.nr_blocks, frame.nr_channels, frame.nr_subbands)) frame.sb_sample = np.ndarray(shape=(frame.nr_blocks, frame.nr_channels, frame.nr_subbands))
index = 0
for ch in range(frame.nr_channels): for ch in range(frame.nr_channels):
index = 0
for blk in range(frame.nr_blocks): for blk in range(frame.nr_blocks):
for sb in range(frame.nr_subbands): for sb in range(frame.nr_subbands):
frame.EX[sb] = np.int16(frame.pcm[ch][index]) frame.EX[sb] = np.int16(frame.pcm[ch][index])
@ -84,39 +80,9 @@ def sbc_encode(frame):
err = sbc_quantization(frame) err = sbc_quantization(frame)
return err return err
def calculate_joint_stereo_signal(frame):
sb_sample = np.zeros(shape = (frame.nr_blocks,frame.nr_channels,frame.nr_subbands), dtype = np.uint32)
scale_factor = np.zeros(shape=(frame.nr_channels, frame.nr_subbands), dtype = np.int32)
scalefactor = np.zeros(shape=(frame.nr_channels, frame.nr_subbands), dtype = np.int32)
for sb in range(frame.nr_subbands-1):
for blk in range(frame.nr_blocks):
sb_sample[blk][0][sb] = (frame.sb_sample_f[blk][0][sb] + frame.sb_sample_f[blk][1][sb]) >> 1
sb_sample[blk][1][sb] = (frame.sb_sample_f[blk][0][sb] - frame.sb_sample_f[blk][1][sb]) >> 1
for ch in range(frame.nr_channels):
for sb in range(frame.nr_subbands-1):
frame.scale_factor[ch][sb] = 0
frame.scalefactor[ch][sb] = 2
for blk in range(frame.nr_blocks):
while frame.scalefactor[ch][sb] < abs(frame.sb_sample[blk][ch][sb]):
frame.scale_factor[ch][sb]+=1
frame.scalefactor[ch][sb] *= 2
for sb in range(frame.nr_subbands-1):
if (frame.scalefactor[0][sb] + frame.scalefactor[1][sb]) > (scalefactor[0][sb] + scalefactor[1][sb]):
frame.join[sb] = 1
frame.scale_factor[0][sb] = scale_factor[0][sb]
frame.scale_factor[1][sb] = scale_factor[1][sb]
frame.scalefactor[0][sb] = scalefactor[0][sb]
frame.scalefactor[1][sb] = scalefactor[1][sb]
for blk in range(frame.nr_blocks):
frame.sb_sample[blk][0][sb] = sb_sample[blk][0][sb]
frame.sb_sample[blk][1][sb] = sb_sample[blk][1][sb]
def sbc_quantization(frame): def sbc_quantization(frame):
calculate_scalefactors_and_channel_mode(frame) frame.scale_factor, frame.scalefactor = calculate_scalefactors(frame.nr_blocks, frame.nr_channels, frame.nr_subbands, frame.sb_sample)
calculate_channel_mode(frame)
frame.bits = sbc_bit_allocation(frame) frame.bits = sbc_bit_allocation(frame)
# Reconstruct the Audio Samples # Reconstruct the Audio Samples
@ -127,12 +93,6 @@ def sbc_quantization(frame):
frame.syncword = 156 frame.syncword = 156
frame.crc_check = calculate_crc(frame) frame.crc_check = calculate_crc(frame)
frame.join = np.zeros(frame.nr_subbands, dtype = np.uint8)
if frame.channel_mode == JOINT_STEREO:
calculate_joint_stereo_signal(frame)
for blk in range(frame.nr_blocks): for blk in range(frame.nr_blocks):
for ch in range(frame.nr_channels): for ch in range(frame.nr_channels):
@ -195,9 +155,12 @@ if __name__ == "__main__":
sbc_encode(sbc_encoder_frame) sbc_encode(sbc_encoder_frame)
sbc_write_frame(fout, sbc_encoder_frame) sbc_write_frame(fout, sbc_encoder_frame)
# if subband_frame_count == 0:
# exit(0)
audio_frame_count += nr_samples audio_frame_count += nr_samples
subband_frame_count += 1 subband_frame_count += 1
fin.close() fin.close()
fout.close() fout.close()
print("DONE, WAV file %s encoded into SBC file %s " % (infile, sbcfile)) print("DONE, WAV file %s encoded into SBC file %s " % (infile, sbcfile))

20
test/sbc/sbc_encoder_test.py
View File

@ -12,17 +12,16 @@ max_error = -1
def sbc_compare_audio_frames(frame_count, actual_frame, expected_frame): def sbc_compare_audio_frames(frame_count, actual_frame, expected_frame):
global error, max_error 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
M = mse(actual_frame.audio_sample, expected_frame.audio_sample) if M > error:
if M > max_error: print "audio_sample error (%d, %f ) " % (frame_count, M)
max_error = M return -1
if M > error:
print "audio_sample error (%d, %d ) " % (frame_count, M)
return -1
return 0 return 0
def sbc_compare_headers(frame_count, actual_frame, expected_frame): def sbc_compare_headers(frame_count, actual_frame, expected_frame):
if actual_frame.syncword != expected_frame.syncword: if actual_frame.syncword != expected_frame.syncword:
print "syncword wrong ", actual_frame.syncword print "syncword wrong ", actual_frame.syncword
@ -60,6 +59,7 @@ def sbc_compare_headers(frame_count, actual_frame, expected_frame):
print "join error \nE:\n %s \nD:\n %s" % (actual_frame.join, expected_frame.join) print "join error \nE:\n %s \nD:\n %s" % (actual_frame.join, expected_frame.join)
return -1 return -1
if mse(actual_frame.scale_factor, expected_frame.scale_factor) > 0: if mse(actual_frame.scale_factor, expected_frame.scale_factor) > 0:
print "scale_factor error \nE:\n %s \nD:\n %s" % (actual_frame.scale_factor, expected_frame.scale_factor) print "scale_factor error \nE:\n %s \nD:\n %s" % (actual_frame.scale_factor, expected_frame.scale_factor)
return -1 return -1
@ -89,7 +89,6 @@ file_size = 0
def get_expected_frame(fin_expected): def get_expected_frame(fin_expected):
global file_size global file_size
expected_frame = SBCFrame() expected_frame = SBCFrame()
sbc_unpack_frame(fin_expected, file_size - fin_expected.tell(), expected_frame) sbc_unpack_frame(fin_expected, file_size - fin_expected.tell(), expected_frame)
return expected_frame return expected_frame
@ -139,9 +138,6 @@ try:
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)
expected_frame = get_expected_frame(fin_expected) expected_frame = get_expected_frame(fin_expected)
print actual_frame.sb_sample
print expected_frame.sb_sample
err = sbc_compare_headers(subband_frame_count, actual_frame, expected_frame) err = sbc_compare_headers(subband_frame_count, actual_frame, expected_frame)
if err < 0: if err < 0:
exit(1) exit(1)