{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"},"kaggle":{"accelerator":"nvidiaTeslaT4","dataSources":[{"sourceType":"competition","sourceId":7634,"databundleVersionId":46676}],"dockerImageVersionId":31329,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"!mkdir -p /kaggle/working/dataset/train\n\n!7z x /kaggle/input/competitions/tensorflow-speech-recognition-challenge/train.7z -o/kaggle/working/dataset/train -y -bd\n\n!pip install -q transformers evaluate soundfile accelerate matplotlib","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================================== #\n#         UNIVERSAL ROBUSTNESS ANALYSIS: WAV2VEC 2.0 (ALL STRATEGIES)            #\n# ============================================================================== #\n\nimport os\nimport random\nimport gc\nimport torch\nimport numpy as np\nimport soundfile as sf\nimport matplotlib.pyplot as plt\nimport seaborn as sns\nimport glob\nimport evaluate\nfrom torch.utils.data import Dataset\nfrom transformers import (\n    Wav2Vec2FeatureExtractor, \n    Wav2Vec2ForSequenceClassification,\n    Trainer, \n    TrainingArguments\n)\nfrom sklearn.metrics import confusion_matrix\n\n# ------------------------------------------------------------------------------ #\n# [1] GLOBAL SETUP & DATASET SPLITS\n# ------------------------------------------------------------------------------ #\nfound_paths = glob.glob('/kaggle/working/**/validation_list.txt', recursive=True)\nif not found_paths:\n    raise FileNotFoundError(\"Error, .txt files not accesible.\")\n\nBASE_TRAIN_DIR = os.path.dirname(found_paths[0])\nAUDIO_DIR = os.path.join(BASE_TRAIN_DIR, \"audio\")\n\nCORE_CLASSES = ['yes', 'no', 'up', 'down', 'left', 'right', 'on', 'off', 'stop', 'go']\nALL_CLASSES = CORE_CLASSES + ['silence', 'unknown']\nlabel2id = {label: i for i, label in enumerate(ALL_CLASSES)}\nid2label = {i: label for i, label in enumerate(ALL_CLASSES)}\n\ndef load_split_list(file_path):\n    if not os.path.exists(file_path): return set()\n    with open(file_path, 'r') as f: return set(line.strip() for line in f if line.strip())\n\nval_files = load_split_list(os.path.join(BASE_TRAIN_DIR, \"validation_list.txt\"))\ntest_files = load_split_list(os.path.join(BASE_TRAIN_DIR, \"testing_list.txt\"))\n\n# ------------------------------------------------------------------------------ #\n# [2] DATASET DEFINITION\n# ------------------------------------------------------------------------------ #\nclass SpeechCommandsDataset(Dataset):\n    def __init__(self, audio_dir, feature_extractor, split='train', max_samples=None):\n        self.feature_extractor = feature_extractor\n        self.samples = []\n        self.background_noises = []\n        \n        noise_dir = os.path.join(audio_dir, \"_background_noise_\")\n        if os.path.exists(noise_dir):\n            for f in os.listdir(noise_dir):\n                if f.endswith('.wav') and f != 'README.md':\n                    noise_array, _ = sf.read(os.path.join(noise_dir, f))\n                    self.background_noises.append(noise_array)\n        \n        unknown_files = []\n        class_counts = {label: 0 for label in ALL_CLASSES}\n        \n        for folder_name in os.listdir(audio_dir):\n            folder_path = os.path.join(audio_dir, folder_name)\n            if not os.path.isdir(folder_path) or folder_name == '_background_noise_': continue\n                \n            for f in os.listdir(folder_path):\n                if not f.endswith('.wav'): continue\n                    \n                rel_path = f\"{folder_name}/{f}\"\n                is_val = rel_path in val_files\n                is_test = rel_path in test_files\n                \n                if split == 'val' and not is_val: continue\n                if split == 'test' and not is_test: continue\n                if split == 'train' and (is_val or is_test): continue\n                \n                full_path = os.path.join(folder_path, f)\n                \n                if folder_name in CORE_CLASSES:\n                    if max_samples is None or class_counts[folder_name] < max_samples:\n                        self.samples.append({\"path\": full_path, \"label\": label2id[folder_name]})\n                        class_counts[folder_name] += 1\n                else:\n                    unknown_files.append(full_path)\n        \n        random.seed(42)\n        random.shuffle(unknown_files)\n        core_counts = [count for label, count in class_counts.items() if label in CORE_CLASSES and count > 0]\n        target_unknown_count = max_samples if max_samples else (min(core_counts) if core_counts else len(unknown_files))\n        if target_unknown_count == 0: target_unknown_count = len(unknown_files)\n        \n        for f in unknown_files[:target_unknown_count]:\n            self.samples.append({\"path\": f, \"label\": label2id['unknown']})\n        for _ in range(target_unknown_count):\n            self.samples.append({\"path\": \"REAL_NOISE\", \"label\": label2id['silence']})\n\n    def __len__(self): return len(self.samples)\n\n    def __getitem__(self, idx):\n        item = self.samples[idx]\n        target_length = 16000\n        \n        if item[\"path\"] == \"REAL_NOISE\":\n            if self.background_noises:\n                noise_audio = random.choice(self.background_noises)\n                start_idx = random.randint(0, len(noise_audio) - target_length)\n                speech = noise_audio[start_idx : start_idx + target_length] * random.uniform(0.0, 0.8)\n            else:\n                speech = np.random.normal(0, 0.001, target_length)\n        else:\n            speech, _ = sf.read(item[\"path\"])\n            \n        if len(speech) < target_length: speech = np.pad(speech, (0, target_length - len(speech)), 'constant')\n        else: speech = speech[:target_length]\n            \n        inputs = self.feature_extractor(speech, sampling_rate=16000, return_tensors=\"pt\", padding=True)\n        return {\"input_values\": inputs.input_values.squeeze(0), \"labels\": torch.tensor(item[\"label\"], dtype=torch.long)}\n\n# ------------------------------------------------------------------------------ #\n# [3] PREPARE DATA & METRICS\n# ------------------------------------------------------------------------------ #\nprint(\"[INFO] Loading Feature Extractor & Datasets...\")\nfeature_extractor = Wav2Vec2FeatureExtractor.from_pretrained(\"facebook/wav2vec2-base\")\n\ntrain_dataset = SpeechCommandsDataset(AUDIO_DIR, feature_extractor, split='train', max_samples=800)\neval_dataset = SpeechCommandsDataset(AUDIO_DIR, feature_extractor, split='val', max_samples=800)\ntest_dataset = SpeechCommandsDataset(AUDIO_DIR, feature_extractor, split='test', max_samples=800)\n\naccuracy_metric = evaluate.load(\"accuracy\")\ndef compute_metrics(eval_pred):\n    predictions = np.argmax(eval_pred.predictions, axis=1)\n    return accuracy_metric.compute(predictions=predictions, references=eval_pred.label_ids)\n\n# ------------------------------------------------------------------------------ #\n# [4] THE MASTER EXPERIMENT FUNCTION\n# ------------------------------------------------------------------------------ #\nfrom sklearn.metrics import confusion_matrix\nimport seaborn as sns\n\ndef run_robustness_experiment(exp_name, strategy, lr_candidates, epochs, num_runs=5):\n    print(f\"\\n\\n{'='*80}\")\n    print(f\"STARTING EXPERIMENT: {exp_name.upper()} (Strategy: {strategy})\")\n    print(f\"{'='*80}\")\n    \n    all_results = {lr: [] for lr in lr_candidates}\n    best_overall_acc = 0\n    best_lr = None\n    \n    best_run_history = None\n    best_run_y_true = None\n    best_run_y_pred = None\n    \n    color_palette = ['red', 'blue', 'green', 'orange', 'purple']\n    lr_colors = {lr: color_palette[i % len(color_palette)] for i, lr in enumerate(lr_candidates)}\n\n    for lr in lr_candidates:\n        print(f\"\\n[{exp_name}] TESTING LEARNING RATE: {lr}\")\n        \n        for run in range(num_runs):\n            current_seed = 42 + run\n            torch.manual_seed(current_seed)\n            np.random.seed(current_seed)\n            random.seed(current_seed)\n\n            model = Wav2Vec2ForSequenceClassification.from_pretrained(\n                \"facebook/wav2vec2-base\", num_labels=len(ALL_CLASSES), label2id=label2id, id2label=id2label\n            )\n            \n            model.freeze_feature_encoder()\n            if strategy == 'full':\n                for param in model.wav2vec2.parameters(): param.requires_grad = False\n                for param in model.classifier.parameters(): param.requires_grad = True\n            elif strategy == 'partial':\n                for layer in model.wav2vec2.encoder.layers[:6]:\n                    for param in layer.parameters(): param.requires_grad = False\n            elif strategy == 'none': pass\n\n            training_args = TrainingArguments(\n                output_dir=f\"/kaggle/tmp/{strategy}_lr_{lr}_run_{run}\",\n                eval_strategy=\"epoch\",\n                logging_strategy=\"epoch\",\n                save_strategy=\"no\",\n                learning_rate=lr,\n                per_device_train_batch_size=32,\n                per_device_eval_batch_size=32,\n                num_train_epochs=epochs,\n                fp16=True,\n            )\n\n            trainer = Trainer(\n                model=model, args=training_args,\n                train_dataset=train_dataset, eval_dataset=eval_dataset,\n                compute_metrics=compute_metrics\n            )\n\n            trainer.train()\n            \n            metrics = trainer.evaluate(eval_dataset=test_dataset)\n            final_test_acc = metrics['eval_accuracy']\n            \n            pred_output = trainer.predict(test_dataset)\n            y_pred = np.argmax(pred_output.predictions, axis=1)\n            y_true = pred_output.label_ids\n            history = trainer.state.log_history\n            \n\n            val_acc_list = [x['eval_accuracy'] for x in history if 'eval_accuracy' in x]\n            val_loss_list = [x['eval_loss'] for x in history if 'eval_loss' in x] \n            \n            if len(val_acc_list) > epochs: val_acc_list = val_acc_list[:epochs]\n            if len(val_loss_list) > epochs: val_loss_list = val_loss_list[:epochs] \n            \n            all_results[lr].append({\n                'val_acc': val_acc_list,\n                'val_loss': val_loss_list, \n                'test_acc': final_test_acc\n            })\n            \n            print(f\"   --> Run {run+1}/{num_runs} completed. Test Acc: {final_test_acc:.4f}\")\n            \n            if final_test_acc > best_overall_acc:\n                best_overall_acc = final_test_acc\n                best_lr = lr\n                best_run_history = history\n                best_run_y_true = y_true\n                best_run_y_pred = y_pred\n\n            del model\n            del trainer\n            gc.collect()\n            torch.cuda.empty_cache()\n\n    print(f\"\\nBEST LR FOR {exp_name}: {best_lr} (Acc: {best_overall_acc:.4f})\")\n\n# ========================================================================= #\n    #                             PLOTTING SECTION\n    # ========================================================================= #\n    \n    # --- PLOT 1: ROBUSTNESS ANALYSIS (LOSS & ACCURACY) ---\n    plt.figure(figsize=(16, 6)) \n    \n    # --- Subplot 1: Validation Loss ---\n    plt.subplot(1, 2, 1)\n    for lr in lr_candidates:\n        lr_losses = [run['val_loss'] for run in all_results[lr]]\n        min_len = min([len(lst) for lst in lr_losses]) if lr_losses else 0\n        if min_len > 0:\n            lr_losses_uniform = np.array([lst[:min_len] for lst in lr_losses])\n            mean_loss = np.mean(lr_losses_uniform, axis=0)\n            x_axis = range(1, min_len + 1)\n            \n            for run_loss in lr_losses_uniform:\n                plt.plot(x_axis, run_loss, color=lr_colors[lr], alpha=0.15)\n            plt.plot(x_axis, mean_loss, color=lr_colors[lr], linewidth=3, label=f\"Mean LR {lr}\")\n\n    plt.title(f'Validation Loss Reproducibility ({exp_name})', fontsize=14)\n    plt.xlabel('Epochs')\n    plt.ylabel('Validation Loss')\n    plt.legend()\n    plt.grid(True, alpha=0.3)\n\n    # --- Subplot 2: Validation Accuracy ---\n    plt.subplot(1, 2, 2)\n    for lr in lr_candidates:\n        lr_accs = [run['val_acc'] for run in all_results[lr]]\n        min_len = min([len(lst) for lst in lr_accs]) if lr_accs else 0\n        if min_len > 0:\n            lr_accs_uniform = np.array([lst[:min_len] for lst in lr_accs])\n            mean_acc = np.mean(lr_accs_uniform, axis=0)\n            x_axis = range(1, min_len + 1)\n            \n            for run_acc in lr_accs_uniform:\n                plt.plot(x_axis, run_acc, color=lr_colors[lr], alpha=0.15)\n            plt.plot(x_axis, mean_acc, color=lr_colors[lr], linewidth=3, label=f\"Mean LR {lr}\")\n\n    plt.title(f'Validation Accuracy Reproducibility ({exp_name})', fontsize=14)\n    plt.xlabel('Epochs')\n    plt.ylabel('Validation Accuracy')\n    plt.legend()\n    plt.grid(True, alpha=0.3)\n\n    plt.tight_layout()\n    plt.savefig(f\"robustness_{strategy}.png\", dpi=300)\n    plt.show()\n\n\n        \n# ------------------------------------------------------------------------------ #\n# [5] EXPERIMENT SWITCHBOARD (UNCOMMENT TO RUN)\n# ------------------------------------------------------------------------------ #\n\n# --- EXPERIMENT 1: FULL FREEZING ---\n# (Needs higher LR to move the rigid classification head. Trains fast, so 8 epochs are fine)\nrun_robustness_experiment(\n     exp_name=\"Full Freezing\", \n     strategy=\"full\", \n     lr_candidates=[1e-1, 1e-2, 1e-3], \n     epochs=5, \n     num_runs=3\n)\n\n# --- EXPERIMENT 2: PARTIAL FREEZING ---\n# (Needs medium LR. 5 epochs is usually enough to see the trend without overfitting)\n#run_robustness_experiment(\n#    exp_name=\"Partial Freezing\", \n#    strategy=\"partial\", \n#    lr_candidates=[1e-3, 1e-4, 1e-5], \n#    epochs=5, \n#    num_runs=3\n#)\n\n# --- EXPERIMENT 3: NO FREEZING (FULL FINE-TUNING) ---\n# (Needs lower LR to not destroy pre-trained weights. Can be prone to overfitting)\n#run_robustness_experiment(\n#    exp_name=\"No Freezing\", \n#    strategy=\"none\", \n#    lr_candidates=[1e-3, 1e-4, 1e-5], \n#    epochs=5, \n#    num_runs=3\n#)\n","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"outputs":[],"execution_count":null}]}