| """Waveform/byte streaming bridge with no predeclared utterance length.""" |
| import torch |
|
|
| from tinyasr.codec_live_bitstream_v1 import LiveBitReader,LiveBitWriter |
| from tinyasr.codec_streaming_v1 import StreamingDecoder,StreamingEncoder |
|
|
|
|
| class LiveEncoder: |
| def __init__(self,codec,model_sha256): |
| self.audio=StreamingEncoder(codec);self.transport=LiveBitWriter(model_sha256) |
| self.header_sent=False |
|
|
| def start(self): |
| if self.header_sent:return b'' |
| self.header_sent=True;return self.transport.header |
|
|
| def push(self,audio): |
| if (self.audio.samples+audio.numel()+319)//320>self.transport.max_frames: |
| raise ValueError('Live input exceeds frame limit') |
| codes=self.audio.push(audio) |
| return self.start()+self.transport.push(codes[0].numpy()) |
|
|
| def finish(self): |
| codes=self.audio.finish() |
| return self.start()+self.transport.push(codes[0].numpy())+self.transport.finish(self.audio.samples) |
|
|
| @property |
| def float_state_bytes(self):return self.audio.state_bytes |
|
|
|
|
| class LiveDecoder: |
| def __init__(self,codec,model_sha256): |
| self.audio=StreamingDecoder(codec);self.transport=LiveBitReader(model_sha256) |
|
|
| def push(self,data): |
| codes=self.transport.push(data) |
| return self.audio.push(torch.from_numpy(codes[None])) |
|
|
| def finish(self): |
| end=self.transport.finish() |
| return self.audio.finish(end.samples),end |
|
|
| @property |
| def float_state_bytes(self):return self.audio.state_bytes |
|
|
