langdetect.py

#!/usr/local/bin/python3
import datetime
import itertools
import json
import logging
import os
import time
import uuid
from argparse import ArgumentParser
from datetime import datetime
import grpc
import speech_pb2
import speech_pb2_grpc
from webex_ci_sso import sso_access

parser = ArgumentParser(description='Audio Client')
parser.add_argument('--expected_language_codes', type=str, default="")
parser.add_argument('--language_code', type=str, default="en")
parser.add_argument('--sample_rate', type=int, default=16000)
parser.add_argument('--encoding', type=str, default="LINEAR16")
parser.add_argument('--enable_automatic_punctuation', type=bool, default=True)
parser.add_argument('--enable_multilang_detection', type=str, default=True)
parser.add_argument('--enable_training_deprecated', type=bool, default=False)
parser.add_argument('--final-only', type=bool, default=False),
parser.add_argument('--file', type=str, default="")
parser.add_argument('--longrunning', type=bool, default=True)
parser.add_argument('--url', type=str, default="")
parser.add_argument('--callback_url', type=str, default="")
parser.add_argument('--engine', type=str, default="k2")
parser.add_argument('--enable_phonetic_output', type=bool, default=False)
parser.add_argument('--server', type=str, default="asr-proxy.uscentral1-1.cint.vcra.co:443")
parser.add_argument('--config_key_name', type=str, default="")
parser.add_argument('--recording_upload', type=bool, default=False)
parser.add_argument('--csi_file', type=str, default="")
parser.add_argument('--csi_file_url', type=str, default="")
parser.add_argument('--meeting_id', type=str, default="")
parser.add_argument('--get_transcript', type=bool, default=False)
parser.add_argument('--cleanup', type=str, default="yes")
parser.add_argument('--json_output', type=str, default="yes")

args = parser.parse_args()
supportedAudioFormats = {
    "mp3": "audio/mp3",
    "mpeg": "audio/mpeg",
    "ogg": "audio/ogg",
    "wav": "audio/wav",
    "flac": "audio/flac",
}

_LOGGER = logging.getLogger(__name__)
_LOGGER.setLevel(logging.INFO)
_SERVER_ADDR_TEMPLATE = args.server

USER_ID=os.getenv("CI_USER_ID","voicea-asr-sample-client-bts")
USER_PASSWD=os.getenv("CI_USER_PASSWD", "UGMP.stgh.38.AGXI.vwym.37.XHKL.rwkp.2578") # Vault path: cint/asrproxy/sample_python_client
USER_REALM=os.getenv("CI_ORG_ID","a93dde14-65b7-4a59-81dd-28962a8473e3")
BROKER_HOST=os.getenv("BORKER_HOST", "idbrokerbts.webex.com")
CLIENT_ID=os.getenv("CI_CLIENT_ID", "C6ac166d70e8ea8a59e0a341c7690ddc0e3de836bc1c702d31c211501358b594b")
CLIENT_SECRET=os.getenv("CI_CLIENT_SECRET", "19d82f38891800bcd9bc636d986b6a0d037ec0fc7aae72161e60c7eb0c67e930")
SCOPE=os.getenv("CI_SCOPE", "Identity:SCIM Identity:Organization Identity:Config webex-voice-assistant:asr")
__ci_token__ = "" 
__token_expire__ = None
__sso__ = sso_access(orgId=USER_REALM, machineAccount=USER_ID, machineAccountPW=USER_PASSWD, oauth_host=BROKER_HOST)
 
def make_channel():
    """Creates a secure channel with auth credentials from the environment."""
    # Grab application default credentials from the environment
    creds = grpc.ssl_channel_credentials()
    #options = (('grpc.ssl_target_name_override', 'nutcracker.grpc.staging.vcra.co',),)
    max_message_length = 16 * 1024 * 1024  # Example: 16 MB
    options=[
        ('grpc.max_send_message_length', max_message_length),
        ('grpc.max_receive_message_length', max_message_length)
    ]
    #secure
    if _SERVER_ADDR_TEMPLATE.endswith("443"):
        return grpc.secure_channel(_SERVER_ADDR_TEMPLATE, creds, options=options)
    else:
        return grpc.insecure_channel(_SERVER_ADDR_TEMPLATE, options=options)


def getAccessToken():
    global __ci_token__
    global __token_expire__
    if 1==1:
        if __ci_token__ == "" or \
           (__token_expire__ - datetime.now()).total_seconds() < 5:
            __ci_token__, __token_expire__ = __sso__.getAccessToken(CLIENT_ID, CLIENT_SECRET, scope=SCOPE)
        return __ci_token__
    return None

