기술/기타

[python/AzureAI] 발음평가(Pronunciation Assessment) API 사용해보기(cognitive-services-speech-sdk)

하기싫지만어떡해해야지 2024. 9. 9. 21:48

외주 개발 프로젝트에서

front에서 user의 영어 녹음 파일을 받아와서

pronunciation assessment API에 보내

발음평가 결과를 받아와야했다

 

이 작업을 하면서 삽질을 너무 많이해서

30분이면 끝날 작업을

3일에 걸쳐서 완성하게 됐는데 ...

 

삽질의 과정은 딴것보다는 wav파일을 보내는 부분 때문이었는데 ,,,

 

삽질 기록과 해결법은 다른 게시물에

이미 기록해뒀으니 참고해두면 좋을 것 같다

 

아무튼 그것만 빼면 그렇게 어려운 작업은 아니었던

이번 작업을 기록에 남겨두려고한다

 

왜냐면 얘네 Microsoft라 공식문서가 잘돼있을 줄 알았는데

그렇지 않았기때문에 ㅎ,,,


일단 이 발음평가 API를 사용하려면

미리 세팅해야하는게 2가지가 있다

 

1. speech_key와 service_region

2. cognitive-services-speech-sdk 설치

 

우선 본격적인 코딩에 앞서 1번부터 시작해보도록 하겠다

 


AzureAI speech_key, service_region 받기

https://portal.azure.com/

 

Microsoft Azure

 

portal.azure.com

 

우선 위 홈페이지에 들어간 후

로그인을 해준다

 

처음으로는 구독을 생성해줘야한다



홈페이지에서 구독 탭에 들어간 후

새로운 구독을 추가해준다

 

기존 구독이 있다면 pass해도 된다

 

대충 구독 이름은 아무거나 해주고

청구 계정이나 프로필, 섹션, 플랜은 자동으로 세팅이 된다

 

 

그 다음 구독이 생성되었다면 우린 azureAI의

speech service를 이용해야하기 때문에

음성 서비스의 리소스를 생성해주어야한다

 

 

 

이 리소스 만들기를 클릭해준 다음

 

AI+ 기계학습 을 선택한 뒤

요 음성을 클릭해준다

 

아까 새로 만들어줬던 구독을 선택해주고

리소스 그룹은 새로 만들어주면 되는데

대충 프로젝트 이름으로 만들어줬다

 

지역은 대한민국에서 이용할거니

Korea Central로 해주고

 

이름은 그냥 진짜 아무이름..으로 해줬다

흔한 이름으로 하면 이미 존재하는 endpoint라고 빠꾸먹으니

정말 잘 고안해서 해야한다

 

가격 책정 계층은 옵션이 한 개밖에 없어서 그걸로 해줬다

 

그런 다음 가장 아래의 검토+만들기 버튼을 클릭해주면

시간이 조금 지나면 음성 리소스가 생성이 된다

 

이제 speech_key와 service_region을 확인하러가보자

 

방금 생성한 음성 리소스에 들어가면

 

이런 화면이 뜨는데 저 오른쪽 아래의

키관리를 선택해준다

 

그럼 위와같은 화면이 나오는데

키 1이나 키 2중에서 아무거나 사용하면 되는 것 같다

 

나는 키 1을 사용했는데 정상적으로 API 요청이 갔다

키 1 or 키 2 = speech_key

위치/지역 = service_region

이 된다

 

나중에 코드에서 API 요청할 때 rq에 담아서 보내줘야하니

저 친구들을 꼭 잘 저장하고 있으면 된다


cognitive-services-speech-sdk 설치

 

이제 우리가 사용할 sdk를 프로젝트에 설치해야한다

https://github.com/Azure-Samples/cognitive-services-speech-sdk?tab=readme-ov-file

 

GitHub - Azure-Samples/cognitive-services-speech-sdk: Sample code for the Microsoft Cognitive Services Speech SDK

Sample code for the Microsoft Cognitive Services Speech SDK - Azure-Samples/cognitive-services-speech-sdk

github.com

 

요 github 페이지에 들어가면

내 프로젝트에 맞는 예시 코드를 찾을 수 있을 것이다

 

