02_Preprocess_03_fill_startoken_monologue_auto.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "5/10 (Fri) | Preprocess\n",
    "\n",
    "# Post-Process for Forced Alignment Outputs of Monologue Data (ASR Transcript)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Introduction\n",
    "\n",
    "This notebook fills star tokens in the forced alignment outputs focusing on monologue data.\n",
    "The procedure consists of the followign stages.\n",
    "\n",
    "1. Load a rev-format transcript\n",
    "2. Load a corresponding forced alignment output file\n",
    "3. Preprocess the rev-format transcript\n",
    "4. Fill star tokens (i.e., unsupported characters in wav2vec forced alignment)\n",
    "5. Save the result\n",
    "\n",
    "The following code block loads required packages and define global variables."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing import List, Generator\n",
    "from pathlib import Path\n",
    "import re\n",
    "\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "DATA_DIR = Path(\"/home/matsuura/Development/app/feature_extraction_api/experiment/data\")\n",
    "\n",
    "TASK = [\"Arg_Oly\", \"Cartoon\", \"RtSwithoutRAA\", \"RtSwithRAA\"]\n",
    "\n",
    "PUNCTUATIONS = [\".\", \",\", \":\", \"?\", \"!\"]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "## 2. Define Functions\n",
    "\n",
    "This section defines functions to complete the post-process.\n",
    "The following code block defines a generator of rev-format transcript path."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def rev_transcript_path_generator(task: str) -> Generator[Path, None, None]:\n",
    "    load_dir = DATA_DIR / f\"{task}/02_Rev_Transcript\"\n",
    "\n",
    "    for rev_transcript_path in load_dir.glob(\"*_long.csv\"):\n",
    "        yield rev_transcript_path"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following code blocks define functions to load a rev-format transcript and forced alignment output."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def load_rev_transcript(rev_transcript_path: Path) -> pd.DataFrame:\n",
    "    df_transcript = pd.read_csv(rev_transcript_path, index_col=0, na_values=[\"\", \" \"], keep_default_na=False)\n",
    "\n",
    "    return df_transcript"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def load_fa_output(rev_transcript_path: Path, task: str) -> pd.DataFrame:\n",
    "    filename = rev_transcript_path.stem.removesuffix(\"_long\")\n",
    "\n",
    "    fa_output_path = DATA_DIR / f\"{task}/04_FA_csv_Auto/{filename}.csv\"\n",
    "    df_fa = pd.read_csv(fa_output_path)\n",
    "\n",
    "    return df_fa"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following code block defines functions to transform a rev-format transcript."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def extract_word_list(df_transcript: pd.DataFrame) -> List[str]:\n",
    "    punct_mask = (df_transcript[\"type\"] == \"punct\")\n",
    "    df_transcript_wo_punct = df_transcript[~punct_mask]\n",
    "\n",
    "    word_list = df_transcript_wo_punct[\"text\"].to_list()\n",
    "\n",
    "    return word_list\n",
    "\n",
    "def convert_word_list(word_list: List[str]) -> np.ndarray:\n",
    "    # 1. get word list\n",
    "    word_list_conv = []\n",
    "    number_pattern = r\"\\d\"\n",
    "    for word in word_list:\n",
    "        if re.match(number_pattern, word) is None:\n",
    "            word_list_conv.append(word)\n",
    "            continue\n",
    "        \n",
    "        # if the word consists of numbers, split them\n",
    "        numbers = word.split(\" \")\n",
    "        for number in numbers:\n",
    "            word_list_conv.append(number)\n",
    "    \n",
    "    # 2. temporally change an inaudible tags to star token\n",
    "    fa_transcript = \" \".join(word_list_conv)\n",
    "    fa_transcript = fa_transcript.replace(\"<inaudible>\", \"*\")\n",
    "\n",
    "    # 3. remove other tags\n",
    "    tag_pattern = r\"\\<.*?\\>\"\n",
    "    fa_transcript = re.sub(tag_pattern, \" \", fa_transcript) \n",
    "\n",
    "    # 4. recover star token to inaudible tags\n",
    "    fa_transcript = fa_transcript.replace(\"*\", \"<inaudible>\")\n",
    "\n",
    "    # 5. lower transcript\n",
    "    fa_transcript = fa_transcript.lower()\n",
    "\n",
    "    # 6. remove punctuations\n",
    "    for punct in PUNCTUATIONS:\n",
    "       fa_transcript = fa_transcript.replace(punct, \" \") \n",
    "\n",
    "    # 7. remove extra pauses\n",
    "    while \"  \" in fa_transcript:\n",
    "        fa_transcript = fa_transcript.replace(\"  \", \" \")\n",
    "\n",
    "    if fa_transcript[0] == \" \":\n",
    "        fa_transcript = fa_transcript[1:]\n",
    "    if fa_transcript[-1] == \" \":\n",
    "        fa_transcript = fa_transcript[:-1]\n",
    "\n",
    "    return np.array(fa_transcript.split(\" \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following code block defines a function to check the number of words in a rev-format transcript and the corresponding forced alignment output."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "def is_same_length(word_list: np.ndarray, df_fa: pd.DataFrame) -> bool:\n",
    "    return len(word_list) == len(df_fa)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following code block defines a function to fill star tokens in a forced alignment output."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "def fill_startokens(word_list: np.ndarray, df_fa: pd.DataFrame) -> pd.DataFrame:\n",
    "    df_fa_filled = df_fa.copy(deep=True)\n",
    "\n",
    "    mask_star = (df_fa[\"word\"] == \"*\").to_numpy()\n",
    "\n",
    "    df_fa_filled.loc[mask_star, \"word\"] = word_list[mask_star]\n",
    "\n",
    "    return df_fa_filled"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following code block defines a function to save a forced alignment output in which star tokens are filled."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "def save_df_fa_filled(df_fa_filled: pd.DataFrame, rev_transcript_path: Path, task: str) -> None:\n",
    "    filename = rev_transcript_path.stem.removesuffix(\"_long\")\n",
    "    save_path = DATA_DIR / f\"{task}/04_FA_csv_Auto/{filename}_filled.csv\"\n",
    "\n",
    "    df_fa_filled.to_csv(save_path, index=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "## 3. Conduct Post-Process\n",
    "\n",
    "The following code block conducts post-process."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "for task in TASK:\n",
    "    for rev_transcript_path in rev_transcript_path_generator(task):\n",
    "        df_transcript = load_rev_transcript(rev_transcript_path)\n",
    "        df_fa = load_fa_output(rev_transcript_path, task)\n",
    "\n",
    "        word_list = extract_word_list(df_transcript)\n",
    "        word_list = convert_word_list(word_list)\n",
    "\n",
    "        if is_same_length(word_list, df_fa):\n",
    "            df_fa_filled = fill_startokens(word_list, df_fa)\n",
    "            save_df_fa_filled(df_fa_filled, rev_transcript_path, task)\n",
    "\n",
    "        else:\n",
    "            print(f\"[Error] {rev_transcript_path.stem}\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "teai-incremental-classifier",
   "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.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}