LoginSignup
5
7

More than 3 years have passed since last update.

@BA-KAZUin株式会社ソントレーゾ

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

Last updated at Posted at 2020-01-10

前置き

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

5
7
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
5
7