在按鍵按鍵事件中多次調用on_press回調

Question

我試圖從我的麥克風錄音，同時按下按鈕，問題是我正在使用的庫無法檢測到保持事件。它只能檢測在印刷機上，其中發生一次，這意味着麥克風只記錄一個樣本..

import pyaudio 
import wave 
from pynput import keyboard 

CHUNK = 8192 
FORMAT = pyaudio.paInt16 
CHANNELS = 2 
RATE = 44100 
RECORD_SECONDS = 5 
WAVE_OUTPUT_FILENAME = "output.wav" 

p = pyaudio.PyAudio() 
stream = p.open(format=FORMAT, 
       channels=CHANNELS, 
       rate=RATE, 
       input=True, 
       frames_per_buffer=CHUNK)   
frames = [] 

def on_press(key): 
    if key == keyboard.Key.cmd_l: 
     print('- Started recording -'.format(key)) 
     try: 
      data = stream.read(CHUNK) 
      frames.append(data) 
     except IOError: 
      print 'warning: dropped frame' # can replace with 'pass' if no message desired 
    else: 
     print('incorrect character {0}, press cmd_l'.format(key)) 


def on_release(key): 
    print('{0} released'.format(
     key)) 
    if key == keyboard.Key.cmd_l: 
     print('{0} stop'.format(key)) 
     keyboard.Listener.stop 
     return False 

print("* recording") 


with keyboard.Listener(on_press=on_press, on_release=on_release) as listener: 
    listener.join() 

print("* done recording") 

stream.stop_stream() 
stream.close() 
p.terminate() 

wf = wave.open(WAVE_OUTPUT_FILENAME, 'wb') 
wf.setnchannels(CHANNELS) 
wf.setsampwidth(p.get_sample_size(FORMAT)) 
wf.setframerate(RATE) 
wf.writeframes(b''.join(frames)) 
wf.close() 

我使用鍵盤事件和pyaudio記錄。 它似乎記錄了cmd_l被按下時停止，當我釋放。問題是生成的音頻文件不包含任何內容，或者很短，如0.19s長。

我想這可能與stream.read有關，必須多次調用，並且按鍵僅被記錄一次，因此只有一個樣本正在被記錄。

但是，如果是這種情況，我怎麼讓它調用stream.read多次，同時也檢測到它也檢測到它被釋放時？ 或者是否有支持鍵盤上的on_hold機制的庫？

與線程更新方法：

from pynput import keyboard 
import time 
import pyaudio 
import wave 

CHUNK = 8192 
FORMAT = pyaudio.paInt16 
CHANNELS = 2 
RATE = 44100 
RECORD_SECONDS = 5 
WAVE_OUTPUT_FILENAME = "output.wav" 

p = pyaudio.PyAudio() 
frames = [] 

def callback(in_data, frame_count, time_info, status): 
    return (in_data, pyaudio.paContinue) 

class MyListener(keyboard.Listener): 
    def __init__(self): 
     super(MyListener, self).__init__(self.on_press, self.on_release) 
     self.key_pressed = None 

     self.stream = p.open(format=FORMAT, 
          channels=CHANNELS, 
          rate=RATE, 
          input=True, 
          frames_per_buffer=CHUNK, 
          stream_callback = self.callback) 
     print self.stream.is_active() 

    def on_press(self, key): 
     if key == keyboard.Key.cmd_l: 
      self.key_pressed = True 

    def on_release(self, key): 
     if key == keyboard.Key.cmd_l: 
      self.key_pressed = False 

    def callback(self,in_data, frame_count, time_info, status): 
     if self.key_pressed == True: 
      return (in_data, pyaudio.paContinue) 
     elif self.key_pressed == False: 
      return (in_data, pyaudio.paComplete) 
     else: 
      return (in_data,pyaudio.paAbort) 


listener = MyListener() 
listener.start() 
started = False 

while True: 
    time.sleep(0.1) 
    if listener.key_pressed == True and started == False: 
     started = True 
     listener.stream.start_stream() 
     print "start Stream" 

    elif listener.key_pressed == False and started == True: 
     print "Something coocked" 
     listener.stream.stop_stream() 
     listener.stream.close() 
     p.terminate() 

     wf = wave.open(WAVE_OUTPUT_FILENAME, 'wb') 
     wf.setnchannels(CHANNELS) 
     wf.setsampwidth(p.get_sample_size(FORMAT)) 
     wf.setframerate(RATE) 
     wf.writeframes(b''.join(frames)) 
     wf.close() 

     started = False 

我嘗試了不同的方法引入兩個線程，其中一個監測鍵盤，而另一個挑選出的記錄。但是這個解決方案不斷下降幀，因此沒有什麼被實際記錄？

怎麼回事？

Answer 1

下面是一個如何做到這一點的示例：我已經通過使單獨的線程進行記錄來修改代碼。鍵盤按下/釋放設置事件的狀態，並且當事件被設置時，線程在循環中記錄塊。一個單獨的事件用於告訴線程在完成記錄後終止。這爲我記錄了多個32Kb塊（8192 * 2 chans * 16位）。

import pyaudio 
import wave 
from pynput import keyboard 
import threading 

CHUNK = 8192 
FORMAT = pyaudio.paInt16 
CHANNELS = 2 
RATE = 44100 
RECORD_SECONDS = 5 
WAVE_OUTPUT_FILENAME = "output.wav" 

p = pyaudio.PyAudio() 

stream = p.open(format=FORMAT, 
       channels=CHANNELS, 
       rate=RATE, 
       input=True, 
       frames_per_buffer=CHUNK)   
frames = [] 

recordingEvent = threading.Event() # set to activate recording 
exitEvent = threading.Event()   # set to stop recording thread 


def on_press(key): 
    if key == keyboard.Key.ctrl: 
     print('- Started recording -'.format(key)) 
     recordingEvent.set() 
    else: 
     print('incorrect character {0}, press cmd_l'.format(key)) 


def on_release(key): 
    print('{0} released'.format(key)) 
    if key == keyboard.Key.ctrl: 
     print('{0} stop'.format(key)) 
     recordingEvent.clear() 
     keyboard.Listener.stop 
     return False 


def do_recording(): 
    while (not exitEvent.is_set()): 
     if (recordingEvent.wait(0.1)): 
      try: 
       data = stream.read(CHUNK) 
       # print len(data) 
       frames.append(data) 
      except IOError: 
       print 'warning: dropped frame' # can replace with 'pass' if no message desired 


class myRecorder(threading.Thread): 
    def __init__(self): 
     threading.Thread.__init__(self) 
    def run(self): 
     do_recording() 


# start recorder thread 
recordingThread = myRecorder() 
recordingThread.start() 

# monitor keyboard 
with keyboard.Listener(on_press=on_press, on_release=on_release) as listener: 
    listener.join() 

# stop recorder thread 
exitEvent.set()  
recordingThread.join() 

print("* done recording") 

stream.stop_stream() 
stream.close() 
p.terminate() 

wf = wave.open(WAVE_OUTPUT_FILENAME, 'wb') 
wf.setnchannels(CHANNELS) 
wf.setsampwidth(p.get_sample_size(FORMAT)) 
wf.setframerate(RATE) 
wf.writeframes(b''.join(frames)) 
wf.close() 

NB我用Ctrl鍵代替了Cmd鍵，因爲它更適合我的機器。希望這是有用的。

Answer 2

您可以在this answer中執行類似於在單獨線程上定期運行函數的功能。似乎是最簡單的解決方案。