나는 backend에 심어줘야했는데

backend가 python으로 되어있었어서

QuickStart Python으로 들어가줬다

https://github.com/Azure-Samples/cognitive-services-speech-sdk/tree/master/quickstart/python/from-microphone

 

cognitive-services-speech-sdk/quickstart/python/from-microphone at master · Azure-Samples/cognitive-services-speech-sdk

Sample code for the Microsoft Cognitive Services Speech SDK - Azure-Samples/cognitive-services-speech-sdk

github.com

 

우선 프로젝트에 sdk 설치를 해주자

pip3 install azure-cognitiveservices-speech

 

README를 보면 우분투나 데비안은

아래 패키지도 다운받아야한다는데

난 둘다 아니므로 pass

 

아무튼 이렇게 sdk를 정상적으로 설치해주면

import azure.cognitiveservices.speech as speechsdk

 

요 import문이 정상적으로 실행되는 것을 확인해볼 수 있다


pronunciation assessment API 요청하고 결과받기

이제 본격적으로 API 요청을 날리고 결과를 받아와보자

 

https://ai.azure.com/explore/aiservices/speech/pronunciationassessment?tid=56b5b06f-62d4-4c16-b193-36e8379dae27

 

Azure AI Studio

 

ai.azure.com

 

위에서 공식문서가 잘 안나와있다고 불평했지만

위 페이지의 코드는 나름 잘나와있다

(javascript 코드는 오류가 좀 있긴했는데 ㅎ,,)

 

사실 코드 자체는 거의 위 예시 코드를 복붙했다

내가 이식한 전체 코드는 다음과 같다

 

조금 바뀐 부분이 있다면 내가 따로 추가한

각 단어별 점수를 보기 위해서 words라는 json array를

넣어주는 부분을 추가했다

 

reference_text는 정답인 text이고

temp_file_path는 audio file이 있는 경로이다

