04_SCTK_06_preprocess_auto_roberta_L1.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "5/13 (Mon) | SCTK\n",
    "\n",
    "# Preprocess of Automatic Annotation Result for SCTK Alignment "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Introduction\n",
    "\n",
    "This notebook coducts the preprocess of automatic annotation results for SCTK alignment.\n",
    "The goal of the preprocess is to generate csv files which have the following four columns:\n",
    "\n",
    "- start_time ... the start time of the row in sec\n",
    "- end_time ... the end time of the row in sec\n",
    "- type ... the type of the row (01_text, 02_pause, 03_disfl, 04_filler)\n",
    "- text ... the text of the row (word, \\<CI\\>, \\<CE\\>, \\<DISFLUENCU\\>, \\<FILLER\\>)\n",
    "\n",
    "The preprocess consists of the following procedures.\n",
    "\n",
    "1. Load a turn object\n",
    "2. Load a textgrid corresponding to the turn\n",
    "3. get 01_text type rows from the turn object\n",
    "4. get 03_disfl type rows from the turn object\n",
    "5. get 04_filler type rows from the turn object\n",
    "6. get 02_pause type roes from the textgrid\n",
    "7. concat rows as a pandas' DataFrame object and sort them by start_time\n",
    "8. save the DataFrame as a csv file\n",
    "\n",
    "Before starting the preprocess, the following code block load required packages and defines global variables."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "from typing import List, Tuple, Generator\n",
    "from pathlib import Path\n",
    "\n",
    "import pandas as pd\n",
    "import pickle as pkl\n",
    "from textgrids import TextGrid\n",
    "\n",
    "sys.path.append(\n",
    "    \"/home/matsuura/Development/app/feature_extraction_api/app/modules\"\n",
    ")\n",
    "\n",
    "from fluency import Turn\n",
    "\n",
    "DATA_DIR = Path(\"/home/matsuura/Development/app/feature_extraction_api/experiment/data\")\n",
    "\n",
    "TASK = [\"WoZ_Interview\"]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "## 2. Define Functions\n",
    "\n",
    "This section defines functions to conduct the preprocess.\n",
    "The following code block defines a generator to yield file paths of Turn object and TextGrid and a function to load the Turn object and TextGrid."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def turn_textgrid_path_generator(task: str) -> Generator[Tuple[Path, Path], None, None]:\n",
    "    load_dir = DATA_DIR / f\"{task}/08_Auto_Annotation\"\n",
    "\n",
    "    suffix = \"*_roberta_L1.pkl\"\n",
    "\n",
    "    for turn_path in load_dir.glob(suffix):\n",
    "        textgrid_path = load_dir / f\"{turn_path.stem}.TextGrid\"\n",
    "\n",
    "        yield turn_path, textgrid_path\n",
    "\n",
    "\n",
    "def load_turn_and_textgrid(turn_path: Path, textgrid_path: Path) -> Tuple[Turn, TextGrid]:\n",
    "    with open(turn_path, \"rb\") as f:\n",
    "        turn = pkl.load(f)\n",
    "\n",
    "    textgrid = TextGrid(str(textgrid_path))\n",
    "\n",
    "    return turn, textgrid"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following code block defines a function to get rows from a Turn object."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def extract_rows_from_turn(turn: Turn) -> List[dict]:\n",
    "    turn.show_disfluency()\n",
    "\n",
    "    rows = []\n",
    "    for word in turn.words:\n",
    "        text_row = {\n",
    "            \"start_time\": word.start_time,\n",
    "            \"end_time\": word.end_time,\n",
    "            \"type\": \"01_text\",\n",
    "            \"text\": str(word)\n",
    "        }\n",
    "\n",
    "        rows.append(text_row)\n",
    "\n",
    "        if word.idx != -1:\n",
    "            continue\n",
    "\n",
    "        disfl_type = word.disfluency.name\n",
    "        if disfl_type == \"FILLER\":\n",
    "            filler_row = {\n",
    "                \"start_time\": word.start_time,\n",
    "                \"end_time\": word.end_time,\n",
    "                \"type\": \"04_filler\",\n",
    "                \"text\": \"<FILLER>\"\n",
    "            }\n",
    "            rows.append(filler_row)\n",
    "            continue\n",
    "\n",
    "        disfl_row = {\n",
    "            \"start_time\": word.start_time,\n",
    "            \"end_time\": word.end_time,\n",
    "            \"type\": \"03_disfl\",\n",
    "            \"text\": \"<DISFLUENCY>\"\n",
    "        }\n",
    "        rows.append(disfl_row)\n",
    "\n",
    "    return rows"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following code block defines a function to get pause rows from a TextGrid."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def extract_rows_from_textgrid(textgrid: TextGrid) -> List[dict]:\n",
    "    pause_tier = textgrid[\"pause\"]\n",
    "    \n",
    "    rows = []\n",
    "    for interval in pause_tier:\n",
    "        pause_type = interval.text\n",
    "\n",
    "        if pause_type == \"\":\n",
    "            continue\n",
    "        \n",
    "        row = {\n",
    "            \"start_time\": interval.xmin,\n",
    "            \"end_time\": interval.xmax,\n",
    "            \"type\": \"02_pause\",\n",
    "            \"text\": f\"<{pause_type}>\"\n",
    "        }\n",
    "        rows.append(row)\n",
    "\n",
    "    return rows"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "## 3. Preprocess\n",
    "\n",
    "This section conducts the preprocess."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "for task in TASK:\n",
    "    save_dir = DATA_DIR / f\"{task}/10_SCTK_Inputs\"\n",
    "\n",
    "    for turn_path, textgrid_path in turn_textgrid_path_generator(task):\n",
    "        data = []\n",
    "        \n",
    "        turn, textgrid = load_turn_and_textgrid(turn_path, textgrid_path)\n",
    "\n",
    "        data += extract_rows_from_turn(turn)\n",
    "        data += extract_rows_from_textgrid(textgrid)\n",
    "\n",
    "        df_annotation = pd.DataFrame.from_dict(data)\n",
    "        df_annotation = df_annotation.sort_values(\"start_time\").reset_index(drop=True)\n",
    "\n",
    "        filename = turn_path.stem.removesuffix(\"_roberta_L1\").removesuffix(\"_long\")\n",
    "        save_path = save_dir / f\"{filename}_auto_roberta_L1.csv\"\n",
    "        df_annotation.to_csv(save_path, index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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
}