从麦克风实时录音发送到服务端（一般单路测试使用）
()
| 120 | |
| 121 | |
| 122 | async def record_microphone(): |
| 123 | """从麦克风实时录音发送到服务端（一般单路测试使用）""" |
| 124 | try: |
| 125 | import pyaudio |
| 126 | except ImportError as e: |
| 127 | raise ImportError( |
| 128 | "缺少 PyAudio，麦克风推流前请先运行 `pip install pyaudio`" |
| 129 | ) from e |
| 130 | |
| 131 | global voices |
| 132 | FORMAT = pyaudio.paInt16 |
| 133 | CHANNELS = 1 |
| 134 | RATE = 16000 |
| 135 | chunk_size = 60 * args.chunk_size[1] / args.chunk_interval |
| 136 | CHUNK = int(RATE / 1000 * chunk_size) |
| 137 | |
| 138 | p = pyaudio.PyAudio() |
| 139 | |
| 140 | stream = p.open( |
| 141 | format=FORMAT, channels=CHANNELS, rate=RATE, input=True, frames_per_buffer=CHUNK |
| 142 | ) |
| 143 | # hotwords |
| 144 | fst_dict = {} |
| 145 | hotword_msg = "" |
| 146 | if args.hotword.strip() != "": |
| 147 | if os.path.exists(args.hotword): |
| 148 | f_scp = open(args.hotword, encoding="utf-8") |
| 149 | hot_lines = f_scp.readlines() |
| 150 | for line in hot_lines: |
| 151 | words = line.strip().split(" ") |
| 152 | if len(words) < 2: |
| 153 | print("Please checkout format of hotwords") |
| 154 | continue |
| 155 | try: |
| 156 | fst_dict[" ".join(words[:-1])] = int(words[-1]) |
| 157 | except ValueError: |
| 158 | print("Please checkout format of hotwords") |
| 159 | hotword_msg = json.dumps(fst_dict, ensure_ascii=False) |
| 160 | else: |
| 161 | hotword_msg = args.hotword |
| 162 | |
| 163 | use_itn = True |
| 164 | if args.use_itn == 0: |
| 165 | use_itn = False |
| 166 | |
| 167 | message = json.dumps( |
| 168 | { |
| 169 | "mode": args.mode, |
| 170 | "chunk_size": args.chunk_size, |
| 171 | "chunk_interval": args.chunk_interval, |
| 172 | "encoder_chunk_look_back": args.encoder_chunk_look_back, |
| 173 | "decoder_chunk_look_back": args.decoder_chunk_look_back, |
| 174 | "wav_name": "microphone", |
| 175 | "is_speaking": True, |
| 176 | "hotwords": hotword_msg, |
| 177 | "itn": use_itn, |
| 178 | }, |
| 179 | ensure_ascii=False, |