def azure_pronunciation_assessment(reference_text, temp_file_path):
    import string
    import time
    # Copyright (c) Microsoft. All rights reserved.
    # Licensed under the MIT license. See LICENSE.md file in the project root for full license information.
    try:
        import azure.cognitiveservices.speech as speechsdk
    except ImportError:
        print("""
        Importing the Speech SDK for Python failed.
        Refer to
        https://docs.microsoft.com/azure/cognitive-services/speech-service/quickstart-python for
        installation instructions.
        """)
        import sys
        sys.exit(1)
    """Performs continuous pronunciation assessment asynchronously with input from an audio file.
        See more information at https://aka.ms/csspeech/pa"""

    import difflib
    import json

    # Creates an instance of a speech config with specified subscription key and service region.
    # Replace with your own subscription key and service region (e.g., "westus").
    # Note: The sample is for en-US language.
    speech_key, service_region = "key", "koreacentral"
    # Specify the path to an audio file containing speech (mono WAV / PCM with a sampling rate of 16kHz).
    speech_config = speechsdk.SpeechConfig(subscription=speech_key, region=service_region)
    audio_config = speechsdk.audio.AudioConfig(filename=temp_file_path)

    # Create pronunciation assessment config, set grading system, granularity and if enable miscue based on your requirement.
    enable_miscue = True
    enable_prosody_assessment = True
    pronunciation_config = speechsdk.PronunciationAssessmentConfig(
        reference_text=reference_text,
        grading_system=speechsdk.PronunciationAssessmentGradingSystem.HundredMark,
        granularity=speechsdk.PronunciationAssessmentGranularity.Phoneme,
        enable_miscue=enable_miscue)
    if enable_prosody_assessment:
        pronunciation_config.enable_prosody_assessment()

    # Creates a speech recognizer using a file as audio input.
    language = 'en-US'
    speech_recognizer = speechsdk.SpeechRecognizer(speech_config=speech_config, language=language, audio_config=audio_config)
    # Apply pronunciation assessment config to speech recognizer
    pronunciation_config.apply_to(speech_recognizer)

    done = False
    recognized_words = []
    prosody_scores = []
    fluency_scores = []
    durations = []

    def stop_cb(evt: speechsdk.SessionEventArgs):
        """callback that signals to stop continuous recognition upon receiving an event `evt`"""
        print('CLOSING on {}'.format(evt))
        nonlocal done
        done = True

    def recognized(evt: speechsdk.SpeechRecognitionEventArgs):
        print("pronunciation assessment for: {}".format(evt.result.text))
        pronunciation_result = speechsdk.PronunciationAssessmentResult(evt.result)
        print("    Accuracy score: {}, prosody score: {}, pronunciation score: {}, completeness score : {}, fluency score: {}".format(
            pronunciation_result.accuracy_score, pronunciation_result.prosody_score, pronunciation_result.pronunciation_score,
            pronunciation_result.completeness_score, pronunciation_result.fluency_score
        ))
        nonlocal recognized_words, prosody_scores, fluency_scores, durations
        recognized_words += pronunciation_result.words
        fluency_scores.append(pronunciation_result.fluency_score)
        if pronunciation_result.prosody_score is not None:
            prosody_scores.append(pronunciation_result.prosody_score)
        json_result = evt.result.properties.get(speechsdk.PropertyId.SpeechServiceResponse_JsonResult)
        jo = json.loads(json_result)
        nb = jo["NBest"][0]
        durations.append(sum([int(w["Duration"]) for w in nb["Words"]]))
    # Connect callbacks to the events fired by the speech recognizer
    speech_recognizer.recognized.connect(recognized)
    speech_recognizer.session_started.connect(lambda evt: print('SESSION STARTED: {}'.format(evt)))
    speech_recognizer.session_stopped.connect(lambda evt: print('SESSION STOPPED {}'.format(evt)))
    speech_recognizer.canceled.connect(lambda evt: print('CANCELED {}'.format(evt)))
    # Stop continuous recognition on either session stopped or canceled events
    speech_recognizer.session_stopped.connect(stop_cb)
    speech_recognizer.canceled.connect(stop_cb)

    # Start continuous pronunciation assessment
    speech_recognizer.start_continuous_recognition()
    while not done:
        time.sleep(.5)

    speech_recognizer.stop_continuous_recognition()
    reference_words = [w.strip(string.punctuation) for w in reference_text.lower().split()]

    # For continuous pronunciation assessment mode, the service won't return the words with `Insertion` or `Omission`
    # even if miscue is enabled.
    # We need to compare with the reference text after received all recognized words to get these error words.
    if enable_miscue:
        diff = difflib.SequenceMatcher(None, reference_words, [x.word.lower() for x in recognized_words])
        final_words = []
        for tag, i1, i2, j1, j2 in diff.get_opcodes():
            if tag in ['insert', 'replace']:
                for word in recognized_words[j1:j2]:
                    if word.error_type == 'None':
                        word._error_type = 'Insertion'
                    final_words.append(word)
            if tag in ['delete', 'replace']:
                for word_text in reference_words[i1:i2]:
                    word = speechsdk.PronunciationAssessmentWordResult({
                        'Word': word_text,
                        'PronunciationAssessment': {
                            'ErrorType': 'Omission',
                        }
                    })
                    final_words.append(word)
            if tag == 'equal':
                final_words += recognized_words[j1:j2]
    else:
        final_words = recognized_words

    # We can calculate whole accuracy by averaging
    final_accuracy_scores = []
    for word in final_words:
        if word.error_type == 'Insertion':
            continue
        else:
            final_accuracy_scores.append(word.accuracy_score)
    accuracy_score = sum(final_accuracy_scores) / len(final_accuracy_scores)
    # Re-calculate the prosody score by averaging
    if len(prosody_scores) == 0:
        prosody_score = float("nan")
    else:
        prosody_score = sum(prosody_scores) / len(prosody_scores)
    # Re-calculate fluency score
    fluency_score = sum([x * y for (x, y) in zip(fluency_scores, durations)]) / sum(durations)
    # Calculate whole completeness score
    completeness_score = len([w for w in recognized_words if w.error_type == "None"]) / len(reference_words) * 100
    completeness_score = completeness_score if completeness_score <= 100 else 100

    print('    Paragraph accuracy score: {}, prosody score: {}, completeness score: {}, fluency score: {}'.format(
        accuracy_score, prosody_score, completeness_score, fluency_score
    ))

	#각 단어별 점수를 array로 얻고싶어서 개인적으로 추가한 코드
    words = []
    for idx, word in enumerate(final_words):
        print('    {}: word: {}\taccuracy score: {}\terror type: {};'.format(
            idx + 1, word.word, word.accuracy_score, word.error_type
        ))
        words.append({
            "idx": idx + 1,
            "word": word.word,
            "accuracy_score": word.accuracy_score,
            "error_type": word.error_type
        })
    
    return(
            {
                "accuracy_score": round(accuracy_score, 1),
                "prosody_score": round(prosody_score, 1),
                "fluency_score": round(fluency_score, 1),
                "completeness_score": round(completeness_score, 1),
                "text": '{}: word: {}\taccuracy score: {}\terror type: {};'.format( idx + 1, word.word, word.accuracy_score, word.error_type),
                "words": words
            }
    )

 

