adaptation.py 文件源码

def write(self, audio_data, length):
        # Pass audio_data=None to write silence
        if length == 0:
            return 0

        with self._lock:
            write_ptr = self._ds_buffer.lock(self._write_cursor_ring, length)
            assert 0 < length <= self._buffer_size
            assert length == write_ptr.audio_length_1.value + write_ptr.audio_length_2.value

            if audio_data:
                ctypes.memmove(write_ptr.audio_ptr_1, audio_data.data, write_ptr.audio_length_1.value)
                audio_data.consume(write_ptr.audio_length_1.value, self.source_group.audio_format)
                if write_ptr.audio_length_2.value > 0:
                    ctypes.memmove(write_ptr.audio_ptr_2, audio_data.data, write_ptr.audio_length_2.value)
                    audio_data.consume(write_ptr.audio_length_2.value, self.source_group.audio_format)
            else:
                if self.source_group.audio_format.sample_size == 8:
                    c = 0x80
                else:
                    c = 0
                ctypes.memset(write_ptr.audio_ptr_1, c, write_ptr.audio_length_1.value)
                if write_ptr.audio_length_2.value > 0:
                    ctypes.memset(write_ptr.audio_ptr_2, c, write_ptr.audio_length_2.value)
            self._ds_buffer.unlock(write_ptr)

            self._write_cursor += length
            self._write_cursor_ring += length
            self._write_cursor_ring %= self._buffer_size