{"metadata":{"kernelspec":{"name":"python3","display_name":"Python 3","language":"python"},"language_info":{"name":"python","version":"3.12.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"\n# =============================================================================\n# CELL 1 — UTILITIES + PHASE-AWARE ATTENTION MODEL (IMUModel)\n# Feature engineering is identical to solution2-imu-triet.ipynb cell 1.\n# =============================================================================\nimport os\nimport pathlib\nimport re\nimport json\nimport warnings\n\nimport numpy as np\nimport pandas as pd\nimport joblib\n\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom torch import Tensor\nfrom torch.nn.modules.batchnorm import _BatchNorm\nfrom torch.utils.data import Dataset, DataLoader\n\nfrom scipy.spatial.transform import Rotation as R\nimport kaggle_evaluation.cmi_inference_server\n\nwarnings.filterwarnings(\"ignore\")\n\n# -----------------------------------------------------------------------------\n# CONFIG\n# -----------------------------------------------------------------------------\nclass Config:\n    DEVICE      = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n    ACC_COLS    = [\"acc_x\", \"acc_y\", \"acc_z\"]\n    ROT_COLS    = [\"rot_w\", \"rot_x\", \"rot_y\", \"rot_z\"]   # [w,x,y,z]\n    BATCH_SIZE  = 1\n    MAX_SEQ_LEN = 200\n\nFEATURE_NAMES = [\n    \"acc_x\", \"acc_y\", \"acc_z\",\n    \"rot_6d_0\", \"rot_6d_1\", \"rot_6d_2\", \"rot_6d_3\", \"rot_6d_4\", \"rot_6d_5\",\n    \"angular_vel_x\", \"angular_vel_y\", \"angular_vel_z\",\n    \"linear_acc_x\", \"linear_acc_y\", \"linear_acc_z\",\n]\n\n# -----------------------------------------------------------------------------\n# Feature engineering — identical to training (utils.py / cell 1)\n# All functions take xyzw after make_features() converts from wxyz.\n# -----------------------------------------------------------------------------\ndef normalize_quaternion(quat):\n    norm = np.linalg.norm(quat, axis=-1, keepdims=True)\n    norm = np.where(norm > 1e-8, norm, 1.0)\n    return quat / norm\n\ndef quaternion_to_6d_rotation(quat: np.ndarray) -> np.ndarray:\n    \"\"\"quat shape (...,4) in [x,y,z,w]. Returns (...,6).\"\"\"\n    if quat.ndim == 1:\n        quat = quat.reshape(1, -1)\n    has_nan = np.any(np.isnan(quat), axis=-1)\n    result = np.full((*quat.shape[:-1], 6), np.nan)\n    valid = ~has_nan & ~np.all(np.isclose(quat, 0), axis=-1)\n    if not np.any(valid):\n        return result\n    try:\n        q_norm = normalize_quaternion(quat[valid])\n        mats = R.from_quat(q_norm).as_matrix()\n        result[valid] = mats[:, :, :2].reshape(-1, 6)\n    except (ValueError, RuntimeError):\n        pass\n    return result\n\ndef calculate_angular_velocity_from_quat(rot_data: np.ndarray, time_delta=1/200) -> np.ndarray:\n    \"\"\"rot_data shape (T,4) in [x,y,z,w]. Returns (T,3).\"\"\"\n    quat = rot_data\n    n = quat.shape[0]\n    angular_vel = np.full((n, 3), np.nan)\n    for i in range(n - 1):\n        q_t, q_t1 = quat[i], quat[i+1]\n        if (np.any(np.isnan(q_t)) or np.any(np.isnan(q_t1)) or\n                np.all(np.isclose(q_t, 0)) or np.all(np.isclose(q_t1, 0))):\n            continue\n        try:\n            r_t  = R.from_quat(normalize_quaternion(q_t.reshape(1, -1))[0])\n            r_t1 = R.from_quat(normalize_quaternion(q_t1.reshape(1, -1))[0])\n            angular_vel[i] = (r_t.inv() * r_t1).as_rotvec() / time_delta\n        except (ValueError, RuntimeError):\n            pass\n    return angular_vel\n\ndef remove_gravity_from_acc(acc_values: np.ndarray, quat_xyzw: np.ndarray,\n                            gravity_world=np.array([0., 0., 9.81])) -> np.ndarray:\n    \"\"\"acc_values (T,3), quat_xyzw (T,4) in [x,y,z,w]. Returns (T,3).\"\"\"\n    n = acc_values.shape[0]\n    linear_accel = np.full_like(acc_values, np.nan)\n    for i in range(n):\n        if np.any(np.isnan(acc_values[i])) or np.any(np.isnan(quat_xyzw[i])):\n            continue\n        if np.all(np.isclose(quat_xyzw[i], 0)):\n            linear_accel[i] = acc_values[i]\n            continue\n        try:\n            q = normalize_quaternion(quat_xyzw[i:i+1])[0]\n            gsensor = R.from_quat(q).apply(gravity_world, inverse=True)\n            linear_accel[i] = acc_values[i] - gsensor\n        except (ValueError, RuntimeError):\n            pass\n    return linear_accel\n\n# Special-subject coordinate flip (identical to training cell 1)\n_SPECIAL_SUBJECTS = {\"SUBJ_019262\", \"SUBJ_045235\"}\n_FLIP_COLS = list(range(13))  # feature indices 0-12\n\ndef make_features(acc: np.ndarray, rot_wxyz: np.ndarray,\n                  handedness: int, subject: str) -> np.ndarray:\n    \"\"\"\n    Build 15-feature array — identical to training make_features().\n    acc (T,3), rot_wxyz (T,4) in [w,x,y,z].\n    \"\"\"\n    rot_xyzw = rot_wxyz[:, [1, 2, 3, 0]]   # convert to [x,y,z,w] for scipy\n\n    feat = acc.copy().astype(np.float32)\n    feat = np.concatenate([feat, quaternion_to_6d_rotation(rot_xyzw)], axis=1)\n    feat = np.concatenate([feat, calculate_angular_velocity_from_quat(rot_xyzw)], axis=1)\n    feat = np.concatenate([feat, remove_gravity_from_acc(acc, rot_xyzw)], axis=1)\n    feat = np.nan_to_num(feat, nan=0.0).astype(np.float32)\n\n    # Left-handed flip (handedness == 0) — same column indices as training\n    if handedness == 0:\n        feat[:, 0]  *= -1.   # acc_x\n        feat[:, 3]  *= -1.   # rot_6d_0\n        feat[:, 7]  *= -1.   # angular_vel_x\n        feat[:, 10] *= -1.   # linear_acc_x\n        feat[:, 11] *= -1.   # linear_acc_y\n        feat[:, 12] *= -1.   # linear_acc_z\n\n    # Special-subject flip\n    if subject in _SPECIAL_SUBJECTS:\n        for c in _FLIP_COLS:\n            feat[:, c] *= -1.\n\n    return feat\n\n# -----------------------------------------------------------------------------\n# Masked BatchNorm — identical to training (masked_batchnorm.py)\n# -----------------------------------------------------------------------------\ndef masked_batch_norm(input: Tensor, mask: Tensor, weight, bias,\n                      running_mean, running_var, training: bool,\n                      momentum: float, eps: float = 1e-5) -> Tensor:\n    num_dims = len(input.shape[2:])\n    _dims  = (0,) + tuple(range(-num_dims, 0))\n    _slice = (None, ...) + (None,) * num_dims\n    if training:\n        num_elements = mask.sum(_dims).clamp_min(1.0)\n        mean = (input * mask).sum(_dims) / num_elements\n        var  = (((input - mean[_slice]) * mask) ** 2).sum(_dims) / num_elements\n        if running_mean is not None:\n            running_mean.copy_(running_mean * (1 - momentum) + momentum * mean.detach())\n        if running_var is not None:\n            running_var.copy_(running_var  * (1 - momentum) + momentum * var.detach())\n    else:\n        if running_mean is None or running_var is None:\n            raise ValueError(\"running stats required in eval mode\")\n        mean, var = running_mean, running_var\n    out = (input - mean[_slice]) / torch.sqrt(var[_slice] + eps)\n    if weight is not None and bias is not None:\n        out = out * weight[_slice] + bias[_slice]\n    return out\n\nclass _MaskedBatchNorm(_BatchNorm):\n    def forward(self, input: Tensor, mask: Tensor = None) -> Tensor:\n        self._check_input_dim(input)\n        momentum = 0.0 if self.momentum is None else self.momentum\n        if self.training and self.track_running_stats and self.num_batches_tracked is not None:\n            self.num_batches_tracked += 1\n            momentum = self.momentum if self.momentum is not None else 1.0 / float(self.num_batches_tracked)\n        bn_training = self.training or (self.running_mean is None and self.running_var is None)\n        if mask is None:\n            return F.batch_norm(input,\n                self.running_mean if not self.training or self.track_running_stats else None,\n                self.running_var  if not self.training or self.track_running_stats else None,\n                self.weight, self.bias, bn_training, momentum, self.eps)\n        return masked_batch_norm(input, mask, self.weight, self.bias,\n                self.running_mean if not self.training or self.track_running_stats else None,\n                self.running_var  if not self.training or self.track_running_stats else None,\n                bn_training, momentum, self.eps)\n\nclass MaskedBatchNorm1d(torch.nn.BatchNorm1d, _MaskedBatchNorm):\n    \"\"\"Auto-unsqueezes 2-D mask (B,T) → (B,1,T) before broadcasting.\"\"\"\n    def forward(self, inputs, mask=None):\n        if mask is not None and mask.dim() == 2:\n            mask = mask[:, None, :]\n        return super().forward(inputs, mask)\n\n# -----------------------------------------------------------------------------\n# Shared model building blocks — identical to training (model.py)\n# -----------------------------------------------------------------------------\ndef create_mask(lengths, max_len):\n    batch_size = lengths.size(0)\n    mask = torch.arange(max_len, device=lengths.device).expand(batch_size, max_len) < lengths.unsqueeze(1)\n    return mask.float()\n\nclass SEBlock(nn.Module):\n    def __init__(self, channels, reduction=8):\n        super().__init__()\n        self.excitation = nn.Sequential(\n            nn.Linear(channels, channels // reduction, bias=False),\n            nn.ReLU(inplace=True),\n            nn.Linear(channels // reduction, channels, bias=False),\n            nn.Sigmoid(),\n        )\n    def forward(self, x, mask):\n        mask_  = mask.unsqueeze(1)\n        masked = x * mask_\n        slen   = mask_.sum(dim=-1, keepdim=True)\n        y = masked.sum(dim=-1, keepdim=True) / (slen + 1e-8)\n        y = self.excitation(y.squeeze(-1)).unsqueeze(-1)\n        return x * y.expand_as(x)\n\nclass ResidualSECNNBlock(nn.Module):\n    def __init__(self, in_channels, out_channels, kernel_size, dropout=0.3):\n        super().__init__()\n        self.conv1       = nn.Conv1d(in_channels,  out_channels, kernel_size, padding=kernel_size//2, bias=False)\n        self.bn1         = MaskedBatchNorm1d(out_channels)\n        self.conv2       = nn.Conv1d(out_channels, out_channels, kernel_size, padding=kernel_size//2, bias=False)\n        self.bn2         = MaskedBatchNorm1d(out_channels)\n        self.se          = SEBlock(out_channels)\n        self.shortcut    = nn.Conv1d(in_channels,  out_channels, 1, bias=False)\n        self.shortcut_bn = MaskedBatchNorm1d(out_channels)\n        self.dropout     = nn.Dropout(dropout)\n    def forward(self, x, mask):\n        m1 = mask.unsqueeze(1)\n        sc  = self.shortcut_bn(self.shortcut(x) * m1, mask)\n        out = F.relu(self.bn1(self.conv1(x) * m1, mask))\n        out = self.bn2(self.conv2(out) * m1, mask)\n        out = self.se(out, mask)\n        out = F.relu(out + sc)\n        return self.dropout(out) * m1\n\n# -----------------------------------------------------------------------------\n# Phase-aware attention model — identical to training cell 2 (model.py)\n# Uses nn.Sequential so state_dict keys match saved checkpoints.\n# -----------------------------------------------------------------------------\nclass PhaseAttentionLayer(nn.Module):\n    def __init__(self, hidden_dim):\n        super().__init__()\n        self.attention = nn.Linear(hidden_dim, 1)\n    def forward(self, x, phase_weights, mask):\n        scores = torch.tanh(self.attention(x)).squeeze(-1)\n        scores = scores.masked_fill(~mask.bool(), -1e9)\n        attn   = F.softmax(scores, dim=1)\n        combined = attn * phase_weights * mask\n        combined = combined / (combined.sum(dim=1, keepdim=True) + 1e-8)\n        return torch.sum(x * combined.unsqueeze(-1), dim=1)\n\nclass IMUModel(nn.Module):\n    def __init__(self, input_size: int, n_classes: int):\n        super().__init__()\n        self.acc_block = nn.Sequential(\n            ResidualSECNNBlock(3,   64,  1),\n            ResidualSECNNBlock(64,  128, 3),\n            ResidualSECNNBlock(128, 256, 5),\n        )\n        self.rot_block = nn.Sequential(\n            ResidualSECNNBlock(9,   64,  1),\n            ResidualSECNNBlock(64,  128, 3),\n            ResidualSECNNBlock(128, 256, 5),\n        )\n        self.combined_block = nn.Sequential(\n            ResidualSECNNBlock(3,   64,  1),\n            ResidualSECNNBlock(64,  128, 3),\n            ResidualSECNNBlock(128, 256, 5),\n        )\n        self.imu_block        = ResidualSECNNBlock(256 * 3, 256, 3)\n        self.phase_head       = nn.Linear(256, 3)\n        self.phase1_attention = PhaseAttentionLayer(256)\n        self.phase2_attention = PhaseAttentionLayer(256)\n        self.phase3_attention = PhaseAttentionLayer(256)\n        self.dense1    = nn.Linear(256 * 3, 512, bias=False)\n        self.bn_dense1 = nn.BatchNorm1d(512)\n        self.drop1     = nn.Dropout(0.5)\n        self.dense2    = nn.Linear(512, 256, bias=False)\n        self.bn_dense2 = nn.BatchNorm1d(256)\n        self.drop2     = nn.Dropout(0.3)\n        self.classifier = nn.Linear(256, n_classes)\n\n    def _run_seq(self, layers, x, mask):\n        for layer in layers:\n            x = layer(x, mask)\n        return x\n\n    def forward(self, x, lengths):\n        x    = x.transpose(1, 2)\n        mask = create_mask(lengths, x.size(2))\n        x1 = self._run_seq(self.acc_block,      x[:, :3,   :], mask)\n        x2 = self._run_seq(self.rot_block,      x[:, 3:12, :], mask)\n        x3 = self._run_seq(self.combined_block, x[:, 12:,  :], mask)\n        x1 = self.imu_block(torch.cat([x1, x2, x3], dim=1), mask)\n        x1 = x1.transpose(1, 2)\n        phase_logits = self.phase_head(x1)\n        phase_logits = phase_logits.masked_fill(~mask.unsqueeze(-1).bool(), 0)\n        phase_probs  = F.softmax(phase_logits, dim=-1)\n        p1 = self.phase1_attention(x1, phase_probs[:, :, 0], mask)\n        p2 = self.phase2_attention(x1, phase_probs[:, :, 1], mask)\n        p3 = self.phase3_attention(x1, phase_probs[:, :, 2], mask)\n        feat = torch.cat([p1, p2, p3], dim=-1)\n        h = F.relu(self.bn_dense1(self.dense1(feat))); h = self.drop1(h)\n        h = F.relu(self.bn_dense2(self.dense2(h)));    h = self.drop2(h)\n        return {\"gesture_logits\": self.classifier(h), \"phase_logits\": phase_logits}\n\nprint(\"[Cell 1] utilities + IMUModel (phase-aware) ready.\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-25T12:28:15.958917Z","iopub.execute_input":"2026-06-25T12:28:15.959980Z","iopub.status.idle":"2026-06-25T12:28:23.113498Z","shell.execute_reply.started":"2026-06-25T12:28:15.959941Z","shell.execute_reply":"2026-06-25T12:28:23.112383Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\n# =============================================================================\n# CELL 2 — UTILITIES + NORMAL ATTENTION MODEL (IMUModelNormalAttn)\n# Self-contained re-import. Identical utilities to Cell 1.\n# =============================================================================\nimport os\nimport pathlib\nimport re\nimport json\nimport warnings\n\nimport numpy as np\nimport pandas as pd\nimport joblib\n\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom torch import Tensor\nfrom torch.nn.modules.batchnorm import _BatchNorm\nfrom torch.utils.data import Dataset, DataLoader\n\nfrom scipy.spatial.transform import Rotation as R\nimport kaggle_evaluation.cmi_inference_server\n\nwarnings.filterwarnings(\"ignore\")\n\nclass Config:\n    DEVICE      = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n    ACC_COLS    = [\"acc_x\", \"acc_y\", \"acc_z\"]\n    ROT_COLS    = [\"rot_w\", \"rot_x\", \"rot_y\", \"rot_z\"]\n    BATCH_SIZE  = 1\n    MAX_SEQ_LEN = 200\n\nFEATURE_NAMES = [\n    \"acc_x\", \"acc_y\", \"acc_z\",\n    \"rot_6d_0\", \"rot_6d_1\", \"rot_6d_2\", \"rot_6d_3\", \"rot_6d_4\", \"rot_6d_5\",\n    \"angular_vel_x\", \"angular_vel_y\", \"angular_vel_z\",\n    \"linear_acc_x\", \"linear_acc_y\", \"linear_acc_z\",\n]\n\ndef normalize_quaternion(quat):\n    norm = np.linalg.norm(quat, axis=-1, keepdims=True)\n    norm = np.where(norm > 1e-8, norm, 1.0)\n    return quat / norm\n\ndef quaternion_to_6d_rotation(quat: np.ndarray) -> np.ndarray:\n    if quat.ndim == 1:\n        quat = quat.reshape(1, -1)\n    has_nan = np.any(np.isnan(quat), axis=-1)\n    result = np.full((*quat.shape[:-1], 6), np.nan)\n    valid = ~has_nan & ~np.all(np.isclose(quat, 0), axis=-1)\n    if not np.any(valid):\n        return result\n    try:\n        q_norm = normalize_quaternion(quat[valid])\n        mats = R.from_quat(q_norm).as_matrix()\n        result[valid] = mats[:, :, :2].reshape(-1, 6)\n    except (ValueError, RuntimeError):\n        pass\n    return result\n\ndef calculate_angular_velocity_from_quat(rot_data: np.ndarray, time_delta=1/200) -> np.ndarray:\n    quat = rot_data\n    n = quat.shape[0]\n    angular_vel = np.full((n, 3), np.nan)\n    for i in range(n - 1):\n        q_t, q_t1 = quat[i], quat[i+1]\n        if (np.any(np.isnan(q_t)) or np.any(np.isnan(q_t1)) or\n                np.all(np.isclose(q_t, 0)) or np.all(np.isclose(q_t1, 0))):\n            continue\n        try:\n            r_t  = R.from_quat(normalize_quaternion(q_t.reshape(1, -1))[0])\n            r_t1 = R.from_quat(normalize_quaternion(q_t1.reshape(1, -1))[0])\n            angular_vel[i] = (r_t.inv() * r_t1).as_rotvec() / time_delta\n        except (ValueError, RuntimeError):\n            pass\n    return angular_vel\n\ndef remove_gravity_from_acc(acc_values: np.ndarray, quat_xyzw: np.ndarray,\n                            gravity_world=np.array([0., 0., 9.81])) -> np.ndarray:\n    n = acc_values.shape[0]\n    linear_accel = np.full_like(acc_values, np.nan)\n    for i in range(n):\n        if np.any(np.isnan(acc_values[i])) or np.any(np.isnan(quat_xyzw[i])):\n            continue\n        if np.all(np.isclose(quat_xyzw[i], 0)):\n            linear_accel[i] = acc_values[i]\n            continue\n        try:\n            q = normalize_quaternion(quat_xyzw[i:i+1])[0]\n            gsensor = R.from_quat(q).apply(gravity_world, inverse=True)\n            linear_accel[i] = acc_values[i] - gsensor\n        except (ValueError, RuntimeError):\n            pass\n    return linear_accel\n\n_SPECIAL_SUBJECTS = {\"SUBJ_019262\", \"SUBJ_045235\"}\n_FLIP_COLS = list(range(13))\n\ndef make_features(acc: np.ndarray, rot_wxyz: np.ndarray,\n                  handedness: int, subject: str) -> np.ndarray:\n    rot_xyzw = rot_wxyz[:, [1, 2, 3, 0]]\n    feat = acc.copy().astype(np.float32)\n    feat = np.concatenate([feat, quaternion_to_6d_rotation(rot_xyzw)], axis=1)\n    feat = np.concatenate([feat, calculate_angular_velocity_from_quat(rot_xyzw)], axis=1)\n    feat = np.concatenate([feat, remove_gravity_from_acc(acc, rot_xyzw)], axis=1)\n    feat = np.nan_to_num(feat, nan=0.0).astype(np.float32)\n    if handedness == 0:\n        feat[:, 0]  *= -1.\n        feat[:, 3]  *= -1.\n        feat[:, 7]  *= -1.\n        feat[:, 10] *= -1.\n        feat[:, 11] *= -1.\n        feat[:, 12] *= -1.\n    if subject in _SPECIAL_SUBJECTS:\n        for c in _FLIP_COLS:\n            feat[:, c] *= -1.\n    return feat\n\ndef masked_batch_norm(input: Tensor, mask: Tensor, weight, bias,\n                      running_mean, running_var, training: bool,\n                      momentum: float, eps: float = 1e-5) -> Tensor:\n    num_dims = len(input.shape[2:])\n    _dims  = (0,) + tuple(range(-num_dims, 0))\n    _slice = (None, ...) + (None,) * num_dims\n    if training:\n        num_elements = mask.sum(_dims).clamp_min(1.0)\n        mean = (input * mask).sum(_dims) / num_elements\n        var  = (((input - mean[_slice]) * mask) ** 2).sum(_dims) / num_elements\n        if running_mean is not None:\n            running_mean.copy_(running_mean * (1 - momentum) + momentum * mean.detach())\n        if running_var is not None:\n            running_var.copy_(running_var  * (1 - momentum) + momentum * var.detach())\n    else:\n        if running_mean is None or running_var is None:\n            raise ValueError(\"running stats required in eval mode\")\n        mean, var = running_mean, running_var\n    out = (input - mean[_slice]) / torch.sqrt(var[_slice] + eps)\n    if weight is not None and bias is not None:\n        out = out * weight[_slice] + bias[_slice]\n    return out\n\nclass _MaskedBatchNorm(_BatchNorm):\n    def forward(self, input: Tensor, mask: Tensor = None) -> Tensor:\n        self._check_input_dim(input)\n        momentum = 0.0 if self.momentum is None else self.momentum\n        if self.training and self.track_running_stats and self.num_batches_tracked is not None:\n            self.num_batches_tracked += 1\n            momentum = self.momentum if self.momentum is not None else 1.0 / float(self.num_batches_tracked)\n        bn_training = self.training or (self.running_mean is None and self.running_var is None)\n        if mask is None:\n            return F.batch_norm(input,\n                self.running_mean if not self.training or self.track_running_stats else None,\n                self.running_var  if not self.training or self.track_running_stats else None,\n                self.weight, self.bias, bn_training, momentum, self.eps)\n        return masked_batch_norm(input, mask, self.weight, self.bias,\n                self.running_mean if not self.training or self.track_running_stats else None,\n                self.running_var  if not self.training or self.track_running_stats else None,\n                bn_training, momentum, self.eps)\n\nclass MaskedBatchNorm1d(torch.nn.BatchNorm1d, _MaskedBatchNorm):\n    def forward(self, inputs, mask=None):\n        if mask is not None and mask.dim() == 2:\n            mask = mask[:, None, :]\n        return super().forward(inputs, mask)\n\ndef create_mask(lengths, max_len):\n    batch_size = lengths.size(0)\n    mask = torch.arange(max_len, device=lengths.device).expand(batch_size, max_len) < lengths.unsqueeze(1)\n    return mask.float()\n\nclass SEBlock(nn.Module):\n    def __init__(self, channels, reduction=8):\n        super().__init__()\n        self.excitation = nn.Sequential(\n            nn.Linear(channels, channels // reduction, bias=False),\n            nn.ReLU(inplace=True),\n            nn.Linear(channels // reduction, channels, bias=False),\n            nn.Sigmoid(),\n        )\n    def forward(self, x, mask):\n        mask_  = mask.unsqueeze(1)\n        masked = x * mask_\n        slen   = mask_.sum(dim=-1, keepdim=True)\n        y = masked.sum(dim=-1, keepdim=True) / (slen + 1e-8)\n        y = self.excitation(y.squeeze(-1)).unsqueeze(-1)\n        return x * y.expand_as(x)\n\nclass ResidualSECNNBlock(nn.Module):\n    def __init__(self, in_channels, out_channels, kernel_size, dropout=0.3):\n        super().__init__()\n        self.conv1       = nn.Conv1d(in_channels,  out_channels, kernel_size, padding=kernel_size//2, bias=False)\n        self.bn1         = MaskedBatchNorm1d(out_channels)\n        self.conv2       = nn.Conv1d(out_channels, out_channels, kernel_size, padding=kernel_size//2, bias=False)\n        self.bn2         = MaskedBatchNorm1d(out_channels)\n        self.se          = SEBlock(out_channels)\n        self.shortcut    = nn.Conv1d(in_channels,  out_channels, 1, bias=False)\n        self.shortcut_bn = MaskedBatchNorm1d(out_channels)\n        self.dropout     = nn.Dropout(dropout)\n    def forward(self, x, mask):\n        m1 = mask.unsqueeze(1)\n        sc  = self.shortcut_bn(self.shortcut(x) * m1, mask)\n        out = F.relu(self.bn1(self.conv1(x) * m1, mask))\n        out = self.bn2(self.conv2(out) * m1, mask)\n        out = self.se(out, mask)\n        out = F.relu(out + sc)\n        return self.dropout(out) * m1\n\n# Normal attention model — identical to training cell 3\nclass AttentionLayer(nn.Module):\n    def __init__(self, hidden_dim):\n        super().__init__()\n        self.attention = nn.Linear(hidden_dim, 1)\n    def forward(self, x, mask):\n        scores = torch.tanh(self.attention(x)).squeeze(-1)\n        scores = scores.masked_fill(~mask.bool(), -1e9)\n        attn   = F.softmax(scores, dim=1)\n        return torch.sum(x * attn.unsqueeze(-1), dim=1)\n\nclass IMUModelNormalAttn(nn.Module):\n    def __init__(self, input_size: int, n_classes: int):\n        super().__init__()\n        self.acc_block = nn.Sequential(\n            ResidualSECNNBlock(3,   64,  1),\n            ResidualSECNNBlock(64,  128, 3),\n            ResidualSECNNBlock(128, 256, 5),\n        )\n        self.rot_block = nn.Sequential(\n            ResidualSECNNBlock(9,   64,  1),\n            ResidualSECNNBlock(64,  128, 3),\n            ResidualSECNNBlock(128, 256, 5),\n        )\n        self.combined_block = nn.Sequential(\n            ResidualSECNNBlock(3,   64,  1),\n            ResidualSECNNBlock(64,  128, 3),\n            ResidualSECNNBlock(128, 256, 5),\n        )\n        self.imu_block = ResidualSECNNBlock(256 * 3, 256, 3)\n        self.attn1 = AttentionLayer(256)\n        self.attn2 = AttentionLayer(256)\n        self.attn3 = AttentionLayer(256)\n        self.dense1    = nn.Linear(256 * 3, 512, bias=False)\n        self.bn_dense1 = nn.BatchNorm1d(512)\n        self.drop1     = nn.Dropout(0.5)\n        self.dense2    = nn.Linear(512, 256, bias=False)\n        self.bn_dense2 = nn.BatchNorm1d(256)\n        self.drop2     = nn.Dropout(0.3)\n        self.classifier = nn.Linear(256, n_classes)\n\n    def _run_seq(self, layers, x, mask):\n        for layer in layers:\n            x = layer(x, mask)\n        return x\n\n    def forward(self, x, lengths):\n        x    = x.transpose(1, 2)\n        mask = create_mask(lengths, x.size(2))\n        x1 = self._run_seq(self.acc_block,      x[:, :3,   :], mask)\n        x2 = self._run_seq(self.rot_block,      x[:, 3:12, :], mask)\n        x3 = self._run_seq(self.combined_block, x[:, 12:,  :], mask)\n        x1 = self.imu_block(torch.cat([x1, x2, x3], dim=1), mask)\n        x1 = x1.transpose(1, 2)\n        p1 = self.attn1(x1, mask)\n        p2 = self.attn2(x1, mask)\n        p3 = self.attn3(x1, mask)\n        feat = torch.cat([p1, p2, p3], dim=-1)\n        h = F.relu(self.bn_dense1(self.dense1(feat))); h = self.drop1(h)\n        h = F.relu(self.bn_dense2(self.dense2(h)));    h = self.drop2(h)\n        return {\"gesture_logits\": self.classifier(h)}\n\nprint(\"[Cell 2] utilities + IMUModelNormalAttn (normal attention) ready.\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-25T12:28:23.116760Z","iopub.execute_input":"2026-06-25T12:28:23.117630Z","iopub.status.idle":"2026-06-25T12:28:23.210207Z","shell.execute_reply.started":"2026-06-25T12:28:23.117586Z","shell.execute_reply":"2026-06-25T12:28:23.209181Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\n# =============================================================================\n# CELL 3 — HUNGARIAN POST-PROCESSING\n# Ported from solution2 test.py: solve_capacity1_with_hungarian()\n#\n# Key idea: within one subject the same gesture never repeats.\n# By accumulating probability vectors for all sequences of a subject into an\n# N×K matrix (N sequences, K gesture classes) and running linear_sum_assignment,\n# we find the globally optimal unique-label assignment.\n# The prediction for the *current* (most-recent) sequence is the last row's\n# assigned label.  Earlier predictions can shift as more sequences arrive —\n# this is the intended \"online Hungarian\" behaviour from the 2nd-place solution.\n# =============================================================================\nfrom scipy.optimize import linear_sum_assignment\n\ndef solve_capacity1_with_hungarian(S):\n    \"\"\"\n    S : 2-D array shape (N, K)\n        S[i, j] = ensemble probability of sequence i belonging to gesture class j.\n    Returns the integer class index for the LAST row (most-recent sequence).\n    Constraint: each class is assigned at most once (capacity-1).\n    Requires N <= K (a subject has at most as many sequences as gesture classes).\n    \"\"\"\n    S = np.asarray(S, dtype=np.float64)\n    N, K = S.shape\n    if N > K:\n        # Fallback: plain argmax if constraint is violated (shouldn't happen)\n        return int(np.argmax(S[-1]))\n    cost        = S.max() - S                    # convert max → min\n    rows, cols  = linear_sum_assignment(cost)    # optimal unique assignment\n    assign      = (cols % K).tolist()\n    return assign[-1]                            # prediction for newest sequence\n\n\n# Per-subject state — reset once at session start, persists across predict() calls\n_hungarian_subject_to_idx  : dict  = {}\n_hungarian_logits_list     : list  = []\n\ndef reset_hungarian_state():\n    \"\"\"Call once before starting inference to clear per-subject history.\"\"\"\n    global _hungarian_subject_to_idx, _hungarian_logits_list\n    _hungarian_subject_to_idx  = {}\n    _hungarian_logits_list     = []\n    print(\"[Hungarian] State reset.\")\n\nreset_hungarian_state()\nprint(\"[Cell 3] Hungarian post-processing ready.\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-25T12:28:23.211851Z","iopub.execute_input":"2026-06-25T12:28:23.212242Z","iopub.status.idle":"2026-06-25T12:28:23.379547Z","shell.execute_reply.started":"2026-06-25T12:28:23.212210Z","shell.execute_reply":"2026-06-25T12:28:23.378337Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\n# =============================================================================\n# CELL 4 — MAIN INFERENCE\n# Ensemble of phase-aware + normal-attention folds with Hungarian post-processing.\n#\n# Pipeline per sequence:\n#   1. _prepare_sequence()  — feature engineering + truncate last N + pad\n#   2. Ensemble 10 models   — average softmax probs → probability vector (1×K)\n#   3. Append to subject's accumulated matrix (N×K)\n#   4. solve_capacity1_with_hungarian() → unique-label assignment\n#   5. Return pred_to_label[pred_idx]  (int → gesture name)\n#\n# Backward-compatible: supports both old prep[\"categories\"] (list)\n# and new prep[\"pred_to_label\"] (dict {int: gesture_name}).\n# =============================================================================\n\nclass CMIInferenceDataset(Dataset):\n    def __init__(self, X, length):\n        self.X      = X\n        self.length = length\n    def __len__(self): return 1\n    def __getitem__(self, idx):\n        return torch.FloatTensor(self.X), torch.tensor(self.length, dtype=torch.long)\n\n\ndef _prepare_sequence(sequence_df: pd.DataFrame, demo_df: pd.DataFrame,\n                      features: list, max_length: int):\n    \"\"\"\n    Matches training GestureDataset.__getitem__ exactly:\n      1. make_features()  — feature engineering + inline flips\n      2. Truncate to LAST max_length steps\n      3. Zero-pad to max_length, return effective length\n    \"\"\"\n    sequence_df = sequence_df.copy()\n    if \"sequence_counter\" in sequence_df.columns:\n        sequence_df = sequence_df.sort_values(\"sequence_counter\")\n\n    acc      = sequence_df[Config.ACC_COLS].to_numpy(dtype=np.float64)\n    rot_wxyz = sequence_df[Config.ROT_COLS].to_numpy(dtype=np.float64)\n\n    subject    = str(sequence_df[\"subject\"].iloc[0]) if \"subject\" in sequence_df.columns else \"\"\n    handedness = 1\n    if \"subject\" in sequence_df.columns and not demo_df.empty:\n        subj = sequence_df[\"subject\"].iloc[0]\n        row  = demo_df[demo_df[\"subject\"] == subj]\n        if not row.empty and \"handedness\" in row.columns:\n            handedness = int(row[\"handedness\"].iloc[0])\n\n    feat = make_features(acc, rot_wxyz, handedness, subject)   # (T, 15)\n    if len(feat) > max_length:\n        feat = feat[-max_length:]\n\n    eff_len = len(feat)\n    padded  = np.zeros((max_length, feat.shape[1]), dtype=np.float32)\n    padded[:eff_len] = feat\n    return padded, eff_len, subject\n\n\n# -----------------------------------------------------------------------------\n# Artifact discovery\n# -----------------------------------------------------------------------------\nSEARCH_DIRS = [\n    pathlib.Path(\"/kaggle/working\"),\n    pathlib.Path(\"/kaggle/input/datasets/triet1208/2nd-solution-phase-aware-attention\"),\n    pathlib.Path(\"/kaggle/input/datasets/triet1208/2nd-solution-normalattention\"),\n]\n\ndef _scan(prep_pattern: str, model_pattern: str):\n    found, seen_keys = [], set()\n    for base in SEARCH_DIRS:\n        if not base.exists(): continue\n        for prep_path in sorted(base.glob(prep_pattern)):\n            m = re.search(r\"fold_(\\d+)\\.joblib$\", prep_path.name)\n            if not m: continue\n            k          = int(m.group(1))\n            model_path = base / model_pattern.format(k=k)\n            if not model_path.exists(): continue\n            key = (prep_path.name, model_path.name)\n            if key in seen_keys: continue\n            seen_keys.add(key)\n            found.append((base, k, prep_path, model_path))\n    return found\n\ndef _load_variant(model_cls, prep_pattern, model_pattern, label):\n    loaded = []\n    for base, k, prep_path, model_path in _scan(prep_pattern, model_pattern):\n        prep  = joblib.load(prep_path)\n        # Support both new pred_to_label (dict) and old categories (list)\n        n_cls = len(prep.get(\"pred_to_label\", prep.get(\"categories\", {})))\n        model = model_cls(input_size=len(prep[\"features\"]), n_classes=n_cls).to(Config.DEVICE)\n        state = torch.load(model_path, map_location=Config.DEVICE)\n        model.load_state_dict(state)\n        model.eval()\n        loaded.append({\"model\": model, \"prep\": prep, \"fold\": k,\n                       \"variant\": label, \"dir\": str(base)})\n        print(f\"  [{label}] fold {k}  <-  {base}\")\n    return loaded\n\n\n# -----------------------------------------------------------------------------\n# Predict function — ensemble + Hungarian post-processing\n# -----------------------------------------------------------------------------\ndef make_predict_fn(entries, pred_to_label: dict):\n    \"\"\"\n    pred_to_label: {int → gesture_name}\n    Handles both 102-class (new) and 18-class (old categories list → dict).\n    \"\"\"\n    if not entries:\n        raise RuntimeError(\"No model artifacts found.\")\n\n    def predict(sequence: \"pl.DataFrame\", demographics: \"pl.DataFrame\") -> str:\n        import traceback\n        try:\n            seq_df  = sequence.to_pandas()\n            demo_df = demographics.to_pandas()\n\n            # ── Step 1: get ensemble probability vector ───────────────────\n            probs_all = []\n            subject   = None\n            for ent in entries:\n                prep  = ent[\"prep\"]\n                model = ent[\"model\"]\n                X_pad, eff_len, subj = _prepare_sequence(\n                    seq_df, demo_df,\n                    features=prep[\"features\"],\n                    max_length=prep[\"max_length\"],\n                )\n                if subject is None:\n                    subject = subj\n                ds     = CMIInferenceDataset(X_pad, eff_len)\n                loader = DataLoader(ds, batch_size=Config.BATCH_SIZE, num_workers=0)\n                with torch.no_grad():\n                    for batch_X, batch_len in loader:\n                        out   = model(batch_X.to(Config.DEVICE), batch_len.to(Config.DEVICE))\n                        probs = torch.softmax(out[\"gesture_logits\"], dim=1).cpu().numpy()\n                        probs_all.append(probs[0])\n\n            probs_mean = np.mean(probs_all, axis=0)   # (K,)\n\n            # ── Step 2: accumulate per-subject and run Hungarian ──────────\n            if subject not in _hungarian_subject_to_idx:\n                _hungarian_subject_to_idx[subject] = len(_hungarian_subject_to_idx)\n                _hungarian_logits_list.append([])\n\n            subj_idx = _hungarian_subject_to_idx[subject]\n            _hungarian_logits_list[subj_idx].append(probs_mean)   # append (K,)\n\n            pred_idx = solve_capacity1_with_hungarian(_hungarian_logits_list[subj_idx])\n            pred     = pred_to_label[pred_idx]\n\n            n_seqs = len(_hungarian_logits_list[subj_idx])\n            print(f\"[Predict] subject={subject}  seq#{n_seqs}  -> {pred}  \"\n                  f\"(confidence={probs_mean.max():.3f})\")\n            return pred\n\n        except Exception as e:\n            print(f\"[Predict Error] {e}\")\n            traceback.print_exc()\n            return pred_to_label[0]\n\n    return predict\n\n\n# -----------------------------------------------------------------------------\n# Main\n# -----------------------------------------------------------------------------\ndef main():\n    print(\"Scanning for artifacts...\")\n    phase_entries = _load_variant(\n        IMUModel,\n        prep_pattern=\"prep_phaseattn_fold_*.joblib\",\n        model_pattern=\"model_phaseattn_best_fold_{k}.pt\",\n        label=\"phase\",\n    )\n    norm_entries = _load_variant(\n        IMUModelNormalAttn,\n        prep_pattern=\"prep_normattn_fold_*.joblib\",\n        model_pattern=\"model_normattn_best_fold_{k}.pt\",\n        label=\"normattn\",\n    )\n    # all_entries = phase_entries + norm_entries\n    all_entries = phase_entries\n    if not all_entries:\n        raise FileNotFoundError(\n            \"No artifacts found in: \" + \", \".join(str(p) for p in SEARCH_DIRS)\n        )\n    print(f\"Loaded {len(phase_entries)} phase-aware + {len(norm_entries)} normal-attn \"\n          f\"= {len(all_entries)} models.\")\n\n    # Build pred_to_label from first entry (consistent across folds trained on same data)\n    first_prep = all_entries[0][\"prep\"]\n    if \"pred_to_label\" in first_prep:\n        pred_to_label = first_prep[\"pred_to_label\"]\n    else:\n        # Backward compat: old prep stored categories as a list\n        cats = first_prep[\"categories\"]\n        pred_to_label = {i: c for i, c in enumerate(cats)}\n    print(f\"Label mapping: {len(pred_to_label)} classes.\")\n\n    reset_hungarian_state()   # clear any leftover state before inference starts\n    return make_predict_fn(all_entries, pred_to_label)\n\n\nimport polars as pl\npredict_function = main()\ninference_server = kaggle_evaluation.cmi_inference_server.CMIInferenceServer(predict_function)\n\nif os.getenv(\"KAGGLE_IS_COMPETITION_RERUN\"):\n    inference_server.serve()\nelse:\n    test_csv_path  = pathlib.Path(\"/kaggle/input/competitions/cmi-detect-behavior-with-sensor-data/test.csv\")\n    test_demo_path = pathlib.Path(\"/kaggle/input/competitions/cmi-detect-behavior-with-sensor-data/test_demographics.csv\")\n    if test_csv_path.exists():\n        inference_server.run_local_gateway(data_paths=(test_csv_path, test_demo_path))\n    else:\n        print(\"Local test data not found.\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-25T12:28:37.307857Z","iopub.execute_input":"2026-06-25T12:28:37.308931Z","iopub.status.idle":"2026-06-25T12:29:01.217146Z","shell.execute_reply.started":"2026-06-25T12:28:37.308892Z","shell.execute_reply":"2026-06-25T12:29:01.216035Z"}},"outputs":[],"execution_count":null}]}