def chunk_array(arr, length):
    return [arr[0+i:length+i] for i in range(0, len(arr), length)]


def word_time_to_seconds(word_time):
    return word_time.seconds + (word_time.nanos / 1_000_000_000)


def print_utterances(transcripts):
    for tr in transcripts:
        _time_from = (
            word_time_to_seconds(tr.start_time)
            if tr.start_time
            else None
        )
        _time_to = (
            word_time_to_seconds(tr.end_time)
            if tr.end_time
            else None
        )
        _speaker_id = tr.speaker_ids[0] if len(tr.speaker_ids) > 0 else None
        _speaker_label = tr.speaker_labels[0] if len(tr.speaker_labels) > 0 else None
        
        meta = ""
        if _speaker_id:
            meta += f"[speaker_id:{_speaker_id}]"
        if _speaker_label:
            meta += f"[speaker_label:{_speaker_label}]"
        if tr.language_code:
            meta += f"[language:{tr.language_code}]"

        time_range = ""
        if _time_from and _time_to:
            time_range = f"[{_time_from}-{_time_to}] "

        text = tr.alternatives[0].transcript
        
        if meta:
            print(f"{meta}")
        print(f"{time_range}{text}")
        print("=" * 80)

def longRunningRecognize(stub, url, fileName, callbackUrl, expectedLanguageCodes, csi_file_url=""):
    trackingId = 'asr_client_python_'+str(uuid.uuid4())
    # print("Tracking ID will be "+ str(trackingId))
    if fileName:
        with open(fileName, 'rb') as f:
            data = f.read()
        RATE = args.sample_rate
        metadata = (('authorization', 'Bearer ' + getAccessToken()),
                    ('trackingid', str(trackingId)),
                    ('configkeyname', str(args.config_key_name)))
        request_config = speech_pb2.RecognitionConfig(sample_rate_hertz=RATE,max_alternatives=1)
        request_config.enable_multilang_detection = args.enable_multilang_detection
        request_config.language_code = args.language_code
        if(len(expectedLanguageCodes)>0):
            request_config.enable_automatic_language_detection = True
            request_config.expected_language_codes.extend(expectedLanguageCodes)
        request_config.enable_training_deprecated = args.enable_training_deprecated        
        request_config.enable_phonetic_output = args.enable_phonetic_output
        request_config.encoding = speech_pb2.RecognitionConfig.LINEAR16 
        request_audio = speech_pb2.RecognitionAudio(content=data)
        
        recognize_request = speech_pb2.LongRunningRecognizeRequest(
                config=request_config, audio=request_audio, callback_url=callbackUrl, csi_file_url=csi_file_url)
        
    else:
        RATE = args.sample_rate
        metadata = (('authorization', 'Bearer ' + getAccessToken()),
                    ('trackingid', str(trackingId)),
                    ('configkeyname', str(args.config_key_name)))
        request_config = speech_pb2.RecognitionConfig(sample_rate_hertz=RATE, max_alternatives=1)
        request_config.enable_automatic_punctuation = args.enable_automatic_punctuation
        request_config.enable_multilang_detection = args.enable_multilang_detection
        request_config.enable_training_deprecated = args.enable_training_deprecated
        request_config.language_code = args.language_code
        if len(expectedLanguageCodes) > 0:
            request_config.enable_automatic_language_detection = True
            request_config.expected_language_codes.extend(expectedLanguageCodes)

        request_audio = speech_pb2.RecognitionAudio(uri=url)
        recognize_request = speech_pb2.LongRunningRecognizeRequest(
                config=request_config, audio=request_audio, callback_url=callbackUrl, csi_file_url=csi_file_url)
    response = stub.LongRunningRecognize(recognize_request, metadata=metadata)
    if len(response.name) > 0:
        print("Job Id is "+ str(response.name))
        return response.name
    return ""


