Azure CognitiveServicesのSpeakerRecognition APIによる話者識別をPythonで検証してみた。#2

## 前置き

という訳で「Speaker Recognition API」を使った話者識別の実際の検証について記載していきます。
（変なところがあったら教えてください！）

## 処理の流れ

話者識別をするには、以下の3ステップが必要になります。

1. ユーザのプロファイルを作成する
2. ユーザのプロファイルに音声を登録する
3. 登録した音声を元に、誰が発言したのか識別する

なので今回は自分が解りやすいように、それぞれのステップ毎で3つの処理を作成していきたいと思います。

## ステップ1 ユーザのプロファイルを作成する

まずは話者識別させたいユーザを作成していきます。
API機能としては「Identification Profile」の「Create Profile」を使います。
これを使用するとユーザのプロファイルを作成して、ユーザのプロファイルIDを返してきます。
（名前は登録されないので、別でリスト管理する必要があります）

検証スクリプトでは引数にユーザ名を指定する形にして「Profile_List.csv」というファイルにユーザ名とIDを紐づけて出力しています。

__`CreateProfile.py`__

    ########### module #############
    import sys                  # 引数を格納するためのライブラリ
    import requests             # HTTPの通信を行うためのライブラリ
    import json                 # データをjson形式で使うためのライブラリ
    import base64
    import csv

    ########### Args & variable #########################
    args = sys.argv
    Profile_Name = args[1]
    Profile_List = 'Profile_List.csv'

    ########### Create Profile #########################
    with open(Profile_List) as fp:
        lst = list(csv.reader(fp))

    for i in lst:
        if Profile_Name in i:
            print('指定されたユーザは既に登録されています。')
            sys.exit()

    ApiPath = 'https://speaker-recognitionapi.cognitiveservices.azure.com/spid/v1.0/identificationProfiles'

    headers = {
        # Request headers
        'Content-Type': 'application/json',
        'Ocp-Apim-Subscription-Key': '<サブスクリプションキー>',
    }

    body = {
        'locale':'en-us',
    }

    r = requests.post(
        ApiPath,            # URL 
        headers = headers,  # ヘッダー
        json = body         # ボディ
    )

    try:
        ProfileId = r.json()['identificationProfileId']
    except Exception:
        print('Error:{}'.format(r.status_code))
        print(r.json()['error'])
        sys.exit()

    print(ProfileId)

    f = open(Profile_List, 'a')
    writer = csv.writer(f, lineterminator='\n')
    writer.writerow([Profile_Name, ProfileId])
    ####################################

## ステップ2 ユーザのプロファイルに音声を登録する

上記で作成したユーザに音声を登録していきます。
（話者認証とは違ってフレーズが指定されていないので、内容はなんでもOKなようです）

ここでは以下の機能を使用しています。

1.「Identification Profile」の「Create Enrollment」（音声登録）
2.「Speaker Recognition」の「Get Operation Status」（登録状況の確認）

あと個人的にめちゃくちゃハマったのですが、利用できるオーディオファイルにかなり厳しめな制約があります。

|プロパティ|必須値|
|:--|:--|
|コンテナー|WAV|
|エンコード|PCM|
|レート|16K|
|サンプル形式|16 ビット|
|チャネル|モノラル|

なかなか条件を満たす音声が取れなかったのですが、「Audacity」という無料ソフトでなんとか録音することが出来ました。（これめっちゃ便利）

スクリプトの引数はユーザ名にしています。
（音声ファイルにユーザ名が付いている前提ですが、検証だしいいよね）

__`CreateEnrollment.py`__

    ########### module #############
    import sys                  # 引数を格納するためのライブラリ
    import requests             # HTTPの通信を行うためのライブラリ
    import json                 # データをjson形式で使うためのライブラリ
    import base64
    import csv
    import time

    ########### Args & variable #########################
    args = sys.argv
    Profile_Name = args[1]
    Profile_List = 'Profile_List.csv'
    WavFile = f'{Profile_Name}.wav'

    with open(Profile_List) as fp:
        lst = list(csv.reader(fp))

    for i in lst:
        if Profile_Name in i:
            break

    j = lst.index(i)
    ProfileId = lst[j][1]

    ########### Create Enrollment #########################
    ApiPath = f'https://speaker-recognitionapi.cognitiveservices.azure.com/spid/v1.0/identificationProfiles/{ProfileId}/enroll?shortAudio=true'

    headers = {
        # Request headers
        'Content-Type': 'application/octet-stream',
        'Ocp-Apim-Subscription-Key': '<サブスクリプションキー>',
    }

    with open(WavFile, 'rb') as f:
        body = f.read()

    r = requests.post(
        ApiPath,            # URL 
        headers = headers,  # ヘッダー
        data = body         # ボディ
    )

    try:
        response = r
        print('response:', response.status_code)
        if response.status_code == 202:
            print(response.headers['Operation-Location'])
            operation_url = response.headers['Operation-Location']
        else:
            print(response.json()['error'])
            sys.exit()
    except Exception:
        print(r.json()['error'])
        sys.exit()
    ####################################
    ########### Get Operation Status #########################
    url = operation_url

    headers = {
        # Request headers
        'Ocp-Apim-Subscription-Key': '<サブスクリプションキー>',
    }

    status = ''
    while status != 'succeeded':
        
        r = requests.get(
            url,            # URL 
            headers = headers,  # ヘッダー
        )

        try:
            response = r
            print('response:', response.status_code)
            if response.status_code == 200:
                status = response.json()['status']
                print(f'現在の状態；{status}')
                if status == 'failed':
                    message = response.json()['message']
                    print(f'error:{message}')
                    sys.exit()
                elif status != 'succeeded':
                    time.sleep(3)
            else:
                print(r.json()['error'])
                sys.exit()
        except Exception:
            print(r.json()['error'])
            sys.exit()

    enrollmentStatus = response.json()['processingResult']['enrollmentStatus']
    remainingEnrollmentSpeechTime = response.json()['processingResult']['remainingEnrollmentSpeechTime']
    speechTime = response.json()['processingResult']['speechTime']

    if enrollmentStatus == 'enrolling':
        status = 'プロファイルは現在、登録中であり、識別の準備はできていません。'
    elif enrollmentStatus == 'training':
        status = 'プロファイルは現在、トレーニング中であり、識別の準備はできていません。'
    else:
        status = 'プロファイルは現在、登録中であり、識別の準備ができています。'

    print(f'\nステータス；{enrollmentStatus}')
    print(f'現在の状態；{status}')
    print(f'有効な音声の合計時間(秒):{speechTime}')
    print(f'登録を成功させるのに必要な残りの音声時間(秒):{remainingEnrollmentSpeechTime}')

