Porcupine demo code 분석
이 글은 Raspberry Pi, Google STT를 이용한 한국어 음성 인식 프로젝트 의 일부입니다.
이전 글에서 porcupine demo를 이용하여 원하는 hot word를 탐지할 수 있도록 설정하였습니다. 이번에는 demo code를 분석하여 hot word가 인식되면 원하는 행위를 하도록 진행해 보겠습니다.
#
# Copyright 2018-2021 Picovoice Inc.
#
# You may not use this file except in compliance with the license. A copy of the license is located in the "LICENSE"
# file accompanying this source.
#
# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
# an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
#
import argparse
import os
import struct
import wave
from datetime import datetime
from threading import Thread
import pvporcupine
from pvrecorder import PvRecorder
class PorcupineDemo(Thread):
"""
Microphone Demo for Porcupine wake word engine. It creates an input audio stream from a microphone, monitors it, and
upon detecting the specified wake word(s) prints the detection time and wake word on console. It optionally saves
the recorded audio into a file for further debugging.
"""
def __init__(
self,
access_key,
library_path,
model_path,
keyword_paths,
sensitivities,
input_device_index=None,
output_path=None):
"""
Constructor.
:param library_path: Absolute path to Porcupine's dynamic library.
:param model_path: Absolute path to the file containing model parameters.
:param keyword_paths: Absolute paths to keyword model files.
:param sensitivities: Sensitivities for detecting keywords. Each value should be a number within [0, 1]. A
higher sensitivity results in fewer misses at the cost of increasing the false alarm rate. If not set 0.5 will
be used.
:param input_device_index: Optional argument. If provided, audio is recorded from this input device. Otherwise,
the default audio input device is used.
:param output_path: If provided recorded audio will be stored in this location at the end of the run.
"""
super(PorcupineDemo, self).__init__()
self._access_key = access_key
self._library_path = library_path
self._model_path = model_path
self._keyword_paths = keyword_paths
self._sensitivities = sensitivities
self._input_device_index = input_device_index
self._output_path = output_path
def run(self):
"""
Creates an input audio stream, instantiates an instance of Porcupine object, and monitors the audio stream for
occurrences of the wake word(s). It prints the time of detection for each occurrence and the wake word.
"""
keywords = list()
for x in self._keyword_paths:
keyword_phrase_part = os.path.basename(x).replace('.ppn', '').split('_')
if len(keyword_phrase_part) > 6:
keywords.append(' '.join(keyword_phrase_part[0:-6]))
else:
keywords.append(keyword_phrase_part[0])
porcupine = None
recorder = None
wav_file = None
try:
porcupine = pvporcupine.create(
access_key=self._access_key,
library_path=self._library_path,
model_path=self._model_path,
keyword_paths=self._keyword_paths,
sensitivities=self._sensitivities)
recorder = PvRecorder(device_index=self._input_device_index, frame_length=porcupine.frame_length)
recorder.start()
if self._output_path is not None:
wav_file = wave.open(self._output_path, "w")
wav_file.setparams((1, 2, 16000, 512, "NONE", "NONE"))
print('Using device: %s', recorder.selected_device)
print('Listening {')
for keyword, sensitivity in zip(keywords, self._sensitivities):
print(' %s (%.2f)' % (keyword, sensitivity))
print('}')
while True:
pcm = recorder.read()
if wav_file is not None:
wav_file.writeframes(struct.pack("h" * len(pcm), *pcm))
result = porcupine.process(pcm)
if result >= 0:
print('[%s] Detected %s' % (str(datetime.now()), keywords[result]))
except pvporcupine.PorcupineInvalidArgumentError as e:
args = (
self._access_key,
self._library_path,
self._model_path,
self._keyword_paths,
self._sensitivities,
)
print("One or more arguments provided to Porcupine is invalid: ", args)
print("If all other arguments seem valid, ensure that '%s' is a valid AccessKey" % self._access_key)
raise e
except pvporcupine.PorcupineActivationError as e:
print("AccessKey activation error")
raise e
except pvporcupine.PorcupineActivationLimitError as e:
print("AccessKey '%s' has reached it's temporary device limit" % self._access_key)
raise e
except pvporcupine.PorcupineActivationRefusedError as e:
print("AccessKey '%s' refused" % self._access_key)
raise e
except pvporcupine.PorcupineActivationThrottledError as e:
print("AccessKey '%s' has been throttled" % self._access_key)
raise e
except pvporcupine.PorcupineError as e:
print("Failed to initialize Porcupine")
raise e
except KeyboardInterrupt:
print('Stopping ...')
finally:
if porcupine is not None:
porcupine.delete()
if recorder is not None:
recorder.delete()
if wav_file is not None:
wav_file.close()
@classmethod
def show_audio_devices(cls):
devices = PvRecorder.get_audio_devices()
for i in range(len(devices)):
print('index: %d, device name: %s' % (i, devices[i]))
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--access_key',
help='AccessKey obtained from Picovoice Console (https://console.picovoice.ai/)')
parser.add_argument(
'--keywords',
nargs='+',
help='List of default keywords for detection. Available keywords: %s' % ', '.join(sorted(pvporcupine.KEYWORDS)),
choices=sorted(pvporcupine.KEYWORDS),
metavar='')
parser.add_argument(
'--keyword_paths',
nargs='+',
help="Absolute paths to keyword model files. If not set it will be populated from `--keywords` argument")
parser.add_argument('--library_path', help='Absolute path to dynamic library.', default=pvporcupine.LIBRARY_PATH)
parser.add_argument(
'--model_path',
help='Absolute path to the file containing model parameters.',
default=pvporcupine.MODEL_PATH)
parser.add_argument(
'--sensitivities',
nargs='+',
help="Sensitivities for detecting keywords. Each value should be a number within [0, 1]. A higher " +
"sensitivity results in fewer misses at the cost of increasing the false alarm rate. If not set 0.5 " +
"will be used.",
type=float,
default=None)
parser.add_argument('--audio_device_index', help='Index of input audio device.', type=int, default=-1)
parser.add_argument('--output_path', help='Absolute path to recorded audio for debugging.', default=None)
parser.add_argument('--show_audio_devices', action='store_true')
args = parser.parse_args()
if args.show_audio_devices:
PorcupineDemo.show_audio_devices()
else:
if args.access_key is None:
raise ValueError("AccessKey (--access_key) is required")
if args.keyword_paths is None:
if args.keywords is None:
raise ValueError("Either `--keywords` or `--keyword_paths` must be set.")
keyword_paths = [pvporcupine.KEYWORD_PATHS[x] for x in args.keywords]
else:
keyword_paths = args.keyword_paths
if args.sensitivities is None:
args.sensitivities = [0.5] * len(keyword_paths)
if len(keyword_paths) != len(args.sensitivities):
raise ValueError('Number of keywords does not match the number of sensitivities.')
PorcupineDemo(
access_key=args.access_key,
library_path=args.library_path,
model_path=args.model_path,
keyword_paths=keyword_paths,
sensitivities=args.sensitivities,
output_path=args.output_path,
input_device_index=args.audio_device_index).run()
if __name__ == '__main__':
main()
- 제공된 코드에서 class는 Thread class를 상속합니다.
- Multi threading을 하기 위해 Thread class를 사용합니다.
- Thread class에서는
__init__에서parameter로 받은 값을 이용하여 초기 세팅을 진행합니다. - thread.start를 하면 자동으로 run 함수를 실행하게 됩니다.
# 예시 th1 = Thread() th2 = Thread() threads.append(th1) threads.append(th2) th1.start() th2.start() - main 함수에서는 argument로 받은 값들을 parsing하고 제대로 입력을 받았는지 확인합니다.
- 모든 argument에 문제가 없다면 PorcupineDemo를 실행합니다.
- 여기에선 thread를 사용하지 않았습니다. 이후 hot word가 감지된 후 google_stt에서 결과를 받아오는 부분을 thread로 처리할 예정입니다.
- PvPorcupine class의 porcupine 객체를 생성합니다.
- PvRecorder를 이용하여 녹음한 내용을 porcupine으로 처리(porcupine.process)합니다.
-
return 받은 result가 존재한다면 hot word가 감지되었다고 판단합니다.
if result >= 0: print('[%s] Detected %s' % (str(datetime.now()), keywords[result])) - 결론적으로 result ≥ 0에 해당하는 부분에 hot word가 감지되었을 때 수행할 내용을 작성하면 됩니다!
댓글남기기