def getLongRunningRecognizeOperation(stub, meetingId):
    transcripts = []
    try:
        trackingId = 'asr_client_python_'+str(uuid.uuid4())
        metadata = (('authorization', 'Bearer ' + getAccessToken()),
                    ('trackingid', str(trackingId)))
        operationRequest = speech_pb2.LongRunningRecognizeOperationRequest(name=meetingId)
        operation = stub.GetLongRunningRecognizeOperationV2(operationRequest, metadata=metadata)
        if operation.done:
            if operation.error is not None and operation.error.code > 0:
                print("Response error:",operation.error)
                return (True, None, None, None)

            # Ucomment below line to see the full response
            jsonOutput = {}
            jsonTranscripts = []
            languagesDuration = {}
            jsonLanguages = {}
            jsonOutput["transcripts"] = jsonTranscripts
            jsonOutput["languages"] = jsonLanguages
            jsonOutput["meeting_id"] = meetingId

            for r in operation.response.results:
                words = []
                for w in r.alternatives[0].words:
                    start_time = w.start_time
                    end_time = w.end_time
                    words.append({
                        "word": w.word,
                        "start_time": { "seconds": start_time.seconds, "nanos": start_time.nanos },
                        "end_time": { "seconds": end_time.seconds, "nanos": end_time.nanos },
                    })
                transcripts.append(r)
                alt={}
                alt['transcript'] = r.alternatives[0].transcript
                alt['confidence'] = r.alternatives[0].confidence
                if r.language_code:
                    alt['language_code'] = r.language_code  
                if r.start_time:
                    alt['start_time'] = r.start_time.seconds + (r.start_time.nanos / 1_000_000_000)
                else:
                    alt['start_time'] = 0.0
                if r.end_time:
                    alt['end_time'] = r.end_time.seconds + (r.end_time.nanos / 1_000_000_000)
                if languagesDuration.get(r.language_code, None) is None: 
                    languagesDuration[r.language_code] = alt['end_time'] - alt['start_time']
                else:
                    languagesDuration[r.language_code] += alt['end_time'] - alt['start_time']
                jsonTranscripts.append(alt)

            #calculate percentage per language
            total_duration = 0.0
            for lang in languagesDuration.keys():
                total_duration += languagesDuration[lang]
            for lang in languagesDuration.keys():
                jsonLanguages[lang] = languagesDuration[lang] / total_duration
            print("final result:", jsonOutput)
            return True, transcripts, words, jsonOutput
        else:
            # print("Transcript is not ready yet, current progress: %i, meetingId=%s" % (operation.metadata.progress_percent, meetingId))
            return False, None, None, None
    except Exception as ex:
        print("failed at", datetime.now())
        print("failed ex %s" % ex)

def deleteLongRunningRecognizeOperation(stub, meetingId):
    try:
        trackingId = 'asr_client_python_'+str(uuid.uuid4())
        metadata = (('authorization', 'Bearer ' + getAccessToken()),
                    ('trackingid', str(trackingId))) 
        operationRequest = speech_pb2.LongRunningRecognizeOperationRequest(name=meetingId)
        stub.DeleteLongRunningRecognizeOperation(operationRequest, metadata=metadata)
    except Exception as ex:
        print ("failed to delete the meeting %s" % ex)

    
def long_running_recognize(stub, expected_language_codes):
    url = args.url
    
    file = args.file
    if (file == None or len(file)==0) and (url == None or len(url)==0):
        url = "https://www2.cs.uic.edu/~i101/SoundFiles/preamble.wav"
    callback = args.callback_url
    csi_file_url = args.csi_file_url
    meetingId = longRunningRecognize(stub, 
                                     url=url, 
                                     fileName=file, 
                                     callbackUrl=callback,
                                     expectedLanguageCodes=expected_language_codes,
                                     csi_file_url=csi_file_url)
    done = False
    words = []
    transcripts = ""
    while not done:
        current = datetime.now()
        (done, transcripts, words, jsonOutput) =  getLongRunningRecognizeOperation(stub, meetingId)
        # print("time-elapsed", (current-starttime))
        time.sleep(5)
    if args.json_output == "yes" and jsonOutput is not None:
        with open(f'{meetingId}.json', 'w+') as fd:
            json.dump(jsonOutput, fd)
        # print(f"JSON output written to {meetingId}.json")
        langcode = jsonOutput.get("languages", {})
        print("langcode:", langcode)
        # print(jsonOutput)
    else:
        # print("Final Utterances:")
        print_utterances(transcripts)

    with open(f'{file}.words.json', 'w+') as fd:
        json.dump(words, fd)

    # print("args: %s" % args)
    if args.cleanup == "yes":
        # print("meeting delete: %s" % meetingId)
        deleteLongRunningRecognizeOperation(stub, meetingId)


def main():
    expected_language_codes = []
    if args.expected_language_codes != "":
       expected_language_codes = args.expected_language_codes.split(",")

    channel = make_channel()
    stub = speech_pb2_grpc.SpeechStub(channel)
    long_running_recognize(stub, expected_language_codes)
    
if __name__ == '__main__':
    main()