Paraformer 实时语音识别 Python SDK

import os
import signal  # for keyboard events handling (press "Ctrl+C" to terminate recording)
import sys

import dashscope
import pyaudio
from dashscope.audio.asr import *

mic = None
stream = None

# Set recording parameters
sample_rate = 16000  # sampling rate (Hz)
channels = 1  # mono channel
dtype = 'int16'  # data type
format_pcm = 'pcm'  # the format of the audio data
block_size = 3200  # number of frames per buffer


def init_dashscope_api_key():
  """
  Set your DashScope API-key. More information:
  https://github.com/aliyun/alibabacloud-bailian-speech-demo/blob/master/PREREQUISITES.md
  """
  if 'DASHSCOPE_API_KEY' in os.environ:
    dashscope.api_key = os.environ['DASHSCOPE_API_KEY']
  else:
    dashscope.api_key = '<your-dashscope-api-key>'


# Real-time speech recognition callback
class Callback(RecognitionCallback):
  def on_open(self) -> None:
    global mic
    global stream
    print('RecognitionCallback open.')
    mic = pyaudio.PyAudio()
    stream = mic.open(format=pyaudio.paInt16,
      channels=1,
      rate=16000,
      input=True)

  def on_close(self) -> None:
    global mic
    global stream
    print('RecognitionCallback close.')
    stream.stop_stream()
    stream.close()
    mic.terminate()
    stream = None
    mic = None

  def on_complete(self) -> None:
    print('RecognitionCallback completed.')

  def on_error(self, message) -> None:
    print('RecognitionCallback task_id: ', message.request_id)
    print('RecognitionCallback error: ', message.message)
    if 'stream' in globals() and stream.active:
      stream.stop()
      stream.close()
    sys.exit(1)

  def on_event(self, result: RecognitionResult) -> None:
    sentence = result.get_sentence()
    if 'text' in sentence:
      print('RecognitionCallback text: ', sentence['text'])
      if RecognitionResult.is_sentence_end(sentence):
        print(
          'RecognitionCallback sentence end, request_id:%s, usage:%s'
          % (result.get_request_id(), result.get_usage(sentence)))


def signal_handler(sig, frame):
  print('Ctrl+C pressed, stop recognition ...')
  recognition.stop()
  print('Recognition stopped.')
  print(
    '[Metric] requestId: {}, first package delay ms: {}, last package delay ms: {}'
    .format(
      recognition.get_last_request_id(),
      recognition.get_first_package_delay(),
      recognition.get_last_package_delay(),
    ))
  sys.exit(0)


if __name__ == '__main__':
  init_dashscope_api_key()
  print('Initializing ...')

  callback = Callback()

  recognition = Recognition(
    model='paraformer-realtime-v2',
    format=format_pcm,
    sample_rate=sample_rate,
    semantic_punctuation_enabled=False,
    callback=callback)

  recognition.start()

  signal.signal(signal.SIGINT, signal_handler)
  print("Press 'Ctrl+C' to stop recording and recognition...")

  while True:
    if stream:
      data = stream.read(3200, exception_on_overflow=False)
      recognition.send_audio_frame(data)
    else:
      break

  recognition.stop()