transcription_service.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def transcribe_gcs(gcs_uri):\n",
    "    \"\"\"Asynchronously transcribes the audio file specified by the gcs_uri.\"\"\"\n",
    "    from google.cloud import speech\n",
    "    from google.cloud.speech import enums\n",
    "    \n",
    "    client = speech.SpeechClient()\n",
    "    audio  = speech.types.RecognitionAudio(uri = gcs_uri)\n",
    "    config = speech.types.RecognitionConfig(\n",
    "        encoding = enums.RecognitionConfig.AudioEncoding.FLAC,\n",
    "        language_code = 'en-US')\n",
    "    \n",
    "    operation = client.long_running_recognize(config, audio)\n",
    "    print('Waiting for operation to complete...')\n",
    "    response = operation.result()\n",
    "    return response\n",
    "\n",
    "def set_google_app_credentials(root):\n",
    "    from os import environ\n",
    "    from os.path import join\n",
    "    credentials_json = \"PATH/TO/APPLICATION/CREDENTIALS\"\n",
    "    credentials = join(root, credentials_json)\n",
    "    try:\n",
    "        environ[\"GOOGLE_APPLICATION_CREDENTIALS\"] = credentials\n",
    "        print(\"Set!\")\n",
    "    except:\n",
    "        print(\"NOT Set!\")\n",
    "        \n",
    "def get_dirs_and_wav_files():\n",
    "    from os import getcwd, listdir\n",
    "    from os.path import join, splitext\n",
    "    root = getcwd()\n",
    "    wav_dir = join(root, 'wav-files')\n",
    "    wav_dir_success = join(wav_dir, 'success')\n",
    "    wav_dir_fail = join(wav_dir, 'fail')\n",
    "    wav_files = [join(wav_dir, wav) for wav in listdir(wav_dir) if splitext(wav)[1] == '.wav']\n",
    "    flac_dir = join(root, 'flac-files')\n",
    "    transcriptions_dir = join(root, 'transcriptions')\n",
    "    return root, wav_dir, wav_dir_success, wav_dir_fail, wav_files, flac_dir, transcriptions_dir\n",
    "\n",
    "def move_wav_file(conversion_outcome, wav, wav_dir_success, wav_dir_fail):\n",
    "    from shutil import move\n",
    "    if conversion_outcome:\n",
    "        move(wav, wav_dir_success)\n",
    "    else:\n",
    "        move(wav, wav_dir_fail)\n",
    "        \n",
    "def get_sample_rate(wav):\n",
    "    from pydub.utils import mediainfo\n",
    "    info = mediainfo(wav)\n",
    "    sample_rate = int(info['sample_rate'])\n",
    "    return sample_rate\n",
    "\n",
    "def reset_sample_and_channel(wav, sample_rate):\n",
    "    from pydub import AudioSegment\n",
    "    audio = AudioSegment.from_file(wav, format = \"wav\", frame_rate = sample_rate)\n",
    "    audio = audio.set_frame_rate(16000)\n",
    "    audio = audio.set_channels(1)\n",
    "    return audio\n",
    "\n",
    "def convert_to_flac(wav, flac_dir):\n",
    "    from os.path import basename, splitext, join\n",
    "    basename = basename(wav)\n",
    "    filename = splitext(basename)[0]\n",
    "    flac_path = join(flac_dir, filename + '.flac')\n",
    "    x = 1\n",
    "    try:\n",
    "        audio.export(flac_path, format = 'flac')\n",
    "        return 1\n",
    "    except:\n",
    "        return 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "'''GET DIRS'''\n",
    "\n",
    "root, wav_dir, wav_dir_success, wav_dir_fail, wav_files, flac_dir, transcriptions_dir = get_dirs_and_wav_files()\n",
    "wav_files = wav_files[0:1]\n",
    "wav_files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "'''CONVERT .WAV to .FLAC'''\n",
    "\n",
    "for wav in wav_files:\n",
    "    sample_rate = get_sample_rate(wav)\n",
    "    audio = reset_sample_and_channel(wav, sample_rate)\n",
    "    conversion_outcome = convert_to_flac(wav, flac_dir)\n",
    "    move_wav_file(conversion_outcome, wav, wav_dir_success, wav_dir_fail)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "'''Manually upload the converted .flac files in /flac-files to GC storage bucket'''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "'''TRANSCRIPTION SERVICE'''\n",
    "\n",
    "import os\n",
    "import pandas as pd\n",
    "\n",
    "set_google_app_credentials(root)\n",
    "bucket = \"PATH/TO/BUCKET\"\n",
    "\n",
    "samples = ['BC13-58_SFAudio_144_a_cleaned.flac', 'BC13-58_SFAudio_144_b_cleaned.flac']\n",
    "results_ledger = os.path.join(root, 'results_ledger' + '.txt')\n",
    "\n",
    "for sample in samples:\n",
    "    '''Build Paths'''\n",
    "    sample_basename = os.path.splitext(sample)[0]\n",
    "    sample_filename = os.path.basename(sample_basename)\n",
    "    \n",
    "    csv_dir = os.path.join(transcriptions_dir, sample_filename)\n",
    "    csv_file = os.path.join(csv_dir, sample_filename + '.csv')\n",
    "    txt_file = os.path.join(csv_dir, sample_filename + '.txt')\n",
    "    \n",
    "    '''Transcribe'''\n",
    "    gcs_uri = os.path.join(bucket, sample)\n",
    "    results = transcribe_gcs(gcs_uri).results\n",
    "    all_transcriptions = [result.alternatives[0].transcript for result in results]\n",
    "    all_confidence = [result.alternatives[0].confidence for result in results]\n",
    "    result_dict = {'transcription': all_transcriptions, 'confidence': all_confidence}\n",
    "    \n",
    "    '''Write to dir'''\n",
    "    if not os.path.exists(csv_dir):\n",
    "        os.mkdir(csv_dir)\n",
    "    \n",
    "    df = pd.DataFrame(result_dict)\n",
    "    df.to_csv(txt_file, encoding = 'utf-8', index = False, header = False, sep = ' ')\n",
    "    \n",
    "    '''Write results to ledger'''\n",
    "    row = {'sample_basename': sample_basename, 'confidence_mean': df.confidence.mean()}\n",
    "    df_row = pd.DataFrame(row, index=[0])\n",
    "    df_row.to_csv(results_ledger, encoding='utf-8', mode = 'a', index = False, header = False, sep = ' ')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}