{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.11","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":102335,"databundleVersionId":12518947,"sourceType":"competition"}],"dockerImageVersionId":31041,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import os\nimport random\nimport math\nimport time\nimport warnings\nfrom typing import Dict, Any, Optional, List, Tuple\nfrom dataclasses import dataclass\nimport numpy as np\nimport pandas as pd\nfrom sklearn.preprocessing import StandardScaler, LabelEncoder\nfrom sklearn.model_selection import train_test_split, StratifiedKFold\nfrom sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score, classification_report\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom torch.utils.data import Dataset, DataLoader\nfrom torch.optim import AdamW, Adam\nfrom torch.optim.lr_scheduler import CosineAnnealingLR, ReduceLROnPlateau\nimport timm\nfrom tqdm.auto import tqdm\nimport matplotlib.pyplot as plt\nimport seaborn as sns\n\nwarnings.filterwarnings('ignore')\n\n# ================================\n# Configuration & Hyperparameters\n# ================================\n\n@dataclass\nclass Config:\n # Data paths\n kaggle_root: str = \"/kaggle/input/cmi-detect-behavior-with-sensor-data\"\n train_csv: str = \"train.csv\"\n test_csv: str = \"test.csv\"\n test_demo: str = \"test_demographics.csv\"\n \n # Model configuration\n model_name: str = \"resnet1d\"\n pretrained: bool = False\n \n # Training hyperparameters\n seed: int = 42\n batch_size: int = 64\n learning_rate: float = 2e-3\n max_epochs: int = 100\n patience: int = 15\n val_split: float = 0.2\n label_smoothing: float = 0.05\n weight_decay: float = 1e-3\n \n # Model specific hyperparameters\n hidden_dim: int = 128 # Reduced for smaller models\n num_layers: int = 2 # Reduced for smaller models\n dropout: float = 0.4\n \n # Data processing\n sequence_percentile: int = 85\n use_augmentation: bool = True\n augmentation_prob: float = 0.2\n \n # Cross-validation\n use_cv: bool = False\n cv_folds: int = 5\n \n # Hardware - Fixed for Kaggle\n device: str = \"cuda\" if torch.cuda.is_available() else \"cpu\"\n num_workers: int = 0 # Set to 0 to avoid multiprocessing issues in Kaggle\n pin_memory: bool = False # Disabled for Kaggle\n\ndef set_random_seed(seed: int):\n \"\"\"Set random seeds for reproducibility\"\"\"\n random.seed(seed)\n np.random.seed(seed)\n torch.manual_seed(seed)\n torch.cuda.manual_seed(seed)\n torch.cuda.manual_seed_all(seed)\n torch.backends.cudnn.deterministic = True\n torch.backends.cudnn.benchmark = False\n\ndef setup_environment(config: Config):\n \"\"\"Setup environment and configuration\"\"\"\n set_random_seed(config.seed)\n return config\n\n# ================================\n# Data Loading & Preprocessing\n# ================================\n\nclass GestureDataset(Dataset):\n def __init__(self, X: np.ndarray, y: np.ndarray, augment: bool = False, augment_prob: float = 0.3):\n self.X = X\n self.y = y\n self.augment = augment\n self.augment_prob = augment_prob\n \n def __len__(self):\n return len(self.X)\n \n def __getitem__(self, idx: int):\n x = torch.from_numpy(self.X[idx]).float()\n y = torch.tensor(self.y[idx], dtype=torch.long)\n \n if self.augment and random.random() < self.augment_prob:\n x = self._apply_augmentation(x)\n \n x = x.transpose(0, 1) # (n_features, seq_len) for Conv1d\n return x, y\n \n def _apply_augmentation(self, x: torch.Tensor) -> torch.Tensor:\n \"\"\"Apply data augmentation techniques\"\"\"\n # Gaussian noise\n if random.random() < 0.4:\n noise = torch.randn_like(x) * 0.005\n x = x + noise\n \n # Time shift\n if random.random() < 0.3:\n shift = random.randint(-3, 3)\n x = torch.roll(x, shift, dims=0)\n \n # Scaling\n if random.random() < 0.3:\n scale = random.uniform(0.95, 1.05)\n x = x * scale\n \n # Mixup on feature level\n if random.random() < 0.2:\n alpha = 0.1\n lam = np.random.beta(alpha, alpha)\n noise_features = torch.randn_like(x) * 0.01\n x = lam * x + (1 - lam) * noise_features\n \n return x\n\ndef clip_outliers(data: pd.DataFrame, columns: List[str]) -> pd.DataFrame:\n \"\"\"Clip outliers using quantile-based method\"\"\"\n for col in columns:\n q1, q99 = data[col].quantile([0.01, 0.99])\n data[col] = data[col].clip(q1, q99)\n return data\n\ndef handle_missing_values(data: pd.DataFrame) -> pd.DataFrame:\n \"\"\"Handle missing values with robust interpolation\"\"\"\n return data.interpolate(method='linear').ffill().bfill().fillna(0)\n\ndef normalize_features(data: pd.DataFrame) -> np.ndarray:\n \"\"\"Normalize features using StandardScaler\"\"\"\n scaler = StandardScaler()\n return scaler.fit_transform(data).astype(np.float32)\n\ndef preprocess_single_sequence(seq_df: pd.DataFrame, imu_cols: List[str]) -> np.ndarray:\n \"\"\"Preprocess a single sequence\"\"\"\n data = seq_df[imu_cols].copy()\n data = handle_missing_values(data)\n data = clip_outliers(data, imu_cols)\n return normalize_features(data)\n\ndef pad_or_truncate_sequence(arr: np.ndarray, maxlen: int, dtype=np.float32) -> np.ndarray:\n \"\"\"Pad or truncate array to fixed length\"\"\"\n t, c = arr.shape\n if t >= maxlen:\n return arr[:maxlen].astype(dtype, copy=False)\n out = np.zeros((maxlen, c), dtype=dtype)\n out[:t] = arr\n return out\n\ndef encode_labels(df: pd.DataFrame) -> Tuple[pd.DataFrame, LabelEncoder]:\n \"\"\"Encode gesture labels\"\"\"\n le = LabelEncoder()\n df[\"gesture\"] = le.fit_transform(df[\"gesture\"].astype(str))\n np.save(\"gesture_classes.npy\", le.classes_)\n return df, le\n\ndef calculate_padding_length(lengths: List[int], percentile: int) -> int:\n \"\"\"Calculate optimal padding length based on sequence lengths\"\"\"\n pad_len = int(np.percentile(lengths, percentile))\n print(f\"Padding/truncating to length: {pad_len}\")\n print(f\"Length statistics: min={np.min(lengths)}, max={np.max(lengths)}, mean={np.mean(lengths):.1f}\")\n np.save(\"sequence_maxlen.npy\", pad_len)\n return pad_len\n\ndef process_sequences(seq_groups, imu_cols: List[str]) -> Tuple[List[np.ndarray], List[int], List[int]]:\n \"\"\"Process all sequences\"\"\"\n X_list, y_list, lengths = [], [], []\n \n for seq_id, seq_df in tqdm(seq_groups, desc=\"Processing sequences\"):\n X_seq = preprocess_single_sequence(seq_df, imu_cols)\n X_list.append(X_seq)\n lengths.append(X_seq.shape[0])\n y_list.append(seq_df[\"gesture\"].iloc[0])\n \n return X_list, y_list, lengths\n\ndef load_and_process_data(config: Config):\n \"\"\"Load and preprocess the training data\"\"\"\n print(\"Loading and processing data...\")\n \n # Load data\n train_path = os.path.join(config.kaggle_root, config.train_csv)\n df = pd.read_csv(train_path)\n print(f\"Loaded {len(df):,} rows.\")\n \n # Encode labels\n df, le = encode_labels(df)\n \n # Define sensor columns\n imu_cols = [\"acc_x\", \"acc_y\", \"acc_z\", \"rot_w\", \"rot_x\", \"rot_y\", \"rot_z\"]\n \n # Process sequences\n print(\"Building sequences...\")\n seq_groups = df.groupby(\"sequence_id\")\n X_list, y_list, lengths = process_sequences(seq_groups, imu_cols)\n \n # Determine padding length\n pad_len = calculate_padding_length(lengths, config.sequence_percentile)\n \n # Pad sequences\n X = np.stack([pad_or_truncate_sequence(arr, pad_len) for arr in X_list])\n y = np.array(y_list, dtype=np.int64)\n \n print(f\"Final data shape: X={X.shape}, y={y.shape}\")\n print(f\"Number of classes: {len(le.classes_)}\")\n print(f\"Class distribution: {np.bincount(y)}\")\n \n return X, y, len(le.classes_), pad_len, le.classes_\n\n# ================================\n# Model Architectures - Simplified\n# ================================\n\nclass ResNet1DBlock(nn.Module):\n def __init__(self, in_channels: int, out_channels: int, kernel_size: int = 3, \n stride: int = 1, dropout: float = 0.3):\n super().__init__()\n self.conv1 = nn.Conv1d(in_channels, out_channels, kernel_size, stride, \n padding=kernel_size//2, bias=False)\n self.bn1 = nn.BatchNorm1d(out_channels)\n self.conv2 = nn.Conv1d(out_channels, out_channels, kernel_size, 1, \n padding=kernel_size//2, bias=False)\n self.bn2 = nn.BatchNorm1d(out_channels)\n self.dropout = nn.Dropout(dropout)\n \n # Skip connection\n self.skip = nn.Sequential()\n if stride != 1 or in_channels != out_channels:\n self.skip = nn.Sequential(\n nn.Conv1d(in_channels, out_channels, 1, stride, bias=False),\n nn.BatchNorm1d(out_channels)\n )\n \n def forward(self, x):\n residual = self.skip(x)\n out = F.relu(self.bn1(self.conv1(x)))\n out = self.dropout(out)\n out = self.bn2(self.conv2(out))\n out = self.dropout(out)\n return F.relu(out + residual)\n\nclass ResNet1D(nn.Module):\n def __init__(self, n_features: int, n_classes: int, channels: List[int] = [16, 32, 64],\n dropout: float = 0.3):\n super().__init__()\n self.conv1 = nn.Conv1d(n_features, channels[0], 7, 2, 3, bias=False)\n self.bn1 = nn.BatchNorm1d(channels[0])\n self.maxpool = nn.MaxPool1d(3, 2, 1)\n self.dropout = nn.Dropout(dropout)\n \n # ResNet blocks\n layers = []\n in_ch = channels[0]\n for i, out_ch in enumerate(channels):\n stride = 2 if i > 0 else 1\n layers.append(ResNet1DBlock(in_ch, out_ch, dropout=dropout))\n in_ch = out_ch\n \n self.layers = nn.Sequential(*layers)\n self.global_pool = nn.AdaptiveAvgPool1d(1)\n self.classifier = nn.Sequential(\n nn.Flatten(),\n nn.Dropout(dropout),\n nn.Linear(channels[-1], 32),\n nn.ReLU(),\n nn.Dropout(dropout),\n nn.Linear(32, n_classes)\n )\n\n def forward(self, x):\n x = F.relu(self.bn1(self.conv1(x)))\n x = self.maxpool(x)\n x = self.dropout(x)\n x = self.layers(x)\n x = self.global_pool(x)\n return self.classifier(x)\n\nclass EfficientNet1D(nn.Module):\n def __init__(self, n_features: int, n_classes: int, dropout: float = 0.3):\n super().__init__()\n # Simplified stem\n self.stem = nn.Sequential(\n nn.Conv1d(n_features, 16, 3, 2, 1, bias=False),\n nn.BatchNorm1d(16),\n nn.SiLU(),\n nn.Dropout(dropout)\n )\n \n # Simplified blocks\n self.blocks = nn.Sequential(\n self._make_simple_block(16, 32, 3, 1, dropout),\n self._make_simple_block(32, 64, 3, 2, dropout),\n )\n \n # Simplified head\n self.head = nn.Sequential(\n nn.AdaptiveAvgPool1d(1),\n nn.Flatten(),\n nn.Dropout(dropout),\n nn.Linear(64, 32),\n nn.ReLU(),\n nn.Dropout(dropout),\n nn.Linear(32, n_classes)\n )\n\n def _make_simple_block(self, in_ch: int, out_ch: int, kernel_size: int, \n stride: int, dropout: float):\n return nn.Sequential(\n nn.Conv1d(in_ch, out_ch, kernel_size, stride, kernel_size//2, bias=False),\n nn.BatchNorm1d(out_ch),\n nn.ReLU(),\n nn.Dropout(dropout)\n )\n \n def forward(self, x):\n x = self.stem(x)\n x = self.blocks(x)\n return self.head(x)\n\nclass LSTMAttention(nn.Module):\n def __init__(self, n_features: int, n_classes: int, hidden_dim: int = 64,\n num_layers: int = 1, dropout: float = 0.3):\n super().__init__()\n self.lstm = nn.LSTM(n_features, hidden_dim, num_layers, \n batch_first=True, dropout=dropout if num_layers > 1 else 0, \n bidirectional=True)\n \n # Simplified attention\n self.attention = nn.Sequential(\n nn.Linear(hidden_dim * 2, hidden_dim // 2),\n nn.Tanh(),\n nn.Dropout(dropout),\n nn.Linear(hidden_dim // 2, 1),\n nn.Softmax(dim=1)\n )\n \n self.classifier = nn.Sequential(\n nn.Dropout(dropout),\n nn.Linear(hidden_dim * 2, 32),\n nn.ReLU(),\n nn.Dropout(dropout),\n nn.Linear(32, n_classes)\n )\n \n def forward(self, x):\n x = x.transpose(1, 2) # (batch, seq_len, features)\n lstm_out, _ = self.lstm(x)\n \n # Attention weights\n attention_weights = self.attention(lstm_out)\n weighted_output = torch.sum(lstm_out * attention_weights, dim=1)\n \n return self.classifier(weighted_output)\n\nclass CNNLSTM(nn.Module):\n def __init__(self, n_features: int, n_classes: int, hidden_dim: int = 32,\n dropout: float = 0.4):\n super().__init__()\n # Minimal CNN feature extractor\n self.cnn = nn.Sequential(\n nn.Conv1d(n_features, 16, 3, padding=1),\n nn.BatchNorm1d(16),\n nn.ReLU(),\n nn.Dropout(dropout),\n nn.MaxPool1d(2),\n \n nn.Conv1d(16, 32, 3, padding=1),\n nn.BatchNorm1d(32),\n nn.ReLU(),\n nn.Dropout(dropout),\n nn.AdaptiveAvgPool1d(16)\n )\n \n # Single LSTM layer\n self.lstm = nn.LSTM(32, hidden_dim, 1, batch_first=True, bidirectional=True)\n \n # Classifier\n self.classifier = nn.Sequential(\n nn.Dropout(dropout),\n nn.Linear(hidden_dim * 2, 32),\n nn.ReLU(),\n nn.Dropout(dropout),\n nn.Linear(32, n_classes)\n )\n \n def forward(self, x):\n # CNN features\n cnn_out = self.cnn(x)\n cnn_out = cnn_out.transpose(1, 2) # (batch, seq_len, features)\n \n # LSTM\n lstm_out, (h_n, _) = self.lstm(cnn_out)\n # Use last hidden state from both directions\n output = torch.cat([h_n[-2], h_n[-1]], dim=1)\n \n return self.classifier(output)\n\nclass TCNBlock(nn.Module):\n def __init__(self, in_ch: int, out_ch: int, kernel_size: int, dilation: int, dropout: float):\n super().__init__()\n pad = (kernel_size - 1) * dilation\n self.conv1 = nn.Conv1d(in_ch, out_ch, kernel_size, padding=pad, dilation=dilation)\n self.bn1 = nn.BatchNorm1d(out_ch)\n self.conv2 = nn.Conv1d(out_ch, out_ch, kernel_size, padding=pad, dilation=dilation)\n self.bn2 = nn.BatchNorm1d(out_ch)\n self.dropout = nn.Dropout(dropout)\n self.downsample = nn.Conv1d(in_ch, out_ch, 1) if in_ch != out_ch else nn.Identity()\n self.crop = pad * 2\n \n def forward(self, x):\n out = F.relu(self.bn1(self.conv1(x)))\n out = self.dropout(out)\n out = self.bn2(self.conv2(out))\n out = self.dropout(out)\n res = self.downsample(x)\n out = out[:, :, :-self.crop] if self.crop > 0 else out\n res = res[:, :, -out.shape[2]:] if res.shape[2] > out.shape[2] else res\n return F.relu(out + res)\n\nclass TCN(nn.Module):\n def __init__(self, n_features: int, n_classes: int, channels: List[int] = [16, 32],\n kernel_size: int = 3, dropout: float = 0.4):\n super().__init__()\n layers = []\n in_ch = n_features\n for i, out_ch in enumerate(channels):\n layers.append(TCNBlock(in_ch, out_ch, kernel_size, dilation=2**i, dropout=dropout))\n in_ch = out_ch\n \n self.tcn = nn.Sequential(*layers)\n self.pool = nn.AdaptiveAvgPool1d(1)\n self.classifier = nn.Sequential(\n nn.Flatten(),\n nn.Dropout(dropout),\n nn.Linear(in_ch, 32),\n nn.ReLU(),\n nn.Dropout(dropout),\n nn.Linear(32, n_classes)\n )\n \n def forward(self, x):\n x = self.tcn(x)\n x = self.pool(x)\n return self.classifier(x)\n\ndef create_model(model_name: str, n_features: int, n_classes: int, config: Config):\n \"\"\"Factory function to create models\"\"\"\n models = {\n \"resnet1d\": lambda: ResNet1D(n_features, n_classes, dropout=config.dropout),\n \"efficientnet1d\": lambda: EfficientNet1D(n_features, n_classes, dropout=config.dropout),\n \"lstm_attention\": lambda: LSTMAttention(n_features, n_classes, \n hidden_dim=config.hidden_dim, \n num_layers=config.num_layers, \n dropout=config.dropout),\n \"cnn_lstm\": lambda: CNNLSTM(n_features, n_classes, \n hidden_dim=config.hidden_dim, \n dropout=config.dropout),\n \"tcn\": lambda: TCN(n_features, n_classes, dropout=config.dropout)\n }\n \n if model_name not in models:\n raise ValueError(f\"Unknown model: {model_name}. Available: {list(models.keys())}\")\n \n return models[model_name]()\n\n# ================================\n# Training Functions\n# ================================\n\ndef calculate_metrics(y_true: np.ndarray, y_pred: np.ndarray) -> Dict[str, float]:\n \"\"\"Calculate comprehensive metrics\"\"\"\n return {\n 'accuracy': accuracy_score(y_true, y_pred),\n 'f1_macro': f1_score(y_true, y_pred, average='macro'),\n 'f1_weighted': f1_score(y_true, y_pred, average='weighted'),\n 'precision_macro': precision_score(y_true, y_pred, average='macro', zero_division=0),\n 'recall_macro': recall_score(y_true, y_pred, average='macro', zero_division=0),\n 'precision_weighted': precision_score(y_true, y_pred, average='weighted', zero_division=0),\n 'recall_weighted': recall_score(y_true, y_pred, average='weighted', zero_division=0)\n }\n\nclass Trainer:\n def __init__(self, model: nn.Module, config: Config, num_classes: int):\n self.model = model.to(config.device)\n self.config = config\n self.num_classes = num_classes\n self.device = config.device\n \n # Loss and optimizer\n self.criterion = nn.CrossEntropyLoss(label_smoothing=config.label_smoothing)\n self.optimizer = AdamW(model.parameters(), lr=config.learning_rate, \n weight_decay=config.weight_decay, eps=1e-8)\n \n # Scheduler\n self.scheduler = ReduceLROnPlateau(self.optimizer, mode='max', factor=0.7,\n patience=5, verbose=True, min_lr=1e-6)\n \n # Metrics tracking\n self.train_losses = []\n self.val_losses = []\n self.train_metrics = []\n self.val_metrics = []\n self.best_val_f1 = 0.0\n self.patience_counter = 0\n \n def train_single_epoch(self, train_loader: DataLoader) -> Tuple[float, Dict[str, float]]:\n \"\"\"Train for one epoch\"\"\"\n self.model.train()\n total_loss = 0.0\n all_preds = []\n all_targets = []\n \n pbar = tqdm(train_loader, desc=\"Training\")\n for batch_idx, (data, target) in enumerate(pbar):\n data, target = data.to(self.device), target.to(self.device)\n \n self.optimizer.zero_grad()\n output = self.model(data)\n loss = self.criterion(output, target)\n \n loss.backward()\n torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=1.0)\n self.optimizer.step()\n \n total_loss += loss.item()\n pred = output.argmax(dim=1)\n all_preds.extend(pred.cpu().numpy())\n all_targets.extend(target.cpu().numpy())\n \n # Update progress bar\n if len(all_targets) > 0:\n current_acc = accuracy_score(all_targets, all_preds)\n pbar.set_postfix({\n 'Loss': f'{loss.item():.4f}',\n 'Acc': f'{current_acc:.3f}',\n 'LR': f'{self.optimizer.param_groups[0][\"lr\"]:.1e}'\n })\n \n avg_loss = total_loss / len(train_loader)\n metrics = calculate_metrics(np.array(all_targets), np.array(all_preds))\n return avg_loss, metrics\n \n def validate_single_epoch(self, val_loader: DataLoader) -> Tuple[float, Dict[str, float]]:\n \"\"\"Validate for one epoch\"\"\"\n self.model.eval()\n total_loss = 0.0\n all_preds = []\n all_targets = []\n \n with torch.no_grad():\n pbar = tqdm(val_loader, desc=\"Validation\")\n for data, target in pbar:\n data, target = data.to(self.device), target.to(self.device)\n output = self.model(data)\n loss = self.criterion(output, target)\n \n total_loss += loss.item()\n pred = output.argmax(dim=1)\n all_preds.extend(pred.cpu().numpy())\n all_targets.extend(target.cpu().numpy())\n \n if len(all_targets) > 0:\n current_acc = accuracy_score(all_targets, all_preds)\n current_f1 = f1_score(all_targets, all_preds, average='weighted')\n pbar.set_postfix({\n 'Loss': f'{loss.item():.4f}',\n 'Acc': f'{current_acc:.3f}',\n 'F1': f'{current_f1:.3f}'\n })\n \n avg_loss = total_loss / len(val_loader)\n metrics = calculate_metrics(np.array(all_targets), np.array(all_preds))\n return avg_loss, metrics\n \n def save_checkpoint(self, epoch: int, is_best: bool = False):\n \"\"\"Save model checkpoint\"\"\"\n checkpoint_data = {\n 'model_state_dict': self.model.state_dict(),\n 'optimizer_state_dict': self.optimizer.state_dict(),\n 'scheduler_state_dict': self.scheduler.state_dict(),\n 'epoch': epoch,\n 'best_val_f1': self.best_val_f1,\n 'config': self.config,\n 'train_losses': self.train_losses,\n 'val_losses': self.val_losses,\n 'train_metrics': self.train_metrics,\n 'val_metrics': self.val_metrics\n }\n \n filename = f\"best_{self.config.model_name}_checkpoint.pth\"\n torch.save(checkpoint_data, filename)\n return filename\n \n def check_early_stopping(self, val_metrics: Dict[str, float], epoch: int) -> bool:\n \"\"\"Check if early stopping should be triggered\"\"\"\n current_f1 = val_metrics['f1_weighted']\n \n if current_f1 > self.best_val_f1:\n self.best_val_f1 = current_f1\n self.best_val_acc = val_metrics['accuracy']\n self.patience_counter = 0\n \n self.save_checkpoint(epoch, is_best=True)\n print(f\"ā˜… New best F1: {current_f1:.4f} (Acc: {val_metrics['accuracy']:.4f}) - Model saved!\")\n return False\n else:\n self.patience_counter += 1\n print(f\"No improvement for {self.patience_counter}/{self.config.patience} epochs\")\n \n if self.patience_counter >= self.config.patience:\n print(\"Early stopping triggered!\")\n return True\n \n return False\n \n def print_epoch_results(self, epoch: int, train_loss: float, train_metrics: Dict[str, float],\n val_loss: float, val_metrics: Dict[str, float]):\n \"\"\"Print results for current epoch\"\"\"\n print(f\"\\nEpoch {epoch+1}/{self.config.max_epochs}\")\n print(\"-\" * 70)\n print(f\"Train - Loss: {train_loss:.4f}, Acc: {train_metrics['accuracy']:.4f}, \"\n f\"F1: {train_metrics['f1_weighted']:.4f}\")\n print(f\"Val - Loss: {val_loss:.4f}, Acc: {val_metrics['accuracy']:.4f}, \"\n f\"F1: {val_metrics['f1_weighted']:.4f}\")\n print(f\"Val Precision: {val_metrics['precision_weighted']:.4f}, \"\n f\"Recall: {val_metrics['recall_weighted']:.4f}\")\n print(f\"Learning Rate: {self.optimizer.param_groups[0]['lr']:.6f}\")\n \n def load_best_model(self):\n \"\"\"Load the best saved model\"\"\"\n checkpoint_path = f\"best_{self.config.model_name}_checkpoint.pth\"\n if os.path.exists(checkpoint_path):\n checkpoint = torch.load(checkpoint_path, weights_only=False)\n self.model.load_state_dict(checkpoint['model_state_dict'])\n print(f\"Loaded best model from {checkpoint_path}\")\n \n def fit(self, train_loader: DataLoader, val_loader: DataLoader):\n \"\"\"Main training loop\"\"\"\n print(f\"Starting training for {self.config.max_epochs} epochs...\")\n print(f\"Model parameters: {sum(p.numel() for p in self.model.parameters()):,}\")\n \n for epoch in range(self.config.max_epochs):\n # Training\n train_loss, train_metrics = self.train_single_epoch(train_loader)\n self.train_losses.append(train_loss)\n self.train_metrics.append(train_metrics)\n \n # Validation\n val_loss, val_metrics = self.validate_single_epoch(val_loader)\n self.val_losses.append(val_loss)\n self.val_metrics.append(val_metrics)\n \n # Learning rate scheduling\n self.scheduler.step(val_metrics['f1_weighted'])\n \n# Check early stopping\n if self.check_early_stopping(val_metrics, epoch):\n break\n \n print(f\"\\nTraining completed!\")\n print(f\"Best validation F1: {self.best_val_f1:.4f}\")\n print(f\"Best validation Accuracy: {self.best_val_acc:.4f}\")\n \n # Load best model\n self.load_best_model()\n \n def plot_history(self):\n \"\"\"Plot training history\"\"\"\n fig, axes = plt.subplots(2, 2, figsize=(15, 10))\n \n # Loss curves\n axes[0, 0].plot(self.train_losses, label='Train Loss', color='blue')\n axes[0, 0].plot(self.val_losses, label='Val Loss', color='red')\n axes[0, 0].set_title('Training and Validation Loss')\n axes[0, 0].set_xlabel('Epoch')\n axes[0, 0].set_ylabel('Loss')\n axes[0, 0].legend()\n axes[0, 0].grid(True)\n \n # Accuracy curves\n train_accs = [m['accuracy'] for m in self.train_metrics]\n val_accs = [m['accuracy'] for m in self.val_metrics]\n axes[0, 1].plot(train_accs, label='Train Accuracy', color='blue')\n axes[0, 1].plot(val_accs, label='Val Accuracy', color='red')\n axes[0, 1].set_title('Training and Validation Accuracy')\n axes[0, 1].set_xlabel('Epoch')\n axes[0, 1].set_ylabel('Accuracy')\n axes[0, 1].legend()\n axes[0, 1].grid(True)\n \n # F1 curves\n train_f1s = [m['f1_weighted'] for m in self.train_metrics]\n val_f1s = [m['f1_weighted'] for m in self.val_metrics]\n axes[1, 0].plot(train_f1s, label='Train F1', color='blue')\n axes[1, 0].plot(val_f1s, label='Val F1', color='red')\n axes[1, 0].set_title('Training and Validation F1 Score')\n axes[1, 0].set_xlabel('Epoch')\n axes[1, 0].set_ylabel('F1 Score')\n axes[1, 0].legend()\n axes[1, 0].grid(True)\n \n # Learning rate\n if hasattr(self.scheduler, 'get_last_lr'):\n lrs = [group['lr'] for group in self.optimizer.param_groups]\n axes[1, 1].plot(lrs, color='green')\n axes[1, 1].set_title('Learning Rate Schedule')\n axes[1, 1].set_xlabel('Epoch')\n axes[1, 1].set_ylabel('Learning Rate')\n axes[1, 1].set_yscale('log')\n axes[1, 1].grid(True)\n else:\n axes[1, 1].text(0.5, 0.5, 'Learning Rate\\nSchedule\\nNot Available', \n ha='center', va='center', transform=axes[1, 1].transAxes)\n \n plt.tight_layout()\n plt.savefig(f'{self.config.model_name}_training_history.png', dpi=300, bbox_inches='tight')\n plt.show()\n\n# ================================\n# Main Training Pipeline\n# ================================\n\ndef main():\n \"\"\"Main training pipeline\"\"\"\n print(\"šŸš€ Starting Gesture Recognition Training Pipeline\")\n print(\"=\" * 60)\n \n # Setup environment\n config = setup_environment(Config())\n \n # Load and process data\n print(\"šŸ“Š Loading and processing data...\")\n X, y, num_classes, seq_len, class_names = load_and_process_data(config)\n \n # Train-validation split\n X_train, X_val, y_train, y_val = train_test_split(\n X, y, test_size=config.val_split, random_state=config.seed, stratify=y\n )\n \n print(f\"šŸ“ˆ Dataset Statistics:\")\n print(f\" Train set: {X_train.shape[0]} samples\")\n print(f\" Validation set: {X_val.shape[0]} samples\")\n print(f\" Sequence length: {seq_len}\")\n print(f\" Features: {X.shape[2]}\")\n print(f\" Classes: {num_classes}\")\n print(f\" Class names: {class_names}\")\n \n # Create datasets and dataloaders\n train_dataset = GestureDataset(X_train, y_train, \n augment=config.use_augmentation, \n augment_prob=config.augmentation_prob)\n val_dataset = GestureDataset(X_val, y_val, augment=False)\n \n train_loader = DataLoader(\n train_dataset, \n batch_size=config.batch_size, \n shuffle=True,\n num_workers=config.num_workers, \n pin_memory=config.pin_memory,\n drop_last=True\n )\n \n val_loader = DataLoader(\n val_dataset, \n batch_size=config.batch_size, \n shuffle=False,\n num_workers=config.num_workers, \n pin_memory=config.pin_memory\n )\n \n # Create model\n print(f\"šŸ—ļø Creating {config.model_name} model...\")\n model = create_model(config.model_name, X.shape[2], num_classes, config)\n print(f\" Model parameters: {sum(p.numel() for p in model.parameters()):,}\")\n \n # Create trainer and start training\n print(\"šŸŽÆ Starting training...\")\n trainer = Trainer(model, config, num_classes)\n trainer.fit(train_loader, val_loader)\n \n # Plot training history\n print(\"šŸ“Š Generating training plots...\")\n trainer.plot_history()\n \n # Save final artifacts\n print(\"šŸ’¾ Saving final artifacts...\")\n final_checkpoint = {\n 'model_state_dict': trainer.model.state_dict(),\n 'config': config,\n 'class_names': class_names,\n 'num_classes': num_classes,\n 'sequence_length': seq_len,\n 'best_val_f1': trainer.best_val_f1,\n 'best_val_acc': trainer.best_val_acc,\n 'final_train_metrics': trainer.train_metrics[-1] if trainer.train_metrics else None,\n 'final_val_metrics': trainer.val_metrics[-1] if trainer.val_metrics else None,\n 'training_history': {\n 'train_losses': trainer.train_losses,\n 'val_losses': trainer.val_losses,\n 'train_metrics': trainer.train_metrics,\n 'val_metrics': trainer.val_metrics\n }\n }\n \n torch.save(final_checkpoint, f\"final_{config.model_name}_checkpoint.pth\")\n \n # Save additional artifacts\n np.save(f\"{config.model_name}_training_losses.npy\", trainer.train_losses)\n np.save(f\"{config.model_name}_validation_losses.npy\", trainer.val_losses)\n \n print(\"āœ… Training completed and artifacts saved!\")\n print(f\" Best validation F1: {trainer.best_val_f1:.4f}\")\n print(f\" Best validation Accuracy: {trainer.best_val_acc:.4f}\")\n \n return trainer, final_checkpoint\n\n# ================================\n# Model Comparison Utility\n# ================================\n\ndef compare_models(models_to_test: List[str] = [\"resnet1d\", \"efficientnet1d\", \"lstm_attention\", \"tcn\"]):\n \"\"\"Compare different model architectures\"\"\"\n print(\"šŸ” Starting Model Comparison\")\n print(\"=\" * 60)\n \n results = {}\n config = setup_environment(Config())\n \n # Load data once\n print(\"šŸ“Š Loading data for comparison...\")\n X, y, num_classes, seq_len, class_names = load_and_process_data(config)\n X_train, X_val, y_train, y_val = train_test_split(\n X, y, test_size=config.val_split, random_state=config.seed, stratify=y\n )\n \n # Create datasets\n train_dataset = GestureDataset(X_train, y_train, \n augment=config.use_augmentation,\n augment_prob=config.augmentation_prob)\n val_dataset = GestureDataset(X_val, y_val, augment=False)\n \n train_loader = DataLoader(train_dataset, batch_size=config.batch_size, \n shuffle=True, num_workers=config.num_workers)\n val_loader = DataLoader(val_dataset, batch_size=config.batch_size, \n shuffle=False, num_workers=config.num_workers)\n \n # Test each model\n for i, model_name in enumerate(models_to_test):\n print(f\"\\n{'='*50}\")\n print(f\"Testing {model_name.upper()} ({i+1}/{len(models_to_test)})\")\n print('='*50)\n \n # Update config for current model\n original_model_name = config.model_name\n config.model_name = model_name\n \n try:\n # Create and train model\n model = create_model(model_name, X.shape[2], num_classes, config)\n trainer = Trainer(model, config, num_classes)\n \n print(f\"Model parameters: {sum(p.numel() for p in model.parameters()):,}\")\n \n # Train with reduced epochs for comparison\n original_epochs = config.max_epochs\n config.max_epochs = min(50, config.max_epochs) # Limit epochs for comparison\n \n trainer.fit(train_loader, val_loader)\n \n # Store results\n results[model_name] = {\n 'best_val_f1': trainer.best_val_f1,\n 'best_val_acc': trainer.best_val_acc,\n 'final_train_acc': trainer.train_metrics[-1]['accuracy'] if trainer.train_metrics else 0,\n 'final_val_acc': trainer.val_metrics[-1]['accuracy'] if trainer.val_metrics else 0,\n 'final_train_f1': trainer.train_metrics[-1]['f1_weighted'] if trainer.train_metrics else 0,\n 'final_val_f1': trainer.val_metrics[-1]['f1_weighted'] if trainer.val_metrics else 0,\n 'num_params': sum(p.numel() for p in model.parameters()),\n 'epochs_trained': len(trainer.train_losses)\n }\n \n # Save individual model\n torch.save({\n 'model_state_dict': trainer.model.state_dict(),\n 'results': results[model_name],\n 'config': config\n }, f\"comparison_{model_name}_checkpoint.pth\")\n \n # Restore original epochs\n config.max_epochs = original_epochs\n \n except Exception as e:\n print(f\"āŒ Error training {model_name}: {str(e)}\")\n results[model_name] = {\n 'error': str(e),\n 'best_val_f1': 0,\n 'best_val_acc': 0,\n 'num_params': 0\n }\n \n # Restore original model name\n config.model_name = original_model_name\n \n # Print comparison results\n print(f\"\\n{'='*80}\")\n print(\"šŸ† MODEL COMPARISON RESULTS\")\n print('='*80)\n print(f\"{'Model':<15} {'Best F1':<10} {'Best Acc':<10} {'Params':<12} {'Final F1':<10} {'Epochs':<8}\")\n print('-'*80)\n \n # Sort by best F1 score\n sorted_results = sorted(results.items(), key=lambda x: x[1].get('best_val_f1', 0), reverse=True)\n \n for model_name, metrics in sorted_results:\n if 'error' not in metrics:\n print(f\"{model_name:<15} {metrics['best_val_f1']:<10.4f} \"\n f\"{metrics['best_val_acc']:<10.4f} {metrics['num_params']:<12,} \"\n f\"{metrics['final_val_f1']:<10.4f} {metrics['epochs_trained']:<8}\")\n else:\n print(f\"{model_name:<15} {'ERROR':<10} {'ERROR':<10} {'ERROR':<12} {'ERROR':<10} {'ERROR':<8}\")\n \n # Save comparison results\n comparison_results = {\n 'results': results,\n 'config_used': config,\n 'dataset_info': {\n 'num_classes': num_classes,\n 'sequence_length': seq_len,\n 'train_samples': len(X_train),\n 'val_samples': len(X_val),\n 'class_names': class_names\n }\n }\n \n torch.save(comparison_results, \"model_comparison_results.pth\")\n \n # Create comparison plot\n plt.figure(figsize=(12, 8))\n \n # Plot F1 scores\n plt.subplot(2, 2, 1)\n models = [name for name, res in sorted_results if 'error' not in res]\n f1_scores = [res['best_val_f1'] for name, res in sorted_results if 'error' not in res]\n plt.bar(models, f1_scores, color='skyblue')\n plt.title('Best Validation F1 Score by Model')\n plt.xticks(rotation=45)\n plt.ylabel('F1 Score')\n \n # Plot accuracy\n plt.subplot(2, 2, 2)\n acc_scores = [res['best_val_acc'] for name, res in sorted_results if 'error' not in res]\n plt.bar(models, acc_scores, color='lightgreen')\n plt.title('Best Validation Accuracy by Model')\n plt.xticks(rotation=45)\n plt.ylabel('Accuracy')\n \n # Plot parameters\n plt.subplot(2, 2, 3)\n param_counts = [res['num_params'] for name, res in sorted_results if 'error' not in res]\n plt.bar(models, param_counts, color='coral')\n plt.title('Model Parameters Count')\n plt.xticks(rotation=45)\n plt.ylabel('Parameters')\n plt.yscale('log')\n \n # Plot efficiency (F1 / params)\n plt.subplot(2, 2, 4)\n efficiency = [f1/params*1000000 for f1, params in zip(f1_scores, param_counts) if params > 0]\n plt.bar(models, efficiency, color='gold')\n plt.title('Model Efficiency (F1 Score per Million Parameters)')\n plt.xticks(rotation=45)\n plt.ylabel('Efficiency')\n \n plt.tight_layout()\n plt.savefig('model_comparison.png', dpi=300, bbox_inches='tight')\n plt.show()\n \n print(f\"\\nāœ… Model comparison completed!\")\n print(f\" Best model: {sorted_results[0][0]} (F1: {sorted_results[0][1]['best_val_f1']:.4f})\")\n \n return results\n\n# ================================\n# Inference Functions\n# ================================\n\ndef load_trained_model(checkpoint_path: str, device: str = None):\n \"\"\"Load a trained model from checkpoint\"\"\"\n if device is None:\n device = \"cuda\" if torch.cuda.is_available() else \"cpu\"\n \n checkpoint = torch.load(checkpoint_path, map_location=device, weights_only=False)\n config_used = checkpoint['config']\n \n # Create model\n model = create_model(\n config_used.model_name, \n 7, # Number of IMU features\n checkpoint['num_classes'], \n config_used\n )\n \n model.load_state_dict(checkpoint['model_state_dict'])\n model.to(device)\n model.eval()\n \n return model, checkpoint\n\ndef predict_gesture(model, sequence_data: np.ndarray, class_names: List[str], device: str = \"cpu\"):\n \"\"\"Predict gesture from sequence data\"\"\"\n model.eval()\n with torch.no_grad():\n # Preprocess sequence\n if len(sequence_data.shape) == 2:\n sequence_data = sequence_data[np.newaxis, :] # Add batch dimension\n \n x = torch.from_numpy(sequence_data).float().to(device)\n x = x.transpose(1, 2) # (batch, features, seq_len)\n \n # Get predictions\n outputs = model(x)\n probabilities = F.softmax(outputs, dim=1)\n predicted_class = outputs.argmax(dim=1).item()\n confidence = probabilities[0][predicted_class].item()\n \n return {\n 'predicted_class': predicted_class,\n 'predicted_gesture': class_names[predicted_class],\n 'confidence': confidence,\n 'all_probabilities': probabilities[0].cpu().numpy()\n }\n\n# ================================\n# Run Training\n# ================================\n\nif __name__ == \"__main__\":\n # Setup environment\n config = setup_environment(Config())\n \n print(\"šŸŽÆ Gesture Recognition Training\")\n print(f\"Device: {config.device}\")\n print(f\"Model: {config.model_name}\")\n print(f\"Batch size: {config.batch_size}\")\n print(f\"Learning rate: {config.learning_rate}\")\n print(f\"Max epochs: {config.max_epochs}\")\n \n # Run main training\n trainer, checkpoint = main()\n \n # Optionally run model comparison\n print(\"\\n\" + \"=\"*60)\n comparison_results = compare_models()\n \n print(\"\\nšŸŽ‰ All done! Check the saved artifacts:\")\n print(\" - final_*_checkpoint.pth (best model)\")\n print(\" - *_training_history.png (training plots)\")\n print(\" - gesture_classes.npy (class names)\")\n print(\" - sequence_maxlen.npy (sequence length)\")\n if 'comparison_results' in locals():\n print(\" - model_comparison_results.pth (comparison results)\")\n print(\" - model_comparison.png (comparison plots)\")","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-06-08T01:44:49.772209Z","iopub.execute_input":"2025-06-08T01:44:49.772408Z","iopub.status.idle":"2025-06-08T01:45:05.614224Z","shell.execute_reply.started":"2025-06-08T01:44:49.772391Z","shell.execute_reply":"2025-06-08T01:45:05.613587Z"}},"outputs":[],"execution_count":null}]}