참고로 내가 필요한 부분은

정답이 있는 영어 문단 + 녹음한파일에 대한 발음평가

였기에

위와 같은 함수를 이식했다

https://github.com/Azure-Samples/cognitive-services-speech-sdk/blob/master/samples/csharp/sharedcontent/console/speech_recognition_samples.cs

 

cognitive-services-speech-sdk/samples/csharp/sharedcontent/console/speech_recognition_samples.cs at master · Azure-Samples/cogn

Sample code for the Microsoft Cognitive Services Speech SDK - Azure-Samples/cognitive-services-speech-sdk

github.com

 

위 github에서 python 코드를 보면

다양한 상황에서 사용할 수 있는 함수들이 나와있다

 

긴 글 + 마이크 음성 실시간 평가

짧은 글 + 마이크 음성 실시간 평가

긴 글+ 녹음 파일 평가

짧은 글 + 녹음 파일 평가

 

다양한 상황에서 사용할 수 있는 메소드들이 구현되어있으니

주석을 잘 읽은다음 필요한 메소드를 찾아 이식하면 될 것 같다


error가 뜨더라..

이렇게해서 순탄할 줄 알았던 나의 작업이

자꾸만 에러가 뜨는 것이다

 

audio객체를 담아서 보내주는

speech_recognizer = speechsdk.SpeechRecognizer(speech_config=speech_config, language=language, audio_config=audio_config)

 

이 부분에서 자꾸

0xa (SPXERR_INVALID_HEADER)

위와같은 에러가 떠서 진짜 너무나 삽질을 했는데

 

이 삽질 일기는

https://think0905.tistory.com/entry/python-ffmpeg%EB%A1%9C-wav%ED%8C%8C%EC%9D%BC%EC%97%90-header-%EB%84%A3%EC%96%B4%EC%A3%BC%EA%B8%B0ffmpeg-python-feat-microsoft-speech-cognitive-%EC%97%90%EB%9F%AC

 

[python] ffmpeg로 wav파일에 header 넣어주기(ffmpeg-python) (feat. microsoft speech cogni

react에서 user가 녹음한 파일을 백엔드로 받아와서 백엔드에서 발음평가 API를 날려 점수를 받아오는 기능을 구현해야했는데,, 내가 이용한 발음평가 API는 Azure AI의 pronunciation assessment였고 cognitive-

think0905.tistory.com

이 게시글에 따로 정리를 해뒀다..

 

결론부터 말하면 user의 마이크에서 받아와서

녹음파일을 저장해줬는데

새로 생성한 wav파일이다보니

header정보가 누락되어서 

ffmpeg 라이브러리를 통해 wav header를

새로 생성해주었다

 

같은 에러가 뜬다면 위 게시글을 참고하면 좋을 것 같다


아무튼 에러까지 해결해서 API로 발음 평가를 날려주면

 

위와같이 눈물겹게 result가 잘 나오는 것을 확인할 수 있다

 

wav파일 설정만 잘해주면 생각보다 간단한

Azure AI pronunciation assessment 이용하기,,

 

기록은 여기까지 -!