{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"nvidiaTeslaT4","dataSources":[{"sourceId":106809,"databundleVersionId":13056355,"sourceType":"competition"}],"dockerImageVersionId":31192,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# --- Cell 1: 匯入、安裝與全域設定 ---\n\n# 0. 安裝必要的解碼套件\nimport subprocess\nimport sys\ntry:\n    import pyctcdecode\n    print(\"✅ pyctcdecode 已安裝\")\nexcept ImportError:\n    print(\"正在安裝 pyctcdecode...\")\n    subprocess.check_call([sys.executable, \"-m\", \"pip\", \"install\", \"pyctcdecode\"])\n    import pyctcdecode\n\nimport os, glob, h5py, gc, re, urllib.request\nimport numpy as np\nimport pandas as pd\nimport torch\nimport torch.nn as nn\nfrom torch.utils.data import Dataset, DataLoader\nfrom torch.nn.utils.rnn import pad_sequence\nfrom tqdm.notebook import tqdm\nfrom pyctcdecode import build_ctcdecoder\n\n# 1. 設定與清理\ngc.collect()\ntorch.cuda.empty_cache()\n\n# Character List (29 classes)\nCHAR_LIST = ['BLANK', ' ', 'a','b','c','d','e','f','g','h','i','j','k','l','m','n',\n             'o','p','q','r','s','t','u','v','w','x','y','z',\"'\"]\nCHAR_TO_IDX = {c: i for i, c in enumerate(CHAR_LIST)}\nNUM_CLASSES = len(CHAR_LIST)\n\nCONFIG = {\n    'data_root': '/kaggle/input/brain-to-text-25', # 確保指向官方數據集\n    'input_dim': 512,\n    'd_dim': 14,\n    'hidden_dim': 256,\n    'num_layers': 1,     # 單層 GRU 最穩定\n    'batch_size': 4,\n    'lr': 1e-4,\n    'epochs': 15,        # 建議至少跑 15 輪\n    'device': torch.device('cuda' if torch.cuda.is_available() else 'cpu')\n}\n\nprint(f\"✅ 設定完成。使用設備：{CONFIG['device']}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-12-08T19:04:12.479454Z","iopub.execute_input":"2025-12-08T19:04:12.480098Z","iopub.status.idle":"2025-12-08T19:04:23.002735Z","shell.execute_reply.started":"2025-12-08T19:04:12.480072Z","shell.execute_reply":"2025-12-08T19:04:23.002103Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# --- Cell 2: 語言模型設定與建構 ---\n\nprint(\"準備語言模型...\")\nlm_path = \"count_1w.txt\"\nif not os.path.exists(lm_path):\n    print(\"下載單字頻率表...\")\n    # 注意：這個下載步驟可能需要 Notebook 開啟網路連線\n    urllib.request.urlretrieve(\"http://norvig.com/ngrams/count_1w.txt\", lm_path)\n\nunigrams = {}\nwith open(lm_path, \"r\") as f:\n    for line in f:\n        parts = line.strip().split(\"\\t\")\n        if len(parts) == 2: unigrams[parts[0]] = int(parts[1])\n\n# 為 pyctcdecode 準備標籤列表\nvocab = [c for c in CHAR_LIST]\nvocab[0] = \"\" # pyctcdecode 專用修正：將 'BLANK' 替換為空字串\n\n# 建立 Beam Search 解碼器\ndecoder = build_ctcdecoder(labels=vocab, unigrams=unigrams, alpha=0.5, beta=1.0)\nprint(\"✅ 智慧解碼器 (pyctcdecode) 準備完成。\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-12-08T19:04:23.79699Z","iopub.execute_input":"2025-12-08T19:04:23.797563Z","iopub.status.idle":"2025-12-08T19:04:24.132018Z","shell.execute_reply.started":"2025-12-08T19:04:23.797535Z","shell.execute_reply":"2025-12-08T19:04:24.131419Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# --- Cell 3: 數據集與數據加載器 ---\n\nclass BrainDataset(Dataset):\n    def __init__(self, root, mode='train'):\n        self.mode = mode\n        self.samples = []\n        self.stride = 1\n        self.patch_size = CONFIG['d_dim']\n\n        print(f\"掃描 {mode} 數據中...\")\n        all_files = glob.glob(os.path.join(root, \"**/*.hdf5\"), recursive=True)\n        \n        for f_path in tqdm(all_files, desc=\"Indexing\"):\n            fname = os.path.basename(f_path)\n            is_test = 'test' in fname\n            if mode == 'train' and is_test: continue\n            if mode == 'test' and not is_test: continue\n            self.samples.append(f_path)\n\n        print(f\"✔ {mode} 共找到 {len(self.samples)} 個檔案\")\n\n    def __len__(self): return len(self.samples)\n\n    def __getitem__(self, idx):\n        f_path = self.samples[idx]\n        try:\n            with h5py.File(f_path, 'r') as f:\n                keys = list(f.keys())\n                trial_keys = [k for k in keys if k.startswith('trial_')]\n                if not trial_keys: return None\n                \n                # 訓練時隨機取一個 trial，測試時取第一個\n                key = np.random.choice(trial_keys) if self.mode == 'train' else trial_keys[0]\n                group = f[key]\n\n                # 1. 特徵處理 (Z-Score + Tanh)\n                raw = group['input_features'][:]\n                neural = torch.tensor(raw, dtype=torch.float32)\n                neural = (neural - neural.mean()) / (neural.std() + 1e-8)\n                neural = torch.tanh(neural)\n\n                # 2. Patching 處理\n                T, D = neural.shape\n                d = CONFIG['d_dim']\n                num_patches = (T - d) // 1 + 1\n                \n                # 處理過短序列：用 0 填充到 patch_size\n                if num_patches <= 0:\n                    pad = torch.zeros((d - T, CONFIG['input_dim']))\n                    neural = torch.cat([neural, pad], 0)\n                    num_patches = 1\n                \n                patches = torch.stack([neural[i:i+d] for i in range(num_patches)], 0)\n                \n                # 3. 標籤處理\n                if self.mode == 'train':\n                    if 'transcription' not in group: return None\n                    raw_bytes = group['transcription'][:]\n                    raw_bytes = raw_bytes[raw_bytes != 0]\n                    text = \"\".join([chr(c) for c in raw_bytes]).lower()\n                    label = [CHAR_TO_IDX[c] for c in text if c in CHAR_TO_IDX]\n                    \n                    # CTC 長度過濾\n                    if patches.shape[0] < len(label): return None\n                    \n                    return patches, torch.tensor(label, dtype=torch.long)\n                \n                # Test mode 回傳 key (用於提交 ID)\n                return patches, key\n        except Exception: return None\n\ndef collate_fn(batch):\n    batch = [x for x in batch if x is not None]\n    if len(batch) == 0: return None, None, None, None\n    inputs, targets = zip(*batch)\n    input_lens = torch.tensor([x.shape[0] for x in inputs])\n    inputs_pad = pad_sequence(inputs, batch_first=True)\n    \n    if isinstance(targets[0], str): # Test mode\n        return inputs_pad, list(targets), input_lens, None\n    \n    # Train mode\n    target_lens = torch.tensor([len(t) for t in targets])\n    targets_pad = pad_sequence(targets, batch_first=True)\n    return inputs_pad, targets_pad, input_lens, target_lens","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-12-08T19:04:32.731185Z","iopub.execute_input":"2025-12-08T19:04:32.73171Z","iopub.status.idle":"2025-12-08T19:04:32.742867Z","shell.execute_reply.started":"2025-12-08T19:04:32.731686Z","shell.execute_reply":"2025-12-08T19:04:32.742117Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# --- Cell 4: 模型定義 ---\n\nclass BrainModel(nn.Module):\n    # [Image of GRU-CTC Architecture Diagram]\n    def __init__(self, cfg, num_classes):\n        super().__init__()\n        # Patching 後的輸入維度: 14 * 512\n        input_size = cfg['d_dim'] * cfg['input_dim']\n        \n        # 降維 Projection Layer\n        self.proj = nn.Sequential(\n            nn.Linear(input_size, cfg['hidden_dim']),\n            nn.LayerNorm(cfg['hidden_dim']),\n            nn.GELU(),\n            nn.Dropout(0.2)\n        )\n        # 雙向 GRU\n        self.rnn = nn.GRU(\n            cfg['hidden_dim'], cfg['hidden_dim'],\n            num_layers=cfg['num_layers'], bidirectional=True, batch_first=True\n        )\n        # 輸出層\n        self.fc = nn.Linear(cfg['hidden_dim'] * 2, num_classes) # * 2 因為是雙向\n        self.lsm = nn.LogSoftmax(dim=2) # 轉為 Log 機率\n\n    def forward(self, x):\n        B,T,D,F = x.shape\n        x = x.view(B, T, D*F) # 攤平 Patch (B, T, D*F)\n        x = self.proj(x)\n        x,_ = self.rnn(x)\n        return self.lsm(self.fc(x))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-12-08T19:04:36.483775Z","iopub.execute_input":"2025-12-08T19:04:36.484306Z","iopub.status.idle":"2025-12-08T19:04:36.490118Z","shell.execute_reply.started":"2025-12-08T19:04:36.484281Z","shell.execute_reply":"2025-12-08T19:04:36.489358Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# --- Cell 5: 訓練迴圈 ---\n\ndef run_training():\n    torch.backends.cudnn.enabled = False \n    print(\"🚀 準備訓練資料...\")\n    ds = BrainDataset(CONFIG['data_root'], mode='train')\n    dl = DataLoader(ds, batch_size=CONFIG['batch_size'], shuffle=True, collate_fn=collate_fn)\n    \n    model = BrainModel(CONFIG, NUM_CLASSES).to(CONFIG['device'])\n    opt = torch.optim.AdamW(model.parameters(), lr=CONFIG['lr'])\n    crit = nn.CTCLoss(blank=0, zero_infinity=True)\n    \n    print(\"🔥 開始訓練...\")\n    model.train()\n    \n    for ep in range(CONFIG['epochs']):\n        pbar = tqdm(dl, desc=f\"Epoch {ep+1}\")\n        total_loss = 0\n        steps = 0\n        for batch in pbar:\n            if batch[0] is None: continue\n            x, y, xl, yl = batch\n            x, y = x.to(CONFIG['device']), y.to(CONFIG['device'])\n            \n            opt.zero_grad()\n            out = model(x).permute(1, 0, 2) # [T, B, C]\n            \n            try:\n                loss = crit(out, y, xl, yl)\n            except: continue\n            \n            if torch.isnan(loss): continue\n            loss.backward()\n            torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)\n            opt.step()\n            \n            total_loss += loss.item()\n            steps += 1\n            pbar.set_postfix({'loss': f\"{loss.item():.4f}\"})\n            \n        if steps > 0:\n            print(f\"✅ Epoch {ep+1} Avg Loss: {total_loss/steps:.4f}\")\n            \n    return model\n\n# 執行訓練\ntrained_model = run_training()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-12-08T19:04:40.113068Z","iopub.execute_input":"2025-12-08T19:04:40.113729Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# --- Cell 6: 推理與提交迴圈 ---\n\ndef generate_submission(model):\n    print(\"🚀 生成 submission...\")\n    ds = BrainDataset(CONFIG['data_root'], mode='test')\n    dl = DataLoader(ds, batch_size=1, collate_fn=collate_fn)\n    \n    model.eval()\n    results = []\n    idx = 0\n    \n    with torch.no_grad():\n        for batch in tqdm(dl):\n            # 必須保持 ID 連續，即使遇到壞檔\n            if batch[0] is None: \n                results.append({\"id\": idx, \"text\": \"\"})\n                idx += 1\n                continue\n                \n            x, keys, xl, _ = batch\n            x = x.to(CONFIG['device'])\n            \n            # 取得 Logits\n            logits = model(x).squeeze(0).cpu().numpy()\n            \n            # 使用 Beam Search 智慧解碼\n            text = decoder.decode(logits, beam_width=20)\n            \n            results.append({\"id\": idx, \"text\": text})\n            idx += 1\n            \n    df = pd.DataFrame(results)\n    df.to_csv(\"submission.csv\", index=False)\n    print(\"✔ submission.csv 已輸出\")\n    print(\"--- 提交檔案預覽 ---\")\n    print(df.head())\n\n# 執行推理\ngenerate_submission(trained_model)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# --- Cell 7: 檢查提交檔案 ---\n\n# 確認 submission.csv 存在且非空\nif os.path.exists(\"submission.csv\"):\n    df_check = pd.read_csv(\"submission.csv","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}