{"cells": [{"cell_type": "markdown", "metadata": {}, "source": "# Sign Language Feature Encoder Training\n\n## Overview \ud83d\ude80\n\nThis notebook trains a reusable hand-motion encoder for dynamic sign retrieval. It first learns general temporal representations from the Google Isolated Sign Language Recognition corpus, then adapts the encoder to an object-NPY landmark collection. The temporary classification heads provide training supervision; exported features are not tied to a fixed vocabulary.\n\n**Input contract:** two hands \u00d7 21 landmarks \u00d7 `(x, y, z, velocity)`, up to 64 frames.  \n**Output contract:** 128-dimensional L2-normalized features for every valid frame and zeros for padding.\n\nRun the notebook from top to bottom on a Kaggle **single T4** runtime. Production defaults follow the `hand168-temporal` preprocessing contract used by the C++ runtime; set `SMOKE_TEST = True` for a short integration run. Dataset paths can be set explicitly in the configuration cell or discovered under `/kaggle/input`.\n\nData sources:\n\n- [Google - Isolated Sign Language Recognition](https://www.kaggle.com/competitions/asl-signs/data)\n- [ASL-preprocessing 7 output](https://www.kaggle.com/code/abdelrhmankaram/asl-preprocessing-7/output)\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Configuration \u2699\ufe0f\n\nAll user-editable settings live in this cell. Explicit paths are recommended for reproducible public notebooks.\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "from dataclasses import asdict, dataclass, field\nfrom pathlib import Path\nfrom typing import Any, Dict, Iterable, Iterator, List, Mapping, Optional, Sequence, Tuple\nimport contextlib, csv, gc, hashlib, io, json, math, os, pickle, random, shutil, sys, time\nfrom concurrent.futures import ThreadPoolExecutor, as_completed\n\nimport numpy as np\nimport pandas as pd\nimport torch\nfrom torch import nn\nfrom torch.nn import functional as F\nfrom torch.utils.data import Dataset, DataLoader, Sampler\nfrom tqdm.auto import tqdm\n\nPREPROCESSING_CONTRACT = \"hand168-temporal\"\nCHECKPOINT_SCHEMA = 1\nSMOKE_TEST = False\n\n@dataclass\nclass TrainSettings:\n    epochs: int\n    batches_per_epoch: int\n    classes_per_batch: int\n    samples_per_class: int\n    learning_rate: float\n    freeze_epochs: int = 0\n    dtw_every: int = 0\n\n@dataclass\nclass Config:\n    source_root: Optional[str] = None\n    target_landmarks: Optional[str] = None\n    work_dir: str = \"/kaggle/working/signlang-det\"\n    seed: int = 20260712\n    max_frames: int = 64\n    min_frames: int = 12\n    max_input_frames: int = 120\n    feature_dim: int = 168\n    embedding_dim: int = 128\n    padding_value: float = -100.0\n    num_workers: int = 0\n    cache_workers: int = 4\n    cache_chunk_size: int = 256\n    cache_flush_chunks: int = 4\n    weight_decay: float = 1e-4\n    use_amp: bool = True\n    use_data_parallel: bool = False\n    resume: bool = True\n    cleanup_epoch_checkpoints: bool = True\n    keep_latest_checkpoint: bool = False\n    top_k: int = 20\n    dtw_window: int = 12\n    source_reference_per_class: int = 1\n    source_query_per_class: int = 4\n    final_query_per_class: int = 2\n    target_right_slice: Tuple[int, int] = (0, 21)\n    target_left_slice: Tuple[int, int] = (21, 42)\n    source: TrainSettings = field(default_factory=lambda: TrainSettings(30, 300, 16, 4, 3e-4))\n    adaptation: TrainSettings = field(default_factory=lambda: TrainSettings(40, 200, 16, 2, 1e-4, 2, 5))\n\nCFG = Config()\nif SMOKE_TEST:\n    CFG.source = TrainSettings(1, 2, 2, 2, 3e-4)\n    CFG.adaptation = TrainSettings(1, 2, 2, 2, 1e-4, 0, 1)\n\nWORK_DIR = Path(CFG.work_dir)\nWORK_DIR.mkdir(parents=True, exist_ok=True)\n\ndef seed_everything(seed: int) -> None:\n    random.seed(seed)\n    np.random.seed(seed)\n    torch.manual_seed(seed)\n    if torch.cuda.is_available():\n        torch.cuda.manual_seed_all(seed)\n\nseed_everything(CFG.seed)\nprint(json.dumps(asdict(CFG), indent=2, default=str))\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Environment Check \ud83d\udd0d\n\nCUDA availability alone is insufficient. The selected GPU architecture must also be present in the installed PyTorch build.\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "GOOGLE_ASL_URL = \"https://www.kaggle.com/competitions/asl-signs/data\"\nTARGET_OUTPUT_URL = \"https://www.kaggle.com/code/abdelrhmankaram/asl-preprocessing-7/output\"\nSTANDARD_SOURCE_RELATIVE = \"competitions/asl-signs\"\nSTANDARD_TARGET_RELATIVE = \"notebooks/abdelrhmankaram/asl-preprocessing-7/landmarks\"\n\ndef validate_source_root(path: Path) -> Path:\n    path = path.expanduser().resolve()\n    required = [path / \"train.csv\", path / \"sign_to_prediction_index_map.json\", path / \"train_landmark_files\"]\n    missing = [str(item) for item in required if not item.exists()]\n    if missing:\n        raise FileNotFoundError(\n            f\"Google ASL dataset validation failed at {path}. Missing: {missing}. Source: {GOOGLE_ASL_URL}\"\n        )\n    with (path / \"train.csv\").open(\"r\", newline=\"\", encoding=\"utf-8\") as handle:\n        columns = set(next(csv.reader(handle), []))\n    expected = {\"path\", \"sign\", \"participant_id\"}\n    if not expected.issubset(columns):\n        raise ValueError(f\"Google ASL train.csv is missing columns {sorted(expected - columns)} at {path}\")\n    return path\n\ndef validate_target_landmarks(path: Path) -> Path:\n    path = path.expanduser().resolve()\n    if not path.is_dir() or next(path.glob(\"*.npy\"), None) is None:\n        raise FileNotFoundError(\n            f\"ASL-preprocessing 7 validation failed at {path}; expected landmarks/*.npy. Source: {TARGET_OUTPUT_URL}\"\n        )\n    return path\n\ndef _valid_source_candidates(input_root: Path) -> List[Path]:\n    candidates = []\n    for marker in input_root.rglob(\"sign_to_prediction_index_map.json\"):\n        try: candidates.append(validate_source_root(marker.parent))\n        except (FileNotFoundError, ValueError): pass\n    return sorted(set(candidates))\n\ndef _valid_target_candidates(input_root: Path) -> List[Path]:\n    candidates = []\n    for path in input_root.rglob(\"landmarks\"):\n        try: candidates.append(validate_target_landmarks(path))\n        except FileNotFoundError: pass\n    return sorted(set(candidates))\n\ndef _standard_or_error(input_root: Path, relative: str, validator, candidates, description: str, url: str) -> Path:\n    standard = input_root / relative\n    if standard.exists():\n        return validator(standard)\n    alternatives = candidates(input_root)\n    preview = \"\\n\".join(f\"  - {path}\" for path in alternatives[:20])\n    if len(alternatives) > 1:\n        raise FileNotFoundError(\n            f\"Multiple {description} candidates were found, but the standard Kaggle mount is missing: {standard}. \"\n            f\"Set the corresponding Config path explicitly. Source: {url}\\n{preview}\"\n        )\n    if len(alternatives) == 1:\n        raise FileNotFoundError(\n            f\"The standard Kaggle mount is missing: {standard}. A non-standard candidate was found at \"\n            f\"{alternatives[0]}; set the corresponding Config path explicitly. Source: {url}\"\n        )\n    raise FileNotFoundError(f\"The standard Kaggle mount is missing: {standard}. Attach the data from {url}\")\n\ndef resolve_inputs(cfg: Config, input_root: Path = Path(\"/kaggle/input\")) -> Tuple[Path, Path]:\n    if not input_root.is_dir():\n        raise FileNotFoundError(\"/kaggle/input is unavailable. Attach both required Kaggle data sources.\")\n    source = validate_source_root(Path(cfg.source_root)) if cfg.source_root else _standard_or_error(\n        input_root, STANDARD_SOURCE_RELATIVE, validate_source_root, _valid_source_candidates,\n        \"source dataset\", GOOGLE_ASL_URL,\n    )\n    target = validate_target_landmarks(Path(cfg.target_landmarks)) if cfg.target_landmarks else _standard_or_error(\n        input_root, STANDARD_TARGET_RELATIVE, validate_target_landmarks, _valid_target_candidates,\n        \"target landmark dataset\", TARGET_OUTPUT_URL,\n    )\n    return source, target\n\ndef validate_cuda_build() -> torch.device:\n    if not torch.cuda.is_available():\n        raise RuntimeError(\"A CUDA GPU is required. Select a Kaggle T4 accelerator and restart the session.\")\n    index = torch.cuda.current_device()\n    capability = torch.cuda.get_device_capability(index)\n    compiled = set(torch.cuda.get_arch_list())\n    arch = f\"sm_{capability[0]}{capability[1]}\"\n    if compiled and arch not in compiled:\n        raise RuntimeError(\n            f\"{torch.cuda.get_device_name(index)} requires {arch}, but this PyTorch build contains {sorted(compiled)}. \"\n            \"Use a T4-compatible Kaggle image.\"\n        )\n    print(f\"GPU: {torch.cuda.get_device_name(index)} | capability: {arch} | PyTorch: {torch.__version__}\")\n    return torch.device(\"cuda\", index)\n\nDEVICE = validate_cuda_build()\nSOURCE_ROOT, TARGET_LANDMARKS = resolve_inputs(CFG)\nprint(f\"Source dataset: {SOURCE_ROOT}\\nTarget landmarks: {TARGET_LANDMARKS}\\nOutputs: {WORK_DIR}\")\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Shared Preprocessing \ud83e\udde9\n\n`hand168-temporal` matches the `SegmentPreprocessor` contract used by the C++ runtime. Hand identity comes from source metadata, never horizontal image position.\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "def hand_present(hand: np.ndarray) -> np.ndarray:\n    return np.any(np.abs(hand) > 0, axis=(1, 2))\n\ndef trim_empty_edges(hands: np.ndarray) -> np.ndarray:\n    present = hand_present(hands[:, 0]) | hand_present(hands[:, 1])\n    indices = np.flatnonzero(present)\n    return hands[indices[0]:indices[-1] + 1] if len(indices) else hands[:0]\n\ndef resample_normalized(coords: np.ndarray, present: np.ndarray, length: int = 64) -> Tuple[np.ndarray, np.ndarray]:\n    if len(coords) == length:\n        return coords.copy(), present.copy()\n    output = np.zeros((length, 2, 21, 3), np.float32)\n    output_present = np.zeros((length, 2), bool)\n    for out_index in range(length):\n        source = out_index * (len(coords) - 1) / (length - 1)\n        lower = int(math.floor(source)); upper = min(lower + 1, len(coords) - 1)\n        weight = source - lower\n        nearest = min(int(math.floor(source + 0.5)), len(coords) - 1)\n        for hand in range(2):\n            output_present[out_index, hand] = present[nearest, hand]\n            if not output_present[out_index, hand]:\n                continue\n            left = coords[lower, hand] if present[lower, hand] else coords[nearest, hand]\n            right = coords[upper, hand] if present[upper, hand] else coords[nearest, hand]\n            output[out_index, hand] = left + (right - left) * weight\n    return output, output_present\n\ndef normalize_coordinates(hands: np.ndarray, eps: float = 1e-6) -> Tuple[np.ndarray, np.ndarray]:\n    coords = np.zeros_like(hands, dtype=np.float32)\n    present = np.zeros((len(hands), 2), dtype=bool)\n    for h in range(2):\n        present[:, h] = hand_present(hands[:, h])\n        relative = hands[:, h] - hands[:, h, :1]\n        scale = np.linalg.norm(relative, axis=-1).max(axis=1)\n        coords[present[:, h], h] = relative[present[:, h]] / (scale[present[:, h], None, None] + eps)\n    return coords, present\n\ndef preprocess_hands(\n    hands: np.ndarray, max_frames: int = 64, min_frames: int = 12,\n    padding_value: float = -100.0, max_input_frames: int = 120\n) -> Tuple[np.ndarray, int]:\n    hands = np.asarray(hands, dtype=np.float32)\n    if hands.ndim != 4 or hands.shape[1:] != (2, 21, 3):\n        raise ValueError(f\"Expected T x 2 x 21 x 3 hands, received {hands.shape}\")\n    if not np.isfinite(hands).all():\n        raise ValueError(\"Landmarks contain non-finite values\")\n    hands = trim_empty_edges(hands)\n    if len(hands) < min_frames:\n        raise ValueError(f\"Sequence has {len(hands)} valid action frames; minimum is {min_frames}\")\n    if len(hands) > max_input_frames:\n        raise ValueError(f\"Sequence has {len(hands)} valid action frames; maximum is {max_input_frames}\")\n    coords, present = normalize_coordinates(hands)\n    if len(hands) > max_frames:\n        coords, present = resample_normalized(coords, present, max_frames)\n    velocity = np.zeros_like(coords)\n    continuous = present[1:] & present[:-1]\n    delta = coords[1:] - coords[:-1]\n    for h in range(2):\n        velocity[1:, h][continuous[:, h]] = delta[:, h][continuous[:, h]]\n    features = np.concatenate([coords, np.linalg.norm(velocity, axis=-1, keepdims=True)], axis=-1)\n    valid_length = len(features)\n    padded = np.full((max_frames, 2, 21, 4), padding_value, dtype=np.float32)\n    padded[:valid_length] = features\n    return padded.reshape(max_frames, 168), valid_length\n\ndef pack_hand_rows(frame_values, hand_types, landmark_indices, coordinates) -> np.ndarray:\n    frame_values = np.asarray(frame_values)\n    hand_types = np.asarray(hand_types)\n    landmark_indices = np.asarray(landmark_indices, dtype=np.int64)\n    coordinates = np.asarray(coordinates, dtype=np.float32)\n    if not (len(frame_values) == len(hand_types) == len(landmark_indices) == len(coordinates)):\n        raise ValueError(\"Parquet hand columns have inconsistent lengths\")\n    if not len(frame_values):\n        return np.zeros((0, 2, 21, 3), np.float32)\n    _, frame_indices = np.unique(frame_values, return_inverse=True)\n    hand_indices = np.full(len(hand_types), -1, dtype=np.int8)\n    hand_indices[hand_types == \"left_hand\"] = 0\n    hand_indices[hand_types == \"right_hand\"] = 1\n    valid = (\n        (hand_indices >= 0) & (landmark_indices >= 0) & (landmark_indices < 21)\n        & np.isfinite(coordinates).all(axis=1)\n    )\n    output = np.zeros((int(frame_indices.max()) + 1, 2, 21, 3), np.float32)\n    output[frame_indices[valid], hand_indices[valid], landmark_indices[valid]] = coordinates[valid]\n    return output\n\ndef source_parquet_hands(path: Path) -> np.ndarray:\n    frame = pd.read_parquet(path, columns=[\"frame\", \"type\", \"landmark_index\", \"x\", \"y\", \"z\"])\n    frame = frame[frame[\"type\"].isin([\"left_hand\", \"right_hand\"])].drop_duplicates(\n        [\"frame\", \"type\", \"landmark_index\"], keep=\"last\"\n    )\n    return pack_hand_rows(\n        frame[\"frame\"].to_numpy(), frame[\"type\"].to_numpy(),\n        frame[\"landmark_index\"].to_numpy(), frame[[\"x\", \"y\", \"z\"]].to_numpy(dtype=np.float32),\n    )\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Source Dataset Preparation \ud83d\udcbe\n\nData is loaded from [Google - Isolated Sign Language Recognition](https://www.kaggle.com/competitions/asl-signs/data). Parquet sequences are vectorized with bounded I/O concurrency and written to a resumable float16 memory map in recoverable chunks. Cache progress and every rejected sample remain inspectable.\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "def append_jsonl(path: Path, row: Mapping[str, Any]) -> None:\n    path.parent.mkdir(parents=True, exist_ok=True)\n    with path.open(\"a\", encoding=\"utf-8\") as handle:\n        handle.write(json.dumps(dict(row), ensure_ascii=False, default=str) + \"\\n\")\n\ndef append_csv_row(path: Path, row: Mapping[str, Any]) -> None:\n    path.parent.mkdir(parents=True, exist_ok=True)\n    flat = {key: value for key, value in row.items() if isinstance(value, (str, int, float, bool)) or value is None}\n    write_header = not path.exists()\n    with path.open(\"a\", newline=\"\", encoding=\"utf-8\") as handle:\n        writer = csv.DictWriter(handle, fieldnames=list(flat))\n        if write_header: writer.writeheader()\n        writer.writerow(flat)\n\ndef atomic_json_write(path: Path, payload: Mapping[str, Any]) -> None:\n    temporary = path.with_suffix(path.suffix + \".tmp\")\n    temporary.write_text(json.dumps(dict(payload), indent=2, default=str), encoding=\"utf-8\")\n    os.replace(temporary, path)\n\ndef prepare_source_cache_item(path: Path, cfg: Config) -> Tuple[np.ndarray, int]:\n    return preprocess_hands(\n        source_parquet_hands(path), cfg.max_frames, cfg.min_frames,\n        cfg.padding_value, cfg.max_input_frames,\n    )\n\ndef build_source_cache(source_root: Path, cache_dir: Path, cfg: Config) -> Tuple[pd.DataFrame, Path, Path]:\n    if cfg.cache_workers <= 0 or cfg.cache_chunk_size <= 0 or cfg.cache_flush_chunks <= 0:\n        raise ValueError(\"cache_workers, cache_chunk_size, and cache_flush_chunks must be positive\")\n    cache_dir.mkdir(parents=True, exist_ok=True)\n    metadata = pd.read_csv(source_root / \"train.csv\")\n    required = {\"path\", \"sign\", \"participant_id\"}\n    if not required.issubset(metadata.columns):\n        raise ValueError(f\"train.csv must contain {sorted(required)}\")\n    relative_paths = metadata[\"path\"].astype(str).tolist()\n    data_path, lengths_path = cache_dir / \"features.f16\", cache_dir / \"lengths.i16\"\n    accepted_path, state_path = cache_dir / \"accepted.csv\", cache_dir / \"state.json\"\n    rejected_path = cache_dir / \"rejected.jsonl\"\n    shape = (len(metadata), cfg.max_frames, cfg.feature_dim)\n    resumable = data_path.exists() and lengths_path.exists() and state_path.exists()\n    mode = \"r+\" if resumable else \"w+\"\n    features = np.memmap(data_path, dtype=np.float16, mode=mode, shape=shape)\n    lengths = np.memmap(lengths_path, dtype=np.int16, mode=mode, shape=(len(metadata),))\n    start = 0\n    if mode == \"w+\":\n        lengths[:] = 0\n        features.flush(); lengths.flush()\n        rejected_path.unlink(missing_ok=True)\n    else:\n        state = json.loads(state_path.read_text())\n        if state.get(\"preprocessing\") != PREPROCESSING_CONTRACT or state.get(\"rows\") != len(metadata):\n            raise RuntimeError(\"Source cache metadata does not match this dataset or preprocessing contract\")\n        start = int(state.get(\"next_index\", 0))\n        if not 0 <= start <= len(metadata):\n            raise RuntimeError(f\"Invalid source cache next_index: {start}\")\n\n    worker_count = max(1, min(cfg.cache_workers, os.cpu_count() or 1))\n    progress = tqdm(total=len(metadata) - start, desc=\"Building temporal hand-feature cache\")\n    chunks_since_flush = 0\n    with ThreadPoolExecutor(max_workers=worker_count) as executor:\n        for chunk_start in range(start, len(metadata), cfg.cache_chunk_size):\n            chunk_end = min(len(metadata), chunk_start + cfg.cache_chunk_size)\n            futures = {\n                executor.submit(prepare_source_cache_item, source_root / relative_paths[index], cfg): index\n                for index in range(chunk_start, chunk_end)\n            }\n            chunk_rejections = []\n            for future in as_completed(futures):\n                index = futures[future]\n                try:\n                    sample, valid_length = future.result()\n                    features[index] = sample.astype(np.float16)\n                    lengths[index] = valid_length\n                except Exception as exc:\n                    lengths[index] = 0\n                    chunk_rejections.append(\n                        {\"index\": index, \"path\": relative_paths[index], \"reason\": type(exc).__name__, \"detail\": str(exc)}\n                    )\n                progress.update(1)\n            if chunk_rejections:\n                with rejected_path.open(\"a\", encoding=\"utf-8\") as handle:\n                    for rejection in sorted(chunk_rejections, key=lambda item: item[\"index\"]):\n                        handle.write(json.dumps(rejection, ensure_ascii=False, default=str) + \"\\n\")\n            chunks_since_flush += 1\n            should_flush = chunks_since_flush >= cfg.cache_flush_chunks or chunk_end == len(metadata)\n            if should_flush:\n                features.flush(); lengths.flush()\n                accepted_count = int(np.count_nonzero(lengths[:chunk_end]))\n                atomic_json_write(state_path, {\n                    \"preprocessing\": PREPROCESSING_CONTRACT,\n                    \"rows\": len(metadata),\n                    \"next_index\": chunk_end,\n                    \"accepted\": accepted_count,\n                    \"rejected\": chunk_end - accepted_count,\n                    \"complete\": chunk_end == len(metadata),\n                    \"cache_workers\": worker_count,\n                    \"cache_chunk_size\": cfg.cache_chunk_size,\n                })\n                chunks_since_flush = 0\n    progress.close()\n\n    accepted_indices = np.flatnonzero(np.asarray(lengths) > 0)\n    accepted_rows = metadata.iloc[accepted_indices][[\"sign\", \"participant_id\", \"path\"]].copy()\n    accepted_rows.insert(0, \"cache_index\", accepted_indices)\n    temporary_accepted = accepted_path.with_suffix(accepted_path.suffix + \".tmp\")\n    accepted_rows.to_csv(temporary_accepted, index=False)\n    os.replace(temporary_accepted, accepted_path)\n    return accepted_rows.reset_index(drop=True), data_path, lengths_path\n\nclass MemmapDataset(Dataset):\n    def __init__(self, rows: pd.DataFrame, data_path: Path, lengths_path: Path, total_rows: int, label_map: Mapping[str, int], cfg: Config):\n        self.rows = rows.reset_index(drop=True)\n        self.data = np.memmap(data_path, np.float16, \"r\", shape=(total_rows, cfg.max_frames, cfg.feature_dim))\n        self.lengths = np.memmap(lengths_path, np.int16, \"r\", shape=(total_rows,))\n        self.label_map, self.cfg = dict(label_map), cfg\n    def __len__(self): return len(self.rows)\n    def __getitem__(self, index):\n        row = self.rows.iloc[index]\n        cache_index = int(row.cache_index)\n        return torch.from_numpy(np.array(self.data[cache_index], dtype=np.float32)), int(self.lengths[cache_index]), self.label_map[str(row.sign)]\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Target Dataset Audit \ud83d\udee1\ufe0f\n\nLandmark records come from [ASL-preprocessing 7 output](https://www.kaggle.com/code/abdelrhmankaram/asl-preprocessing-7/output). Object arrays are deserialized with an explicit global allow-list. Invalid files are reported instead of silently entering training.\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "class RestrictedNumpyUnpickler(pickle.Unpickler):\n    ALLOWED = {\n        (\"numpy\", \"ndarray\"), (\"numpy\", \"dtype\"),\n        (\"numpy.core.multiarray\", \"_reconstruct\"), (\"numpy._core.multiarray\", \"_reconstruct\"),\n        (\"builtins\", \"set\"), (\"builtins\", \"slice\"),\n    }\n    def find_class(self, module, name):\n        if (module, name) not in self.ALLOWED:\n            raise pickle.UnpicklingError(f\"Blocked pickle global: {module}.{name}\")\n        return super().find_class(module, name)\n\ndef restricted_load_npy(path: Path) -> Any:\n    if path.stat().st_size == 0:\n        raise ValueError(\"Empty NPY file\")\n    with path.open(\"rb\") as handle:\n        version = np.lib.format.read_magic(handle)\n        if version == (1, 0):\n            shape, fortran, dtype = np.lib.format.read_array_header_1_0(handle)\n        elif version in {(2, 0), (3, 0)}:\n            shape, fortran, dtype = np.lib.format.read_array_header_2_0(handle)\n        else:\n            raise ValueError(f\"Unsupported NPY version: {version}\")\n        if not dtype.hasobject:\n            handle.seek(0)\n            return np.load(handle, allow_pickle=False)\n        value = RestrictedNumpyUnpickler(handle).load()\n        if isinstance(value, np.ndarray) and value.shape != shape:\n            raise ValueError(f\"Object array header shape {shape} does not match payload {value.shape}\")\n        return value\n\ndef unpack_target_object(value: Any, cfg: Config) -> Tuple[str, np.ndarray]:\n    if isinstance(value, np.ndarray) and value.dtype == object:\n        value = value.tolist()\n    if isinstance(value, dict):\n        label = value.get(\"label\", value.get(\"sign\"))\n        landmarks = value.get(\"landmark\", value.get(\"landmarks\"))\n    elif isinstance(value, (list, tuple)) and len(value) >= 2:\n        label, landmarks = value[0], value[1]\n    else:\n        raise ValueError(\"Expected an object containing label and landmarks\")\n    label = str(label).strip()\n    landmarks = np.asarray(landmarks, dtype=np.float32)\n    if not label: raise ValueError(\"Label is empty\")\n    if landmarks.ndim != 3 or landmarks.shape[1:] != (100, 3):\n        raise ValueError(f\"Expected T x 100 x 3 landmarks, received {landmarks.shape}\")\n    if not np.isfinite(landmarks).all(): raise ValueError(\"Landmarks contain non-finite values\")\n    rs, re = cfg.target_right_slice; ls, le = cfg.target_left_slice\n    if re - rs != 21 or le - ls != 21: raise ValueError(\"Each configured target hand slice must contain 21 landmarks\")\n    hands = np.stack([landmarks[:, ls:le], landmarks[:, rs:re]], axis=1)  # canonical left, right\n    return label, hands\n\ndef audit_target_dataset(root: Path, output_dir: Path, cfg: Config) -> Tuple[np.ndarray, np.ndarray, List[str], List[str]]:\n    samples, lengths, labels, paths = [], [], [], []\n    rejected = output_dir / \"target_rejected.jsonl\"\n    for path in tqdm(sorted(root.rglob(\"*.npy\")), desc=\"Auditing target NPY\"):\n        try:\n            label, hands = unpack_target_object(restricted_load_npy(path), cfg)\n            sample, length = preprocess_hands(hands, cfg.max_frames, cfg.min_frames, cfg.padding_value, cfg.max_input_frames)\n            samples.append(sample); lengths.append(length); labels.append(label); paths.append(str(path))\n        except Exception as exc:\n            append_jsonl(rejected, {\"path\": str(path), \"reason\": type(exc).__name__, \"detail\": str(exc)})\n    if not samples: raise RuntimeError(f\"No valid target samples were found under {root}\")\n    return np.stack(samples), np.asarray(lengths, np.int64), labels, paths\n\nclass ArrayDataset(Dataset):\n    def __init__(self, samples, lengths, labels, indices, label_map):\n        self.samples, self.lengths, self.labels = samples, lengths, labels\n        self.indices, self.label_map = np.asarray(indices), dict(label_map)\n    def __len__(self): return len(self.indices)\n    def __getitem__(self, index):\n        i = int(self.indices[index])\n        return torch.from_numpy(np.asarray(self.samples[i], np.float32)), int(self.lengths[i]), self.label_map[self.labels[i]]\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Model and Objectives \ud83e\udde0\n\n+The encoder combines a shared per-hand temporal branch with masked Transformer fusion. Classification is used only as an auxiliary training signal.\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "def split_labels(labels: Sequence[str], seed: int, ratios=(0.8, 0.1, 0.1)) -> Dict[str, List[str]]:\n    unique = sorted(set(map(str, labels)))\n    if len(unique) < 3: raise ValueError(\"At least three distinct labels are required for class-disjoint splits\")\n    rng = random.Random(seed); rng.shuffle(unique)\n    n_val = max(1, round(len(unique) * ratios[1])); n_test = max(1, round(len(unique) * ratios[2]))\n    n_train = len(unique) - n_val - n_test\n    if n_train < 1: n_train, n_val = 1, len(unique) - n_test - 1\n    result = {\"train\": unique[:n_train], \"validation\": unique[n_train:n_train+n_val], \"test\": unique[n_train+n_val:]}\n    if set(result[\"train\"]) & set(result[\"validation\"]) or set(result[\"train\"]) & set(result[\"test\"]) or set(result[\"validation\"]) & set(result[\"test\"]):\n        raise AssertionError(\"Class leakage detected\")\n    return result\n\nclass PKBatchSampler(Sampler[List[int]]):\n    def __init__(self, labels: Sequence[int], p: int, k: int, batches: int, seed: int):\n        self.p, self.k, self.batches, self.seed, self.epoch = p, k, batches, seed, 0\n        self.groups: Dict[int, List[int]] = {}\n        for index, label in enumerate(labels): self.groups.setdefault(int(label), []).append(index)\n        if len(self.groups) < p: raise ValueError(f\"PK sampling needs {p} classes, found {len(self.groups)}\")\n    def set_epoch(self, epoch): self.epoch = int(epoch)\n    def __len__(self): return self.batches\n    def __iter__(self):\n        rng = random.Random(self.seed + self.epoch)\n        classes = list(self.groups)\n        for _ in range(self.batches):\n            batch = []\n            for label in rng.sample(classes, self.p):\n                members = self.groups[label]\n                batch.extend(rng.sample(members, self.k) if len(members) >= self.k else rng.choices(members, k=self.k))\n            yield batch\n\ndef pk_batches(labels, p, k, batches, seed):\n    return PKBatchSampler(labels, p, k, batches, seed)\n\nclass TemporalBlock(nn.Module):\n    def __init__(self, channels: int, dilation: int):\n        super().__init__()\n        self.depthwise = nn.Conv1d(channels, channels, 3, padding=dilation, dilation=dilation, groups=channels)\n        self.pointwise = nn.Conv1d(channels, channels, 1)\n        self.norm = nn.BatchNorm1d(channels)\n    def forward(self, x): return F.gelu(self.norm(self.pointwise(self.depthwise(x)))) + x\n\nclass HandEncoder(nn.Module):\n    def __init__(self, feature_dim=168, hand_dim=96, fusion_dim=192, embedding_dim=128, max_frames=64):\n        super().__init__()\n        if feature_dim != 168: raise ValueError(\"hand168-temporal requires feature_dim=168\")\n        self.model_config = dict(feature_dim=feature_dim, hand_dim=hand_dim, fusion_dim=fusion_dim, embedding_dim=embedding_dim, max_frames=max_frames)\n        self.input_projection = nn.Linear(84, hand_dim)\n        self.tcn = nn.Sequential(TemporalBlock(hand_dim, 1), TemporalBlock(hand_dim, 2), TemporalBlock(hand_dim, 4))\n        self.fusion = nn.Linear(hand_dim * 2, fusion_dim)\n        self.position = nn.Parameter(torch.zeros(1, max_frames, fusion_dim))\n        layer = nn.TransformerEncoderLayer(fusion_dim, 4, fusion_dim * 4, 0.1, batch_first=True, norm_first=True, activation=\"gelu\")\n        self.transformer = nn.TransformerEncoder(layer, 2, enable_nested_tensor=False)\n        self.output = nn.Linear(fusion_dim, embedding_dim)\n        nn.init.trunc_normal_(self.position, std=0.02)\n    def forward(self, x, lengths):\n        if x.ndim != 3 or x.shape[-1] != 168: raise ValueError(f\"Expected B x T x 168, received {tuple(x.shape)}\")\n        steps = x.shape[1]; mask = torch.arange(steps, device=x.device)[None] >= lengths[:, None]\n        x = x.masked_fill(mask[..., None], 0.0).reshape(x.shape[0], steps, 2, 84)\n        streams = []\n        for hand in range(2):\n            stream = self.input_projection(x[:, :, hand]).transpose(1, 2)\n            streams.append(self.tcn(stream).transpose(1, 2))\n        fused = self.fusion(torch.cat(streams, dim=-1)) + self.position[:, :steps]\n        fused = self.transformer(fused, src_key_padding_mask=mask)\n        output = F.normalize(self.output(fused), dim=-1)\n        return output.masked_fill(mask[..., None], 0.0)\n\nclass TrainingModel(nn.Module):\n    def __init__(self, encoder: HandEncoder, classes: int):\n        super().__init__(); self.encoder, self.classifier = encoder, nn.Linear(encoder.model_config[\"embedding_dim\"], classes)\n    def forward(self, x, lengths):\n        frames = self.encoder(x, lengths)\n        mask = torch.arange(frames.shape[1], device=frames.device)[None] < lengths[:, None]\n        pooled = (frames * mask[..., None]).sum(1) / lengths.clamp_min(1)[:, None]\n        pooled = F.normalize(pooled, dim=-1)\n        return frames, pooled, self.classifier(pooled)\n\ndef supervised_contrastive_loss(features, labels, temperature=0.1):\n    features = F.normalize(features, dim=-1); logits = features @ features.T / temperature\n    eye = torch.eye(len(labels), dtype=torch.bool, device=labels.device)\n    positives = labels[:, None].eq(labels[None]) & ~eye\n    logits = logits.masked_fill(eye, -torch.inf)\n    log_prob = logits - torch.logsumexp(logits, dim=1, keepdim=True)\n    valid = positives.any(1)\n    return -(log_prob.masked_fill(~positives, 0).sum(1)[valid] / positives.sum(1)[valid]).mean() if valid.any() else features.sum() * 0\n\ndef batch_hard_triplet_loss(features, labels, margin=0.2):\n    distance = 1.0 - F.normalize(features, dim=-1) @ F.normalize(features, dim=-1).T\n    eye = torch.eye(len(labels), dtype=torch.bool, device=labels.device)\n    pos = labels[:, None].eq(labels[None]) & ~eye; neg = ~labels[:, None].eq(labels[None])\n    valid = pos.any(1) & neg.any(1)\n    hardest_pos = distance.masked_fill(~pos, -torch.inf).max(1).values\n    hardest_neg = distance.masked_fill(~neg, torch.inf).min(1).values\n    return F.relu(hardest_pos[valid] - hardest_neg[valid] + margin).mean() if valid.any() else features.sum() * 0\n\ndef augment_features(x, lengths):\n    x = x.clone(); batch = x.shape[0]\n    for i in range(batch):\n        length = int(lengths[i]); view = x[i, :length].reshape(length, 2, 21, 4)\n        if torch.rand(()) < 0.5: view[..., 0].mul_(-1)\n        if torch.rand(()) < 0.5:\n            view[..., :3].mul_(torch.empty((), device=x.device).uniform_(0.95, 1.05))\n        if length > 4 and torch.rand(()) < 0.3:\n            start = int(torch.randint(0, length - 1, ()).item()); view[start:min(length, start + max(1, length // 10))].zero_()\n        if torch.rand(()) < 0.3:\n            hand = int(torch.randint(0, 2, ()).item()); landmark = int(torch.randint(1, 21, ()).item()); view[:, hand, landmark].zero_()\n    return x\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Representation Training \ud83c\udfcb\ufe0f\n\n+Classes are split before sampling. Unseen-class validation selects the checkpoint using prototype-style retrieval, not the temporary classifier.\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "@torch.inference_mode()\ndef encode_loader(model, loader, device):\n    model.eval(); vectors, labels, sequences, lengths_out = [], [], [], []\n    for x, lengths, y in loader:\n        x, lengths = x.to(device), lengths.to(device)\n        frames, pooled, _ = model(x, lengths)\n        vectors.append(pooled.cpu().numpy()); labels.extend(y.numpy().tolist())\n        sequences.extend([frames[i, :int(lengths[i])].cpu().numpy() for i in range(len(x))]); lengths_out.extend(lengths.cpu().tolist())\n    return np.concatenate(vectors), np.asarray(labels), sequences, np.asarray(lengths_out)\n\ndef cosine_distance(a, b): return 1.0 - np.clip(a @ b.T, -1.0, 1.0)\n\ndef make_reference_query(labels, seed, references=1, max_queries=4, groups=None):\n    rng = random.Random(seed); ref, query = [], []\n    for label in sorted(set(labels.tolist())):\n        indices = np.flatnonzero(labels == label).tolist(); rng.shuffle(indices)\n        selected = indices[:references]\n        if groups is None:\n            candidates = indices[references:]\n        else:\n            reference_groups = {str(groups[i]) for i in selected}\n            candidates = [i for i in indices if i not in selected and str(groups[i]) not in reference_groups]\n        if selected and candidates:\n            ref.extend(selected); query.extend(candidates[:max_queries])\n    return np.asarray(ref), np.asarray(query)\n\ndef pooled_retrieval(vectors, labels, seed=0, references=1, max_queries=4, groups=None):\n    ref, query = make_reference_query(labels, seed, references, max_queries, groups)\n    if not len(query): return {\"recall_at_1\": 0.0, \"queries\": 0, \"method\": \"pooled_cosine\"}\n    distances = cosine_distance(vectors[query], vectors[ref])\n    predictions = labels[ref][distances.argmin(1)]\n    return {\"recall_at_1\": float(np.mean(predictions == labels[query])), \"queries\": int(len(query)), \"method\": \"pooled_cosine\"}\n\ndef constrained_dtw(a: np.ndarray, b: np.ndarray, window: int = 12) -> float:\n    n, m = len(a), len(b); window = max(window, abs(n - m))\n    costs = np.full((n + 1, m + 1), np.inf, np.float64); steps = np.zeros((n + 1, m + 1), np.int32); costs[0, 0] = 0\n    for i in range(1, n + 1):\n        for j in range(max(1, i - window), min(m, i + window) + 1):\n            options = [(costs[i-1, j], steps[i-1, j]), (costs[i, j-1], steps[i, j-1]), (costs[i-1, j-1], steps[i-1, j-1])]\n            best_cost, best_steps = min(options, key=lambda item: item[0])\n            costs[i, j] = best_cost + float(np.linalg.norm(a[i-1] - b[j-1])); steps[i, j] = best_steps + 1\n    return float(costs[n, m] / max(1, steps[n, m]))\n\ndef topk_dtw_retrieval(vectors, sequences, labels, seed, top_k=20, window=12, max_queries=2, groups=None):\n    ref, query = make_reference_query(labels, seed, 1, max_queries, groups)\n    correct = 0\n    for qi in query:\n        pooled = cosine_distance(vectors[qi:qi+1], vectors[ref])[0]\n        shortlist = ref[np.argsort(pooled)[:min(top_k, len(ref))]]\n        prediction = labels[min(shortlist, key=lambda ri: constrained_dtw(sequences[qi], sequences[ri], window))]\n        correct += int(prediction == labels[qi])\n    return {\"recall_at_1\": correct / len(query) if len(query) else 0.0, \"queries\": int(len(query)), \"method\": \"topk_dtw\"}\n\ndef atomic_torch_save(payload, path: Path):\n    path.parent.mkdir(parents=True, exist_ok=True); temporary = path.with_suffix(path.suffix + \".tmp\")\n    torch.save(payload, temporary); os.replace(temporary, path)\n\ndef split_identity(split):\n    return hashlib.sha256(json.dumps(split, sort_keys=True).encode()).hexdigest()\n\ndef train_recipe(name, model, train_dataset, train_labels, validation_loader, settings, split, output_dir, device, cfg, validation_groups=None):\n    output_dir.mkdir(parents=True, exist_ok=True)\n    sampler = PKBatchSampler(train_labels, settings.classes_per_batch, settings.samples_per_class, settings.batches_per_epoch, cfg.seed)\n    loader = DataLoader(train_dataset, batch_sampler=sampler, num_workers=cfg.num_workers, pin_memory=True)\n    model.to(device)\n    optimizer = torch.optim.AdamW(model.parameters(), lr=settings.learning_rate, weight_decay=cfg.weight_decay)\n    scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, settings.epochs)\n    scaler = torch.amp.GradScaler(\"cuda\", enabled=cfg.use_amp)\n    latest = output_dir / \"latest.pt\"; start_epoch, best = 0, -1.0\n    expected_split = split_identity(split)\n    if cfg.resume and latest.exists():\n        state = torch.load(latest, map_location=\"cpu\", weights_only=False)\n        if state[\"schema\"] != CHECKPOINT_SCHEMA or state[\"preprocessing\"] != PREPROCESSING_CONTRACT or state[\"split_identity\"] != expected_split:\n            raise RuntimeError(f\"Refusing incompatible resume checkpoint: {latest}\")\n        model.load_state_dict(state[\"model\"]); optimizer.load_state_dict(state[\"optimizer\"]); scheduler.load_state_dict(state[\"scheduler\"]); scaler.load_state_dict(state[\"scaler\"])\n        start_epoch, best = state[\"epoch\"] + 1, state[\"best_recall\"]\n    if cfg.use_data_parallel:\n        if torch.cuda.device_count() < 2: raise RuntimeError(\"use_data_parallel=True requires at least two visible CUDA devices\")\n        model = nn.DataParallel(model)\n    core_model = model.module if isinstance(model, nn.DataParallel) else model\n    for epoch in range(start_epoch, settings.epochs):\n        sampler.set_epoch(epoch); model.train()\n        frozen = epoch < settings.freeze_epochs\n        for parameter in list(core_model.encoder.input_projection.parameters()) + list(core_model.encoder.tcn[0].parameters()): parameter.requires_grad_(not frozen)\n        running = 0.0\n        progress = tqdm(loader, desc=f\"{name} {epoch+1}/{settings.epochs}\", leave=False)\n        for batch_index, (x, lengths, labels) in enumerate(progress):\n            x, lengths, labels = x.to(device, non_blocking=True), lengths.to(device), labels.to(device)\n            x = augment_features(x, lengths); optimizer.zero_grad(set_to_none=True)\n            with torch.autocast(device_type=\"cuda\", dtype=torch.float16, enabled=cfg.use_amp):\n                _, pooled, logits = model(x, lengths)\n                ce = F.cross_entropy(logits, labels); supcon = supervised_contrastive_loss(pooled, labels); triplet = batch_hard_triplet_loss(pooled, labels)\n                loss = ce + 0.25 * supcon + 0.5 * triplet\n            scaler.scale(loss).backward(); scaler.step(optimizer); scaler.update(); running += loss.detach().item()\n            progress.set_postfix(loss=f\"{running/(batch_index+1):.4f}\")\n            if (batch_index + 1) % 25 == 0:\n                (output_dir / \"status.json\").write_text(json.dumps({\"epoch\": epoch, \"last_batch\": batch_index, \"cuda_bytes\": torch.cuda.memory_allocated()}, indent=2))\n        scheduler.step()\n        vectors, labels_np, sequences, _ = encode_loader(model, validation_loader, device)\n        use_dtw = settings.dtw_every and (epoch + 1) % settings.dtw_every == 0\n        metrics = topk_dtw_retrieval(vectors, sequences, labels_np, cfg.seed, cfg.top_k, cfg.dtw_window, groups=validation_groups) if use_dtw else pooled_retrieval(vectors, labels_np, cfg.seed, cfg.source_reference_per_class, cfg.source_query_per_class, validation_groups)\n        recall = metrics[\"recall_at_1\"]; best = max(best, recall)\n        payload = {\"schema\": CHECKPOINT_SCHEMA, \"preprocessing\": PREPROCESSING_CONTRACT, \"model_config\": core_model.encoder.model_config, \"model\": core_model.state_dict(), \"encoder\": core_model.encoder.state_dict(), \"optimizer\": optimizer.state_dict(), \"scheduler\": scheduler.state_dict(), \"scaler\": scaler.state_dict(), \"epoch\": epoch, \"best_recall\": best, \"metrics\": metrics, \"split_identity\": expected_split, \"seed\": cfg.seed}\n        epoch_path = output_dir / f\"epoch_{epoch+1:03d}.pt\"; atomic_torch_save(payload, epoch_path); atomic_torch_save(payload, latest)\n        if recall >= best: atomic_torch_save(payload, output_dir / \"best.pt\")\n        metric_row = {\"epoch\": epoch + 1, \"loss\": running / len(loader), **metrics}\n        append_jsonl(output_dir / \"metrics.jsonl\", metric_row); append_csv_row(output_dir / \"metrics.csv\", metric_row)\n        log_line = f\"{name}: epoch={epoch+1} loss={running/len(loader):.4f} recall@1={recall:.4f}\"\n        with (output_dir / \"train.log\").open(\"a\", encoding=\"utf-8\") as handle: handle.write(log_line + \"\\n\")\n        print(log_line)\n        gc.collect(); torch.cuda.empty_cache()\n    if not (output_dir / \"best.pt\").exists(): raise RuntimeError(f\"No best checkpoint was produced for {name}\")\n    return output_dir / \"best.pt\"\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "# Prepare the source corpus and deterministic class-disjoint splits.\nSOURCE_OUT = WORK_DIR / \"representation_training\"\nsource_rows, source_data_path, source_lengths_path = build_source_cache(SOURCE_ROOT, WORK_DIR / \"source_cache\", CFG)\nsource_split = split_labels(source_rows.sign.tolist(), CFG.seed)\n(SOURCE_OUT / \"splits.json\").parent.mkdir(parents=True, exist_ok=True)\n(SOURCE_OUT / \"splits.json\").write_text(json.dumps(source_split, indent=2))\n\nsource_train = source_rows[source_rows.sign.isin(source_split[\"train\"])].reset_index(drop=True)\nsource_val = source_rows[source_rows.sign.isin(source_split[\"validation\"])].reset_index(drop=True)\nsource_train_map = {label: i for i, label in enumerate(source_split[\"train\"])}\nsource_val_map = {label: i for i, label in enumerate(source_split[\"validation\"])}\ntotal_source_rows = len(pd.read_csv(SOURCE_ROOT / \"train.csv\"))\nsource_train_ds = MemmapDataset(source_train, source_data_path, source_lengths_path, total_source_rows, source_train_map, CFG)\nsource_val_ds = MemmapDataset(source_val, source_data_path, source_lengths_path, total_source_rows, source_val_map, CFG)\nsource_val_loader = DataLoader(source_val_ds, batch_size=128, shuffle=False, num_workers=0)\nsource_train_labels = [source_train_map[str(label)] for label in source_train.sign]\n\nencoder = HandEncoder(embedding_dim=CFG.embedding_dim, max_frames=CFG.max_frames)\nsource_model = TrainingModel(encoder, len(source_train_map))\nsource_best = train_recipe(\n    \"representation\", source_model, source_train_ds, source_train_labels, source_val_loader,\n    CFG.source, source_split, SOURCE_OUT, DEVICE, CFG, source_val.participant_id.astype(str).tolist()\n)\n\n# Evaluate the selected source encoder once with participant-disjoint references and queries.\nsource_test = source_rows[source_rows.sign.isin(source_split[\"test\"])].reset_index(drop=True)\nsource_test_map = {label: i for i, label in enumerate(source_split[\"test\"])}\nsource_test_ds = MemmapDataset(source_test, source_data_path, source_lengths_path, total_source_rows, source_test_map, CFG)\nsource_test_loader = DataLoader(source_test_ds, batch_size=128, shuffle=False, num_workers=0)\nsource_eval_state = torch.load(source_best, map_location=\"cpu\", weights_only=False)\nsource_model.load_state_dict(source_eval_state[\"model\"])\nsource_vectors, source_test_labels, source_sequences, _ = encode_loader(source_model.to(DEVICE), source_test_loader, DEVICE)\nsource_test_metrics = topk_dtw_retrieval(\n    source_vectors, source_sequences, source_test_labels, CFG.seed + 1, CFG.top_k, CFG.dtw_window,\n    CFG.final_query_per_class, source_test.participant_id.astype(str).tolist()\n)\n(SOURCE_OUT / \"test_metrics.json\").write_text(json.dumps(source_test_metrics, indent=2))\nprint(\"Source test:\", json.dumps(source_test_metrics))\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Domain Adaptation \ud83d\udd27\n\n+Only the best encoder is transferred. A new temporary classifier is created for target training labels; early encoder layers are briefly frozen, then fine-tuned.\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "TARGET_OUT = WORK_DIR / \"domain_adaptation\"\ntarget_samples, target_lengths, target_labels, target_paths = audit_target_dataset(TARGET_LANDMARKS, TARGET_OUT, CFG)\ntarget_split = split_labels(target_labels, CFG.seed + 1)\nTARGET_OUT.mkdir(parents=True, exist_ok=True)\n(TARGET_OUT / \"splits.json\").write_text(json.dumps(target_split, indent=2))\n\nindices_by_split = {name: [i for i, label in enumerate(target_labels) if label in labels] for name, labels in target_split.items()}\ntarget_train_map = {label: i for i, label in enumerate(target_split[\"train\"])}\ntarget_val_map = {label: i for i, label in enumerate(target_split[\"validation\"])}\ntarget_train_ds = ArrayDataset(target_samples, target_lengths, target_labels, indices_by_split[\"train\"], target_train_map)\ntarget_val_ds = ArrayDataset(target_samples, target_lengths, target_labels, indices_by_split[\"validation\"], target_val_map)\ntarget_val_loader = DataLoader(target_val_ds, batch_size=128, shuffle=False, num_workers=0)\ntarget_train_numeric = [target_train_map[target_labels[i]] for i in indices_by_split[\"train\"]]\n\nsource_state = torch.load(source_best, map_location=\"cpu\", weights_only=False)\nif source_state[\"schema\"] != CHECKPOINT_SCHEMA or source_state[\"preprocessing\"] != PREPROCESSING_CONTRACT:\n    raise RuntimeError(\"The source encoder checkpoint is incompatible with this notebook\")\nadapted_encoder = HandEncoder(**source_state[\"model_config\"])\nadapted_encoder.load_state_dict(source_state[\"encoder\"])\nadaptation_model = TrainingModel(adapted_encoder, len(target_train_map))\nadaptation_best = train_recipe(\"adaptation\", adaptation_model, target_train_ds, target_train_numeric, target_val_loader, CFG.adaptation, target_split, TARGET_OUT, DEVICE, CFG)\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Retrieval Evaluation \ud83d\udcca\n\nFinal evaluation first shortlists prototypes with pooled cosine distance and then reranks them with constrained frame-level DTW.\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "final_state = torch.load(adaptation_best, map_location=\"cpu\", weights_only=False)\nfinal_encoder = HandEncoder(**final_state[\"model_config\"])\nfinal_encoder.load_state_dict(final_state[\"encoder\"])\ntest_map = {label: i for i, label in enumerate(target_split[\"test\"])}\ntarget_test_ds = ArrayDataset(target_samples, target_lengths, target_labels, indices_by_split[\"test\"], test_map)\ntarget_test_loader = DataLoader(target_test_ds, batch_size=128, shuffle=False, num_workers=0)\nevaluation_model = TrainingModel(final_encoder, max(1, len(test_map))).to(DEVICE)\ntest_vectors, test_numeric, test_sequences, _ = encode_loader(evaluation_model, target_test_loader, DEVICE)\nfinal_metrics = topk_dtw_retrieval(test_vectors, test_sequences, test_numeric, CFG.seed + 2, CFG.top_k, CFG.dtw_window, CFG.final_query_per_class)\n(WORK_DIR / \"final_evaluation.json\").write_text(json.dumps(final_metrics, indent=2))\nprint(json.dumps(final_metrics, indent=2))\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Training Curves \ud83d\udcc8\n\nTraining metrics are rendered from the persisted CSV files. Figures are shown inline and saved as reproducible PNG artifacts.\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "def generate_training_charts(\n    work_dir: Path,\n    source_test: Optional[Mapping[str, Any]] = None,\n    target_test: Optional[Mapping[str, Any]] = None,\n) -> List[Path]:\n    import matplotlib.pyplot as plt\n\n    figures_dir = work_dir / \"figures\"\n    figures_dir.mkdir(parents=True, exist_ok=True)\n    runs = [\n        (\"Representation Training\", work_dir / \"representation_training\" / \"metrics.csv\", \"#2563eb\"),\n        (\"Domain Adaptation\", work_dir / \"domain_adaptation\" / \"metrics.csv\", \"#ea580c\"),\n    ]\n    available = []\n    for label, path, color in runs:\n        if not path.exists():\n            print(f\"Chart input is unavailable: {path}\")\n            continue\n        frame = pd.read_csv(path)\n        required = {\"epoch\", \"loss\", \"recall_at_1\"}\n        if frame.empty or not required.issubset(frame.columns):\n            print(f\"Chart input has no usable metrics: {path}\")\n            continue\n        frame = frame.copy()\n        frame[\"training_run\"] = label\n        available.append((label, frame, color))\n    if not available:\n        print(\"No training metrics are available; chart generation was skipped\")\n        return []\n\n    plot_data = pd.concat([frame for _, frame, _ in available], ignore_index=True)\n    plot_data.to_csv(figures_dir / \"plot_data.csv\", index=False)\n    outputs = [figures_dir / \"plot_data.csv\"]\n\n    style_name = \"seaborn-v0_8-whitegrid\" if \"seaborn-v0_8-whitegrid\" in plt.style.available else \"default\"\n    with plt.style.context(style_name):\n        figure, axes = plt.subplots(2, 2, figsize=(13, 8), constrained_layout=True)\n        for row, (expected_label, _, default_color) in enumerate(runs):\n            match = next(((label, frame, color) for label, frame, color in available if label == expected_label), None)\n            if match is None:\n                axes[row, 0].set_visible(False); axes[row, 1].set_visible(False)\n                continue\n            label, frame, color = match\n            axes[row, 0].plot(frame[\"epoch\"], frame[\"loss\"], color=color, linewidth=2)\n            axes[row, 0].set(title=f\"{label} \u2014 Loss\", xlabel=\"Epoch\", ylabel=\"Total loss\")\n            axes[row, 1].plot(frame[\"epoch\"], frame[\"recall_at_1\"], color=color, linewidth=2, marker=\"o\", markersize=3)\n            axes[row, 1].set(title=f\"{label} \u2014 Validation Recall@1\", xlabel=\"Epoch\", ylabel=\"Recall@1\", ylim=(0, 1.02))\n        curves_path = figures_dir / \"training_curves.png\"\n        figure.suptitle(\"Training Progress\", fontsize=15, fontweight=\"bold\")\n        figure.savefig(curves_path, dpi=180, bbox_inches=\"tight\")\n        plt.show(); plt.close(figure); outputs.append(curves_path)\n\n        evaluation = []\n        for label, metrics, color in (\n            (\"Source Retrieval\", source_test, \"#2563eb\"),\n            (\"Target Retrieval\", target_test, \"#ea580c\"),\n        ):\n            if metrics is not None and \"recall_at_1\" in metrics:\n                evaluation.append((label, float(metrics[\"recall_at_1\"]), color))\n        if evaluation:\n            figure, axis = plt.subplots(figsize=(8, 4.8), constrained_layout=True)\n            bars = axis.bar([item[0] for item in evaluation], [item[1] for item in evaluation], color=[item[2] for item in evaluation], width=0.58)\n            axis.set(title=\"Final Retrieval Evaluation\", ylabel=\"Recall@1\", ylim=(0, 1.08))\n            axis.bar_label(bars, labels=[f\"{item[1]:.3f}\" for item in evaluation], padding=4)\n            summary_path = figures_dir / \"retrieval_summary.png\"\n            figure.savefig(summary_path, dpi=180, bbox_inches=\"tight\")\n            plt.show(); plt.close(figure); outputs.append(summary_path)\n        else:\n            print(\"Final retrieval metrics are unavailable; the summary chart was skipped\")\n    return outputs\n\ntry:\n    chart_outputs = generate_training_charts(WORK_DIR, source_test_metrics, final_metrics)\n    for chart_path in chart_outputs: print(f\"Chart artifact: {chart_path}\")\nexcept Exception as exc:\n    chart_outputs = []\n    print(f\"Chart generation was skipped without interrupting export: {type(exc).__name__}: {exc}\")\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Export \ud83d\udce6\n\nThe standalone encoder includes its architecture, preprocessing contract, and SHA-256 fingerprint. Dynamic prototypes must match both the fingerprint and preprocessing contract.\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "def encoder_fingerprint(state_dict: Mapping[str, torch.Tensor]) -> str:\n    digest = hashlib.sha256()\n    for name in sorted(state_dict):\n        tensor = state_dict[name].detach().cpu().contiguous()\n        digest.update(name.encode()); digest.update(str(tensor.dtype).encode()); digest.update(np.asarray(tensor.shape, np.int64).tobytes()); digest.update(tensor.numpy().tobytes())\n    return digest.hexdigest()\n\nencoder_state = final_state[\"encoder\"]\nfingerprint = encoder_fingerprint(encoder_state)\nexport = {\n    \"schema\": CHECKPOINT_SCHEMA,\n    \"preprocessing\": PREPROCESSING_CONTRACT,\n    \"model_config\": final_state[\"model_config\"],\n    \"encoder\": encoder_state,\n    \"encoder_fingerprint\": fingerprint,\n    \"input_contract\": {\"dtype\": \"float32\", \"shape\": [\"B\", 64, 168], \"padding_value\": CFG.padding_value},\n    \"output_contract\": {\"dtype\": \"float32\", \"shape\": [\"B\", 64, 128], \"padding_output\": 0.0},\n}\natomic_torch_save(export, WORK_DIR / \"signlang_det_encoder.pt\")\n(WORK_DIR / \"run_config.json\").write_text(json.dumps(asdict(CFG), indent=2, default=str))\n(WORK_DIR / \"environment.json\").write_text(json.dumps({\"python\": sys.version, \"torch\": torch.__version__, \"cuda\": torch.version.cuda, \"gpu\": torch.cuda.get_device_name()}, indent=2))\nprint(f\"Encoder exported to {WORK_DIR / 'signlang_det_encoder.pt'}\\nSHA-256: {fingerprint}\")\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Checkpoint Cleanup \ud83e\uddf9\n\nCleanup runs only after the exported encoder has been reloaded and verified. Best checkpoints and all reproducibility artifacts are retained.\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "def verify_exported_encoder(path: Path) -> Dict[str, Any]:\n    payload = torch.load(path, map_location=\"cpu\", weights_only=False)\n    required = {\"schema\", \"preprocessing\", \"model_config\", \"encoder\", \"encoder_fingerprint\"}\n    missing = sorted(required - payload.keys())\n    if missing: raise RuntimeError(f\"Exported encoder is missing fields: {missing}\")\n    if payload[\"schema\"] != CHECKPOINT_SCHEMA or payload[\"preprocessing\"] != PREPROCESSING_CONTRACT:\n        raise RuntimeError(\"Exported encoder schema or preprocessing contract is incompatible\")\n    actual_fingerprint = encoder_fingerprint(payload[\"encoder\"])\n    if actual_fingerprint != payload[\"encoder_fingerprint\"]:\n        raise RuntimeError(\"Exported encoder fingerprint verification failed\")\n    model = HandEncoder(**payload[\"model_config\"])\n    model.load_state_dict(payload[\"encoder\"], strict=True); model.eval()\n    sample = torch.full((1, CFG.max_frames, CFG.feature_dim), CFG.padding_value, dtype=torch.float32)\n    sample[:, :CFG.min_frames] = 0.0\n    with torch.inference_mode(): output = model(sample, torch.tensor([CFG.min_frames]))\n    expected_shape = (1, CFG.max_frames, CFG.embedding_dim)\n    if tuple(output.shape) != expected_shape or not torch.isfinite(output).all():\n        raise RuntimeError(f\"Exported encoder smoke test failed: output shape={tuple(output.shape)}\")\n    if torch.count_nonzero(output[:, CFG.min_frames:]).item() != 0:\n        raise RuntimeError(\"Exported encoder produced non-zero padding features\")\n    return {\n        \"path\": str(path), \"fingerprint\": actual_fingerprint,\n        \"output_shape\": list(output.shape), \"verified\": True,\n    }\n\ndef cleanup_checkpoints(work_dir: Path, run_dirs: Sequence[Path], keep_latest: bool = False) -> Dict[str, Any]:\n    run_dirs = [Path(path) for path in run_dirs]\n    missing_best = [str(path / \"best.pt\") for path in run_dirs if not (path / \"best.pt\").is_file()]\n    if missing_best: raise RuntimeError(f\"Checkpoint cleanup refused; missing retained best checkpoints: {missing_best}\")\n    targets = []\n    for run_dir in run_dirs:\n        targets.extend(sorted(run_dir.glob(\"epoch_*.pt\")))\n        latest = run_dir / \"latest.pt\"\n        if not keep_latest and latest.is_file(): targets.append(latest)\n    deleted, released_bytes = [], 0\n    for path in targets:\n        size = path.stat().st_size\n        path.unlink()\n        released_bytes += size\n        try: display_path = str(path.relative_to(work_dir))\n        except ValueError: display_path = str(path)\n        deleted.append({\"path\": display_path, \"bytes\": size})\n    retained = []\n    for run_dir in run_dirs:\n        retained.append(str(run_dir / \"best.pt\"))\n        latest = run_dir / \"latest.pt\"\n        if latest.is_file(): retained.append(str(latest))\n    report = {\n        \"deleted_files\": len(deleted), \"released_bytes\": released_bytes,\n        \"released_gib\": released_bytes / (1024 ** 3), \"deleted\": deleted,\n        \"retained\": retained, \"keep_latest\": keep_latest,\n    }\n    atomic_json_write(work_dir / \"checkpoint_cleanup.json\", report)\n    return report\n\nexport_verification = verify_exported_encoder(WORK_DIR / \"signlang_det_encoder.pt\")\nif CFG.cleanup_epoch_checkpoints:\n    cleanup_report = cleanup_checkpoints(\n        WORK_DIR, [SOURCE_OUT, TARGET_OUT], keep_latest=CFG.keep_latest_checkpoint,\n    )\n    print(\n        f\"Checkpoint cleanup removed {cleanup_report['deleted_files']} files and released \"\n        f\"{cleanup_report['released_gib']:.3f} GiB. Best checkpoints were retained.\"\n    )\nelse:\n    cleanup_report = {\"skipped\": True, \"reason\": \"cleanup_epoch_checkpoints=False\"}\n    atomic_json_write(WORK_DIR / \"checkpoint_cleanup.json\", cleanup_report)\n    print(\"Checkpoint cleanup was skipped by configuration\")\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Final Output Assembly \ud83d\ude9a\n\nThe working directory is reduced to the public output allowlist; any missing required artifact stops the run.\n"}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "FINAL_OUTPUTS = {\n    \"signlang_det_encoder.pt\",\n    \"figures/training_curves.png\",\n    \"figures/retrieval_summary.png\",\n    \"representation_training/metrics.csv\",\n    \"domain_adaptation/metrics.csv\",\n    \"representation_training/train.log\",\n    \"domain_adaptation/train.log\",\n}\n\ndef prune_notebook_outputs(work_dir: Path, allowed: Sequence[str]) -> None:\n    expected = {Path(item) for item in allowed}\n    missing = sorted(str(path) for path in expected if not (work_dir / path).is_file())\n    if missing:\n        raise RuntimeError(f\"Required final notebook outputs are missing: {missing}\")\n    for path in sorted(work_dir.rglob(\"*\"), key=lambda item: len(item.parts), reverse=True):\n        if path.is_file() and path.relative_to(work_dir) not in expected:\n            path.unlink()\n        elif path.is_dir():\n            try: path.rmdir()\n            except OSError: pass\n    actual = {path.relative_to(work_dir) for path in work_dir.rglob(\"*\") if path.is_file()}\n    if actual != expected:\n        raise RuntimeError(f\"Final output allowlist mismatch: expected={sorted(map(str, expected))}, actual={sorted(map(str, actual))}\")\n\nprune_notebook_outputs(WORK_DIR, FINAL_OUTPUTS)\nprint(\"Final notebook outputs:\\n\" + \"\\n\".join(f\"  - {path}\" for path in sorted(FINAL_OUTPUTS)))\n"}, {"cell_type": "markdown", "metadata": {}, "source": "## Usage Notes \ud83d\udcd6\n\n- Attach the Google Isolated Sign Language Recognition dataset and the target dataset containing a `landmarks` directory before starting the session.\n- Edit `CFG.source_root` and `CFG.target_landmarks` when the data is mounted outside the standard Kaggle paths configured by the uploader.\n- Keep `CFG.num_workers = 0` for reliable Kaggle notebook restarts. A single T4 is the default; multi-GPU execution is intentionally opt-in.\n- Interrupted runs resume compatible `latest.pt` checkpoints and continue an incomplete source cache.\n- After successful export verification, the notebook removes every file outside the documented final output allowlist.\n- Keep `signlang_det_encoder.pt`, both charts, both metric CSV files, and both training logs together for delivery.\n- Unknown-sign distance and margin thresholds are **not** derived from training loss. Calibrate them with separate known-query and unknown-action data before deployment.\n- Each dynamic prototype should retain its label, valid length, frame features, pooled feature, preprocessing contract, and encoder fingerprint.\n"}], "metadata": {"kernelspec": {"display_name": "Python 3", "language": "python", "name": "python3"}, "language_info": {"name": "python", "version": "3.11"}}, "nbformat": 4, "nbformat_minor": 5}