1.A. Transcription_models.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "994bb7bb",
   "metadata": {},
   "source": [
    "## Transcription of speech segments (diarized single- or multi-speaker audio)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ff702c55",
   "metadata": {},
   "source": [
    "In this script, we will use the **'Whisper-timestamped' model** (https://github.com/linto-ai/whisper-timestamped) to transcribe speech segments of a **multi-speaker audio file**. This is an adaptation of OpenAi's Whisper that indicates for each word and sentence begin and ending timestamps. We will use these timestamps to write the whisper output to a **.textgrid** file. *Note*: to transcribe a **single-speaker audio file without previous diarization**, please use the model `transcribe_file_total()` defined in the following section."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "32e23be1",
   "metadata": {},
   "source": [
    "The algorithm takes as an input a **diarized .textgrid** file, i.e. a file where speech segments of different speakers are indicated on different tiers. Importanly, the speech segments need to be filled with some characters (e.g. 'speech') to distinguish themselves from the intervals between the speech segments. The algorithm expects that the .textgrid contains multiple tiers that each indicate the speech of one speaker (see parameter 'diarized_speakers='all''), and no other tiers (for example for annotations). If you have only diarized the speech of **one speaker**, please indicate **diarized_speakers='AC'** and name the tier of the speaker 'AC' (Autistic Child). Please indicate the language spoken in the parameter 'language'. The code should work for all languages supported by Whisper. If you don't know which language is spoken, please indicate 'language='unknown'': Whisper will then predict the language on the basis of the audio and use the corresponding language model."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "221c9967",
   "metadata": {},
   "source": [
    "The output of the model is a **new .textgrid**, whose output path to the new file is specified in the parameters of the function. The model will transcribe the speech of every speaker in three different tiers: **an 'utterance' tier, a 'word' tier and a 'warning' tier**. In the 'warning' tier, an interval is created when the utterance only contains one word or when it contains a word that appears in a list of trigger words, that often occur when Whisper tries to generate random words on noise. This list is completed for French and Dutch. If you are using this model for another language, feel free to discover which words Whisper generates on silent or noisy parts and add some trigger words to the `trigger_list`. The name of the tier corresponds to the name of the tier in the diarization textgrid, followed by an underscore and 'word', 'utterance' or 'warning', e.g. 'AC_utterance', 'AC_word' and 'AC_warning'."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c9ac7b26",
   "metadata": {},
   "source": [
    "Part of this notebook is based on Oriane Martin's 'Whispgrid': https://github.com/orianemartin/WhispGrid"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "2a221bb6",
   "metadata": {},
   "outputs": [],
   "source": [
    "def transcribe_file_segments(language, selected_model, audio_file, diarization_textgrid, output_path, speakers_diarized='all'):\n",
    "    \n",
    "    ## Import libraries\n",
    "    from pydub import AudioSegment\n",
    "    from praatio import textgrid\n",
    "    import whisper_timestamped as whisper\n",
    "    from whisper.tokenizer import get_tokenizer\n",
    "    import torch\n",
    "    import tgt\n",
    "\n",
    "    \n",
    "    ## Load audio via 'AudioSegment' library => used to calculate duration of the audio file\n",
    "    audio = AudioSegment.from_wav(audio_file)\n",
    "        \n",
    "    ## Configurate Whisper timestamped model\n",
    "    \n",
    "    # Choose tokenizer\n",
    "    tokenizer = get_tokenizer(multilingual=True)\n",
    "    \n",
    "    # The variable 'number_tokens' will contain the full words for numbers in the language in question,\n",
    "    # so that we can let Whisper spell out numbers instead of writing them in digits\n",
    "    number_tokens = [\n",
    "        i \n",
    "        for i in range(tokenizer.eot)\n",
    "        if all(c in \"0123456789\" for c in tokenizer.decode([i]).strip())]\n",
    "    \n",
    "    # Load selected Whisper timestamped model\n",
    "    model = whisper.load_model(selected_model, device=\"cpu\")\n",
    "    \n",
    "    whisper_languages= whisper.tokenizer.LANGUAGES\n",
    "    \n",
    "    # Get language codes from 'language' attribute input for model configuration\n",
    "    if language=='French':\n",
    "        language_code='fr'\n",
    "    elif language==\"Flemish\":\n",
    "        language_code='nl'\n",
    "    # For other languages:\n",
    "    elif language.lower() in whisper_languages.values():\n",
    "        language_code= str(list(whisper_languages.keys())[list(whisper_languages.values()).index(language.lower())])\n",
    "    # If language unknown: let Whisper detect the language\n",
    "    elif language== 'unknown':\n",
    "        w_audio = whisper.load_audio(audio_file)\n",
    "        w_audio = whisper.pad_or_trim(w_audio)\n",
    "\n",
    "        # make log-Mel spectrogram and move to the same device as the model\n",
    "        mel = whisper.log_mel_spectrogram(w_audio).to(selected_model.device)\n",
    "\n",
    "        # detect the spoken language\n",
    "        _, probs = model.detect_language(mel)\n",
    "        language_code= str(max(probs, key=probs.get))\n",
    "    \n",
    "    # Warning-triggering list\n",
    "    trigger_list= [\"Amara.org\", \"d'Amara.org\",\"sous-titres\",\"ondertitels\",\"ondertiteling\"]\n",
    "    \n",
    "    \n",
    "    ## Transribe the audio file using Whisper timestamped => based on 'Whispgrid' script\n",
    "    ## Use configurations defined above\n",
    "    def transcribe_segment(start_int, end_int):\n",
    "        audio[start_int*1000: end_int*1000].export('segment.wav', format='wav')\n",
    "        \n",
    "        result = whisper.transcribe(\n",
    "            model,\n",
    "            'segment.wav',\n",
    "            #vad= 'auditok', # gives errors and does not seem to improve the alignment\n",
    "            detect_disfluencies= True, \n",
    "            language=language_code,\n",
    "            beam_size=5,\n",
    "            best_of=5,\n",
    "            temperature=(0.0, 0.2, 0.4, 0.6, 0.8, 1.0),\n",
    "            trust_whisper_timestamps=False,\n",
    "            suppress_tokens=[-1] + number_tokens)\n",
    "        #os.remove('segment.wav')\n",
    "        \n",
    "        return result\n",
    "    \n",
    "    # Output 'result' has dictionary structure\n",
    "    \n",
    "    def get_speech_intervals(tier, empty=False):\n",
    "        entries= diar_tg.getTier(tier).entries\n",
    "        intervals= []\n",
    "        for entry in entries:\n",
    "            if (entry.label and empty==False) or (not entry.label and empty==True):\n",
    "                intervals.append((entry.start, entry.end))\n",
    "        return intervals\n",
    "    \n",
    "    # Get speech intervals\n",
    "    diar_tg= textgrid.openTextgrid(diarization_textgrid, includeEmptyIntervals= True)\n",
    "    \n",
    "    if speakers_diarized== 'all':\n",
    "        all_tiers= diar_tg.tierNames\n",
    "        all_intervals= {}\n",
    "        for tier in all_tiers:\n",
    "            all_intervals[tier]= get_speech_intervals(tier)\n",
    "    \n",
    "    elif speakers_diarized== 'AC':\n",
    "        all_intervals['AC']= get_speech_intervals('AC')\n",
    "        all_intervals['other']= get_speech_intervals('AC', empty=True)\n",
    "        \n",
    "    \n",
    "    ## Define functions to check for proper alignment of the time intervals predicted by Whisper\n",
    "    \n",
    "    # check_overlap function: checks if the current interval overlaps with a previous interval\n",
    "    def check_overlap(intervals, time_seconds):\n",
    "        # Iterate through intervals in the tier\n",
    "        for start, end in intervals:\n",
    "            # Check if the time falls within any interval\n",
    "            if start <= time_seconds < end:\n",
    "                return True\n",
    "        # If no interval found:\n",
    "        return False\n",
    "    \n",
    "    # check_alignment function: checks if the start timestamp differs from the end timestamp => should not be the case\n",
    "    # (if Whisper makes up words, the start timestamps are often the same as the end timestamps)\n",
    "    def check_alignment(start_interval, end_interval):\n",
    "        if start_interval == end_interval:\n",
    "            return False \n",
    "        else:\n",
    "            return True\n",
    "        \n",
    "    ## Open a new textgrid for the transcription\n",
    "    tg = tgt.TextGrid()\n",
    "    \n",
    "    ## Iterate over speakers in the diarization textgrid, create output tiers and transcribe their speech\n",
    "    for speaker in all_intervals:\n",
    "        \n",
    "    \n",
    "        # Initialize interval tiers\n",
    "        sentence_tier = tgt.IntervalTier(start_time=0, end_time=len(audio), name=f\"{speaker} utterance\")\n",
    "        word_tier = tgt.IntervalTier(start_time=0, end_time=len(audio), name=f\"{speaker} word\")   \n",
    "        # Intervals will be indicated on the 'warning tier' when it is probable that Whisper's prediction is incorrect\n",
    "        # (for example, if an utterance only contains one word)\n",
    "        warning_tier = tgt.IntervalTier(start_time=0, end_time=len(audio), name=f\"{speaker} WARNING\")\n",
    "        \n",
    "        # Initialize lists of word and sentence time intervals.\n",
    "        # If a predicted time interval overlaps an interval that has been predicted for a previous word\n",
    "        # or sentence, then the current word/ sentence will not be added to the Textgrid\n",
    "        word_intervals=[]\n",
    "        sentence_intervals=[]\n",
    "        \n",
    "        \n",
    "        for (start_int, end_int) in all_intervals[speaker]:\n",
    "            \n",
    "            result= transcribe_segment(start_int, end_int)\n",
    "            \n",
    "        \n",
    "            ## While checking for proper alignment, write the model's predictions to the Textgrid\n",
    "\n",
    "            for segment in result[\"segments\"]: # 'segment' attribute of Whisper = +- utterance\n",
    "\n",
    "                sentence=[] # Initialize sentence text => renews for each whisper segment\n",
    "                sentence_start_s=-1 # Initialize integer value for start of sentence (will be replaced by start time of first word)\n",
    "\n",
    "\n",
    "                # Write words of a segment to textgrid\n",
    "                if \"words\" in segment and segment[\"words\"]: # Check that 'segment' has non-empty attribute 'words'\n",
    "\n",
    "                    for word in segment[\"words\"]:\n",
    "                        \n",
    "                        word_start= start_int + word['start']\n",
    "                        word_end= start_int+ word['end']\n",
    "\n",
    "                        # Check overlap and alignment of word timestamps and add word interval to word tier\n",
    "                        if check_alignment(word_start, word_end) \\\n",
    "                        and not check_overlap(word_intervals, word_start)\\\n",
    "                        and word['text'] != '[*]': # Do not transcribe '[*]': Whisper's indication of pauses\n",
    "\n",
    "                            ## tgt.Interval takes as arguments the start and end time of an interval and the text of the speech,\n",
    "                            ## and creates an interval that can be added to the Praat tier\n",
    "\n",
    "                            # Create the interval\n",
    "                            word_interval = tgt.Interval(start_time=float(word_start), \n",
    "                                                         end_time=float(word_end), text=word[\"text\"])\n",
    "                            # Write the interval to the tier\n",
    "                            word_tier.add_interval(word_interval)\n",
    "                            # Add start and end times to word_intervals, so that future time intervals can be checked for overlap\n",
    "                            word_intervals.append((word_start, word_end))\n",
    "\n",
    "                            # Add the transcribed word to sentence text\n",
    "                            sentence.append(word['text'])\n",
    "\n",
    "                            # Define sentence start timestamp\n",
    "                            if sentence_start_s== -1: # = if the sentence_start has not been modified yet\n",
    "                                sentence_start_s= word_start # change sentence start to start of first word\n",
    "\n",
    "\n",
    "\n",
    "                ## Write sentence to Textgrid\n",
    "\n",
    "                if sentence: # if words have been transcribed for the current segment\n",
    "                    sentence_end_s= word_intervals[-1][1] # end timestamp of sentence = end timestamp of last word\n",
    "\n",
    "                    # Check alignment and add sentence interval to sentence tier (same procedure as above)\n",
    "                    if check_alignment(sentence_start_s, sentence_end_s) \\\n",
    "                    and not check_overlap(sentence_intervals, sentence_start_s):\n",
    "\n",
    "                        sentence_interval= tgt.Interval(start_time=sentence_start_s, \n",
    "                                                        end_time=sentence_end_s, text=' '.join(sentence))\n",
    "                        sentence_tier.add_interval(sentence_interval)\n",
    "                        sentence_intervals.append((sentence_start_s, sentence_end_s))\n",
    "\n",
    "                    # Add an interval on the warning tier if there are signs of nonsense by Whisper\n",
    "                    if len(sentence)==1 or any(trigger_word in sentence for trigger_word in trigger_list):\n",
    "                        warning_interval= tgt.Interval(start_time=sentence_start_s, \n",
    "                                                        end_time=sentence_end_s, text= 'WARNING')\n",
    "                        warning_tier.add_interval(warning_interval)\n",
    "\n",
    "\n",
    "    \n",
    "        ## When all speech intervals are transribed, add tiers to output textgrid \n",
    "        tg.add_tier(warning_tier)\n",
    "        tg.add_tier(sentence_tier)\n",
    "        tg.add_tier(word_tier)\n",
    "\n",
    "    tgt.write_to_file(tg, output_path, format='short')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c968aad2",
   "metadata": {},
   "source": [
    "## Transcription of entire audio (non-diarized single-speaker audio)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9c9c56aa",
   "metadata": {},
   "source": [
    "If you have a **single-speaker audio** that is not yet diarized, you can use the following function to let Whisper **diarize and transcribe** the file. The input of a diarization file is thus not required (nor supported) in this model. The working and output of the model are the same as those of the previous model; the tiers will be named 'utterance', 'word' and 'warning', without indication of the speaker."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e8d2e833",
   "metadata": {},
   "outputs": [],
   "source": [
    "def transcribe_file_total(language, selected_model, audio_file, output_path):\n",
    "    \n",
    "    ## Import libraries\n",
    "    from pydub import AudioSegment\n",
    "    from praatio import textgrid\n",
    "    import whisper_timestamped as whisper\n",
    "    from whisper.tokenizer import get_tokenizer\n",
    "    import torch\n",
    "    import tgt\n",
    "    \n",
    "    ## Load audio via 'AudioSegment' library => used to calculate duration of the audio file\n",
    "    audio = AudioSegment.from_wav(audio_file)\n",
    "        \n",
    "    ## Configurate Whisper timestamped model\n",
    "    \n",
    "    # Choose tokenizer\n",
    "    tokenizer = get_tokenizer(multilingual=True)\n",
    "    \n",
    "    # The variable 'number_tokens' will contain the full words for numbers in the language in question,\n",
    "    # so that we can let Whisper spell out numbers instead of writing them in digits\n",
    "    number_tokens = [\n",
    "        i \n",
    "        for i in range(tokenizer.eot)\n",
    "        if all(c in \"0123456789\" for c in tokenizer.decode([i]).strip())]\n",
    "    \n",
    "    # Load selected Whisper timestamped model\n",
    "    model = whisper.load_model(selected_model, device=\"cpu\")\n",
    "    \n",
    "    whisper_languages= whisper.tokenizer.LANGUAGES\n",
    "    \n",
    "    # Get language codes from 'language' attribute input for model configuration\n",
    "    if language=='French':\n",
    "        language_code='fr'\n",
    "    elif language==\"Flemish\":\n",
    "        language_code='nl'\n",
    "    # For other languages:\n",
    "    elif language.lower() in whisper_languages.values():\n",
    "        language_code= str(list(whisper_languages.keys())[list(whisper_languages.values()).index(language.lower())])\n",
    "    # If language unknown: let Whisper detect the language\n",
    "    elif language== 'unknown':\n",
    "        w_audio = whisper.load_audio(audio_file)\n",
    "        w_audio = whisper.pad_or_trim(w_audio)\n",
    "\n",
    "        # make log-Mel spectrogram and move to the same device as the model\n",
    "        mel = whisper.log_mel_spectrogram(w_audio).to(selected_model.device)\n",
    "\n",
    "        # detect the spoken language\n",
    "        _, probs = model.detect_language(mel)\n",
    "        language_code= str(max(probs, key=probs.get))\n",
    "    \n",
    "    \n",
    "    ## Transribe the audio file using Whisper timestamped => based on 'Whispgrid' script\n",
    "    ## Use configurations defined above\n",
    "    result = whisper.transcribe(\n",
    "        model,\n",
    "        audio_file,\n",
    "        #vad= 'auditok', # gives errors and does not seem to improve the alignment\n",
    "        detect_disfluencies= True, \n",
    "        language=language_code,\n",
    "        beam_size=5,\n",
    "        best_of=5,\n",
    "        temperature=(0.0, 0.2, 0.4, 0.6, 0.8, 1.0),\n",
    "        trust_whisper_timestamps=False,\n",
    "        suppress_tokens=[-1] + number_tokens)\n",
    "    # Output 'result' has dictionary structure\n",
    "    \n",
    "    # Warning-triggering list\n",
    "    trigger_list= [\"Amara.org\", \"d'Amara.org\",\"sous-titres\",\"ondertitels\",\"ondertiteling\"]\n",
    "    \n",
    "    \n",
    "    ## Open a new textgrid for the transcription\n",
    "    tg = tgt.TextGrid()\n",
    "    \n",
    "    # Initialize interval tiers\n",
    "    sentence_tier = tgt.IntervalTier(start_time=0, end_time=len(audio), name=\"utterance\")\n",
    "    word_tier = tgt.IntervalTier(start_time=0, end_time=len(audio), name=\"word\")   \n",
    "    # Intervals will be indicated on the 'warning tier' when it is probable that Whisper's prediction is wrong\n",
    "    # (for example, if an utterance only contains one word)\n",
    "    warning_tier = tgt.IntervalTier(start_time=0, end_time=len(audio), name=\"WARNING\")\n",
    "        \n",
    "        \n",
    "    ## Define functions to check for proper alignment of the time intervals predicted by Whisper\n",
    "    \n",
    "    # Initialize lists of word and sentence time intervals.\n",
    "    # If a predicted time interval overlaps an interval that has been predicted for a previous word\n",
    "    # or sentence, then the current word/ sentence will not be added to the Textgrid\n",
    "    word_intervals=[]\n",
    "    sentence_intervals=[]\n",
    "\n",
    "    # check_overlap function: checks if the current interval overlaps with a previous interval\n",
    "    def check_overlap(intervals, time_seconds):\n",
    "        # Iterate through intervals in the tier\n",
    "        for start, end in intervals:\n",
    "            # Check if the time falls within any interval\n",
    "            if start <= time_seconds < end:\n",
    "                return True\n",
    "        # If no interval found:\n",
    "        return False\n",
    "    \n",
    "    # check_alignment function: checks if the start timestamp differs from the end timestamp => should not be the case\n",
    "    # (if Whisper makes up words, the start timestamps are often the same as the end timestamps)\n",
    "    def check_alignment(start_interval, end_interval):\n",
    "        if start_interval == end_interval:\n",
    "            return False \n",
    "        else:\n",
    "            return True\n",
    "\n",
    "\n",
    "    ## While checking for proper alignment, write the model's predictions to the Textgrid\n",
    "\n",
    "    for segment in result[\"segments\"]: # 'segment' attribute of Whisper = +- utterance\n",
    "\n",
    "        sentence=[] # Initialize sentence text => renews for each whisper segment\n",
    "        sentence_start_s=-1 # Initialize integer value for start of sentence (will be replaced by start time of first word)\n",
    "\n",
    "\n",
    "        # Write words of a segment to textgrid\n",
    "        if \"words\" in segment and segment[\"words\"]: # Check that 'segment' has non-empty attribute 'words'\n",
    "            \n",
    "            for word in segment[\"words\"]:\n",
    "\n",
    "                # Check overlap and alignment of word timestamps and add word interval to word tier\n",
    "                if check_alignment(word['start'], word['end']) \\\n",
    "                and not check_overlap(word_intervals,word[\"start\"])\\\n",
    "                and word['text'] != '[*]': # Do not transcribe '[*]': Whisper's indication of pauses\n",
    "                    \n",
    "                    ## tgt.Interval takes as arguments the start and end time of an interval and the text of the speech,\n",
    "                    ## and creates an interval that can be added to the Praat tier\n",
    "                    \n",
    "                    # Create the interval\n",
    "                    word_interval = tgt.Interval(start_time=float(word['start']), \n",
    "                                                 end_time=float(word['end']), text=word[\"text\"])\n",
    "                    # Write the interval to the tier\n",
    "                    word_tier.add_interval(word_interval)\n",
    "                    # Add start and end times to word_intervals, so that future time intervals can be checked for overlap\n",
    "                    word_intervals.append((word['start'], word['end']))\n",
    "\n",
    "                    # Add the transcribed word to sentence text\n",
    "                    sentence.append(word['text'])\n",
    "                    \n",
    "                    # Define sentence start timestamp\n",
    "                    if sentence_start_s== -1: # = if the sentence_start has not been modified yet\n",
    "                        sentence_start_s= word['start'] # change sentence start to start of first word\n",
    "                        \n",
    "                    # If the alignment and overlap conditions are not fulfilled: do not add word\n",
    "\n",
    "\n",
    "        \n",
    "        ## Write sentence to Textgrid\n",
    "\n",
    "        if sentence: # if words have been transcribed for the current segment\n",
    "            sentence_end_s= word_intervals[-1][1] # end timestamp of sentence = end timestamp of last word\n",
    "            \n",
    "            # Check alignment and add sentence interval to sentence tier (same procedure as above)\n",
    "            if check_alignment(sentence_start_s, sentence_end_s) \\\n",
    "            and not check_overlap(sentence_intervals, sentence_start_s):\n",
    "\n",
    "                sentence_interval= tgt.Interval(start_time=sentence_start_s, \n",
    "                                                end_time=sentence_end_s, text=' '.join(sentence))\n",
    "                sentence_tier.add_interval(sentence_interval)\n",
    "                sentence_intervals.append((sentence_start_s, sentence_end_s))\n",
    "                \n",
    "            # Add an interval on the warning tier if there are signs of nonsense by Whisper\n",
    "            if len(sentence)==1 or any(trigger_word in sentence for trigger_word in trigger_list): \n",
    "                warning_interval= tgt.Interval(start_time=sentence_start_s, \n",
    "                                                end_time=sentence_end_s, text= 'WARNING')\n",
    "                warning_tier.add_interval(warning_interval)\n",
    "\n",
    "\n",
    "    \n",
    "    ## When all speech intervals are transribed, add tiers to output textgrid and output it\n",
    "    tg.add_tier(warning_tier)\n",
    "    tg.add_tier(sentence_tier)\n",
    "    tg.add_tier(word_tier)\n",
    "\n",
    "    tgt.write_to_file(tg, output_path, format='short')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}