mirror/btstack
1
0
Fork 0
mirror of https://github.com/bluekitchen/btstack.git synced 2025-04-23 11:43:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

impl joint stereo signal calc. in encoder

Browse Source
This commit is contained in:
Milanka Ringwald 2016-05-06 16:51:48 +02:00
parent 5f21c38a78
commit ad4708631f
3 changed files with 257 additions and 186 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

149
test/sbc/sbc.py
View File

@ -9,10 +9,36 @@ MONO = 0
DUAL_CHANNEL = 1 DUAL_CHANNEL = 1
STEREO = 2 STEREO = 2
JOINT_STEREO = 3 JOINT_STEREO = 3
channel_modes = ["MONO", "DUAL CHANNEL", "STEREO", "JOINT STEREO"]
# allocation method # allocation method
LOUDNESS = 0 LOUDNESS = 0
SNR = 1 SNR = 1
allocation_methods = ["LOUDNESS", "SNR"]
sampling_frequencies = [16000, 32000, 44100, 48000]
nr_blocks = [4, 8, 12, 16]
nr_subbands = [4, 8]
def allocation_method_to_str(allocation_method):
global allocation_methods
return allocation_methods[allocation_method]
def channel_mode_to_str(channel_mode):
global channel_modes
return channel_modes[channel_mode]
def sampling_frequency_to_str(sampling_frequency):
global sampling_frequencies
return sampling_frequencies[sampling_frequency]
def sampling_frequency_index(sampling_frequency):
global sampling_frequencies
for index, value in enumerate(sampling_frequencies):
if value == sampling_frequency:
return index
return -1
Proto_4_40 = [ Proto_4_40 = [
0.00000000E+00, 5.36548976E-04, 1.49188357E-03, 2.73370904E-03, 0.00000000E+00, 5.36548976E-04, 1.49188357E-03, 2.73370904E-03,
@ -98,10 +124,6 @@ offset8 = np.array([[ -2, 0, 0, 0, 0, 0, 0, 1 ],
[ -4, 0, 0, 0, 0, 0, 1, 2 ] [ -4, 0, 0, 0, 0, 0, 1, 2 ]
]) ])
nr_blocks = [4, 8, 12, 16]
nr_subbands = [4, 8]
sampling_frequency =[16000, 32000, 44100, 48000]
class SBCFrame: class SBCFrame:
syncword = 0 syncword = 0
sampling_frequency = 0 sampling_frequency = 0
@ -124,11 +146,12 @@ class SBCFrame:
bits = np.zeros(shape=(2, 8)) bits = np.zeros(shape=(2, 8))
levels = np.zeros(shape=(2, 8), dtype = np.int32) levels = np.zeros(shape=(2, 8), dtype = np.int32)
def __init__(self, nr_blocks, nr_subbands, nr_channels, sampling_frequency, bitpool):
def __init__(self, nr_blocks=16, nr_subbands=4, nr_channels=1, bitpool=31, sampling_frequency=44100):
self.nr_blocks = nr_blocks self.nr_blocks = nr_blocks
self.nr_subbands = nr_subbands self.nr_subbands = nr_subbands
self.nr_channels = nr_channels self.nr_channels = nr_channels
self.sampling_frequency = sampling_frequency self.sampling_frequency = sampling_frequency_index(sampling_frequency)
self.bitpool = bitpool self.bitpool = bitpool
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)
@ -138,36 +161,37 @@ class SBCFrame:
self.EX = np.zeros(nr_subbands) self.EX = np.zeros(nr_subbands)
return return
def dump_audio_samples(self, blk, ch):
print self.audio_sample[blk][ch]
def dump_subband_samples(self, blk, ch):
print self.sb_sample[blk][ch]
def dump_state(self):
res = "SBCFrameHeader state:"
res += "\n - nr channels %d" % self.nr_channels
res += "\n - nr blocks %d" % self.nr_blocks
res += "\n - nr subbands %d" % self.nr_subbands
res += "\n - scale factors: %s" % self.scale_factor
res += "\n - levels: %s" % self.levels
res += "\n - join: %s" % self.join
res += "\n - bits: %s" % self.bits
print res
def __str__(self): def __str__(self):
res = "SBCFrameHeader:" res = "SBCFrameHeader:"
res = res + "\n - syncword %d" % self.syncword res += "\n - syncword %d" % self.syncword
res = res + "\n - sample frequency %d" % self.sampling_frequency res += "\n - sampling frequency %d Hz" % sampling_frequency_to_str(self.sampling_frequency)
res = res + "\n - nr blocks %d" % self.nr_blocks
if self.channel_mode == MONO: res += "\n - nr channels %d" % self.nr_channels
res = res + "\n - channel mode MONO" res += "\n - nr blocks %d" % self.nr_blocks
elif self.channel_mode == DUAL_CHANNEL: res += "\n - nr subbands %d" % self.nr_subbands
res = res + "\n - channel mode DUAL CHANNEL"
elif self.channel_mode == STEREO:
res = res + "\n - channel mode STEREO"
elif self.channel_mode == JOINT_STEREO:
res = res + "\n - channel mode JOINT STEREO"
else:
res = res + "\n - channel mode %d" % self.channel_mode
res = res + "\n - nr channels %d" % self.nr_channels res += "\n - channel mode %s" % channel_mode_to_str(self.channel_mode)
res += "\n - allocation method %s" % allocation_method_to_str(self.allocation_method)
if self.allocation_method == 1:
res = res + "\n - allocation method SNR"
elif self.allocation_method == 0:
res = res + "\n - allocation method LOUNDNESS"
else:
res = res + "\n - allocation method %d" % self.allocation_method
res = res + "\n - nr subbands %d" % self.nr_subbands
res = res + "\n - bitpool %d" % self.bitpool
res = res + "\n - crc check %d" % self.crc_check
res += "\n - bitpool %d" % self.bitpool
res += "\n - crc check %d" % self.crc_check
return res return res
@ -384,7 +408,7 @@ def sbc_crc8(data, data_len):
return crc return crc
bitstream = [] bitstream = None
bitstream_index = -1 bitstream_index = -1
bitstream_bits_available = 0 bitstream_bits_available = 0
@ -410,6 +434,30 @@ def add_bits(bits, len):
for i in range(len): for i in range(len):
add_bit((bits >> (len-1-i)) & 1) add_bit((bits >> (len-1-i)) & 1)
ibuffer = None
ibuffer_count = 0
def get_bit(fin):
global ibuffer, ibuffer_count
if ibuffer_count == 0:
ibuffer = ord(fin.read(1))
ibuffer_count = 8
bit = (ibuffer >> 7) & 1
ibuffer = ibuffer << 1
ibuffer_count = ibuffer_count - 1
return bit
def drop_remaining_bits():
global ibuffer_count
ibuffer_count = 0
def get_bits(fin, bit_count):
bits = 0
for i in range(bit_count):
bits = (bits << 1) | get_bit(fin)
return bits
def calculate_crc(frame): def calculate_crc(frame):
global bitstream, bitstream_bits_available, bitstream_index global bitstream, bitstream_bits_available, bitstream_index
@ -437,6 +485,43 @@ def calculate_crc(frame):
def frame_to_bitstream(frame):
global bitstream, bitstream_bits_available, bitstream_index
init_bitstream()
add_bits(frame.syncword, 8)
add_bits(frame.sampling_frequency, 2)
add_bits(frame.nr_blocks/4-1, 2)
add_bits(frame.channel_mode, 2)
add_bits(frame.allocation_method, 1)
add_bits(frame.nr_subbands/4-1, 1)
add_bits(frame.bitpool, 8)
add_bits(frame.crc_check, 8)
if frame.channel_mode == JOINT_STEREO:
for sb in range(frame.nr_subbands-1):
add_bits(frame.join[sb],1)
add_bits(0,1)
for ch in range(frame.nr_channels):
for sb in range(frame.nr_subbands):
add_bits(frame.scale_factor[ch][sb], 4)
for blk in range(frame.nr_blocks):
for ch in range(frame.nr_channels):
for sb in range(frame.nr_subbands):
add_bits(frame.audio_sample[blk][ch][sb], frame.bits[ch][sb])
bitstream_bits_available = 0
return bitstream
def mse(a,b):
count = 1
for i in a.shape:
count *= i
delta = a - b
sqr = delta ** 2
res = sqr.sum()*1.0/count
# res = ((a - b) ** 2).mean()
return res

