{"cells":[{"cell_type":"markdown","metadata":{},"source":"# Deep Past MT Kaggle Submission\nGenerated from template.\n"},{"cell_type":"code","execution_count":null,"id":"0b50a7ba","metadata":{},"outputs":[],"source":"import json\nimport os\nimport re\nfrom pathlib import Path\n\nimport pandas as pd\nimport torch\nfrom transformers import AutoModelForSeq2SeqLM, AutoTokenizer\n\nKERNEL_CONFIG = json.loads(\"{\\\"run_id\\\": \\\"a1-20260209-0900-nllb600m\\\", \\\"test_path\\\": \\\"/kaggle/input/deep-past-initiative-machine-translation/test.csv\\\", \\\"model_path\\\": \\\"/kaggle/input/deep-past-mt-artifacts/a1-20260209-0900-nllb600m/best\\\", \\\"output_path\\\": \\\"/kaggle/working/submission.csv\\\", \\\"model_name\\\": \\\"facebook/nllb-200-distilled-600M\\\", \\\"tokenizer_name\\\": \\\"facebook/nllb-200-distilled-600M\\\", \\\"source_lang\\\": \\\"akk\\\", \\\"target_lang\\\": \\\"eng_Latn\\\", \\\"max_source_length\\\": 256, \\\"batch_size\\\": 8, \\\"generation_kwargs\\\": {\\\"num_beams\\\": 4, \\\"max_new_tokens\\\": 128, \\\"length_penalty\\\": 1.0, \\\"no_repeat_ngram_size\\\": 0, \\\"early_stopping\\\": true, \\\"do_sample\\\": false}, \\\"enable_gpu\\\": true, \\\"enable_internet\\\": false, \\\"kernel_schema_version\\\": \\\"kaggle-submit-v3\\\", \\\"build_timestamp_utc\\\": \\\"2026-02-09T14:39:44Z\\\"}\")\n\nprint('Kernel config:')\nprint(json.dumps(KERNEL_CONFIG, indent=2, ensure_ascii=False))\n\nkernel_schema_version = str(KERNEL_CONFIG.get('kernel_schema_version') or '').strip()\nbuild_timestamp_utc = str(KERNEL_CONFIG.get('build_timestamp_utc') or '').strip()\nprint(\n    f'Kernel build marker: schema={kernel_schema_version} '\n    f'build_timestamp_utc={build_timestamp_utc}'\n)\n\nSUBSCRIPT_TRANSLATION = str.maketrans({\n    '₀': '0', '₁': '1', '₂': '2', '₃': '3', '₄': '4',\n    '₅': '5', '₆': '6', '₇': '7', '₈': '8', '₉': '9',\n})\nSCRIBAL_MARKER_RE = re.compile(r\"[!?/:.]\")\nLINE_NUMBER_RE = re.compile(r\"(\\d+)'(?=\\b)\")\nBRACKET_RE = re.compile(r'\\[([^\\]]*)\\]')\nWHITESPACE_RE = re.compile(r\"\\s+\")\n\n\ndef _replace_bracket_gaps(text: str) -> str:\n    def repl(match: re.Match[str]) -> str:\n        content = match.group(1).strip()\n        lowered = content.lower()\n        if not content:\n            return '<big_gap>'\n        if re.fullmatch(r'(x\\s*)+', lowered):\n            return '<gap>'\n        if '...' in content or '…' in content:\n            return '<big_gap>'\n        if re.fullmatch(r'[.\\s]+', content):\n            return '<big_gap>'\n        return content\n\n    return BRACKET_RE.sub(repl, text)\n\n\ndef normalize_transliteration(text: str) -> str:\n    if not text:\n        return ''\n    normalized = text.strip()\n    normalized = normalized.replace('˹', '').replace('˺', '')\n    normalized = normalized.replace('…', '<big_gap>')\n    normalized = normalized.translate(SUBSCRIPT_TRANSLATION)\n    normalized = normalized.replace('Ḫ', 'H').replace('ḫ', 'h')\n    normalized = LINE_NUMBER_RE.sub(r'\\1', normalized)\n    normalized = _replace_bracket_gaps(normalized)\n    normalized = SCRIBAL_MARKER_RE.sub('', normalized)\n    normalized = re.sub(r'<\\s*(?!/?(?:big_)?gap\\b)([^<>]+)\\s*>', r'\\1', normalized)\n    normalized = normalized.replace('<<', '').replace('>>', '')\n    normalized = WHITESPACE_RE.sub(' ', normalized).strip()\n    return normalized\n\n\nprimary_model_path = str(KERNEL_CONFIG.get('model_path') or '').strip()\nrun_id = str(KERNEL_CONFIG.get('run_id') or '').strip()\ntest_path = KERNEL_CONFIG['test_path']\noutput_path = KERNEL_CONFIG['output_path']\nsource_lang = KERNEL_CONFIG.get('source_lang')\ntarget_lang = KERNEL_CONFIG.get('target_lang')\nmax_source_length = int(KERNEL_CONFIG['max_source_length'])\nbatch_size = int(KERNEL_CONFIG['batch_size'])\ngeneration_kwargs = dict(KERNEL_CONFIG['generation_kwargs'])\n\n\ndef resolve_model_path(primary_path: str, run_id: str) -> Path:\n    dataset_slug = ''\n    if primary_path.startswith('/kaggle/input/'):\n        suffix = primary_path[len('/kaggle/input/') :]\n        dataset_slug = suffix.split('/', 1)[0].strip() if suffix else ''\n\n    canonical_candidate = ''\n    if dataset_slug and run_id:\n        canonical_candidate = str(Path(f'/kaggle/input/{dataset_slug}/{run_id}/best'))\n\n    primary_candidate = str(Path(primary_path)) if primary_path else ''\n    legacy_root_best = str(Path(f'/kaggle/input/{dataset_slug}/best')) if dataset_slug else ''\n    runs_compat_candidate = ''\n    if primary_candidate and '/runs/' in primary_candidate:\n        runs_compat_candidate = str(Path(primary_candidate.replace('/runs/', '/', 1)))\n\n    checked: list[str] = []\n    seen: set[str] = set()\n\n    def register(candidate: str) -> str:\n        normalized = str(Path(candidate))\n        if normalized in seen:\n            return ''\n        seen.add(normalized)\n        checked.append(normalized)\n        return normalized\n\n    def has_checkpoint(candidate: str) -> bool:\n        candidate_path = Path(candidate)\n        return candidate_path.is_dir() and (candidate_path / 'config.json').is_file()\n\n    if canonical_candidate:\n        canonical_checked = register(canonical_candidate)\n        if canonical_checked and has_checkpoint(canonical_checked):\n            if primary_candidate and primary_candidate != canonical_checked and primary_candidate == legacy_root_best:\n                print(\n                    f\"Warning: ignoring non-canonical model_path '{primary_candidate}' and using canonical \"\n                    f\"run path '{canonical_checked}'.\"\n                )\n            return Path(canonical_checked)\n\n    if primary_candidate:\n        primary_checked = register(primary_candidate)\n        if primary_checked and has_checkpoint(primary_checked):\n            if canonical_candidate and primary_checked == legacy_root_best:\n                raise FileNotFoundError(\n                    f\"Non-canonical model_path resolved to legacy root path '{primary_checked}', but expected \"\n                    f\"canonical path '{canonical_candidate}' for run_id='{run_id}'. \"\n                    \"Republish artifacts for this RUN_ID and relaunch the Kaggle draft session.\"\n                )\n            return Path(primary_checked)\n\n    if runs_compat_candidate:\n        runs_checked = register(runs_compat_candidate)\n        if runs_checked and has_checkpoint(runs_checked):\n            return Path(runs_checked)\n\n    details = '\\n- '.join(checked) if checked else '<none>'\n    expected = canonical_candidate or '<unavailable>'\n    raise FileNotFoundError(\n        f\"Unable to resolve a local model path. Original={primary_path}. Expected canonical path={expected}. \"\n        f\"Checked:\\n- {details}. Republish artifacts for this RUN_ID and relaunch the Kaggle draft session.\"\n    )\n\n\ndef validate_local_tokenizer_artifacts(model_path: Path) -> None:\n    tokenizer_config_path = model_path / 'tokenizer_config.json'\n    if not tokenizer_config_path.is_file():\n        raise RuntimeError(\n            f'Missing tokenizer_config.json at {tokenizer_config_path}. '\n            f'Republish artifacts for RUN_ID={run_id} and relaunch the Kaggle draft session.'\n        )\n\n    try:\n        tokenizer_config = json.loads(tokenizer_config_path.read_text(encoding='utf-8'))\n    except Exception as exc:\n        raise RuntimeError(\n            f'Invalid tokenizer_config.json at {tokenizer_config_path}: {exc}. '\n            f'Republish artifacts for RUN_ID={run_id} and relaunch the Kaggle draft session.'\n        ) from exc\n\n    tokenizer_class = str(tokenizer_config.get('tokenizer_class') or '').strip()\n    if tokenizer_class == 'TokenizersBackend':\n        raise RuntimeError(\n            f\"Invalid tokenizer metadata at {tokenizer_config_path}: tokenizer_class='TokenizersBackend'. \"\n            f\"Republish artifacts for RUN_ID={run_id} and relaunch the Kaggle draft session.\"\n        )\n\n    extra_special_tokens = tokenizer_config.get('extra_special_tokens')\n    if extra_special_tokens is not None and not isinstance(extra_special_tokens, dict):\n        raise RuntimeError(\n            f'Incompatible tokenizer metadata at {tokenizer_config_path}: '\n            f'extra_special_tokens must be a JSON object/dict for Kaggle runtime, '\n            f'got {type(extra_special_tokens).__name__}. '\n            f'Republish artifacts for RUN_ID={run_id} and relaunch the Kaggle draft session.'\n        )\n\n    tokenizer_asset_names = ('tokenizer.json', 'spiece.model', 'sentencepiece.bpe.model')\n    has_tokenizer_asset = any((model_path / name).is_file() for name in tokenizer_asset_names)\n    if not has_tokenizer_asset:\n        expected = ', '.join(tokenizer_asset_names)\n        raise RuntimeError(\n            f'Missing tokenizer assets in {model_path}. Expected one of: {expected}. '\n            f'Republish artifacts for RUN_ID={run_id} and relaunch the Kaggle draft session.'\n        )\n\n\ndef load_tokenizer(model_path: Path):\n    validate_local_tokenizer_artifacts(model_path)\n    return AutoTokenizer.from_pretrained(\n        str(model_path),\n        local_files_only=True,\n        use_fast=False,\n    )\n\n\ndevice = 'cuda' if torch.cuda.is_available() else 'cpu'\nprint(f'Using device: {device}')\nmodel_path = resolve_model_path(primary_model_path, run_id)\nprint(f'Resolved model path: {model_path}')\n\ntokenizer = load_tokenizer(model_path)\nif getattr(tokenizer, 'pad_token', None) is None and getattr(tokenizer, 'eos_token', None):\n    tokenizer.pad_token = tokenizer.eos_token\n\nif source_lang and hasattr(tokenizer, 'src_lang'):\n    tokenizer.src_lang = source_lang\nif target_lang and hasattr(tokenizer, 'tgt_lang'):\n    tokenizer.tgt_lang = target_lang\nif target_lang and hasattr(tokenizer, 'get_lang_id'):\n    try:\n        generation_kwargs['forced_bos_token_id'] = int(tokenizer.get_lang_id(target_lang))\n    except Exception:\n        pass\n\nmodel = AutoModelForSeq2SeqLM.from_pretrained(\n    str(model_path),\n    local_files_only=True,\n).to(device).eval()\n\ntest_df = pd.read_csv(test_path)\nif 'transliteration' not in test_df.columns or 'id' not in test_df.columns:\n    raise ValueError('Expected test.csv columns: id, transliteration')\n\nsources = [normalize_transliteration(str(text)) for text in test_df['transliteration'].fillna('')]\npredictions: list[str] = []\n\nfor idx in range(0, len(sources), batch_size):\n    batch = sources[idx : idx + batch_size]\n    inputs = tokenizer(\n        batch,\n        return_tensors='pt',\n        padding=True,\n        truncation=True,\n        max_length=max_source_length,\n    )\n    inputs = {k: v.to(device) for k, v in inputs.items()}\n    with torch.no_grad():\n        outputs = model.generate(**inputs, **generation_kwargs)\n    decoded = tokenizer.batch_decode(outputs, skip_special_tokens=True)\n    predictions.extend([re.sub(r'\\s+', ' ', item).strip() for item in decoded])\n\nsubmission = pd.DataFrame({'id': test_df['id'], 'translation': predictions})\nsubmission.to_csv(output_path, index=False)\nprint(f'Wrote {len(submission)} rows to {output_path}')\n"}],"metadata":{"kaggle":{"isGpuEnabled":true,"isInternetEnabled":false,"language":"python","sourceType":"notebook"},"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"name":"python"}},"nbformat":4,"nbformat_minor":5}