## ステップ3 登録した音声を元に、誰が発言したのか識別する

いよいよメインの処理です。
ここでは以下の機能を使用しています。

1.「Speaker Recognition」の「Identification」（話者識別）
2.「Speaker Recognition」の「Get Operation Status」（識別結果の取得）

今回の検証では引数を識別したい音声ファイルにしています。
ちなみに話者識別ですが、今のところ同時に検証できるのは10ユーザ（プロファイル）までのようです。
処理としては「Identification」で識別したい音声とプロファイルID（複数）をPOSTして、戻ってくる`Operation-Location`というURLに対して、「Get Operation Status」を実行し、識別の状況と結果を取得するイメージです。｛検証では識別完了まで最大9秒くらいかかってました）
また、識別結果として返ってくるのは「プロファイルID」なので、別途ユーザ名に置き換える必要があります。なお識別の信頼度も一緒に返ってくるのですが、こちらは低・中・高の3段階あるようですね。

__`Identification.py`__

    ########### module #############
    import sys                  # 引数を格納するためのライブラリ
    import requests             # HTTPの通信を行うためのライブラリ
    import json                 # データをjson形式で使うためのライブラリ
    import base64
    import csv
    import time

    ########### Args & variable #########################
    args = sys.argv
    WavFile = args[1]
    Profile_List = 'Profile_List.csv'

    with open(Profile_List) as fp:
        lst = list(csv.reader(fp))

    ########### Identification #########################
    ProfileIds = ''
    for a, b in lst:
        ProfileIds += b + ','

    ProfileIds = ProfileIds[:-1]

    url = 'https://speaker-recognitionapi.cognitiveservices.azure.com/spid/v1.0/identify'

    params = {
        'identificationProfileIds': ProfileIds,
        'shortAudio': True,
    }

    headers = {
        # Request headers
        'Content-Type': 'application/octet-stream',
        'Ocp-Apim-Subscription-Key': '<サブスクリプションキー>',
    }

    with open(WavFile, 'rb') as f:
        body = f.read()

    r = requests.post(
        url,            # URL 
        params = params,
        headers = headers,  # ヘッダー
        data = body         # ボディ
    )

    try:
        response = r
        print('response:', response.status_code)
        if response.status_code == 202:
            print(response.headers['Operation-Location'])
            operation_url = response.headers['Operation-Location']
        else:
            print(response.json()['error'])
            sys.exit()
    except Exception:
        print(r.json()['error'])
        sys.exit()

    ####################################
    ########### Get Operation Status #########################
    url = operation_url
    #url = 'https://speaker-recognitionapi.cognitiveservices.azure.com/spid/v1.0/operations/ea1edc22-32f4-4fb9-81d6-d597a0072c76'

    headers = {
        # Request headers
        'Ocp-Apim-Subscription-Key': '<サブスクリプションキー>',
    }

    status = ''
    while status != 'succeeded':
        
        r = requests.get(
            url,            # URL 
            headers = headers,  # ヘッダー
        )

        try:
            response = r
            print('response:', response.status_code)
            if response.status_code == 200:
                status = response.json()['status']
                print(f'現在の状態；{status}')
                if status == 'failed':
                    message = response.json()['message']
                    print(f'error:{message}')
                    sys.exit()
                elif status != 'succeeded':
                    time.sleep(3)
            else:
                print(r.json()['error'])
                sys.exit()
        except Exception:
            print(r.json()['error'])
            sys.exit()

    identifiedProfileId = response.json()['processingResult']['identifiedProfileId']
    confidence = response.json()['processingResult']['confidence']

    for i in lst:
        if identifiedProfileId in i:
            break

    j = lst.index(i)
    Profile_Name = lst[j][0]

    print(f'\n発言者；{Profile_Name}')
    print(f'信頼度；{confidence}')
    ####################################

## おわり

という事で今回は「Speaker Recognition API」を検証してみました。
日本語には未対応との事だったのですが、個人的に話者識別はなかなかの精度なんじゃないかと感じました。
上手く活用すれば色々な事ができそうですね！

## 前の記事

[Azure CognitiveServicesのSpeakerRecognition APIによる話者識別をPythonで検証してみた。#1](https://qiita.com/BA-KAZU/items/72e9665524f093f54ede)