Python pyaudio 模块，get_sample_size() 实例源码

def __init__(self, device_index = None, sample_rate = None, chunk_size = None):
            assert device_index is None or isinstance(device_index, int), "Device index must be None or an integer"
            if device_index is not None: # ensure device index is in range
                audio = pyaudio.PyAudio(); count = audio.get_device_count(); audio.terminate() # obtain device count
                assert 0 <= device_index < count, "Device index out of range"
            assert sample_rate is None or isinstance(sample_rate, int) and sample_rate > 0, "Sample rate must be None or a positive integer"
            assert chunk_size is None or isinstance(chunk_size, int) and chunk_size > 0, "Chunk size must be None or a positive integer"
            if sample_rate is None: chunk_size = 16000
            if chunk_size is None: chunk_size = 1024
            self.device_index = device_index
            self.format = pyaudio.paInt16 # 16-bit int sampling
            self.SAMPLE_WIDTH = pyaudio.get_sample_size(self.format) # size of each sample
            self.SAMPLE_RATE = sample_rate # sampling rate in Hertz
            self.CHUNK = chunk_size # number of frames stored in each buffer

            self.audio = None
            self.stream = None

def __init__(self, rt):
        super().__init__(rt)
        config = rt.config['frontends']['speech']['recognizers']
        self.listener_config = config

        self.chunk_size = config['chunk_size']
        self.format = pyaudio.paInt16
        self.sample_width = pyaudio.get_sample_size(self.format)
        self.sample_rate = config['sample_rate']
        self.channels = config['channels']

        self.p = pyaudio.PyAudio()
        self.stream = self.p.open(format=self.format, channels=self.channels,
                                  rate=self.sample_rate, input=True,
                                  frames_per_buffer=self.chunk_size)

        self.buffer_sec = config['wake_word_length']
        self.talking_volume_ratio = config['talking_volume_ratio']
        self.required_integral = config['required_noise_integral']
        self.max_di_dt = config['max_di_dt']
        self.noise_max_out_sec = config['noise_max_out_sec']
        self.sec_between_ww_checks = config['sec_between_ww_checks']
        self.recording_timeout = config['recording_timeout']
        self.energy_weight = 1.0 - pow(1.0 - config['ambient_adjust_speed'],
                                       self.chunk_size / self.sample_rate)

        # For convenience
        self.chunk_sec = self.chunk_size / self.sample_rate

        self.av_energy = None
        self.integral = 0
        self.noise_level = 0
        self._intercept = None

def active_listen_to_all_options(self, THRESHOLD=None, LISTEN=True,
                                     MUSIC=False):
        """
            Records until a second of silence or times out after 12 seconds

            Returns a list of the matching options or None
        """

        RATE = 16000
        CHUNK = 1024
        LISTEN_TIME = 12

        # check if no threshold provided
        if THRESHOLD is None:
            THRESHOLD = self.fetch_threshold()

        self.speaker.play(jessypath.data('audio', 'beep_hi.wav'))

        # prepare recording stream
        stream = self._audio.open(format=pyaudio.paInt16,
                                  channels=1,
                                  rate=RATE,
                                  input=True,
                                  frames_per_buffer=CHUNK)

        frames = []
        # increasing the range # results in longer pause after command
        # generation
        lastN = [THRESHOLD * 1.2 for i in range(30)]

        for i in range(0, RATE / CHUNK * LISTEN_TIME):

            data = stream.read(CHUNK)
            frames.append(data)
            score = self.get_score(data)

            lastN.pop(0)
            lastN.append(score)

            average = sum(lastN) / float(len(lastN))

            # TODO: 0.8 should not be a MAGIC NUMBER!
            if average < THRESHOLD * 0.8:
                break

        self.speaker.play(jessypath.data('audio', 'beep_lo.wav'))

        # save the audio data
        stream.stop_stream()
        stream.close()

        with tempfile.SpooledTemporaryFile(mode='w+b') as f:
            wav_fp = wave.open(f, 'wb')
            wav_fp.setnchannels(1)
            wav_fp.setsampwidth(pyaudio.get_sample_size(pyaudio.paInt16))
            wav_fp.setframerate(RATE)
            wav_fp.writeframes(''.join(frames))
            wav_fp.close()
            f.seek(0)
            return self.active_stt_engine.transcribe(f)