jn_fbm4tv2l_demo.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2a201040-3a18-40c7-9934-3f84f4a26439",
   "metadata": {},
   "outputs": [],
   "source": [
    "from transformers import SeamlessM4Tv2Model, AutoProcessor\n",
    "import torch\n",
    "\n",
    "processor = AutoProcessor.from_pretrained(\"facebook/seamless-m4t-v2-large\")\n",
    "model = SeamlessM4Tv2Model.from_pretrained(\"facebook/seamless-m4t-v2-large\")\n",
    "\n",
    "device = \"cuda:0\" if torch.cuda.is_available() else \"cpu\"\n",
    "model = model.to(device)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b784a64c-9815-4fb8-acee-1bf12ca2e1aa",
   "metadata": {},
   "outputs": [],
   "source": [
    "# This is a partial list.  A complete list is available on hugging face\n",
    "# # https://huggingface.co/facebook/seamless-m4t-v2-large\n",
    "lang_list = [{'sym': 'eng', 'language':'English'},\n",
    "             {'sym': 'cmn', 'language':'Chinese'},\n",
    "             {'sym': 'tha', 'language':'Thai'},\n",
    "             {'sym': 'deu', 'language':'German'},\n",
    "             {'sym': 'rus', 'language':'Russian'},\n",
    "             {'sym': 'fra', 'language':'French'},\n",
    "             {'sym': 'spa', 'language':'Spanish'},\n",
    "             {'sym': 'jpn', 'language':'Japanese'},\n",
    "             {'sym': 'vie', 'language':'Vietnamese'}]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5a309055-60ae-4bbb-96d8-a4074f62f63a",
   "metadata": {},
   "outputs": [],
   "source": [
    "import ipywidgets as widgets\n",
    "from IPython.display import display, HTML\n",
    "# https://huggingface.co/facebook/seamless-m4t-v2-large\n",
    "\n",
    "# Create the dropdown widget\n",
    "dropdown_sts = widgets.Dropdown(options=[(item['language'], index) for index, item in enumerate(lang_list)],\n",
    "                            value=0, description='Select a target language for speech to speech:')\n",
    "dropdown_t_ttt = widgets.Dropdown(options=[(item['language'], index) for index, item in enumerate(lang_list)],\n",
    "                            value=0, description='Target language:')\n",
    "dropdown_s_ttt = widgets.Dropdown(options=[(item['language'], index) for index, item in enumerate(lang_list)],\n",
    "                            value=0, description='Source language:')\n",
    "dropdown_t_tts = widgets.Dropdown(options=[(item['language'], index) for index, item in enumerate(lang_list)],\n",
    "                            value=0, description='Target language:')\n",
    "dropdown_s_tts = widgets.Dropdown(options=[(item['language'], index) for index, item in enumerate(lang_list)],\n",
    "                            value=0, description='Source language:')\n",
    "sym_sts = 'eng'\n",
    "sym_t_ttt = 'eng'\n",
    "sym_s_ttt = 'eng'\n",
    "sym_t_tts = 'eng'\n",
    "sym_s_tts = 'eng'\n",
    "\n",
    "# Function to handle changes in the dropdown_sts value\n",
    "def on_dropdown_sts_change(change):\n",
    "    global sym_sts\n",
    "    selected_index = change['new']\n",
    "    selected_option = lang_list[selected_index]\n",
    "    text = f\"Selected option: {selected_option}\"\n",
    "    sym_sts = selected_option['sym']\n",
    "    language = selected_option['language']\n",
    "    # text_box11.value = sym_sts + '  ' + language\n",
    "\n",
    "def on_dropdown_t_ttt_change(change):\n",
    "    global sym_t_ttt\n",
    "    selected_index = change['new']\n",
    "    selected_option = lang_list[selected_index]\n",
    "    text = f\"Selected option: {selected_option}\"\n",
    "    sym_t_ttt = selected_option['sym']\n",
    "    language = selected_option['language']\n",
    "    # text_box11.value = sym_t_ttt + '  ' + language\n",
    "\n",
    "def on_dropdown_s_ttt_change(change):\n",
    "    global sym_s_ttt\n",
    "    selected_index = change['new']\n",
    "    selected_option = lang_list[selected_index]\n",
    "    text = f\"Selected option: {selected_option}\"\n",
    "    sym_s_ttt = selected_option['sym']\n",
    "    language = selected_option['language']\n",
    "    # text_box11.value = sym_s_ttt + '  ' + language\n",
    "\n",
    "def on_dropdown_t_tts_change(change):\n",
    "    global sym_t_tts\n",
    "    selected_index = change['new']\n",
    "    selected_option = lang_list[selected_index]\n",
    "    text = f\"Selected option: {selected_option}\"\n",
    "    sym_t_tts = selected_option['sym']\n",
    "    language = selected_option['language']\n",
    "    # text_box11.value = sym_t_tts + '  ' + language\n",
    "\n",
    "def on_dropdown_s_tts_change(change):\n",
    "    global sym_s_tts\n",
    "    selected_index = change['new']\n",
    "    selected_option = lang_list[selected_index]\n",
    "    text = f\"Selected option: {selected_option}\"\n",
    "    sym_s_tts = selected_option['sym']\n",
    "    language = selected_option['language']\n",
    "    # text_box11.value = sym_s_tts + '  ' + language\n",
    "    \n",
    "\n",
    "# Attach the function to the dropdown_sts's observe method\n",
    "dropdown_sts.observe(on_dropdown_sts_change, names='value')\n",
    "dropdown_t_ttt.observe(on_dropdown_t_ttt_change, names='value')\n",
    "dropdown_s_ttt.observe(on_dropdown_s_ttt_change, names='value')\n",
    "dropdown_t_tts.observe(on_dropdown_t_tts_change, names='value')\n",
    "dropdown_s_tts.observe(on_dropdown_s_tts_change, names='value')\n",
    "\n",
    "# text_box11 = widgets.Text(value='Initial Text11', description='Text Box11:')\n",
    "# text_box11.layout.width = '50%'  # Adjust the width as needed\n",
    "\n",
    "dropdown_sts.layout.width = '500px'  # Adjust the width as needed\n",
    "# Display the dropdown_sts\n",
    "# display(dropdown_sts)\n",
    "# display(widgets.HBox([text_box11]))\n",
    "\n",
    "# selecte language for three text boxes:  stt then tts\n",
    "txt1_lang = 'eng'\n",
    "txt2_lang = 'cmn'\n",
    "txt3_lang = 'fra'\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "28cf6824-f9e0-4d2e-809d-f28217013c32",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import ipywidgets as widgets\n",
    "from IPython.display import display, HTML\n",
    "import sounddevice as sd\n",
    "import numpy as np\n",
    "import threading\n",
    "\n",
    "# Create an HTML widget for the centered app title\n",
    "centered_app_title = HTML(\"<h2 style='color:blue; text-align:center;'>SeamlessM4T-large v2 Model Demonstration</h2>\")\n",
    "\n",
    "# Display the centered app title\n",
    "display(centered_app_title)\n",
    "\n",
    "# Set the sample rate and other parameters\n",
    "# sample_rate = 44100  # Adjust as needed\n",
    "sample_rate = model.config.sampling_rate\n",
    "\n",
    "# Initial state\n",
    "recording = False\n",
    "\n",
    "# Variables to store the recorded audio\n",
    "audio_data = np.array([])\n",
    "recording_stream = None\n",
    "\n",
    "def tts_translate(change):\n",
    "    # process input\n",
    "    text_inputs = processor(text = text_box_tts.value, src_lang=sym_s_tts, return_tensors=\"pt\").to(device)\n",
    "    # text_inputs = processor(text = text_box_tts.value, return_tensors=\"pt\").to(device)\n",
    "    output_tokens = model.generate(**text_inputs, tgt_lang=sym_t_tts, generate_speech=False)\n",
    "    translated_text_from_text = processor.decode(output_tokens[0].tolist()[0], skip_special_tokens=True)\n",
    "    play_text_src_tgt(translated_text_from_text, sym_s_tts, sym_t_tts)\n",
    "    # text_box_tts.value = f\"{translated_text_from_text}\"\n",
    "\n",
    "def ttt_translate(change):\n",
    "    # process input\n",
    "    text_inputs = processor(text = text_box_ttt.value, src_lang=sym_s_ttt, return_tensors=\"pt\").to(device)\n",
    "    # text_inputs = processor(text = text_box_ttt.value, return_tensors=\"pt\").to(device)\n",
    "    output_tokens = model.generate(**text_inputs, tgt_lang=sym_t_ttt, generate_speech=False)\n",
    "    translated_text_from_text = processor.decode(output_tokens[0].tolist()[0], skip_special_tokens=True)\n",
    "    text_box_ttt.value = f\"{translated_text_from_text}\"\n",
    "\n",
    "def toggle_recording(change):\n",
    "    global recording_stream\n",
    "    global recording\n",
    "    recording = not recording\n",
    "    if recording:\n",
    "        start_button.description = \"Stop Speaking\"\n",
    "        # global audio_data, recording_stream\n",
    "        global audio_data\n",
    "        audio_data = np.array([])  # Clear previous recordings\n",
    "        sd.default.samplerate = sample_rate\n",
    "        sd.default.channels = 1\n",
    "        recording_stream = sd.InputStream(callback=callback)\n",
    "        recording_stream.start()\n",
    "    else:\n",
    "        start_button.description = \"Start Speaking\"\n",
    "        # global recording_stream\n",
    "        if recording_stream:\n",
    "            recording_stream.stop()\n",
    "            recording_stream.close()\n",
    "            play_recorded_audio()\n",
    "    \n",
    "# Callback function for streaming audio\n",
    "def callback(indata, frames, time, status):\n",
    "    global audio_data\n",
    "    if status:\n",
    "        print(status)\n",
    "    audio_data = np.append(audio_data, indata.flatten())\n",
    "\n",
    "# Function to play  from given text\n",
    "def play_text_src_tgt(text, src, tgt):\n",
    "    text_inputs = processor(text = text, src_lang=src, return_tensors=\"pt\").to(device)\n",
    "    # generate translation audio\n",
    "    audio_array_from_text = model.generate(**text_inputs, tgt_lang=tgt)[0].cpu().numpy().squeeze()\n",
    "    # Play the audio using sounddevice\n",
    "    sd.play(audio_array_from_text, sample_rate)\n",
    "    sd.wait()  # Wait for playback to finish\n",
    "\n",
    "# Function to play the recorded audio\n",
    "def play_recorded_audio():\n",
    "    sd.play(audio_data, sample_rate)\n",
    "    sd.wait()\n",
    "    audio_sample = audio_data\n",
    "    # audio_sample = audio_array_from_text\n",
    "    # process input\n",
    "    # audio_inputs = processor(audios=audio_sample[\"array\"], return_tensors=\"pt\").to(device)\n",
    "    # audio_inputs = processor(audios=audio_sample, sampling_rate=sample_rate, return_tensors=\"pt\").to(device)\n",
    "    audio_inputs = processor(audios=audio_sample, sampling_rate=sample_rate, return_tensors=\"pt\").to(device)\n",
    "    text_box1.value = 'processing ...'\n",
    "    text_box2.value = 'processing ...'\n",
    "    text_box3.value = 'processing ...'\n",
    "\n",
    "    # audio_array_from_audio = model.generate(**audio_inputs, tgt_lang=\"cmn\")[0].cpu().numpy().squeeze()\n",
    "    audio_array_from_audio = model.generate(**audio_inputs, tgt_lang=sym_sts)[0].cpu().numpy().squeeze()\n",
    "    sd.play(audio_array_from_audio, sample_rate)\n",
    "    sd.wait()\n",
    "    \n",
    "    # generate translation speech to text\n",
    "    output_tokens_txt1 = model.generate(**audio_inputs, tgt_lang=txt1_lang, generate_speech=False)\n",
    "    output_tokens_txt2 = model.generate(**audio_inputs, tgt_lang=txt2_lang, generate_speech=False)\n",
    "    output_tokens_txt3 = model.generate(**audio_inputs, tgt_lang=txt3_lang, generate_speech=False)\n",
    "    translated_text_from_audio_txt1 = processor.decode(output_tokens_txt1[0].tolist()[0], skip_special_tokens=True)\n",
    "    translated_text_from_audio_txt2 = processor.decode(output_tokens_txt2[0].tolist()[0], skip_special_tokens=True)\n",
    "    translated_text_from_audio_txt3 = processor.decode(output_tokens_txt3[0].tolist()[0], skip_special_tokens=True)\n",
    "\n",
    "    to_text_txt1 = f\"{translated_text_from_audio_txt1}\"\n",
    "    to_text_txt2 = f\"{translated_text_from_audio_txt2}\"\n",
    "    to_text_txt3 = f\"{translated_text_from_audio_txt3}\"\n",
    "    text_box1.value = to_text_txt1\n",
    "    text_box2.value = to_text_txt2\n",
    "    text_box3.value = to_text_txt3\n",
    "    play_text_src_tgt(to_text_txt1, txt1_lang, txt1_lang)\n",
    "    play_text_src_tgt(to_text_txt2, txt2_lang, txt2_lang)\n",
    "    play_text_src_tgt(to_text_txt3, txt3_lang, txt3_lang)\n",
    "\n",
    "# Create buttons\n",
    "start_button = widgets.Button(description=\"Start Speaking\")\n",
    "start_button.on_click(toggle_recording)\n",
    "\n",
    "ttt_button = widgets.Button(description=\"Start TTT\")\n",
    "ttt_button.on_click(ttt_translate)\n",
    "ttt_button.layout.width = '100px'\n",
    "\n",
    "tts_button = widgets.Button(description=\"Start TTS\")\n",
    "tts_button.on_click(tts_translate)\n",
    "tts_button.layout.width = '100px'\n",
    "\n",
    "# stop_button = widgets.Button(description=\"Stop Recording\")\n",
    "# stop_button.on_click(stop_recording)\n",
    "\n",
    "# Extract the language where 'sym' is equal to txt1_lang\n",
    "label1 = widgets.Label(value=next((item['language'] for item in lang_list if item['sym'] == txt1_lang), None))\n",
    "label2 = widgets.Label(value=next((item['language'] for item in lang_list if item['sym'] == txt2_lang), None))\n",
    "label3 = widgets.Label(value=next((item['language'] for item in lang_list if item['sym'] == txt3_lang), None))\n",
    "# txt1_lang, lang_list\n",
    "\n",
    "# Create a text box widget\n",
    "text_box1 = widgets.Text(value='stt then tts', description='', disabled = True)\n",
    "text_box2 = widgets.Text(value='stt then tts', description='', disabled = True)\n",
    "text_box3 = widgets.Text(value='stt then tts', description='', disabled = True)\n",
    "text_box_ttt = widgets.Text(placeholder='Enter text here to translate to target language from the dropdown...', description='', layout=widgets.Layout(width='500px'))\n",
    "text_box_tts = widgets.Text(placeholder='Enter text here to translate to target language from the dropdown...', description='', layout=widgets.Layout(width='500px'))\n",
    "\n",
    "\n",
    "# Display buttons\n",
    "# Set the layout for the HBox\n",
    "dropdown_sts.layout.width = '400px'  # Adjust the width as needed\n",
    "# Set the width for both description label and dropdown box\n",
    "dropdown_sts.style.description_width = 'initial'  # Use 'initial' to allow the label to expand\n",
    "hbox_layout = widgets.Layout(justify_content='space-between', width='100%')\n",
    "hbox = widgets.HBox([start_button, dropdown_sts], layout=hbox_layout)\n",
    "display(hbox)\n",
    "# display(widgets.HBox([start_button, dropdown]))\n",
    "display(widgets.HBox([label1, text_box1, label2, text_box2, label3, text_box3]))\n",
    "\n",
    "dropdown_t_ttt.layout.width = '200px'  # Adjust the width as needed\n",
    "# Set the width for both description label and dropdown box\n",
    "dropdown_t_ttt.style.description_width = 'initial'  # Use 'initial' to allow the label to expand\n",
    "\n",
    "dropdown_s_ttt.layout.width = '200px'  # Adjust the width as needed\n",
    "# Set the width for both description label and dropdown box\n",
    "dropdown_s_ttt.style.description_width = 'initial'  # Use 'initial' to allow the label to expand\n",
    "\n",
    "hbox_layout = widgets.Layout(justify_content='space-between', width='100%')\n",
    "hbox_ttt = widgets.HBox([ttt_button,dropdown_s_ttt, text_box_ttt, dropdown_t_ttt], layout=hbox_layout)\n",
    "display(hbox_ttt)\n",
    "\n",
    "# tts alone\n",
    "dropdown_t_tts.layout.width = '200px'  # Adjust the width as needed\n",
    "# Set the width for both description label and dropdown box\n",
    "dropdown_t_tts.style.description_width = 'initial'  # Use 'initial' to allow the label to expand\n",
    "\n",
    "dropdown_s_tts.layout.width = '200px'  # Adjust the width as needed\n",
    "# Set the width for both description label and dropdown box\n",
    "dropdown_s_tts.style.description_width = 'initial'  # Use 'initial' to allow the label to expand\n",
    "\n",
    "hbox_layout = widgets.Layout(justify_content='space-between', width='100%')\n",
    "hbox_tts = widgets.HBox([tts_button,dropdown_s_tts, text_box_tts, dropdown_t_tts], layout=hbox_layout)\n",
    "display(hbox_tts)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9ae66be2-68a8-461d-8673-98947a6cdf86",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4d178da9-f479-4654-b58a-5d3d90c4838d",
   "metadata": {},
   "outputs": [],
   "source": [
    "SeamlessM4T models support the tasks of:\n",
    "\n",
    "Speech-to-speech translation (S2ST)\n",
    "Speech-to-text translation (S2TT)\n",
    "Text-to-speech translation (T2ST)\n",
    "Text-to-text translation (T2TT)\n",
    "Automatic speech recognition (ASR)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5d74912e-268b-4e78-8fa3-a457225bcc1e",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}