{"cells":[{"cell_type":"markdown","metadata":{},"source":"# CMI Ultra Advanced V2 Model Submission\\nOptimized for better generalization with 61.80% validation accuracy"},{"cell_type":"code","execution_count":null,"metadata":{},"outputs":[],"source":"import sys\\nsys.path.append('/kaggle/input/cmi-proto-lib')"},{"cell_type":"code","execution_count":null,"metadata":{},"outputs":[],"source":"# Install required packages\\n!pip install joblib scikit-learn==1.3.2 -q"},{"cell_type":"code","execution_count":null,"metadata":{},"outputs":[],"source":"import os\\nimport numpy as np\\nimport pandas as pd\\nimport torch\\nimport torch.nn as nn\\nimport torch.nn.functional as F\\nfrom sklearn.preprocessing import StandardScaler\\nimport joblib\\nfrom scipy.spatial.transform import Rotation as R\\n\\n# Import evaluation framework\\nfrom kaggle_evaluation import CMIInferenceServer, CMIGateway\\n\\n# Set device\\ndevice = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\\nprint(f'Using device: {device}')"},{"cell_type":"code","execution_count":null,"metadata":{},"outputs":[],"source":"# Model architecture - same as training\\nclass SEBlock(nn.Module):\\n    def __init__(self, channels, reduction=16):\\n        super().__init__()\\n        self.avg_pool = nn.AdaptiveAvgPool1d(1)\\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    \\n    def forward(self, x):\\n        b, c, t = x.size()\\n        y = self.avg_pool(x).view(b, c)\\n        y = self.excitation(y).view(b, c, 1)\\n        return x * y.expand_as(x)\\n\\nclass MultiScaleConv1d(nn.Module):\\n    def __init__(self, in_channels, out_channels, kernel_sizes=[3, 5, 7]):\\n        super().__init__()\\n        self.convs = nn.ModuleList()\\n        for ks in kernel_sizes:\\n            self.convs.append(nn.Sequential(\\n                nn.Conv1d(in_channels, out_channels, ks, padding=ks//2, bias=False),\\n                nn.BatchNorm1d(out_channels),\\n                nn.ReLU(inplace=True)\\n            ))\\n        \\n    def forward(self, x):\\n        outputs = [conv(x) for conv in self.convs]\\n        return torch.cat(outputs, dim=1)\\n\\nclass ResidualBlock(nn.Module):\\n    def __init__(self, in_channels, out_channels, kernel_size, pool_size=2, dropout=0.5):\\n        super().__init__()\\n        self.conv1 = nn.Conv1d(in_channels, out_channels, kernel_size, padding=kernel_size//2, bias=False)\\n        self.bn1 = nn.BatchNorm1d(out_channels)\\n        \\n        self.conv2 = nn.Conv1d(out_channels, out_channels, kernel_size, padding=kernel_size//2, bias=False)\\n        self.bn2 = nn.BatchNorm1d(out_channels)\\n        \\n        self.se = SEBlock(out_channels, reduction=16)\\n        \\n        self.shortcut = nn.Sequential()\\n        if in_channels != out_channels:\\n            self.shortcut = nn.Sequential(\\n                nn.Conv1d(in_channels, out_channels, 1, bias=False),\\n                nn.BatchNorm1d(out_channels)\\n            )\\n        \\n        self.pool = nn.MaxPool1d(pool_size)\\n        self.dropout = nn.Dropout(dropout)\\n        \\n    def forward(self, x):\\n        shortcut = self.shortcut(x)\\n        \\n        out = F.relu(self.bn1(self.conv1(x)))\\n        out = self.bn2(self.conv2(out))\\n        \\n        out = self.se(out)\\n        \\n        out += shortcut\\n        out = F.relu(out)\\n        \\n        out = self.pool(out)\\n        out = self.dropout(out)\\n        \\n        return out\\n\\nclass AttentionLayer(nn.Module):\\n    def __init__(self, hidden_dim):\\n        super().__init__()\\n        self.attention = nn.Linear(hidden_dim, 1)\\n        \\n    def forward(self, x):\\n        scores = torch.tanh(self.attention(x))\\n        weights = F.softmax(scores.squeeze(-1), dim=1)\\n        context = torch.sum(x * weights.unsqueeze(-1), dim=1)\\n        return context\\n\\nclass UltraAdvancedModelV2(nn.Module):\\n    def __init__(self, num_features, num_classes):\\n        super().__init__()\\n        \\n        self.multi_scale_cnn = MultiScaleConv1d(num_features, 32, kernel_sizes=[3, 5, 7])\\n        \\n        self.cnn_blocks = nn.Sequential(\\n            ResidualBlock(96, 64, 3, dropout=0.5),\\n            ResidualBlock(64, 128, 3, dropout=0.5),\\n            ResidualBlock(128, 128, 3, dropout=0.5)\\n        )\\n        \\n        self.bilstm = nn.LSTM(\\n            128, 128, \\n            num_layers=2, batch_first=True, \\n            bidirectional=True, dropout=0.4\\n        )\\n        \\n        self.bigru = nn.GRU(\\n            128, 128,\\n            num_layers=2, batch_first=True,\\n            bidirectional=True, dropout=0.4\\n        )\\n        \\n        rnn_hidden = 256 * 2\\n        self.attention_pooling = AttentionLayer(rnn_hidden)\\n        \\n        self.classifier = nn.Sequential(\\n            nn.Linear(rnn_hidden, 256),\\n            nn.BatchNorm1d(256),\\n            nn.ReLU(),\\n            nn.Dropout(0.5),\\n            nn.Linear(256, 128),\\n            nn.BatchNorm1d(128),\\n            nn.ReLU(),\\n            nn.Dropout(0.4),\\n            nn.Linear(128, num_classes)\\n        )\\n        \\n    def forward(self, x):\\n        batch_size = x.size(0)\\n        \\n        x = x.transpose(1, 2)\\n        x = self.multi_scale_cnn(x)\\n        x = self.cnn_blocks(x)\\n        x = x.transpose(1, 2)\\n        \\n        lstm_out, _ = self.bilstm(x)\\n        gru_out, _ = self.bigru(x)\\n        \\n        combined = torch.cat([lstm_out, gru_out], dim=2)\\n        pooled = self.attention_pooling(combined)\\n        output = self.classifier(pooled)\\n        \\n        return output"},{"cell_type":"code","execution_count":null,"metadata":{},"outputs":[],"source":"def remove_gravity_from_acc(acc_data, rot_data):\\n    \"\"\"Remove gravity component from accelerometer data using quaternions\"\"\"\\n    num_samples = acc_data.shape[0]\\n    linear_accel = np.zeros_like(acc_data)\\n    \\n    gravity_world = np.array([0, 0, 9.81])\\n    \\n    for i in range(num_samples):\\n        if np.all(np.isnan(rot_data[i])) or np.all(np.isclose(rot_data[i], 0)):\\n            linear_accel[i, :] = acc_data[i, :]\\n            continue\\n            \\n        try:\\n            rotation = R.from_quat(rot_data[i])\\n            gravity_sensor_frame = rotation.apply(gravity_world, inverse=True)\\n            linear_accel[i, :] = acc_data[i, :] - gravity_sensor_frame\\n        except ValueError:\\n            linear_accel[i, :] = acc_data[i, :]\\n    \\n    return linear_accel\\n\\ndef preprocess_sequence(seq_df, scaler, maxlen=103):\\n    \"\"\"Preprocess a single sequence\"\"\"\\n    # Remove gravity\\n    acc_data = seq_df[['acc_x', 'acc_y', 'acc_z']].values\\n    rot_data = seq_df[['rot_x', 'rot_y', 'rot_z', 'rot_w']].values\\n    linear_accel = remove_gravity_from_acc(acc_data, rot_data)\\n    \\n    seq_df = seq_df.copy()\\n    seq_df['linear_acc_x'] = linear_accel[:, 0]\\n    seq_df['linear_acc_y'] = linear_accel[:, 1]\\n    seq_df['linear_acc_z'] = linear_accel[:, 2]\\n    \\n    # TOF aggregation\\n    for i in range(1, 6):\\n        pixel_cols = [f\"tof_{i}_v{p}\" for p in range(64)]\\n        if all(col in seq_df.columns for col in pixel_cols):\\n            tof_data = seq_df[pixel_cols].replace(-1, np.nan)\\n            seq_df[f'tof_{i}_mean'] = tof_data.mean(axis=1)\\n        else:\\n            seq_df[f'tof_{i}_mean'] = 0\\n    \\n    # Select features\\n    imu_features = ['linear_acc_x', 'linear_acc_y', 'linear_acc_z',\\n                   'rot_x', 'rot_y', 'rot_z', 'rot_w']\\n    tof_features = [f'tof_{i}_mean' for i in range(1, 6)]\\n    all_features = imu_features + tof_features\\n    \\n    # Extract and normalize\\n    seq_data = seq_df[all_features].fillna(0).values\\n    seq_data = scaler.transform(seq_data)\\n    \\n    # Pad or truncate\\n    if len(seq_data) > maxlen:\\n        seq_data = seq_data[:maxlen]\\n    elif len(seq_data) < maxlen:\\n        pad_width = ((0, maxlen - len(seq_data)), (0, 0))\\n        seq_data = np.pad(seq_data, pad_width, mode='constant')\\n    \\n    return seq_data"},{"cell_type":"code","execution_count":null,"metadata":{},"outputs":[],"source":"# Load model and preprocessing components\\nprint('Loading model...')\\nmodel = UltraAdvancedModelV2(num_features=12, num_classes=18)\\nmodel.load_state_dict(torch.load('/kaggle/input/cmi-ultra-advanced-v2-model-6180/ultra_advanced_v2_model.pth'))\\nmodel = model.to(device)\\nmodel.eval()\\n\\n# Load preprocessing components\\nscaler = joblib.load('/kaggle/input/cmi-ultra-advanced-v2-model-6180/scaler_ultra_v2.pkl')\\ngesture_classes = np.load('/kaggle/input/cmi-ultra-advanced-v2-model-6180/gesture_classes_ultra_v2.npy', allow_pickle=True)\\nmaxlen = int(np.load('/kaggle/input/cmi-ultra-advanced-v2-model-6180/sequence_maxlen_ultra_v2.npy', allow_pickle=True))\\n\\nprint(f'Model loaded successfully!')\\nprint(f'Number of classes: {len(gesture_classes)}')\\nprint(f'Max sequence length: {maxlen}')"},{"cell_type":"code","execution_count":null,"metadata":{},"outputs":[],"source":"def predict(df: pd.DataFrame) -> str:\\n    \"\"\"Predict gesture from sensor data\"\"\"\\n    try:\\n        # Preprocess\\n        seq_data = preprocess_sequence(df, scaler, maxlen)\\n        \\n        # Convert to tensor\\n        x = torch.FloatTensor(seq_data).unsqueeze(0).to(device)\\n        \\n        # Predict\\n        with torch.no_grad():\\n            output = model(x)\\n            pred_idx = output.argmax(dim=1).cpu().numpy()[0]\\n        \\n        # Return gesture name\\n        return gesture_classes[pred_idx]\\n    \\n    except Exception as e:\\n        print(f\"Error in prediction: {e}\")\\n        # Return most common gesture as fallback\\n        return \"Forehead - scratch\""},{"cell_type":"code","execution_count":null,"metadata":{},"outputs":[],"source":"# Create gateway\\ngateway = CMIGateway(predict)\\n\\n# Create and run server\\nserver = CMIInferenceServer(gateway)\\n\\nprint(\"Starting inference server...\")\\nif os.environ.get('KAGGLE_IS_COMPETITION_RERUN'):\\n    server.serve()\\nelse:\\n    print(\"Running in test mode...\")"}],"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"codemirror_mode":{"name":"ipython","version":3},"file_extension":".py","mimetype":"text/x-python","name":"python","nbconvert_exporter":"python","pygments_lexer":"ipython3","version":"3.10.12"}},"nbformat":4,"nbformat_minor":5}