104
test/sbc/sbc_decoder.py
View File

@ -5,49 +5,17 @@ import struct
import sys import sys
from sbc import * from sbc import *
V = np.zeros(shape = (2, 10*2*8)) V = np.zeros(shape = (2, 10*2*8))
ibuffer = None
ibuffer_count = 0
def get_bit(fin):
global ibuffer, ibuffer_count
if ibuffer_count == 0:
ibuffer = ord(fin.read(1))
ibuffer_count = 8
# print "new byte ", hex(ibuffer)
bit = (ibuffer >> 7) & 1
ibuffer = ibuffer << 1
ibuffer_count = ibuffer_count - 1
# print "bit: ", bit
return bit
def drop_remaining_bits():
global ibuffer_count
ibuffer_count = 0
def get_bits(fin, bit_count):
bits = 0
for i in range(bit_count):
bits = (bits << 1) | get_bit(fin)
# print "collected bits", hex(bits)
return bits
def get_frame_sample_frequences(fin, bit_count):
for i in range(bit_count):
get_bit(fin)
def sbc_unpack_frame(fin, frame): def sbc_unpack_frame(fin, frame):
frame.syncword = get_bits(fin,8) frame.syncword = get_bits(fin,8)
if frame.syncword != 156: if frame.syncword != 156:
print "incorrect syncword ", frame.syncword print "incorrect syncword ", frame.syncword
return -1 return -1
frame.sampling_frequency = get_bits(fin,2) frame.sampling_frequency = get_bits(fin,2)
frame.nr_blocks = nr_blocks[get_bits(fin,2)] frame.nr_blocks = nr_blocks[get_bits(fin,2)]
frame.channel_mode = get_bits(fin,2) frame.channel_mode = get_bits(fin,2)
if frame.channel_mode == MONO: if frame.channel_mode == MONO:
frame.nr_channels = 1 frame.nr_channels = 1
else: else:
@ -66,31 +34,28 @@ def sbc_unpack_frame(fin, frame):
frame.join[sb] = get_bits(fin,1) frame.join[sb] = get_bits(fin,1)
get_bits(fin,1) # RFA get_bits(fin,1) # RFA
# print frame
frame.scale_factor = np.zeros(shape=(frame.nr_channels, frame.nr_subbands), dtype = np.int32) frame.scale_factor = np.zeros(shape=(frame.nr_channels, frame.nr_subbands), dtype = np.int32)
frame.scalefactor = np.zeros(shape=(frame.nr_channels, frame.nr_subbands), dtype = np.int32)
frame.audio_sample = np.ndarray(shape=(frame.nr_blocks, frame.nr_channels, frame.nr_subbands), dtype = np.uint16)
# print frame.audio_sample # print frame.audio_sample
for ch in range(frame.nr_channels): for ch in range(frame.nr_channels):
for sb in range(frame.nr_subbands): for sb in range(frame.nr_subbands):
frame.scale_factor[ch][sb] = get_bits(fin, 4) frame.scale_factor[ch][sb] = get_bits(fin, 4)
crc = calculate_crc(frame) crc = calculate_crc(frame)
if crc != frame.crc_check: if crc != frame.crc_check:
print frame
print "error, crc not equal: ", crc, frame.crc_check print "error, crc not equal: ", crc, frame.crc_check
exit(1) exit(1)
frame.scalefactor = np.zeros(shape=(frame.nr_channels, frame.nr_subbands), dtype = np.int32)
for ch in range(frame.nr_channels): for ch in range(frame.nr_channels):
for sb in range(frame.nr_subbands): for sb in range(frame.nr_subbands):
frame.scalefactor[ch][sb] = 1 << (frame.scale_factor[ch][sb] + 1) frame.scalefactor[ch][sb] = 1 << (frame.scale_factor[ch][sb] + 1)
frame.bits = sbc_bit_allocation(frame)
# print "bits: ", bits
#print "Nr blocks ", frame.nr_blocks, frame.nr_channels, frame.nr_subbands
frame.bits = sbc_bit_allocation(frame)
frame.audio_sample = np.ndarray(shape=(frame.nr_blocks, frame.nr_channels, frame.nr_subbands), dtype = np.uint16)
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):
for sb in range(frame.nr_subbands): for sb in range(frame.nr_subbands):
@ -99,9 +64,9 @@ def sbc_unpack_frame(fin, frame):
# add padding # add padding
drop_remaining_bits() drop_remaining_bits()
return 0
# Reconstruct the Subband Samples def sbc_reconstruct_subband_samples(frame):
frame.levels = np.zeros(shape=(frame.nr_channels, frame.nr_subbands), dtype = np.int32) frame.levels = np.zeros(shape=(frame.nr_channels, frame.nr_subbands), dtype = np.int32)
frame.sb_sample = np.zeros(shape=(frame.nr_blocks, frame.nr_channels, frame.nr_subbands)) frame.sb_sample = np.zeros(shape=(frame.nr_blocks, frame.nr_channels, frame.nr_subbands))
@ -134,7 +99,7 @@ def sbc_unpack_frame(fin, frame):
return 0 return 0
def sbc_synthesis(frame, ch, blk, proto_table): def sbc_frame_synthesis(frame, ch, blk, proto_table):
global V global V
M = frame.nr_subbands M = frame.nr_subbands
L = 10 * M L = 10 * M
@ -172,10 +137,10 @@ def sbc_synthesis(frame, ch, blk, proto_table):
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 = np.concatenate([frame.pcm, frame.X]) frame.pcm = np.concatenate([frame.pcm, np.int16(frame.X)])
def sbc_decode(frame): def sbc_synthesis(frame):
if frame.nr_subbands == 4: if frame.nr_subbands == 4:
proto_table = Proto_4_40 proto_table = Proto_4_40
elif frame.nr_subbands == 8: elif frame.nr_subbands == 8:
@ -185,16 +150,27 @@ def sbc_decode(frame):
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_synthesis(frame, ch, blk, proto_table) sbc_frame_synthesis(frame, ch, blk, proto_table)
return frame.nr_blocks * frame.nr_subbands return frame.nr_blocks * frame.nr_subbands
def sbc_decode(frame):
err = sbc_reconstruct_subband_samples(frame)
if err >= 0:
err = sbc_synthesis(frame)
return err
def write_wav_file(fout, sample):
def write_wav_file(fout, frame):
values = [] values = []
for i in range(len(sample)): for i in range(len(frame.pcm)):
packed_value = struct.pack('h', sample[i]) try:
packed_value = struct.pack('h', frame.pcm[i])
values.append(packed_value) values.append(packed_value)
except struct.error:
print frame
print i, frame.pcm[i], frame.pcm
exit(1)
value_str = ''.join(values) value_str = ''.join(values)
fout.writeframes(value_str) fout.writeframes(value_str)
@ -215,40 +191,42 @@ if __name__ == "__main__":
sys.exit(1) sys.exit(1)
wavfile = infile.replace('.sbc', '-decoded.wav') wavfile = infile.replace('.sbc', '-decoded.wav')
fout = False
with open (infile, 'rb') as fin: with open (infile, 'rb') as fin:
try: try:
frame_count = 0 frame_count = 0
while True: while True:
sbc_frame = SBCFrame(0,0,0,0,0) sbc_decoder_frame = SBCFrame(0,0,0,0,0)
if frame_count % 200 == 0: if frame_count % 200 == 0:
print "== Frame %d ==" % (frame_count) print "== Frame %d ==" % (frame_count)
err = sbc_unpack_frame(fin, sbc_frame)
err = sbc_unpack_frame(fin, sbc_decoder_frame)
if err: if err:
print "error, frame_count: ", frame_count print "error, frame_count: ", frame_count
break break
sbc_decode(sbc_frame) sbc_decode(sbc_decoder_frame)
# print sbc_frame.pcm
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_frame.nr_channels) fout.setnchannels(sbc_decoder_frame.nr_channels)
fout.setsampwidth(2) fout.setsampwidth(2)
fout.setframerate(sampling_frequency[sbc_frame.sampling_frequency]) fout.setframerate(sampling_frequency[sbc_decoder_frame.sampling_frequency])
fout.setnframes(0) fout.setnframes(0)
fout.setcomptype = 'NONE' fout.setcomptype = 'NONE'
write_wav_file(fout, sbc_frame.pcm) write_wav_file(fout, sbc_decoder_frame)
frame_count += 1 frame_count += 1
# if frame_count == 8: except TypeError as err:
# fout.close() if not fout:
# break print err
else:
except TypeError:
fout.close() fout.close()
print "DONE, SBC file %s decoded into WAV file %s ", (infile, wavfile) print ("DONE, SBC file %s decoded into WAV file %s " % (infile, wavfile))
exit(0) exit(0)
except IOError as e: except IOError as e:

180
test/sbc/sbc_encoder.py
View File

@ -5,20 +5,27 @@ import struct
import sys import sys
from sbc import * from sbc import *
X = np.zeros(80) X = np.zeros(80, dtype = np.int16)
def fetch_samples_for_next_sbc_frame(fin, nr_audio_frames, frame):
def fetch_samples_for_next_sbc_frame(fin, frame):
nr_audio_frames = frame.nr_blocks * frame.nr_subbands
raw_data = fin.readframes(nr_audio_frames) # Returns byte data raw_data = fin.readframes(nr_audio_frames) # Returns byte data
total_samples = nr_audio_frames * frame.nr_channels total_samples = nr_audio_frames * frame.nr_channels
fmt = "%ih" % total_samples # read signed 2 byte shorts len_raw_data = len(raw_data) / 2
frame.pcm = np.array(struct.unpack(fmt, raw_data)) padding = np.zeros(total_samples - len_raw_data, dtype=np.int16)
fmt = "%ih" % len_raw_data # read signed 2 byte shorts
frame.pcm = np.concatenate([np.array(struct.unpack(fmt, raw_data)), padding])
del raw_data del raw_data
def sbc_analyse(frame, ch, blk, C, debug): def sbc_frame_analysis(frame, ch, blk, C):
global X global X
M = frame.nr_subbands M = frame.nr_subbands
L = 10 * M L = 10 * M
M2 = 2*M M2 = 2*M
@ -46,23 +53,14 @@ def sbc_analyse(frame, ch, blk, C, debug):
W[i][k] = np.cos((i+0.5)*(k-2)*np.pi/M) W[i][k] = np.cos((i+0.5)*(k-2)*np.pi/M)
S[i] += W[i][k] * Y[k] S[i] += W[i][k] * Y[k]
if debug:
#print "EX:", frame.EX
print "X:", X
# print "Z:"
# print "Y:", Y
# print "W:", W
print "S:", S
for sb in range(M): for sb in range(M):
frame.sb_sample[blk][ch][sb] = S[sb] frame.sb_sample[blk][ch][sb] = S[sb]
def sbc_analysis(frame):
def sbc_encode(frame,debug):
if frame.nr_subbands == 4: if frame.nr_subbands == 4:
proto_table = Proto_4_40 C = Proto_4_40
elif frame.nr_subbands == 8: elif frame.nr_subbands == 8:
proto_table = Proto_8_80 C = Proto_8_80
else: else:
return -1 return -1
@ -71,15 +69,46 @@ def sbc_encode(frame,debug):
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):
for sb in range(frame.nr_subbands): for sb in range(frame.nr_subbands):
frame.EX[sb] = frame.pcm[index] frame.EX[sb] = np.int16(frame.pcm[index])
index+=1 index+=1
sbc_analyse(frame, ch, blk, proto_table,debug) sbc_frame_analysis(frame, ch, blk, C)
sbc_quantization(frame) return 0
def sbc_encode(frame):
err = sbc_analysis(frame)
if err >= 0:
err = sbc_quantization(frame)
return err
def should_use_joint_coding(frame): def calculate_joint_stereo_signal(frame):
# TODO: implement this sb_sample = np.zeros(shape = (frame.nr_blocks,frame.nr_channels,frame.nr_subbands), dtype = np.uint32)
return False 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 calculate_scalefactor(max_subbandsample): def calculate_scalefactor(max_subbandsample):
x = 0 x = 0
@ -91,38 +120,7 @@ def calculate_scalefactor(max_subbandsample):
return (x,y) return (x,y)
def frame_to_bitstream(frame):
global bitstream, bitstream_bits_available
init_bitstream()
add_bits(frame.syncword, 8)
add_bits(frame.sampling_frequency, 2)
add_bits(frame.nr_blocks/4-1, 2)
add_bits(frame.channel_mode, 2)
add_bits(frame.allocation_method, 1)
add_bits(frame.nr_subbands/4-1, 1)
add_bits(frame.bitpool, 8)
add_bits(frame.crc_check, 8)
for sb in range(frame.nr_subbands):
add_bits(frame.join[sb],1)
for ch in range(frame.nr_channels):
for sb in range(frame.nr_subbands):
add_bits(frame.scale_factor[ch][sb], 4)
for blk in range(frame.nr_blocks):
for ch in range(frame.nr_channels):
for sb in range(frame.nr_subbands):
add_bits(frame.audio_sample[blk][ch][sb], frame.bits[ch][sb])
return bitstream
def sbc_quantization(frame): def sbc_quantization(frame):
frame.join = np.zeros(frame.nr_subbands, dtype = np.uint8)
if should_use_joint_coding(frame):
return
max_subbandsample = np.zeros(shape = (frame.nr_channels, frame.nr_subbands)) max_subbandsample = np.zeros(shape = (frame.nr_channels, frame.nr_subbands))
for blk in range(frame.nr_blocks): for blk in range(frame.nr_blocks):
@ -135,7 +133,14 @@ def sbc_quantization(frame):
for ch in range(frame.nr_channels): for ch in range(frame.nr_channels):
for sb in range(frame.nr_subbands): for sb in range(frame.nr_subbands):
(frame.scale_factor[ch][sb], frame.scalefactor[ch][sb]) = calculate_scalefactor(max_subbandsample[ch][sb]) 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
#(frame.scale_factor[ch][sb], frame.scalefactor[ch][sb]) = calculate_scalefactor(max_subbandsample[ch][sb])
frame.bits = sbc_bit_allocation(frame) frame.bits = sbc_bit_allocation(frame)
@ -143,33 +148,42 @@ def sbc_quantization(frame):
frame.levels = np.zeros(shape=(frame.nr_channels, frame.nr_subbands), dtype = np.int32) frame.levels = np.zeros(shape=(frame.nr_channels, frame.nr_subbands), dtype = np.int32)
for ch in range(frame.nr_channels): for ch in range(frame.nr_channels):
for sb in range(frame.nr_subbands): for sb in range(frame.nr_subbands):
frame.levels[ch][sb] = pow(2.0, frame.bits[ch][sb]) - 1 frame.levels[ch][sb] = (1 << frame.bits[ch][sb]) - 1 #pow(2.0, frame.bits[ch][sb]) - 1
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):
for sb in range(frame.nr_subbands): for sb in range(frame.nr_subbands):
if frame.levels[ch][sb] > 0: if frame.levels[ch][sb] > 0:
SB = frame.sb_sample[blk][ch][sb] SB = frame.sb_sample[blk][ch][sb]
SF = frame.scalefactor[ch][sb]
L = frame.levels[ch][sb] L = frame.levels[ch][sb]
SF = frame.scalefactor[ch][sb]
frame.audio_sample[blk][ch][sb] = np.uint16(((SB * L / SF + L) - 1.0)/2.0) frame.audio_sample[blk][ch][sb] = np.uint16(((SB * L / SF + L) - 1.0)/2.0)
else: else:
frame.audio_sample[blk][ch][sb] = 0 frame.audio_sample[blk][ch][sb] = 0
return 0 return 0
def sbc_write_frame(fout, sbc_encoder_frame):
stream = frame_to_bitstream(sbc_encoder_frame)
barray = bytearray(stream)
fout.write(barray)
if __name__ == "__main__": if __name__ == "__main__":
usage = ''' usage = '''
Usage: ./sbc_encoder.py input.wav block_size nr_subbands bitpool Usage: ./sbc_encoder.py input.wav blocks subbands bitpool
Example: ./sbc_encoder.py fanfare.wav 16 4 31
''' '''
nr_blocks = 0 nr_blocks = 0
nr_subbands = 0 nr_subbands = 0
if (len(sys.argv) < 5): if (len(sys.argv) < 5):
print(usage) print(usage)
sys.exit(1) sys.exit(1)
@ -178,43 +192,37 @@ if __name__ == "__main__":
if not infile.endswith('.wav'): if not infile.endswith('.wav'):
print(usage) print(usage)
sys.exit(1) sys.exit(1)
sbcfile = infile.replace('.wav', '-encoded.sbc')
nr_blocks = int(sys.argv[2]) nr_blocks = int(sys.argv[2])
nr_subbands = int(sys.argv[3]) nr_subbands = int(sys.argv[3])
bitpool = int(sys.argv[4]) bitpool = int(sys.argv[4])
sbcfile = infile.replace('.wav', '-encoded.sbc')
fin = wave.open(infile, 'rb') fin = wave.open(infile, 'rb')
nr_channels = fin.getnchannels()
sampling_frequency = fin.getframerate()
nr_audio_frames = fin.getnframes()
wav_nr_channels = fin.getnchannels() subband_frame_count = 0
wav_sample_rate = fin.getframerate()
wav_nr_frames = fin.getnframes()
sbc_sampling_frequency = sbc_sampling_frequency_index(wav_sample_rate)
sbc_frame_count = 0
audio_frame_count = 0 audio_frame_count = 0
nr_samples = nr_blocks * nr_subbands
fout = open(sbcfile, 'wb')
while audio_frame_count < nr_audio_frames:
if subband_frame_count % 200 == 0:
print("== Frame %d ==" % (subband_frame_count))
while audio_frame_count < wav_nr_frames: sbc_encoder_frame = SBCFrame(nr_blocks, nr_subbands, nr_channels, bitpool, sampling_frequency)
# if sbc_frame_count % 200 == 0: fetch_samples_for_next_sbc_frame(fin, sbc_encoder_frame)
print "== Frame %d ==" % (sbc_frame_count)
sbc_encoder_frame = SBCFrame(nr_blocks, nr_subbands, wav_nr_channels, sbc_sampling_frequency, bitpool) sbc_encode(sbc_encoder_frame)
sbc_write_frame(fout, sbc_encoder_frame)
wav_nr_audio_frames = sbc_encoder_frame.nr_blocks * sbc_encoder_frame.nr_subbands audio_frame_count += nr_samples
fetch_samples_for_next_sbc_frame(fin, wav_nr_audio_frames, sbc_encoder_frame) subband_frame_count += 1
sbc_encode(sbc_encoder_frame, 1)
# stream = frame_to_bitstream(frame) fin.close()
audio_frame_count += wav_nr_audio_frames fout.close()
sbc_frame_count += 1 print("DONE, WAV file %s encoded into SBC file %s " % (infile, sbcfile))
if sbc_frame_count == 87:
break;
# except TypeError:
# fin.close()
# print "DONE, WAV file %s encoded into SBC file %s ", (infile, sbcfile)
#channels, num_audio_frames, wav_nr_channels, wav_sample_rate = read_waw_file(wavfile)
except IOError as e: except IOError as e: