{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":73047,"databundleVersionId":8140249,"sourceType":"competition"}],"dockerImageVersionId":30673,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"<h1>This notebook is purely inspired from <a href=\"https://www.kaggle.com/code/smjishanulislam/quickstart-with-whisper-small/notebook\"> here</a><h1>\n   <h1>This notebook is created for utilizing GPU time. Because of limited GPU resources in Kaggle, training and inference in the same notebook is not a good idea. Use this notebook only for training. After training, import the model as a dataset. For inference, use the code in the <a href=\"https://www.kaggle.com/code/samratabduljalil/bengali-speech-recognition-inference/\"> link</a>  <h1> \n       \n       <h2>Must check GPU on or Off. best of luck </h2>","metadata":{}},{"cell_type":"code","source":"import os\n\nimport pandas as pd\n\nimport librosa\nimport librosa.display\n\nimport numpy as np\n\nimport IPython.display as ipd\n\nimport matplotlib.pyplot as plt\n\nimport random\n\nfrom collections import Counter\n\nfrom sklearn.model_selection import train_test_split\n\nimport torch\nimport torchaudio\n\nfrom dataclasses import dataclass\nfrom typing import Any, Dict, List, Union\nfrom datasets import DatasetDict\nfrom datasets import Dataset as DS\n\nfrom transformers import (\n    WhisperFeatureExtractor,\n    WhisperTokenizer,\n    WhisperProcessor,\n    WhisperForConditionalGeneration,\n    Seq2SeqTrainingArguments,\n    Seq2SeqTrainer,\n    TrainerCallback,\n    TrainingArguments,\n    TrainerState,\n    TrainerControl,\n    EarlyStoppingCallback,\n    pipeline\n)\n\nfrom torchmetrics.text import WordErrorRate, CharErrorRate","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"BASE_DIR = '/kaggle/input/ben10/ben10'\ntrain_data_dir = f\"{BASE_DIR}/16_kHz_train_audio/\"\ntest_data_dir = f\"{BASE_DIR}/16_kHz_valid_audio/\"\ndata_path = f\"{BASE_DIR}/train.csv\"","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"split2path = {\n    \"train\": train_data_dir,\n    \"test\": test_data_dir,\n}","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"data = pd.read_csv(data_path)\ndata.sample(10)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def extract_split(filename):\n    filename_ = filename.split(\"_\")\n    split = filename_[0]\n    return split\n\ndef extract_district(filename):\n    filename_ = filename.split(\" \")[0]\n    district = filename_.split(\"_\")[1]\n    return district\n\ndef beautify_dataset(data):\n    splits = []\n    districts = []\n    newpaths = []\n    transcripts = []\n    \n    for i in range(len(data)):\n        filename, transcript = data.iloc[i]\n        split = extract_split(filename)\n        district = extract_district(filename)\n        dir_path = split2path[split]\n        composed_path = f\"{dir_path}{filename}\"\n        \n        if os.path.exists(composed_path) == False:\n            print(f\"{composed_path} does not exist.\")\n            continue\n        \n        # replace any newline characters\n        transcript = transcript.replace(\"\\n\", \" \")\n        transcript = \" \".join(transcript.split())\n        \n        splits.append(split)\n        districts.append(district)\n        newpaths.append(composed_path)\n        transcripts.append(transcript)\n    \n    data['file_path'] = newpaths\n    data['district'] = districts\n    data['split'] = splits\n    data['transcripts'] = transcripts\n    \n#     data.drop(columns=['file_name'], inplace=True)\n    \n    return data","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"data = beautify_dataset(data)\ndata.sample(20)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"data[data[\"transcripts\"] == \"<>\"]\ndata[data[\"transcripts\"] == \"\"]\ndata[data[\"transcripts\"] == \"..\"]","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# print(list(data[data['transcripts'] == ''].index))\ndata.drop(data[data['transcripts'] == ''].index, inplace=True)\n      \n# print(list(data[data['transcripts'] == '<>'].index))\ndata.drop(data[data['transcripts'] == \"<>\"].index, inplace=True)\n      \n# print(list(data[data['transcripts'] == '..'].index))\ndata.drop(data[data['transcripts'] == \"..\"].index, inplace=True)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"data[\"transcripts\"] = data[\"transcripts\"].str.strip()","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"TASK = \"transcribe\"\nMODEL_NAME = \"openai/whisper-small\"#you can use different model from hugging face","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"feature_extractor = WhisperFeatureExtractor.from_pretrained(MODEL_NAME)\ntokenizer = WhisperTokenizer.from_pretrained(MODEL_NAME, language='bn', task=TASK)\nprocessor = WhisperProcessor.from_pretrained(MODEL_NAME, language='bn', task=TASK)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"ids = tokenizer.encode(\"\")\nids","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"tokenizer.decode(ids)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"@dataclass\nclass DataCollatorSpeechSeq2SeqWithPadding:\n    processor: Any\n\n    def __call__(self, features: List[Dict[str, Union[List[int], torch.Tensor]]]) -> Dict[str, torch.Tensor]:\n        # split inputs and labels since they have to be of different lengths and need different padding methods\n        # first treat the audio inputs by simply returning torch tensors\n        input_features = [{\"input_features\": feature[\"input_features\"]} for feature in features]\n        batch = self.processor.feature_extractor.pad(input_features, return_tensors=\"pt\")\n\n        # get the tokenized label sequences\n        label_features = [{\"input_ids\": feature[\"labels\"]} for feature in features]\n        # pad the labels to max length\n        labels_batch = self.processor.tokenizer.pad(label_features, return_tensors=\"pt\")\n\n        # replace padding with -100 to ignore loss correctly\n        labels = labels_batch[\"input_ids\"].masked_fill(labels_batch.attention_mask.ne(1), -100)\n\n        # if bos token is appended in previous tokenization step,\n        # cut bos token here as it's append later anyways\n        if (labels[:, 0] == self.processor.tokenizer.bos_token_id).all().cpu().item():\n            labels = labels[:, 1:]\n\n        batch[\"labels\"] = labels\n        \n        torch.cuda.empty_cache()\n\n        return batch","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"data_collator = DataCollatorSpeechSeq2SeqWithPadding(processor=processor)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def prepare_dataset(example):\n    audio_path = example[\"file_path\"]\n    \n    # load the audio using librosa or torch audio (as you wish)\n    audio, sr = librosa.load(audio_path, sr=16_000)\n    \n    example[\"input_features\"] = feature_extractor(audio, sampling_rate=sr).input_features[0]\n    \n    example[\"labels\"] = tokenizer(f\"{example['transcripts']}\", max_length=448, padding=True, truncation=True).input_ids\n    \n    return example\n\n\ndef filter_inputs(input_audio):\n    \"\"\"filter inputs with zero input length\"\"\"\n    return 0 < len(input_audio)\n\n\ndef filter_labels(input_labels):\n    \"\"\"filter empty label sequences\"\"\"\n    return 0 < len(input_labels)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_df = data[data[\"split\"] == \"train\"]","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\"\"\"\n    adjust test size accordingly.\n\"\"\"\ntrain_df, eval_df = train_test_split(train_df, test_size=0.01, shuffle=True)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"len(train_df), len(eval_df)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"ben_reg_voice_ds = DatasetDict()\n\ntrain_split = DS.from_pandas(train_df)\neval_split = DS.from_pandas(eval_df)\n\nds_splits = DatasetDict({\n    'train': train_split,\n    'eval': eval_split\n})","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"ds_splits = ds_splits.remove_columns([\"split\"])\nprint(ds_splits)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"np.object = object","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"ds_splits = ds_splits.map(prepare_dataset, remove_columns=ds_splits.column_names[\"train\"],\n                          num_proc=2 # open for multithreadding\n)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"len(ds_splits[\"train\"]), len(ds_splits[\"eval\"])","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"cer = CharErrorRate()\nwer = WordErrorRate()","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def compute_metrics(pred):\n    pred_ids = pred.predictions\n    label_ids = pred.label_ids\n\n    label_ids[label_ids == -100] = tokenizer.pad_token_id\n\n    pred_str = tokenizer.batch_decode(pred_ids, skip_special_tokens=True)\n    label_str = tokenizer.batch_decode(label_ids, skip_special_tokens=True)\n\n    wer_res = wer(pred_str, label_str)\n    cer_res = cer(pred_str, label_str)\n    \n    \"\"\"\n        uncomment the next 3 lines if you want to see how the examples look like during eval \n    \"\"\"\n    print(\"WER:\",wer_res,\"| CER:\", cer_res) # to show up during running logs\n    print(\"Pred:\",pred_str[0])\n    print(\"Label:\",label_str[0])\n    \n    return {\"wer\": wer_res, \"cer\": cer_res}","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"model = WhisperForConditionalGeneration.from_pretrained(MODEL_NAME, device_map=\"auto\")","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"model_id = \"whisper-reg-ben\" ","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#finetune this hyperparameter for getting the better result\n\ntraining_args = Seq2SeqTrainingArguments(\n    output_dir=model_id,\n    per_device_train_batch_size=9,\n    per_device_eval_batch_size=8,\n    gradient_accumulation_steps=1,\n    gradient_checkpointing=True,\n    fp16=True,\n    learning_rate=3e-4,\n    weight_decay=1e-2,\n    warmup_steps=100,\n    num_train_epochs=1,\n    evaluation_strategy=\"steps\", # or \"epochs\"\n    predict_with_generate=True,\n#     generation_max_length=448,\n    save_steps=1000,\n    eval_steps=1000,\n    logging_steps=1000,\n    save_total_limit=1,\n    load_best_model_at_end=True,\n    metric_for_best_model=\"wer\",\n    greater_is_better=False,\n    push_to_hub=False,\n    report_to=\"none\",\n    remove_unused_columns=False,\n)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"model.generation_config.language = \"bn\"\nmodel.generation_config.task = \"transcribe\"\n\nmodel.generation_config.forced_decoder_ids = None\nmodel.config.suppress_tokens = [] # added later","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"trainer = Seq2SeqTrainer(\n    args=training_args,\n    model=model,\n    train_dataset=ds_splits[\"train\"],\n    eval_dataset=ds_splits[\"eval\"],\n    data_collator=data_collator,\n    tokenizer=processor.feature_extractor,\n    compute_metrics=compute_metrics,\n#     callbacks=[EarlyStoppingCallback(2, 1.0)]\n)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"trainer.train()\n\n# to use the high-level pipeline, ensure both the processor outputs and model outputs exist in the same dir\ntrainer.save_model(training_args.output_dir)\nprocessor.save_pretrained(training_args.output_dir)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"<h1> go to output of the notebook you can find your best model .just create dataset just clicking on the three dot.then use that in inference notebook. inference notebook :<a href=\"\">Click here</a></h1>","metadata":{}}]}