前置き
という訳で「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