前置き

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

処理の流れ

話者識別をするには、以下の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