def transcribe_multichannel()

in infra-as-code/modules/ingest-pipeline/cf-stt-transcript/lib.py [0:0]

• 47 lines of code
• 7 McCabe index (conditional complexity)

  def transcribe_multichannel(self, audio_file_uri):
    """Calls the Speech Client to do a transcription for a dual-channel file
       located in GCS

    Args:
        audio_file_uri (str): gsutil URI to the audio file in GCS

    Raises:
        Exception: Speech client exception for the LRO

    Returns:
        dict: Dictionary with the audio metadata to send as HTTP response
    """
    print('Starting Speech client')
    creds = self.get_credentials()
    speech_client = SpeechClient(credentials = creds)

    config = cloud_speech.RecognitionConfig(
        auto_decoding_config=cloud_speech.AutoDetectDecodingConfig()
    )

    file_metadata = cloud_speech.BatchRecognizeFileMetadata(uri=audio_file_uri)

    request = cloud_speech.BatchRecognizeRequest(
        recognizer=self.recognizer_path,
        config=config,
        files=[file_metadata],
        recognition_output_config=cloud_speech.RecognitionOutputConfig(
            #inline_response_config=cloud_speech.InlineOutputConfig(),
            gcs_output_config = cloud_speech.GcsOutputConfig(uri=self.transcript_bucket_uri)
        ),
    )

    operation = speech_client.batch_recognize(request=request)
    
    print("Waiting for operation to complete...")
    for i in range(60):
      if operation.done() == True:
        operation_data = operation.operation
        if hasattr(operation_data, 'error') and str(operation_data.error) != '':
            print(f'Operation error: {operation_data.error}')
            raise Exception(str(operation_data.error))
        else:
          print('Transcription finished')
          response = operation.result()
          break
      else:
        print(f'Operation still running, sleeping...')
        time.sleep(30)

    print(f'Response from speech client:  {response}')

    transcript_uri = ''
    for filename, file_result in response.results.items():
        if file_result.cloud_storage_result:
            transcript_uri = file_result.cloud_storage_result.uri
            print(f'Filename: {filename}, Cloud Storage URI: {transcript_uri}')
        else:
            print(f'Filename: {filename} has no cloud_storage_result.')

    transcript_bucket, transcript_filename = self.extract_bucket_and_filename(transcript_uri)
    self.event_dict['transcript_bucket'] = transcript_bucket
    self.event_dict['transcript_filename'] = transcript_filename
    self.event_dict['event_bucket'] = self.formatted_audio_bucket_id
    self.event_dict['event_filename'] = self.formatted_audio_file_name
    self.event_dict['original_file_name'] = self.original_file_name

    self.order_transcript(transcript_bucket, transcript_filename)

    return self.event_dict