{"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"},{"sourceId":12139340,"sourceType":"datasetVersion","datasetId":7645099},{"sourceId":12293285,"sourceType":"datasetVersion","datasetId":7748073},{"sourceId":12411879,"sourceType":"datasetVersion","datasetId":7827797},{"sourceId":12552525,"sourceType":"datasetVersion","datasetId":7925699},{"sourceId":12553874,"sourceType":"datasetVersion","datasetId":7926703},{"sourceId":12579285,"sourceType":"datasetVersion","datasetId":7944408},{"sourceId":242954653,"sourceType":"kernelVersion"},{"sourceId":246893721,"sourceType":"kernelVersion"},{"sourceId":252267585,"sourceType":"kernelVersion"}],"dockerImageVersionId":31041,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"

\n\nOnlyInfKernel\n

","metadata":{}},{"cell_type":"code","source":"fs = 10 # do not change this\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"### **ℹ️INFO**\n* First, we would like to thank our participants for sharing their excellent baselines.\n * [Two‑Branch Human‑Activity‑Recognition Pipeline (IMU + Thermopile/TOF + SE‑CNN + BiLSTM + Attentio](https://www.kaggle.com/code/vonmainstein/imu-tof)\n\n### **ℹ️IMU+THM/TOF Great Related works(Published order)**\n* Thanks for sharing your implementation of the new feature.\n * LB.75 [CMI25 | IMU+THM/TOF |TF BiLSTM+GRU+Attention|LB.75](https://www.kaggle.com/code/hideyukizushi/cmi25-imu-thm-tof-tf-bilstm-gru-attention-lb-75)\n * LB.76 [IMU Signal Processing Optimization](https://www.kaggle.com/code/rktqwe/lb-0-76-imu-thm-tof-tf-bilstm-gru-attention)\n * LB.77 [Gravity Component Removal from Accelerometer Data](https://www.kaggle.com/code/rktqwe/lb-0-77-linear-accel-tf-bilstm-gru-attention)\n * LB.78 [New Feature: Angular Velocity from Quaternion Derivatives](https://www.kaggle.com/code/nksusth/lb-0-78-quaternions-tf-bilstm-gru-attention)\n * LB.80 [IMU × TOF/THM Fusion with Physics‑FE & MixUp|LB0.8](https://www.kaggle.com/code/nksusth/lb-0-78-quaternions-tf-bilstm-gru-attention)\n\n---\n\n### **ℹ️[LB.80 2025/06/28]MyUpdate**\n* In previous notebooks that used \"IMU+TOF\" as features, the validation strategy was TTS, which raised concerns about its robustness.\nFor this reason, we will release an implementation that simply blends five models and a model with a high ValidationScore.\n\n```\npredictions = []\nfor model in models:\n idx = int(model.predict(pad_input, verbose=0).argmax(1)[0])\n predictions.append(idx)\n\nidx = max(set(predictions), key=predictions.count)\nreturn str(gesture_classes[idx])\n```\n\n* The validationScore for the local training model included in this notebook is\n * [ModelWeight](https://www.kaggle.com/datasets/hideyukizushi/20250627-cmi-b-102-b-105)\n\n```\n0.891134700273056\n0.8912659261884439\n0.8914825129445727\n0.8915471835009202\n0.8922128108549205\n```\n\n### **ℹ️[LB.xx 2025/07/09]MyUpdate**\n* {WIP-AfterScoring}","metadata":{}},{"cell_type":"code","source":"import torch\ndeviceGPU = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"

\n\nModel Detail\n

","metadata":{}},{"cell_type":"code","source":"from PIL import Image\nim1 = Image.open('/kaggle/input/cmi25-imu-thmtof-tf-bilstm-gru-attentionlb-xx/CMI-Model.png')\ndef get_concat_h_multi_resize(im_list, resample=Image.BICUBIC):\n min_height = min(im.height for im in im_list)\n im_list_resize = [im.resize((int(im.width * min_height / im.height), min_height),resample=resample)\n for im in im_list]\n total_width = sum(im.width for im in im_list_resize)\n dst = Image.new('RGB', (total_width, min_height))\n pos_x = 0\n for im in im_list_resize:\n dst.paste(im, (pos_x, 0))\n pos_x += im.width\n return dst\nget_concat_h_multi_resize([im1])","metadata":{"trusted":true,"_kg_hide-input":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"

\n\nInfPipeline\n

","metadata":{}},{"cell_type":"markdown","source":"# 》》》**Importing the necessary Libraries**","metadata":{}},{"cell_type":"code","source":"import os, json, joblib, numpy as np, pandas as pd\nfrom pathlib import Path\nimport warnings \nwarnings.filterwarnings(\"ignore\")\n\n\nfrom sklearn.model_selection import train_test_split\nfrom sklearn.preprocessing import StandardScaler, LabelEncoder\nfrom sklearn.utils.class_weight import compute_class_weight\n\nfrom tensorflow.keras.utils import Sequence, to_categorical, pad_sequences\nfrom tensorflow.keras.models import Model, load_model\nfrom tensorflow.keras.layers import (\n Input, Conv1D, BatchNormalization, Activation, add, MaxPooling1D, Dropout,\n Bidirectional, LSTM, GlobalAveragePooling1D, Dense, Multiply, Reshape,\n Lambda, Concatenate, GRU, GaussianNoise\n)\nfrom tensorflow.keras.regularizers import l2\nfrom tensorflow.keras.optimizers import Adam\nfrom tensorflow.keras.callbacks import EarlyStopping\nfrom tensorflow.keras import backend as K\nimport tensorflow as tf\nimport polars as pl\nfrom sklearn.model_selection import StratifiedGroupKFold\nfrom scipy.spatial.transform import Rotation as R\n","metadata":{"trusted":true,"_kg_hide-output":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## importing torch","metadata":{}},{"cell_type":"code","source":"import os, json, joblib, numpy as np, pandas as pd\nfrom pathlib import Path\nimport warnings \nwarnings.filterwarnings(\"ignore\")\n\nfrom datetime import datetime\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nimport torch.nn.functional as F\nfrom torch.utils.data import Dataset, DataLoader\nfrom torch import device\n# For model saving/loading\nimport os\n\n# For creating custom data generators (like Sequence in Keras)\nfrom torch.optim.lr_scheduler import CosineAnnealingWarmRestarts\n# No direct equivalent for Keras' to_categorical, pad_sequences:\n# You will need to handle these manually in your dataset or data loader.\nfrom torch.utils.data import TensorDataset\n\n# No direct GaussianNoise in PyTorch, you can implement it manually as shown earlier.\n\n\n\nimport polars as pl\nfrom sklearn.model_selection import StratifiedGroupKFold\nfrom scipy.spatial.transform import Rotation as R\nfrom sklearn.utils.class_weight import compute_class_weight\nfrom sklearn.preprocessing import StandardScaler, LabelEncoder\n\n\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"","metadata":{}},{"cell_type":"markdown","source":"# 》》》**Fix Seed**","metadata":{}},{"cell_type":"code","source":"import random\ndef seed_everything(seed):\n os.environ['PYTHONHASHSEED'] = str(seed)\n random.seed(seed)\n np.random.seed(seed)\n tf.random.set_seed(seed)\n tf.experimental.numpy.random.seed(seed)\n torch.manual_seed(seed)\n torch.cuda.manual_seed_all(seed) # 如果有多张GPU\n torch.backends.cudnn.deterministic = True\n torch.backends.cudnn.benchmark = False\n os.environ['TF_CUDNN_DETERMINISTIC'] = '1'\n os.environ['TF_DETERMINISTIC_OPS'] = '1'\nseed_everything(seed=42)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 》》》**Configuration**","metadata":{}},{"cell_type":"code","source":"# (Competition metric will only be imported when TRAINing)\nTRAIN = False # ← set to True when you want to train\nRAW_DIR = Path(\"/kaggle/input/cmi-detect-behavior-with-sensor-data\")\nPRETRAINED_DIR = Path(\"/kaggle/input/cmi-d-111\")\nEXPORT_DIR = Path(\"./\") # artefacts will be saved here\nBATCH_SIZE = 64\nPAD_PERCENTILE = 95\nLR_INIT = 5e-4\nWD = 3e-3\nMIXUP_ALPHA = 0.4\nEPOCHS = 160\nPATIENCE = 40\n\n\nprint(\"▶ imports ready · tensorflow\", tf.__version__)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"AVGBaseline = 0.8493577048441893\nHIBaseline = 0.867\nOPT = False\ntrainWithNFold = 10 #do not change this one\nDEVICE = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\nifFreqWidth = True\nifFreqEntropy = True\nifEquivalentToUpright = True\nifFFTMean = False\nifFFTStd = True # This one also a keep\nifFFTMax = True #I think this one is more relevant than others\nifFFTMin = False \nifFFTEnergy = True\nifFFTEnergyAroundMaxFreq = False\nif (ifFFTMax and ifFFTStd) == False:\n ifFFTEnergyAroundMaxFreq = False\n\nifNonOverlapWindow = False\nifOverlapWindow = False\n\nifFFTEnergySegmented = False\nnoSegments = 5\n\nifCurvaturePathLength = False\n\n\nifNonOverlapWindowAcc = False\nifOverlapWindowAcc = False\nifNonOverlapWindowFreq = False\n\nifSRNetAttention = False\n\n\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 》》》**Utility Functions**","metadata":{}},{"cell_type":"code","source":"#Tensor Manipulations\ndef time_sum(x):\n return K.sum(x, axis=1)\n\ndef squeeze_last_axis(x):\n return tf.squeeze(x, axis=-1)\n\ndef expand_last_axis(x):\n return tf.expand_dims(x, axis=-1)\n\ndef se_block(x, reduction=8):\n ch = x.shape[-1]\n se = GlobalAveragePooling1D()(x)\n se = Dense(ch // reduction, activation='relu')(se)\n se = Dense(ch, activation='sigmoid')(se)\n se = Reshape((1, ch))(se)\n return Multiply()([x, se])\n\n# Residual CNN Block with SE\ndef residual_se_cnn_block(x, filters, kernel_size, pool_size=2, drop=0.3, wd=1e-4):\n shortcut = x\n for _ in range(2):\n x = Conv1D(filters, kernel_size, padding='same', use_bias=False,\n kernel_regularizer=l2(wd))(x)\n x = BatchNormalization()(x)\n x = Activation('relu')(x)\n x = se_block(x)\n if shortcut.shape[-1] != filters:\n shortcut = Conv1D(filters, 1, padding='same', use_bias=False,\n kernel_regularizer=l2(wd))(shortcut)\n shortcut = BatchNormalization()(shortcut)\n x = add([x, shortcut])\n x = Activation('relu')(x)\n x = MaxPooling1D(pool_size)(x)\n x = Dropout(drop)(x)\n return x\n\ndef attention_layer(inputs):\n score = Dense(1, activation='tanh')(inputs)\n score = Lambda(squeeze_last_axis)(score)\n weights = Activation('softmax')(score)\n weights = Lambda(expand_last_axis)(weights)\n context = Multiply()([inputs, weights])\n context = Lambda(time_sum)(context)\n return context","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 》》》**Data Helpers**","metadata":{}},{"cell_type":"code","source":"# Normalizes and cleans the time series sequence. \n\ndef preprocess_sequence(df_seq: pd.DataFrame, feature_cols: list[str], scaler: StandardScaler):\n mat = df_seq[feature_cols].ffill().bfill().fillna(0).values\n return scaler.transform(mat).astype('float32')\n\n# MixUp the data argumentation in order to regularize the neural network. \n\nclass MixupGenerator(Sequence):\n def __init__(self, X, y, batch_size, alpha=0.2):\n self.X, self.y = X, y\n self.batch = batch_size\n self.alpha = alpha\n self.indices = np.arange(len(X))\n def __len__(self):\n return int(np.ceil(len(self.X) / self.batch))\n def __getitem__(self, i):\n idx = self.indices[i*self.batch:(i+1)*self.batch]\n Xb, yb = self.X[idx], self.y[idx]\n lam = np.random.beta(self.alpha, self.alpha)\n perm = np.random.permutation(len(Xb))\n X_mix = lam * Xb + (1-lam) * Xb[perm]\n y_mix = lam * yb + (1-lam) * yb[perm]\n return X_mix, y_mix\n def on_epoch_end(self):\n np.random.shuffle(self.indices)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def remove_gravity_from_acc(acc_data, rot_data):\n\n if isinstance(acc_data, pd.DataFrame):\n acc_values = acc_data[['acc_x', 'acc_y', 'acc_z']].values\n else:\n acc_values = acc_data\n\n if isinstance(rot_data, pd.DataFrame):\n quat_values = rot_data[['rot_x', 'rot_y', 'rot_z', 'rot_w']].values\n else:\n quat_values = rot_data\n\n num_samples = acc_values.shape[0]\n linear_accel = np.zeros_like(acc_values)\n \n gravity_world = np.array([0, 0, 9.81])\n\n for i in range(num_samples):\n if np.all(np.isnan(quat_values[i])) or np.all(np.isclose(quat_values[i], 0)):\n linear_accel[i, :] = acc_values[i, :] \n continue\n\n try:\n rotation = R.from_quat(quat_values[i])\n gravity_sensor_frame = rotation.apply(gravity_world, inverse=True)\n linear_accel[i, :] = acc_values[i, :] - gravity_sensor_frame\n except ValueError:\n linear_accel[i, :] = acc_values[i, :]\n \n return linear_accel\n\ndef calculate_angular_velocity_from_quat(rot_data, time_delta=1/200): # Assuming 200Hz sampling rate\n if isinstance(rot_data, pd.DataFrame):\n quat_values = rot_data[['rot_x', 'rot_y', 'rot_z', 'rot_w']].values\n else:\n quat_values = rot_data\n\n num_samples = quat_values.shape[0]\n angular_vel = np.zeros((num_samples, 3))\n\n for i in range(num_samples - 1):\n q_t = quat_values[i]\n q_t_plus_dt = quat_values[i+1]\n\n if np.all(np.isnan(q_t)) or np.all(np.isclose(q_t, 0)) or \\\n np.all(np.isnan(q_t_plus_dt)) or np.all(np.isclose(q_t_plus_dt, 0)):\n continue\n\n try:\n rot_t = R.from_quat(q_t)\n rot_t_plus_dt = R.from_quat(q_t_plus_dt)\n\n # Calculate the relative rotation\n delta_rot = rot_t.inv() * rot_t_plus_dt\n \n # Convert delta rotation to angular velocity vector\n # The rotation vector (Euler axis * angle) scaled by 1/dt\n # is a good approximation for small delta_rot\n angular_vel[i, :] = delta_rot.as_rotvec() / time_delta\n except ValueError:\n # If quaternion is invalid, angular velocity remains zero\n pass\n \n return angular_vel","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def calculate_angular_distance(rot_data):\n if isinstance(rot_data, pd.DataFrame):\n quat_values = rot_data[['rot_x', 'rot_y', 'rot_z', 'rot_w']].values\n else:\n quat_values = rot_data\n\n num_samples = quat_values.shape[0]\n angular_dist = np.zeros(num_samples)\n\n for i in range(num_samples - 1):\n q1 = quat_values[i]\n q2 = quat_values[i+1]\n\n if np.all(np.isnan(q1)) or np.all(np.isclose(q1, 0)) or \\\n np.all(np.isnan(q2)) or np.all(np.isclose(q2, 0)):\n angular_dist[i] = 0 # Или np.nan, в зависимости от желаемого поведения\n continue\n try:\n # Преобразование кватернионов в объекты Rotation\n r1 = R.from_quat(q1)\n r2 = R.from_quat(q2)\n\n # Вычисление углового расстояния: 2 * arccos(|real(p * q*)|)\n # где p* - сопряженный кватернион q\n # В scipy.spatial.transform.Rotation, r1.inv() * r2 дает относительное вращение.\n # Угол этого относительного вращения - это и есть угловое расстояние.\n relative_rotation = r1.inv() * r2\n \n # Угол rotation vector соответствует угловому расстоянию\n # Норма rotation vector - это угол в радианах\n angle = np.linalg.norm(relative_rotation.as_rotvec())\n angular_dist[i] = angle\n except ValueError:\n angular_dist[i] = 0 # В случае недействительных кватернионов\n pass\n \n return angular_dist","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import numpy as np\nfrom scipy.stats import entropy\nimport matplotlib.pyplot as plt\n\n# 假设 fs 是采样率\n # 设置采样率为 100Hz，当然根据你的数据来调整\n\ndef FFTforLinearAccel(linear_accel):\n \"\"\"\n 对加速度的 x, y, z 分量分别进行傅里叶变换，并返回各个分量的 FFT 结果。\n \n 参数:\n - linear_accel: 三维加速度数据，形状为 (n_samples, 3)，每一列对应 x, y, z 方向的加速度。\n \n 返回:\n - fft_results: 包含 x, y, z 分量 FFT 结果的字典。\n \"\"\"\n accelX = linear_accel[:, 0] # x方向的加速度\n accelY = linear_accel[:, 1] # y方向的加速度\n accelZ = linear_accel[:, 2] # z方向的加速度\n \n fft_resultX = np.fft.fft(accelX) # 对加速度数据做傅里叶变换\n freqsX = np.fft.fftfreq(len(accelX), d=1/fs) # 计算对应的频率轴\n\n fft_resultY = np.fft.fft(accelY) # y方向的傅里叶变换\n freqsY = np.fft.fftfreq(len(accelY), d=1/fs)\n\n fft_resultZ = np.fft.fft(accelZ) # z方向的傅里叶变换\n freqsZ = np.fft.fftfreq(len(accelZ), d=1/fs)\n\n fft_results = {\n 'x': (fft_resultX, freqsX),\n 'y': (fft_resultY, freqsY),\n 'z': (fft_resultZ, freqsZ)\n }\n return fft_results\n\n\ndef getEnergyFeaturesFromLinearAccelFFT(fft_results):\n \"\"\"\n 从线性加速度的 FFT 结果中提取能量特征，包括三段频率区间能量（低、中、高）。\n \n 参数:\n - fft_results: 包含 x, y, z 轴的 FFT 结果和频率轴的字典\n 格式: {'x': (fft_result_x, freqs_x), 'y': (fft_result_y, freqs_y), 'z': (fft_result_z, freqs_z)}\n\n 返回:\n - energy_features: 包含能量特征的字典\n \"\"\"\n energy_features = {}\n \n # 获取每个轴的 FFT 结果和频率轴\n fft_result_x, freqs_x = fft_results['x']\n fft_result_y, freqs_y = fft_results['y']\n fft_result_z, freqs_z = fft_results['z']\n \n # 计算每个轴的幅度\n amplitude_x = np.abs(fft_result_x)\n amplitude_y = np.abs(fft_result_y)\n amplitude_z = np.abs(fft_result_z)\n \n # 能量：幅度的平方和\n energy_x = np.sum(amplitude_x**2)\n energy_y = np.sum(amplitude_y**2)\n energy_z = np.sum(amplitude_z**2)\n total_energy = energy_x + energy_y + energy_z\n \n energy_features['freq_energy_x'] = energy_x\n energy_features['freq_energy_y'] = energy_y\n energy_features['freq_energy_z'] = energy_z\n energy_features['freq_energy_total_mag'] = total_energy\n\n \n # 三段频率区间\n low_freq_upper = 2 # 低频 < 2Hz\n mid_freq_lower = 2 # 中频 >= 2Hz\n mid_freq_upper = 8 # 中频 < 8Hz\n high_freq_lower = 8 # 高频 >= 8Hz\n\n # X轴\n low_freq_idx_x = np.where(freqs_x < low_freq_upper)\n mid_freq_idx_x = np.where((freqs_x >= mid_freq_lower) & (freqs_x < mid_freq_upper))\n high_freq_idx_x = np.where(freqs_x >= high_freq_lower)\n # Y轴\n low_freq_idx_y = np.where(freqs_y < low_freq_upper)\n mid_freq_idx_y = np.where((freqs_y >= mid_freq_lower) & (freqs_y < mid_freq_upper))\n high_freq_idx_y = np.where(freqs_y >= high_freq_lower)\n # Z轴\n low_freq_idx_z = np.where(freqs_z < low_freq_upper)\n mid_freq_idx_z = np.where((freqs_z >= mid_freq_lower) & (freqs_z < mid_freq_upper))\n high_freq_idx_z = np.where(freqs_z >= high_freq_lower)\n \n # 计算三段能量\n for axis, amp, low_idx, mid_idx, high_idx in [\n ('x', amplitude_x, low_freq_idx_x, mid_freq_idx_x, high_freq_idx_x),\n ('y', amplitude_y, low_freq_idx_y, mid_freq_idx_y, high_freq_idx_y),\n ('z', amplitude_z, low_freq_idx_z, mid_freq_idx_z, high_freq_idx_z)\n ]:\n energy_features[f'freq_energy_lo_{axis}'] = np.sum(amp[low_idx]**2)\n energy_features[f'freq_energy_mid_{axis}'] = np.sum(amp[mid_idx]**2)\n energy_features[f'freq_energy_hi_{axis}'] = np.sum(amp[high_idx]**2)\n \n # 总低/中/高频能量（xyz和）\n energy_features['freq_energy_lo_total'] = (\n energy_features['freq_energy_lo_x'] +\n energy_features['freq_energy_lo_y'] +\n energy_features['freq_energy_lo_z']\n )\n \n energy_features['freq_energy_mid_total'] = (\n energy_features['freq_energy_mid_x'] +\n energy_features['freq_energy_mid_y'] +\n energy_features['freq_energy_mid_z']\n )\n \n energy_features['freq_energy_hi_total'] = (\n energy_features['freq_energy_hi_x'] +\n energy_features['freq_energy_hi_y'] +\n energy_features['freq_energy_hi_z']\n )\n \n return energy_features\n\ndef freqEnergyAroundFreqMax(FreqMax, FreqSTD, fft_results, k=1.0):\n \"\"\"\n 计算每个轴在 (主频 ± k*std) 区间的能量（幅度平方和）。\n\n 参数:\n - FreqMax: dict，{'x': float, 'y': float, 'z': float}\n - FreqSTD: dict，{'x': float, 'y': float, 'z': float}\n - fft_results: dict，每个轴包含 (fft_result, freqs)\n - k: 标准差倍数 (默认±1 std)\n\n 返回:\n - energy_features: dict，每个轴该区间的能量\n \"\"\"\n energy_features = {}\n\n for axis in ['x', 'y', 'z']:\n fft_result, freqs = fft_results[axis]\n\n amplitude = np.abs(fft_result)\n\n lower = FreqMax[axis] - k * FreqSTD[axis]\n upper = FreqMax[axis] + k * FreqSTD[axis]\n\n # 选出落在区间的频率bin\n idx_in_range = np.where((freqs >= lower) & (freqs <= upper))\n\n # 计算能量\n energy = np.sum(amplitude[idx_in_range] ** 2)\n\n energy_features[f'std_band_energy_{axis}'] = energy\n\n # 总能量\n energy_features['std_band_energy_total'] = (\n energy_features['std_band_energy_x'] +\n energy_features['std_band_energy_y'] +\n energy_features['std_band_energy_z']\n )\n\n return energy_features\n\n\ndef freqEntropyFromLinearAccelFFT(fft_results):\n \"\"\"\n 计算从傅里叶变换结果得到的频域熵。\n \n 参数:\n - fft_results: 包含傅里叶变换结果和频率轴的字典。\n \n 返回:\n - spectral_entropy: 频域熵值。\n \"\"\"\n # 计算幅度\n amplitude_x = np.abs(fft_results['x'][0])\n amplitude_y = np.abs(fft_results['y'][0])\n amplitude_z = np.abs(fft_results['z'][0])\n\n # 归一化幅度\n amplitude_x_norm = amplitude_x / np.sum(amplitude_x)\n amplitude_y_norm = amplitude_y / np.sum(amplitude_y)\n amplitude_z_norm = amplitude_z / np.sum(amplitude_z)\n \n # 计算频域熵\n spectral_entropy_x = entropy(amplitude_x_norm)\n spectral_entropy_y = entropy(amplitude_y_norm)\n spectral_entropy_z = entropy(amplitude_z_norm)\n '''\n print(f\"Spectral Entropy for X: {spectral_entropy_x}\")\n print(f\"Spectral Entropy for Y: {spectral_entropy_y}\")\n print(f\"Spectral Entropy for Z: {spectral_entropy_z}\")\n '''\n \n \n return spectral_entropy_x, spectral_entropy_y, spectral_entropy_z\n\n\n\n# Assume global fs is defined above\n# fs = 10\n\nimport numpy as np\n\ndef nonOverlapWindowSlide(data, window_seconds=1, fs=fs):\n \"\"\"\n Slice data into non-overlapping windows.\n\n Parameters:\n data: numpy array of shape (n_samples, n_features)\n window_seconds: window size in seconds\n fs: sample frequency (Hz), default uses global fs\n\n Returns:\n List of numpy arrays, each with shape (window_size, n_features)\n \"\"\"\n window_size = int(window_seconds * fs)\n windows = []\n n = len(data)\n for start in range(0, n - window_size + 1, window_size):\n window = data[start : start + window_size]\n windows.append(window)\n return windows\n\ndef OverlapWindowSlide(data, window_seconds=1, overlap=0.5, fs=fs):\n \"\"\"\n Slice data into overlapping windows.\n\n Parameters:\n data: numpy array of shape (n_samples, n_features)\n window_seconds: window size in seconds\n overlap: fraction [0, 1) indicating overlap (e.g., 0.5 for 50% overlap)\n fs: sample frequency (Hz), default uses global fs\n\n Returns:\n List of numpy arrays, each with shape (window_size, n_features)\n \"\"\"\n window_size = int(window_seconds * fs)\n stride = int(window_size * (1 - overlap))\n if stride == 0:\n stride = 1 # Avoid infinite loop if overlap is 1.0\n\n windows = []\n n = len(data)\n for start in range(0, n - window_size + 1, stride):\n window = data[start : start + window_size]\n windows.append(window)\n return windows\n\n\n\n\ndef extract_features_for_window(window, fs=10):\n time_features = time_domain_features(window)\n freq_features = frequency_domain_features(window, fs)\n event_features = event_based_features(window)\n imu_features = imu_domain_features(window)\n \n # Combine all features into one dictionary\n combined_features = {**time_features, **freq_features, **event_features, **imu_features}\n \n return combined_features\n\n\ndef time_domain_features(window):\n mean = np.mean(window, axis=0)\n std = np.std(window, axis=0)\n min_val = np.min(window, axis=0)\n max_val = np.max(window, axis=0)\n median = np.median(window, axis=0)\n skewness = np.apply_along_axis(lambda x: np.skew(x), axis=0, arr=window)\n kurtosis = np.apply_along_axis(lambda x: np.kurtosis(x), axis=0, arr=window)\n \n features = {\n 'mean': mean,\n 'std': std,\n 'min': min_val,\n 'max': max_val,\n 'median': median,\n 'skewness': skewness,\n 'kurtosis': kurtosis\n }\n \n return features\n\ndef frequency_domain_features(window, fs):\n # Perform FFT\n fft_result = np.fft.fft(window, axis=0)\n fft_freq = np.fft.fftfreq(len(window), d=1/fs)\n \n # Compute the magnitude of the FFT\n fft_magnitude = np.abs(fft_result)\n \n # Dominant frequency (peak frequency)\n dominant_freq = fft_freq[np.argmax(fft_magnitude)]\n \n # Energy in different frequency bands (e.g., low [0-2Hz], mid [2-8Hz], high [8-10Hz])\n low_freq_energy = np.sum(fft_magnitude[(fft_freq >= 0) & (fft_freq < 2)]**2)\n mid_freq_energy = np.sum(fft_magnitude[(fft_freq >= 2) & (fft_freq < 8)]**2)\n high_freq_energy = np.sum(fft_magnitude[(fft_freq >= 8) & (fft_freq < 10)]**2)\n \n # Spectral entropy\n fft_norm = fft_magnitude / np.sum(fft_magnitude)\n spectral_entropy = -np.sum(fft_norm * np.log(fft_norm + 1e-9)) # avoid log(0)\n \n features = {\n 'dominant_frequency': dominant_freq,\n 'low_freq_energy': low_freq_energy,\n 'mid_freq_energy': mid_freq_energy,\n 'high_freq_energy': high_freq_energy,\n 'spectral_entropy': spectral_entropy\n }\n \n return features\n\n\ndef event_based_features(window):\n zero_crossings = np.sum(np.diff(np.sign(window), axis=0) != 0)\n peaks_count = len(find_peaks(window)[0]) # Assuming find_peaks is imported from scipy.signal\n\n # Jerk (rate of change of acceleration)\n jerk = np.diff(window, axis=0)\n mean_jerk = np.mean(jerk, axis=0)\n std_jerk = np.std(jerk, axis=0)\n\n features = {\n 'zero_crossings': zero_crossings,\n 'peaks_count': peaks_count,\n 'mean_jerk': mean_jerk,\n 'std_jerk': std_jerk\n }\n\n return features\n\ndef imu_domain_features(window):\n # Calculate acceleration magnitude\n accel_magnitude = np.sqrt(np.sum(window**2, axis=1))\n \n # Additional features specific to IMU data\n # For example, calculating orientation or rotation from gyroscope data (simplified here)\n \n features = {\n 'accel_magnitude_mean': np.mean(accel_magnitude),\n 'accel_magnitude_std': np.std(accel_magnitude),\n }\n \n return features\n\ndef extract_features_for_window(window, fs=10):\n time_features = time_domain_features(window)\n freq_features = frequency_domain_features(window, fs)\n event_features = event_based_features(window)\n imu_features = imu_domain_features(window)\n \n # Combine all features into one dictionary\n combined_features = {**time_features, **freq_features, **event_features, **imu_features}\n \n return combined_features\n\ndef extract_features_for_window(window, fs=10):\n time_features = time_domain_features(window)\n freq_features = frequency_domain_features(window, fs)\n event_features = event_based_features(window)\n imu_features = imu_domain_features(window)\n \n # Combine all features into one dictionary\n combined_features = {**time_features, **freq_features, **event_features, **imu_features}\n \n return combined_features\n\nimport numpy as np\n\n\ndef segmented_fft_features_dict(linear_accel, n_segments=noSegments):\n \"\"\"\n 将输入序列分段，对每段提取 FFT 能量和熵特征，返回字典。\n \n 参数:\n - linear_accel: ndarray (n_samples, 3)\n - n_segments: int，分段数\n - fs: float，采样率Hz\n \n 返回:\n - features_dict: dict {列名: 特征值}\n \"\"\"\n global fs\n time_steps = linear_accel.shape[0]\n segment_length = time_steps // n_segments\n\n all_features = {}\n\n for i in range(n_segments):\n start = i * segment_length\n end = (i + 1) * segment_length if i < n_segments - 1 else time_steps\n segment = linear_accel[start:end, :]\n\n # FFT\n fft_results = FFTforLinearAccel(segment)\n\n # 能量特征\n energy_feats = getEnergyFeaturesFromLinearAccelFFT(fft_results)\n # 频域熵\n entropy_x, entropy_y, entropy_z = freqEntropyFromLinearAccelFFT(fft_results)\n\n # 拼接本段所有特征\n seg_prefix = f\"segment_{i+1}_\"\n \n # 能量特征\n for k, v in energy_feats.items():\n all_features[seg_prefix + k] = v\n \n # 熵\n all_features[seg_prefix + \"spectral_entropy_x\"] = entropy_x\n all_features[seg_prefix + \"spectral_entropy_y\"] = entropy_y\n all_features[seg_prefix + \"spectral_entropy_z\"] = entropy_z\n\n return all_features\n\n\ndef computeTrajectoryLengthAndCurvature(accel_sequence):\n \"\"\"\n 参数:\n - accel_sequence: numpy数组，形状 (N,3)，每行是 x,y,z\n\n 返回:\n - dict: {length, curvature}\n \"\"\"\n diffs = accel_sequence[1:] - accel_sequence[:-1] # [N-1,3]\n segment_lengths = np.linalg.norm(diffs, axis=1) # [N-1]\n trajectory_length = np.sum(segment_lengths)\n\n # 起点到终点的欧几里得距离\n start_to_end = np.linalg.norm(accel_sequence[-1] - accel_sequence[0])\n\n curvature = trajectory_length / (start_to_end + 1e-6) # 避免除0\n\n return {\n \"trajectory_length\": trajectory_length,\n \"curvature\": curvature\n }\n\n\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 》》》**Model Definition - Two Branch Architecture**","metadata":{}},{"cell_type":"code","source":"def build_two_branch_model(pad_len, imu_dim, tof_dim, n_classes, wd=1e-4):\n inp = Input(shape=(pad_len, imu_dim+tof_dim))\n imu = Lambda(lambda t: t[:, :, :imu_dim])(inp)\n tof = Lambda(lambda t: t[:, :, imu_dim:])(inp)\n\n # IMU deep branch\n x1 = residual_se_cnn_block(imu, 64, 3, drop=0.1, wd=wd)\n x1 = residual_se_cnn_block(x1, 128, 5, drop=0.1, wd=wd)\n\n # TOF/Thermal lighter branch\n x2 = Conv1D(64, 3, padding='same', use_bias=False, kernel_regularizer=l2(wd))(tof)\n x2 = BatchNormalization()(x2); x2 = Activation('relu')(x2)\n x2 = MaxPooling1D(2)(x2); x2 = Dropout(0.2)(x2)\n x2 = Conv1D(128, 3, padding='same', use_bias=False, kernel_regularizer=l2(wd))(x2)\n x2 = BatchNormalization()(x2); x2 = Activation('relu')(x2)\n x2 = MaxPooling1D(2)(x2); x2 = Dropout(0.2)(x2)\n\n merged = Concatenate()([x1, x2])\n\n xa = Bidirectional(LSTM(128, return_sequences=True, kernel_regularizer=l2(wd)))(merged)\n xb = Bidirectional(GRU(128, return_sequences=True, kernel_regularizer=l2(wd)))(merged)\n xc = GaussianNoise(0.09)(merged)\n xc = Dense(16, activation='elu')(xc)\n \n x = Concatenate()([xa, xb, xc])\n x = Dropout(0.4)(x)\n x = attention_layer(x)\n\n for units, drop in [(256, 0.5), (128, 0.3)]:\n x = Dense(units, use_bias=False, kernel_regularizer=l2(wd))(x)\n x = BatchNormalization()(x); x = Activation('relu')(x)\n x = Dropout(drop)(x)\n\n out = Dense(n_classes, activation='softmax', kernel_regularizer=l2(wd))(x)\n return Model(inp, out)\n\ntmp_model = build_two_branch_model(127,7,325,18)","metadata":{"trusted":true,"_kg_hide-output":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 》》》**Torch Model Classes and Functions**","metadata":{}},{"cell_type":"code","source":"import math\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom einops import rearrange\nimport os\n \nclass MeanShift(nn.Conv2d):\n def __init__(\n self, rgb_range,\n rgb_mean=(0.4488, 0.4371, 0.4040), rgb_std=(1.0, 1.0, 1.0), sign=-1):\n super(MeanShift, self).__init__(3, 3, kernel_size=1)\n std = torch.Tensor(rgb_std)\n self.weight.data = torch.eye(3).view(3, 3, 1, 1) / std.view(3, 1, 1, 1)\n self.bias.data = sign * rgb_range * torch.Tensor(rgb_mean) / std\n for p in self.parameters():\n p.requires_grad = False\n\n\nclass LayerNorm(nn.Module):\n r\"\"\" From ConvNeXt (https://arxiv.org/pdf/2201.03545.pdf)\n \"\"\"\n def __init__(self, normalized_shape, eps=1e-6, data_format=\"channels_last\"):\n super().__init__()\n self.weight = nn.Parameter(torch.ones(normalized_shape))\n self.bias = nn.Parameter(torch.zeros(normalized_shape))\n self.eps = eps\n self.data_format = data_format\n if self.data_format not in [\"channels_last\", \"channels_first\"]:\n raise NotImplementedError\n self.normalized_shape = (normalized_shape,)\n\n def forward(self, x):\n if self.data_format == \"channels_last\":\n return F.layer_norm(x, self.normalized_shape, self.weight, self.bias, self.eps)\n elif self.data_format == \"channels_first\":\n u = x.mean(1, keepdim=True)\n s = (x - u).pow(2).mean(1, keepdim=True)\n x = (x - u) / torch.sqrt(s + self.eps)\n x = self.weight[:, None, None] * x + self.bias[:, None, None]\n return x\n\n\nclass FourierUnit(nn.Module):\n def __init__(self, dim, groups=1, fft_norm='ortho'):\n super().__init__()\n self.groups = groups\n self.fft_norm = fft_norm\n\n self.conv_layer = nn.Conv2d(in_channels=dim * 2, out_channels=dim * 2, kernel_size=1, stride=1,\n padding=0, groups=self.groups, bias=False)\n self.act = nn.GELU()\n\n def forward(self, x):\n batch, c, h, w = x.size()\n r_size = x.size()\n\n # 1️⃣ 新版rfft2：返回 complex tensor\n ffted = torch.fft.rfft2(x, norm=\"ortho\") # shape: (batch, c, h, w//2+1), dtype=complex\n\n # 2️⃣ 将 complex 转成 real tensor: shape (..., 2)\n ffted = torch.view_as_real(ffted) # shape: (batch, c, h, w//2+1, 2)\n\n # 3️⃣ permute + reshape\n ffted = ffted.permute(0, 1, 4, 2, 3).contiguous()\n ffted = ffted.view(batch, -1, h, w//2+1) # (batch, c*2, h, w//2+1)\n\n # 4️⃣ 卷积\n ffted = self.conv_layer(ffted)\n ffted = self.act(ffted)\n\n # 5️⃣ reshape回原来的通道+实虚\n ffted = ffted.view(batch, -1, 2, h, w//2+1).permute(0,1,3,4,2).contiguous()\n\n # 6️⃣ 转回 complex\n ffted = torch.view_as_complex(ffted) # shape: (batch, c, h, w//2+1), complex\n\n # 7️⃣ irfft2\n output = torch.fft.irfft2(ffted, s=r_size[2:], norm=\"ortho\")\n\n return output\n\nclass FConvMod(nn.Module):\n def __init__(self, dim, num_heads):\n super().__init__()\n layer_scale_init_value = 1e-6\n self.num_heads = num_heads\n self.norm = LayerNorm(dim, eps=1e-6, data_format=\"channels_first\")\n self.a = FourierUnit(dim)\n self.v = nn.Conv2d(dim, dim, 1)\n self.act = nn.GELU()\n self.layer_scale = nn.Parameter(layer_scale_init_value * torch.ones(num_heads), requires_grad=True)\n self.CPE = nn.Conv2d(dim, dim, kernel_size=3, stride=1, padding=1, groups=dim)\n self.proj = nn.Conv2d(dim, dim, 1)\n\n def forward(self, x):\n B, C, H, W = x.shape\n N = H * W\n shortcut = x\n pos_embed = self.CPE(x)\n x = self.norm(x)\n a = self.a(x)\n v = self.v(x)\n a = rearrange(a, 'b (head c) h w -> b head c (h w)', head=self.num_heads)\n v = rearrange(v, 'b (head c) h w -> b head c (h w)', head=self.num_heads)\n a_all = torch.split(a, math.ceil(N // 4), dim=-1)\n v_all = torch.split(v, math.ceil(N // 4), dim=-1)\n attns = []\n for a, v in zip(a_all, v_all):\n attn = a * v\n attn = self.layer_scale.unsqueeze(-1).unsqueeze(-1) * attn\n attns.append(attn)\n x = torch.cat(attns, dim=-1)\n x = F.softmax(x, dim=-1)\n x = rearrange(x, 'b head c (h w) -> b (head c) h w', head=self.num_heads, h=H, w=W)\n x = x + pos_embed\n x = self.proj(x)\n out = x + shortcut\n \n return out\n\n\nclass KernelAggregation(nn.Module):\n def __init__(self, dim, kernel_size, groups, num_kernels, bias=True, init_weight=True):\n super().__init__()\n self.groups = groups\n self.bias = bias\n self.num_kernels = num_kernels\n self.kernel_size = kernel_size\n self.dim = dim\n self.weight = nn.Parameter(torch.randn(num_kernels, dim, dim // groups, kernel_size, kernel_size),\n requires_grad=True)\n if bias:\n self.bias = nn.Parameter(torch.zeros(num_kernels, dim))\n else:\n self.bias = None\n\n if init_weight:\n self._initialize_weights()\n\n def _initialize_weights(self):\n for i in range(self.num_kernels):\n nn.init.kaiming_uniform_(self.weight[i])\n\n def forward(self, x, attention):\n B, C, H, W = x.shape\n x = x.contiguous().view(1, B * self.dim, H, W)\n\n weight = self.weight.contiguous().view(self.num_kernels, -1)\n weight = torch.mm(attention, weight).contiguous().view(B * self.dim, self.dim // self.groups,\n self.kernel_size, self.kernel_size)\n if self.bias is not None:\n bias = torch.mm(attention, self.bias).contiguous().view(-1)\n x = F.conv2d(x, weight=weight, bias=bias, stride=1, padding=self.kernel_size // 2,\n groups=self.groups * B)\n else:\n x = F.conv2d(x, weight=weight, bias=None, stride=1, padding=self.kernel_size // 2,\n groups=self.groups * B)\n x = x.contiguous().view(B, self.dim, x.shape[-2], x.shape[-1])\n\n return x\n\n\nclass KernelAttention(nn.Module):\n def __init__(self, dim, reduction=8, num_kernels=8):\n super().__init__()\n if dim != 3:\n mid_channels = dim // reduction\n else:\n mid_channels = num_kernels\n self.avg_pool = nn.AdaptiveAvgPool2d(1)\n self.conv1 = nn.Conv2d(dim, mid_channels, 1)\n self.act = nn.GELU()\n self.conv2 = nn.Conv2d(mid_channels, num_kernels, 1)\n self.sigmoid = nn.Sigmoid()\n\n def forward(self, x):\n x = self.avg_pool(x)\n x = self.conv1(x)\n x = self.act(x)\n x = self.conv2(x)\n x = x.view(x.shape[0], -1)\n x = self.sigmoid(x)\n return x\n\n\nclass DynamicKernelAggregation(nn.Module):\n def __init__(self, dim, kernel_size, groups=1, num_kernels=4):\n super().__init__()\n assert dim % groups == 0\n self.attention = KernelAttention(dim, num_kernels=num_kernels)\n self.aggregation = KernelAggregation(dim, kernel_size=kernel_size, groups=groups, num_kernels=num_kernels)\n\n def forward(self, x):\n attention = x\n attention = self.attention(attention)\n x = self.aggregation(x, attention)\n return x\n\n\nclass DyConv(nn.Module):\n def __init__(self, dim, kernel_size, groups, num_kernels=1):\n super().__init__()\n if num_kernels > 1:\n self.conv = DynamicKernelAggregation(dim, kernel_size=kernel_size, groups=groups,\n num_kernels=num_kernels)\n else:\n self.conv = nn.Conv2d(dim, dim, kernel_size=kernel_size, groups=groups)\n\n def forward(self, x):\n x = self.conv(x)\n return x\n\n\nclass MixFFN(nn.Module):\n def __init__(self, dim, num_kernels):\n super().__init__()\n self.proj_in = nn.Conv2d(dim, dim * 2, 1)\n self.conv1 = DyConv(dim, kernel_size=5, groups=dim, num_kernels=num_kernels)\n self.conv2 = DyConv(dim, kernel_size=7, groups=dim, num_kernels=num_kernels)\n self.proj_out = nn.Conv2d(dim * 2, dim, 1)\n self.norm = LayerNorm(dim, eps=1e-6, data_format=\"channels_first\")\n self.act = nn.GELU()\n\n def forward(self, x):\n shortcut = x\n x = self.norm(x)\n x = self.act(self.proj_in(x))\n x1, x2 = torch.chunk(x, 2, dim=1)\n x1 = self.act(self.conv1(x1)).unsqueeze(dim=2)\n x2 = self.act(self.conv2(x2)).unsqueeze(dim=2)\n x = torch.cat([x1, x2], dim=2)\n x = rearrange(x, 'b c g h w -> b (c g) h w')\n x = self.proj_out(x) \n x = x + shortcut\n return x\n\n\nclass BasicBlock(nn.Module):\n def __init__(self, dim, num_heads, num_kernels):\n super().__init__()\n self.attention = FConvMod(dim, num_heads)\n self.ffn = MixFFN(dim, num_kernels)\n\n def forward(self, x):\n x = self.attention(x)\n x = self.ffn(x)\n \n return x\n\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#rebuild with pytorch\nimport torch\nimport torch.nn as nn\n\nclass GaussianNoise(nn.Module):\n def __init__(self, stddev):\n super(GaussianNoise, self).__init__()\n self.stddev = stddev\n\n def forward(self, x):\n if self.training:\n noise = torch.randn_like(x) * self.stddev\n return x + noise\n else:\n return x\n\nclass EarlyStopping:\n def __init__(self, patience=10, mode=\"max\"):\n self.patience = patience\n self.counter = 0\n self.best_score = None\n self.early_stop = False\n self.mode = mode\n self.best_state = None\n\n def __call__(self, metric, model):\n score = metric\n if self.best_score is None:\n self.best_score = score\n self.best_state = model.state_dict()\n elif (self.mode == 'max' and score <= self.best_score) or \\\n (self.mode == 'min' and score >= self.best_score):\n self.counter += 1\n if self.counter >= self.patience:\n self.early_stop = True\n else:\n self.best_score = score\n self.best_state = model.state_dict()\n self.counter = 0\n\n\ndef soft_cross_entropy(pred, soft_targets):\n log_probs = torch.nn.functional.log_softmax(pred, dim=1)\n return -(soft_targets * log_probs).sum(dim=1).mean()\n\n\n# ===========\n# Mixup helpers\n# ===========\ndef mixup_data(x, y, alpha=MIXUP_ALPHA):\n if alpha > 0:\n lam = np.random.beta(alpha, alpha)\n else:\n lam = 1\n batch_size = x.size()[0]\n index = torch.randperm(batch_size).to(x.device)\n mixed_x = lam * x + (1 - lam) * x[index, :]\n y_a, y_b = y, y[index]\n return mixed_x, y_a, y_b, lam\n\ndef mixup_criterion(criterion, pred, y_a, y_b, lam):\n return lam * criterion(pred, y_a) + (1 - lam) * criterion(pred, y_b, )\n\n\nclass SequenceDataset(Dataset):\n def __init__(self, X, y):\n self.X = torch.tensor(X, dtype=torch.float32)\n self.y = torch.tensor(y, dtype=torch.long) # 假设y已经是整数\n\n def __len__(self):\n return len(self.X)\n\n def __getitem__(self, idx):\n return self.X[idx], self.y[idx]\n\ndef to_categorical(y, num_classes=None):\n y = np.array(y, dtype=np.int64)\n if num_classes is None:\n num_classes = y.max() + 1\n one_hot = np.zeros((y.shape[0], num_classes), dtype=np.float32)\n one_hot[np.arange(y.shape[0]), y] = 1.\n return one_hot\n'''\nAdaptiveAvgPool1d is applied to 1D data (typically a batch of sequences). It will pool over the length (or temporal dimension) of the data.\n\nExample input shape: (batch_size, channels, length). we are using this because this is a time sequence so its 1d\n\nAdaptiveAvgPool2d is applied to 2D data (such as images or 2D feature maps). It pools over both the height and width of the image or feature map.\nI guess we could use this version of pooling for TOF analysis\nExample input shape: (batch_size, channels, height, width). \n'''\ndef pad_sequences(sequences, maxlen, dtype='float32', padding='post', truncating='post', value=0.):\n \"\"\"\n 等价于 Keras 的 pad_sequences（只支持二维特征序列）\n \"\"\"\n num_samples = len(sequences)\n feature_dim = sequences[0].shape[1]\n out = np.full((num_samples, maxlen, feature_dim), fill_value=value, dtype=dtype)\n for idx, seq in enumerate(sequences):\n if len(seq) == 0:\n continue\n if truncating == 'pre':\n trunc = seq[-maxlen:]\n else:\n trunc = seq[:maxlen]\n if padding == 'post':\n out[idx, :len(trunc), :] = trunc\n else:\n out[idx, -len(trunc):, :] = trunc\n return out\n\n#A pytorch module is a subclass of nn.module. Inherits from nn.module. \n#Needs to have def __init__ and def foward\nclass se_block(nn.Module):\n def __init__(self, channels, reduction=8):\n super().__init__()\n self.avg_pool = nn.AdaptiveAvgPool1d(1)\n self.fc1 = nn.Linear(channels, channels // reduction)\n self.fc2 = nn.Linear(channels // reduction, channels)\n self.relu = nn.ReLU(inplace=True)\n self.sigmoid = nn.Sigmoid()\n def forward(self, x):\n b, c, l = x.size() # x: [batch, channels, length]\n y = self.avg_pool(x).view(b, c) # [batch, channels, 1] → [batch, channels]\n y = self.relu(self.fc1(y)) # [batch, channels // reduction]\n y = self.sigmoid(self.fc2(y)) # [batch, channels]\n y = y.view(b, c, 1) # [batch, channels, 1]，这样才能广播\n return x * y # 广播乘法，输出[batch, channels, length]\n\n\nclass ResidualSEBlock(nn.Module):\n def __init__(self, in_channels, out_channels, kernel_size, pool_size=2, dropout=0.3, reduction=8):\n super().__init__()\n # First conv + BN + ReLU\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 self.prelu1 = nn.PReLU()\n # Second conv + BN + ReLU\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 self.prelu2 = nn.PReLU()\n # SE block\n self.se = se_block(out_channels, reduction)\n # Shortcut\n if in_channels != out_channels:\n self.shortcut_conv = nn.Conv1d(in_channels, out_channels, 1, padding=0, bias=False)\n self.shortcut_bn = nn.BatchNorm1d(out_channels)\n else:\n self.shortcut_conv = None\n # Post-process\n self.final_relu = nn.PReLU()\n self.pool = nn.MaxPool1d(pool_size)\n self.dropout = nn.Dropout(dropout)\n\n def forward(self, x):\n identity = x\n\n # Residual branch\n out = self.conv1(x)\n out = self.bn1(out)\n out = self.prelu1(out)\n\n out = self.conv2(out)\n out = self.bn2(out)\n out = self.prelu2(out)\n\n # Squeeze-and-Excitation\n out = self.se(out)\n\n # Shortcut branch\n if self.shortcut_conv is not None:\n identity = self.shortcut_conv(identity)\n identity = self.shortcut_bn(identity)\n\n # Add shortcut\n out = out + identity\n out = self.final_relu(out)\n out = self.pool(out)\n out = self.dropout(out)\n return out\n#changed\n\nclass AttentionLayer(nn.Module):\n def __init__(self, input_dim):\n super().__init__()\n self.fc = nn.Linear(input_dim, 1) # Dense(1)\n self.tanh = nn.Tanh() # activation='tanh'\n self.softmax = nn.Softmax(dim=1) # 按时间/序列softmax\n def forward(self, x):\n score = self.fc(x)\n score = self.tanh(score)\n score = torch.squeeze(score, -1)\n weights =self.softmax(score)\n weights = torch.unsqueeze(weights, -1)\n context = x * weights\n context = torch.sum(context, dim = 1)\n return context\n\nclass AttentionLayer(nn.Module):\n def __init__(self, input_dim):\n super().__init__()\n self.fc = nn.Linear(input_dim, 1) # Dense(1)\n self.tanh = nn.Tanh() # activation='tanh'\n self.softmax = nn.Softmax(dim=1) # 按时间/序列softmax\n def forward(self, x):\n score = self.fc(x)\n score = self.tanh(score)\n score = torch.squeeze(score, -1)\n weights =self.softmax(score)\n weights = torch.unsqueeze(weights, -1)\n context = x * weights\n context = torch.sum(context, dim = 1)\n return context\n\nclass GaussianNoise(nn.Module):\n def __init__(self, stddev):\n super(GaussianNoise, self).__init__()\n self.stddev = stddev\n\n def forward(self, x):\n if self.training:\n noise = torch.randn_like(x) * self.stddev\n return x + noise\n else:\n return x\n\nclass EarlyStopping:\n def __init__(self, patience=10, mode=\"max\"):\n self.patience = patience\n self.counter = 0\n self.best_score = None\n self.early_stop = False\n self.mode = mode\n self.best_state = None\n\n def __call__(self, metric, model):\n score = metric\n if self.best_score is None:\n self.best_score = score\n self.best_state = model.state_dict()\n elif (self.mode == 'max' and score <= self.best_score) or \\\n (self.mode == 'min' and score >= self.best_score):\n self.counter += 1\n if self.counter >= self.patience:\n self.early_stop = True\n else:\n self.best_score = score\n self.best_state = model.state_dict()\n self.counter = 0\n\n\ndef soft_cross_entropy(pred, soft_targets):\n log_probs = torch.nn.functional.log_softmax(pred, dim=1)\n return -(soft_targets * log_probs).sum(dim=1).mean()\n\n\n# ===========\n# Mixup helpers\n# ===========\ndef mixup_data(x, y, alpha=MIXUP_ALPHA):\n if alpha > 0:\n lam = np.random.beta(alpha, alpha)\n else:\n lam = 1\n batch_size = x.size()[0]\n index = torch.randperm(batch_size).to(x.device)\n mixed_x = lam * x + (1 - lam) * x[index, :]\n y_a, y_b = y, y[index]\n return mixed_x, y_a, y_b, lam\n\ndef mixup_criterion(criterion, pred, y_a, y_b, lam):\n return lam * criterion(pred, y_a) + (1 - lam) * criterion(pred, y_b, )\n\n\nclass SequenceDataset(Dataset):\n def __init__(self, X, y):\n self.X = torch.tensor(X, dtype=torch.float32)\n self.y = torch.tensor(y, dtype=torch.long) # 假设y已经是整数\n\n def __len__(self):\n return len(self.X)\n\n def __getitem__(self, idx):\n return self.X[idx], self.y[idx]\n\ndef to_categorical(y, num_classes=None):\n y = np.array(y, dtype=np.int64)\n if num_classes is None:\n num_classes = y.max() + 1\n one_hot = np.zeros((y.shape[0], num_classes), dtype=np.float32)\n one_hot[np.arange(y.shape[0]), y] = 1.\n return one_hot\n'''\nAdaptiveAvgPool1d is applied to 1D data (typically a batch of sequences). It will pool over the length (or temporal dimension) of the data.\n\nExample input shape: (batch_size, channels, length). we are using this because this is a time sequence so its 1d\n\nAdaptiveAvgPool2d is applied to 2D data (such as images or 2D feature maps). It pools over both the height and width of the image or feature map.\nI guess we could use this version of pooling for TOF analysis\nExample input shape: (batch_size, channels, height, width). \n'''\ndef pad_sequences(sequences, maxlen, dtype='float32', padding='post', truncating='post', value=0.):\n \"\"\"\n 等价于 Keras 的 pad_sequences（只支持二维特征序列）\n \"\"\"\n num_samples = len(sequences)\n feature_dim = sequences[0].shape[1]\n out = np.full((num_samples, maxlen, feature_dim), fill_value=value, dtype=dtype)\n for idx, seq in enumerate(sequences):\n if len(seq) == 0:\n continue\n if truncating == 'pre':\n trunc = seq[-maxlen:]\n else:\n trunc = seq[:maxlen]\n if padding == 'post':\n out[idx, :len(trunc), :] = trunc\n else:\n out[idx, -len(trunc):, :] = trunc\n return out\n\n#A pytorch module is a subclass of nn.module. Inherits from nn.module. \n#Needs to have def __init__ and def foward\nclass se_block(nn.Module):\n def __init__(self, channels, reduction=8):\n super().__init__()\n self.avg_pool = nn.AdaptiveAvgPool1d(1)\n self.fc1 = nn.Linear(channels, channels // reduction)\n self.fc2 = nn.Linear(channels // reduction, channels)\n self.relu = nn.ReLU(inplace=True)\n self.sigmoid = nn.Sigmoid()\n def forward(self, x):\n b, c, l = x.size() # x: [batch, channels, length]\n y = self.avg_pool(x).view(b, c) # [batch, channels, 1] → [batch, channels]\n y = self.relu(self.fc1(y)) # [batch, channels // reduction]\n y = self.sigmoid(self.fc2(y)) # [batch, channels]\n y = y.view(b, c, 1) # [batch, channels, 1]，这样才能广播\n return x * y # 广播乘法，输出[batch, channels, length]\n\n\nclass ResidualSEBlock(nn.Module):\n def __init__(self, in_channels, out_channels, kernel_size, pool_size=2, dropout=0.3, reduction=8):\n super().__init__()\n # First conv + BN + ReLU\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 self.prelu1 = nn.PReLU()\n # Second conv + BN + ReLU\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 self.prelu2 = nn.PReLU()\n # SE block\n self.se = se_block(out_channels, reduction)\n # Shortcut\n if in_channels != out_channels:\n self.shortcut_conv = nn.Conv1d(in_channels, out_channels, 1, padding=0, bias=False)\n self.shortcut_bn = nn.BatchNorm1d(out_channels)\n else:\n self.shortcut_conv = None\n # Post-process\n self.final_relu = nn.PReLU()\n self.pool = nn.MaxPool1d(pool_size)\n self.dropout = nn.Dropout(dropout)\n\n def forward(self, x):\n identity = x\n\n # Residual branch\n out = self.conv1(x)\n out = self.bn1(out)\n out = self.prelu1(out)\n\n out = self.conv2(out)\n out = self.bn2(out)\n out = self.prelu2(out)\n\n # Squeeze-and-Excitation\n out = self.se(out)\n\n # Shortcut branch\n if self.shortcut_conv is not None:\n identity = self.shortcut_conv(identity)\n identity = self.shortcut_bn(identity)\n\n # Add shortcut\n out = out + identity\n out = self.final_relu(out)\n out = self.pool(out)\n out = self.dropout(out)\n return out\n\nclass AttentionLayer(nn.Module):\n def __init__(self, input_dim):\n super().__init__()\n self.fc = nn.Linear(input_dim, 1) # Dense(1)\n self.tanh = nn.Tanh() # activation='tanh'\n self.softmax = nn.Softmax(dim=1) # 按时间/序列softmax\n def forward(self, x):\n score = self.fc(x)\n score = self.tanh(score)\n score = torch.squeeze(score, -1)\n weights =self.softmax(score)\n weights = torch.unsqueeze(weights, -1)\n context = x * weights\n context = torch.sum(context, dim = 1)\n return context","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Resolving a problem around missing THM/TOF data in the data sequence:\n\n'''\nKeras in general is more user-friendly and robust than torch. Since I am using Torch and Im basically\ntranslating everything from keras to torch, I only copied the language but failed to recognize fundamental\ndifferences between Keras and Torch. Starting from an open-source Torch package would be a better approach\nImplementing TOF-THM masking \n\nAssuming Batch size B and a masking rate p. p * B lines will be stripped of their THM/TOF lines (replaced with zero) \nand only (1-p) * B lines will have none-zero TOF/THM lines (lines are not the best notation)\n\nEffectively p * B lines will be IMU only\n\nlemme think about implementation. This part should be part of the training data-loader\nquestion: what about post-train eval data-loader? dunno\n\nif some lines are stripped of their THM TOF branches, we need to re-evaluate the network structure and how missing\nTOF-THM parts are handled. Adapt model structure and training loop after this\n\nBATCH_SIZE defines batch size\n\nindicesToMaskList\n\np_mask stores p \n'''\nfrom typing import List\nimport torch.nn as nn\ndef get_mask_indices_for_thm_tof(thm_cols_list, tof_cols_list,all_cols_names):\n cols_to_mask = set(thm_cols_list) | set(tof_cols_list)\n return [i for i, col in enumerate(all_cols_names) if col in cols_to_mask]\n\n\ndef mask_thm_tof_by_p_from_indices (X_tensor: torch.tensor, mask_indices: List[int], p = 0.3):\n B, _, _ = X_tensor.shape\n X = X_tensor.clone()\n num_cut = int(p * B)\n\n indices_to_sheer = np.random.choice(B, size = num_cut, replace = False)\n\n for idx in indices_to_sheer:\n X [idx, : , mask_indices] = 0.0\n return X \n\n# todo: build data augmentation pipeline\n\nclass FeatureAttention(nn.Module):\n def __init__(self, input_dim):\n super().__init__()\n self.attn = nn.Sequential(\n nn.Linear(input_dim, input_dim),\n nn.Tanh(),\n nn.Linear(input_dim, input_dim),\n nn.Sigmoid()\n )\n\n def forward(self, x):\n weights = self.attn(x)\n x = x * weights\n return x\n \n#feature attention for TOF branch\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\nimport math\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom einops import rearrange\nimport os\n \nclass MeanShift(nn.Conv2d):\n def __init__(\n self, rgb_range,\n rgb_mean=(0.4488, 0.4371, 0.4040), rgb_std=(1.0, 1.0, 1.0), sign=-1):\n super(MeanShift, self).__init__(3, 3, kernel_size=1)\n std = torch.Tensor(rgb_std)\n self.weight.data = torch.eye(3).view(3, 3, 1, 1) / std.view(3, 1, 1, 1)\n self.bias.data = sign * rgb_range * torch.Tensor(rgb_mean) / std\n for p in self.parameters():\n p.requires_grad = False\n\n\nclass LayerNorm(nn.Module):\n \n def __init__(self, normalized_shape, eps=1e-6, data_format=\"channels_last\"):\n super().__init__()\n self.weight = nn.Parameter(torch.ones(normalized_shape))\n self.bias = nn.Parameter(torch.zeros(normalized_shape))\n self.eps = eps\n self.data_format = data_format\n if self.data_format not in [\"channels_last\", \"channels_first\"]:\n raise NotImplementedError\n self.normalized_shape = (normalized_shape,)\n\n def forward(self, x):\n if self.data_format == \"channels_last\":\n return F.layer_norm(x, self.normalized_shape, self.weight, self.bias, self.eps)\n elif self.data_format == \"channels_first\":\n u = x.mean(1, keepdim=True)\n s = (x - u).pow(2).mean(1, keepdim=True)\n x = (x - u) / torch.sqrt(s + self.eps)\n x = self.weight[:, None, None] * x + self.bias[:, None, None]\n return x\n\n\nclass FourierUnit(nn.Module):\n def __init__(self, dim, groups=1, fft_norm='ortho'):\n super().__init__()\n self.groups = groups\n self.fft_norm = fft_norm\n\n self.conv_layer = nn.Conv2d(in_channels=dim * 2, out_channels=dim * 2, kernel_size=1, stride=1,\n padding=0, groups=self.groups, bias=False)\n self.act = nn.GELU()\n\n def forward(self, x):\n batch, c, h, w = x.size()\n r_size = x.size()\n\n # 1️⃣ 新版rfft2：返回 complex tensor\n ffted = torch.fft.rfft2(x, norm=\"ortho\") # shape: (batch, c, h, w//2+1), dtype=complex\n\n # 2️⃣ 将 complex 转成 real tensor: shape (..., 2)\n ffted = torch.view_as_real(ffted) # shape: (batch, c, h, w//2+1, 2)\n\n # 3️⃣ permute + reshape\n ffted = ffted.permute(0, 1, 4, 2, 3).contiguous()\n ffted = ffted.view(batch, -1, h, w//2+1) # (batch, c*2, h, w//2+1)\n\n # 4️⃣ 卷积\n ffted = self.conv_layer(ffted)\n ffted = self.act(ffted)\n\n # 5️⃣ reshape回原来的通道+实虚\n ffted = ffted.view(batch, -1, 2, h, w//2+1).permute(0,1,3,4,2).contiguous()\n\n # 6️⃣ 转回 complex\n ffted = torch.view_as_complex(ffted) # shape: (batch, c, h, w//2+1), complex\n\n # 7️⃣ irfft2\n output = torch.fft.irfft2(ffted, s=r_size[2:], norm=\"ortho\")\n\n return output\n\nclass FConvMod(nn.Module):\n def __init__(self, dim, num_heads):\n super().__init__()\n layer_scale_init_value = 1e-6\n self.num_heads = num_heads\n self.norm = LayerNorm(dim, eps=1e-6, data_format=\"channels_first\")\n self.a = FourierUnit(dim)\n self.v = nn.Conv2d(dim, dim, 1)\n self.act = nn.GELU()\n self.layer_scale = nn.Parameter(layer_scale_init_value * torch.ones(num_heads), requires_grad=True)\n self.CPE = nn.Conv2d(dim, dim, kernel_size=3, stride=1, padding=1, groups=dim)\n self.proj = nn.Conv2d(dim, dim, 1)\n\n def forward(self, x):\n B, C, H, W = x.shape\n N = H * W\n shortcut = x\n pos_embed = self.CPE(x)\n x = self.norm(x)\n a = self.a(x)\n v = self.v(x)\n a = rearrange(a, 'b (head c) h w -> b head c (h w)', head=self.num_heads)\n v = rearrange(v, 'b (head c) h w -> b head c (h w)', head=self.num_heads)\n a_all = torch.split(a, math.ceil(N // 4), dim=-1)\n v_all = torch.split(v, math.ceil(N // 4), dim=-1)\n attns = []\n for a, v in zip(a_all, v_all):\n attn = a * v\n attn = self.layer_scale.unsqueeze(-1).unsqueeze(-1) * attn\n attns.append(attn)\n x = torch.cat(attns, dim=-1)\n x = F.softmax(x, dim=-1)\n x = rearrange(x, 'b head c (h w) -> b (head c) h w', head=self.num_heads, h=H, w=W)\n x = x + pos_embed\n x = self.proj(x)\n out = x + shortcut\n \n return out\n\n\nclass KernelAggregation(nn.Module):\n def __init__(self, dim, kernel_size, groups, num_kernels, bias=True, init_weight=True):\n super().__init__()\n self.groups = groups\n self.bias = bias\n self.num_kernels = num_kernels\n self.kernel_size = kernel_size\n self.dim = dim\n self.weight = nn.Parameter(torch.randn(num_kernels, dim, dim // groups, kernel_size, kernel_size),\n requires_grad=True)\n if bias:\n self.bias = nn.Parameter(torch.zeros(num_kernels, dim))\n else:\n self.bias = None\n\n if init_weight:\n self._initialize_weights()\n\n def _initialize_weights(self):\n for i in range(self.num_kernels):\n nn.init.kaiming_uniform_(self.weight[i])\n\n def forward(self, x, attention):\n B, C, H, W = x.shape\n x = x.contiguous().view(1, B * self.dim, H, W)\n\n weight = self.weight.contiguous().view(self.num_kernels, -1)\n weight = torch.mm(attention, weight).contiguous().view(B * self.dim, self.dim // self.groups,\n self.kernel_size, self.kernel_size)\n if self.bias is not None:\n bias = torch.mm(attention, self.bias).contiguous().view(-1)\n x = F.conv2d(x, weight=weight, bias=bias, stride=1, padding=self.kernel_size // 2,\n groups=self.groups * B)\n else:\n x = F.conv2d(x, weight=weight, bias=None, stride=1, padding=self.kernel_size // 2,\n groups=self.groups * B)\n x = x.contiguous().view(B, self.dim, x.shape[-2], x.shape[-1])\n\n return x\n\n\nclass KernelAttention(nn.Module):\n def __init__(self, dim, reduction=8, num_kernels=8):\n super().__init__()\n if dim != 3:\n mid_channels = dim // reduction\n else:\n mid_channels = num_kernels\n self.avg_pool = nn.AdaptiveAvgPool2d(1)\n self.conv1 = nn.Conv2d(dim, mid_channels, 1)\n self.act = nn.GELU()\n self.conv2 = nn.Conv2d(mid_channels, num_kernels, 1)\n self.sigmoid = nn.Sigmoid()\n\n def forward(self, x):\n x = self.avg_pool(x)\n x = self.conv1(x)\n x = self.act(x)\n x = self.conv2(x)\n x = x.view(x.shape[0], -1)\n x = self.sigmoid(x)\n return x\n\n\nclass DynamicKernelAggregation(nn.Module):\n def __init__(self, dim, kernel_size, groups=1, num_kernels=4):\n super().__init__()\n assert dim % groups == 0\n self.attention = KernelAttention(dim, num_kernels=num_kernels)\n self.aggregation = KernelAggregation(dim, kernel_size=kernel_size, groups=groups, num_kernels=num_kernels)\n\n def forward(self, x):\n attention = x\n attention = self.attention(attention)\n x = self.aggregation(x, attention)\n return x\n\n\nclass DyConv(nn.Module):\n def __init__(self, dim, kernel_size, groups, num_kernels=1):\n super().__init__()\n if num_kernels > 1:\n self.conv = DynamicKernelAggregation(dim, kernel_size=kernel_size, groups=groups,\n num_kernels=num_kernels)\n else:\n self.conv = nn.Conv2d(dim, dim, kernel_size=kernel_size, groups=groups)\n\n def forward(self, x):\n x = self.conv(x)\n return x\n\n\nclass MixFFN(nn.Module):\n def __init__(self, dim, num_kernels):\n super().__init__()\n self.proj_in = nn.Conv2d(dim, dim * 2, 1)\n self.conv1 = DyConv(dim, kernel_size=5, groups=dim, num_kernels=num_kernels)\n self.conv2 = DyConv(dim, kernel_size=7, groups=dim, num_kernels=num_kernels)\n self.proj_out = nn.Conv2d(dim * 2, dim, 1)\n self.norm = LayerNorm(dim, eps=1e-6, data_format=\"channels_first\")\n self.act = nn.GELU()\n\n def forward(self, x):\n shortcut = x\n x = self.norm(x)\n x = self.act(self.proj_in(x))\n x1, x2 = torch.chunk(x, 2, dim=1)\n x1 = self.act(self.conv1(x1)).unsqueeze(dim=2)\n x2 = self.act(self.conv2(x2)).unsqueeze(dim=2)\n x = torch.cat([x1, x2], dim=2)\n x = rearrange(x, 'b c g h w -> b (c g) h w')\n x = self.proj_out(x) \n x = x + shortcut\n return x\n\n\nclass BasicBlock(nn.Module):\n def __init__(self, dim, num_heads, num_kernels):\n super().__init__()\n self.attention = FConvMod(dim, num_heads)\n self.ffn = MixFFN(dim, num_kernels)\n\n def forward(self, x):\n x = self.attention(x)\n x = self.ffn(x)\n \n return x\n\n\n\n#rebuild with pytorch\nimport torch\nimport torch.nn as nn\n\nclass GaussianNoise(nn.Module):\n def __init__(self, stddev):\n super(GaussianNoise, self).__init__()\n self.stddev = stddev\n\n def forward(self, x):\n if self.training:\n noise = torch.randn_like(x) * self.stddev\n return x + noise\n else:\n return x\n\nclass EarlyStopping:\n def __init__(self, patience=10, mode=\"max\"):\n self.patience = patience\n self.counter = 0\n self.best_score = None\n self.early_stop = False\n self.mode = mode\n self.best_state = None\n\n def __call__(self, metric, model):\n score = metric\n if self.best_score is None:\n self.best_score = score\n self.best_state = model.state_dict()\n elif (self.mode == 'max' and score <= self.best_score) or \\\n (self.mode == 'min' and score >= self.best_score):\n self.counter += 1\n if self.counter >= self.patience:\n self.early_stop = True\n else:\n self.best_score = score\n self.best_state = model.state_dict()\n self.counter = 0\n\n\ndef soft_cross_entropy(pred, soft_targets):\n log_probs = torch.nn.functional.log_softmax(pred, dim=1)\n return -(soft_targets * log_probs).sum(dim=1).mean()\n\n\n\n# ===========\n# Mixup helpers\n# ===========\nMIXUP_ALPHA = 0.4\ndef mixup_data(x, y, alpha=0.4):\n if alpha > 0:\n lam = np.random.beta(alpha, alpha)\n else:\n lam = 1\n batch_size = x.size()[0]\n index = torch.randperm(batch_size).to(x.device)\n mixed_x = lam * x + (1 - lam) * x[index, :]\n y_a, y_b = y, y[index]\n return mixed_x, y_a, y_b, lam\n\ndef mixup_criterion(criterion, pred, y_a, y_b, lam):\n return lam * criterion(pred, y_a) + (1 - lam) * criterion(pred, y_b, )\n\n\nclass SequenceDataset(Dataset):\n def __init__(self, X, y):\n self.X = torch.tensor(X, dtype=torch.float32)\n self.y = torch.tensor(y, dtype=torch.long) # 假设y已经是整数\n\n def __len__(self):\n return len(self.X)\n\n def __getitem__(self, idx):\n return self.X[idx], self.y[idx]\n\ndef to_categorical(y, num_classes=None):\n y = np.array(y, dtype=np.int64)\n if num_classes is None:\n num_classes = y.max() + 1\n one_hot = np.zeros((y.shape[0], num_classes), dtype=np.float32)\n one_hot[np.arange(y.shape[0]), y] = 1.\n return one_hot\n'''\nAdaptiveAvgPool1d is applied to 1D data (typically a batch of sequences). It will pool over the length (or temporal dimension) of the data.\n\nExample input shape: (batch_size, channels, length). we are using this because this is a time sequence so its 1d\n\nAdaptiveAvgPool2d is applied to 2D data (such as images or 2D feature maps). It pools over both the height and width of the image or feature map.\nI guess we could use this version of pooling for TOF analysis\nExample input shape: (batch_size, channels, height, width). \n'''\ndef pad_sequences(sequences, maxlen, dtype='float32', padding='post', truncating='post', value=0.):\n \n num_samples = len(sequences)\n feature_dim = sequences[0].shape[1]\n out = np.full((num_samples, maxlen, feature_dim), fill_value=value, dtype=dtype)\n for idx, seq in enumerate(sequences):\n if len(seq) == 0:\n continue\n if truncating == 'pre':\n trunc = seq[-maxlen:]\n else:\n trunc = seq[:maxlen]\n if padding == 'post':\n out[idx, :len(trunc), :] = trunc\n else:\n out[idx, -len(trunc):, :] = trunc\n return out\n\n#A pytorch module is a subclass of nn.module. Inherits from nn.module. \n#Needs to have def __init__ and def foward\nclass se_block(nn.Module):\n def __init__(self, channels, reduction=8):\n super().__init__()\n self.avg_pool = nn.AdaptiveAvgPool1d(1)\n self.fc1 = nn.Linear(channels, channels // reduction)\n self.fc2 = nn.Linear(channels // reduction, channels)\n self.relu = nn.ReLU(inplace=True)\n self.sigmoid = nn.Sigmoid()\n def forward(self, x):\n b, c, l = x.size() # x: [batch, channels, length]\n y = self.avg_pool(x).view(b, c) # [batch, channels, 1] → [batch, channels]\n y = self.relu(self.fc1(y)) # [batch, channels // reduction]\n y = self.sigmoid(self.fc2(y)) # [batch, channels]\n y = y.view(b, c, 1) # [batch, channels, 1]，这样才能广播\n return x * y # 广播乘法，输出[batch, channels, length]\n\n\nclass ResidualSEBlock(nn.Module):\n def __init__(self, in_channels, out_channels, kernel_size, pool_size=2, dropout=0.3, reduction=8):\n super().__init__()\n # First conv + BN + ReLU\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 self.prelu1 = nn.PReLU()\n # Second conv + BN + ReLU\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 self.prelu2 = nn.PReLU()\n # SE block\n self.se = se_block(out_channels, reduction)\n # Shortcut\n if in_channels != out_channels:\n self.shortcut_conv = nn.Conv1d(in_channels, out_channels, 1, padding=0, bias=False)\n self.shortcut_bn = nn.BatchNorm1d(out_channels)\n else:\n self.shortcut_conv = None\n # Post-process\n self.final_relu = nn.ReLU()\n self.pool = nn.MaxPool1d(pool_size)\n self.dropout = nn.Dropout(dropout)\n\n def forward(self, x):\n identity = x\n\n # Residual branch\n out = self.conv1(x)\n out = self.bn1(out)\n out = self.prelu1(out)\n\n out = self.conv2(out)\n out = self.bn2(out)\n out = self.prelu2(out)\n\n # Squeeze-and-Excitation\n out = self.se(out)\n\n # Shortcut branch\n if self.shortcut_conv is not None:\n identity = self.shortcut_conv(identity)\n identity = self.shortcut_bn(identity)\n\n # Add shortcut\n out = out + identity\n out = self.final_relu(out)\n out = self.pool(out)\n out = self.dropout(out)\n return out\n\nclass AttentionLayer(nn.Module):\n def __init__(self, input_dim):\n super().__init__()\n self.fc = nn.Linear(input_dim, 1) # Dense(1)\n self.tanh = nn.Tanh() # activation='tanh'\n self.softmax = nn.Softmax(dim=1) # 按时间/序列softmax\n def forward(self, x):\n score = self.fc(x)\n score = self.tanh(score)\n score = torch.squeeze(score, -1)\n weights =self.softmax(score)\n weights = torch.unsqueeze(weights, -1)\n context = x * weights\n context = torch.sum(context, dim = 1)\n return context\n\n\n\n#Resolving a problem around missing THM/TOF data in the data sequence:\n\n\nfrom typing import List\nimport torch.nn as nn\ndef get_mask_indices_for_thm_tof(thm_cols_list, tof_cols_list,all_cols_names):\n cols_to_mask = set(thm_cols_list) | set(tof_cols_list)\n return [i for i, col in enumerate(all_cols_names) if col in cols_to_mask]\n\n\ndef mask_thm_tof_by_p_from_indices (X_tensor: torch.tensor, mask_indices: List[int], p = 0.3):\n B, _, _ = X_tensor.shape\n X = X_tensor.clone()\n num_cut = int(p * B)\n\n indices_to_sheer = np.random.choice(B, size = num_cut, replace = False)\n\n for idx in indices_to_sheer:\n X [idx, : , mask_indices] = 0.0\n return X \n\n# todo: build data augmentation pipeline\n\nclass FeatureAttention(nn.Module):\n def __init__(self, input_dim):\n super().__init__()\n self.attn = nn.Sequential(\n nn.Linear(input_dim, input_dim),\n nn.Tanh(),\n nn.Linear(input_dim, input_dim),\n nn.Sigmoid()\n )\n\n def forward(self, x):\n weights = self.attn(x)\n x = x * weights\n return x\n \n\n \nclass TwoBranchModel(nn.Module):\n def __init__(self, imu_dim, tof_dim, n_classes, ifSRNetAttention = False , focal_gamma = 1.0):\n super().__init__()\n # IMU branch\n self.imu_block1 = ResidualSEBlock(imu_dim, 64, 3, dropout=0.2)\n self.imu_block2 = ResidualSEBlock(64, 128, 5, dropout=0.2)\n # TOF branch\n \n self.tof_conv1 = nn.Conv1d(tof_dim, 64, 3, padding=1, bias=False)\n self.tof_bn1 = nn.BatchNorm1d(64)\n\n self.feature_attention = FeatureAttention(input_dim= 64)\n\n self.tof_conv2 = nn.Conv1d(64, 128, 3, padding=1, bias=False)\n self.tof_bn2 = nn.BatchNorm1d(128)\n self.pool = nn.MaxPool1d(2)\n self.drop = nn.Dropout(0.2)\n\n self.lstm = nn.LSTM(input_size=256, hidden_size=128, batch_first=True, bidirectional=True)\n self.gru = nn.GRU(input_size=256, hidden_size=128, batch_first=True, bidirectional=True)\n \n self.dense_noise = nn.Linear(256, 16)\n self.prelu = nn.PReLU()\n \n self.concat_drop = nn.Dropout(0.4)\n self.attn = AttentionLayer(512 + 16) # 合并后channel数\n \n self.fc1 = nn.Linear(512 + 16, 256, bias=False)\n self.bn_fc1 = nn.BatchNorm1d(256)\n self.fc2 = nn.Linear(256, 128, bias=False)\n self.bn_fc2 = nn.BatchNorm1d(128)\n self.out = nn.Linear(128, n_classes)\n self.tof_attention = BasicBlock (dim=64, num_heads=4, num_kernels=4)\n self.SRNetAttention = ifSRNetAttention\n def forward(self, x):\n # x: [batch, pad_len, imu_dim + tof_dim]\n imu = x[:, :, :self.imu_block1.conv1.in_channels] # [B, T, imu_dim]\n tof = x[:, :, self.imu_block1.conv1.in_channels:] # [B, T, tof_dim]\n \n # PyTorch expects [batch, channel, seq_len]\n imu = imu.transpose(1, 2) # [B, imu_dim, T]\n tof = tof.transpose(1, 2) # [B, tof_dim, T]\n\n # IMU branch\n x1 = self.imu_block1(imu) # [B, 64, ...]\n x1 = self.imu_block2(x1) # [B, 128, ...]\n \n # TOF branch\n x2 = F.relu(self.tof_bn1(self.tof_conv1(tof)))\n \n #simple feature attention\n x2 = x2.transpose(1,2)\n x2 = self.feature_attention(x2)\n x2 = x2.transpose(1,2)\n\n #this part should not be implemented\n if self.SRNetAttention:\n x2 = x2.unsqueeze(-1)\n x2 = self.tof_attention(x2)\n x2 = x2.squeeze(-1)\n x2 = self.pool(x2)\n\n #adopt new attention mech here\n x2 = self.drop(x2)\n x2 = F.relu(self.tof_bn2(self.tof_conv2(x2)))\n x2 = self.pool(x2)\n x2 = self.drop(x2)\n \n # 对齐长度\n min_len = min(x1.shape[-1], x2.shape[-1])\n x1 = x1[..., :min_len]\n x2 = x2[..., :min_len]\n \n # Concatenate along channel dimension\n merged = torch.cat([x1, x2], dim=1) # [B, 256, T']\n merged = merged.transpose(1, 2) # [B, T', 256]\n \n # LSTM/GRU\n xa, _ = self.lstm(merged)\n xb, _ = self.gru(merged)\n \n # Gaussian noise (can be skipped or use dropout as noise)\n xc = merged + 0.09 * torch.randn_like(merged)\n xc = self.dense_noise(xc)\n xc = self.prelu(xc)\n \n # Concatenate\n x_cat = torch.cat([xa, xb, xc], dim=2) # [B, T', 512 + 16]\n x_cat = self.concat_drop(x_cat)\n\n # Focal layer here？？？？\n \n # Attention pooling\n x = self.attn(x_cat) # [B, 512 + 16]\n \n # Dense layers\n x = self.fc1(x)\n x = self.bn_fc1(x)\n x = F.relu(x)\n x = F.dropout(x, 0.5, self.training)\n x = self.fc2(x)\n x = self.bn_fc2(x)\n x = F.relu(x)\n x = F.dropout(x, 0.3, self.training)\n x = self.out(x)\n #x = F.softmax(x, dim=1)\n return x\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"

\n\nBlending Model\n

","metadata":{}},{"cell_type":"markdown","source":"# 》》》**LoadBlendingModels**","metadata":{}},{"cell_type":"code","source":"print(\"▶ INFERENCE MODE – loading artefacts from\", PRETRAINED_DIR)\n\n\ncustom_objs = {\n 'time_sum': time_sum, 'squeeze_last_axis': squeeze_last_axis, 'expand_last_axis': expand_last_axis,\n 'se_block': se_block, 'residual_se_cnn_block': residual_se_cnn_block, 'attention_layer': attention_layer,\n}\n\ndef load_keras_model_set(model_dir: Path, prefix: str, folds: list[int]) -> dict:\n models = []\n for fold in folds:\n model_path = model_dir / f\"{prefix}_{fold}.h5\"\n print(f\">>> Loading Keras Model: {model_path}\")\n model = load_model(model_path, compile=False, custom_objects=custom_objs)\n models.append(model)\n\n # artefacts\n feature_cols = np.load(model_dir / \"feature_cols.npy\", allow_pickle=True).tolist()\n pad_len = int(np.load(model_dir / \"sequence_maxlen.npy\"))\n scaler = joblib.load(model_dir / \"scaler.pkl\")\n gesture_classes = np.load(model_dir / \"gesture_classes.npy\", allow_pickle=True)\n return {\n \"name\": prefix,\n \"models\": models,\n \"feature_cols\": feature_cols,\n \"pad_len\": pad_len,\n \"scaler\": scaler,\n \"gesture_classes\": gesture_classes,\n \"config\": None # 没有 config，不需要 apply_features_by_config\n }\n\n# ----------------------------------------------------------------- #\n# Load any Models\n# * is 2 Train Model Load\n# ----------------------------------------------------------------- #\n\nkeras_model_info = []\nprint(\" Loading models for ensemble inference...\")\n\nMODEL_DIR = Path(\"/kaggle/input/cmi-d-111\")\n\n# 加载两组 10 折模型\nkeras_model_info.append(load_keras_model_set(MODEL_DIR, prefix=\"D-111\", folds=range(10)))\nkeras_model_info.append(load_keras_model_set(MODEL_DIR, prefix=\"v0629\", folds=range(10)))\n\n# 汇总数量\ntotal_keras_models = sum(len(kset[\"models\"]) for kset in keras_model_info)\nprint(\"-\" * 50)\nprint(f\"[INFO] NumUseKerasModels: {total_keras_models}\")\n\n","metadata":{"trusted":true,"scrolled":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## blending custom Torch Models","metadata":{}},{"cell_type":"code","source":"import json\nimport torch\nimport joblib\nimport numpy as np\nfrom pathlib import Path\nimport re\ndef load_feature_config(model_dir: Path) -> dict:\n \"\"\"\n 从指定模型目录读取 feature_config.json 配置字典\n \"\"\"\n config_path = model_dir / \"feature_config.json\"\n if not config_path.exists():\n raise FileNotFoundError(f\"Config not found: {config_path}\")\n with open(config_path, \"r\") as f:\n print(f)\n return json.load(f)\n\ndef load_input_dims(model_dir: Path) -> dict:\n with open(model_dir / \"input_dims.json\", \"r\") as f:\n print(f)\n return json.load(f)\n\ndef load_all_pytorch_models(model_root_dirs: list[Path]):\n \"\"\"\n 加载多个 PyTorch 模型目录，每个目录内多个 .pth 文件，各自配 artefacts。\n 返回 baseline_model_info 列表。\n \"\"\"\n baseline_model_info = []\n \n for model_dir in model_root_dirs:\n # Artefacts 共享\n cfg = load_feature_config(model_dir)\n dims = load_input_dims(model_dir)\n feature_cols = np.load(model_dir / \"feature_cols.npy\", allow_pickle=True).tolist()\n alias_map = {\n \"acc_mag\":\"linear_acc_mag\",\n \"acc_mag_jerk\":\"linear_acc_mag_jerk\"\n }\n \n feature_cols = [alias_map.get(col,col) for col in feature_cols]\n scaler = joblib.load(model_dir / \"scaler.pkl\")\n gesture_classes = np.load(model_dir / \"gesture_classes.npy\", allow_pickle=True)\n pad_len = int(np.load(model_dir / \"sequence_maxlen.npy\"))\n imu_dim = dims[\"imu_dim_final\"]\n tof_dim = dims[\"tof_thm_aggregated_dim_final\"]\n num_classes = dims[\"num_classes\"]\n \n\n \n # 读取该目录下所有模型\n model_paths = sorted(model_dir.glob(\"*.pth\")) + sorted(model_dir.glob(\"*.pt\"))\n \n for model_path in model_paths:\n \n toggleSRNetAttention = str(cfg.get('ifSRNetAttention',\"false\")).lower()== \"true\"\n state_dict = torch.load(model_path, map_location = deviceGPU)\n model = TwoBranchModel(imu_dim, tof_dim, num_classes, ifSRNetAttention = toggleSRNetAttention)\n model.to(deviceGPU)\n \n try:\n model.load_state_dict(state_dict, strict = True)\n except RuntimeError as e:\n print(f\"failed to load model at {model_path.name}:{e}\")\n continue\n \n \n model.eval()\n print(f\"model{model_path} successfully loaded\")\n baseline_model_info.append({\n \"model\": model,\n \"feature_cols\": feature_cols,\n \"scaler\": scaler,\n \"gesture_classes\": gesture_classes,\n \"config\": cfg,\n \"name\": model_path.name,\n \"dir\": model_dir,\n \"pad_len\": pad_len\n })\n\n return baseline_model_info\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#baseline_model_dirs = [Path(\"/kaggle/input/modelwithoutfeature\")]\n#baseline_model_info = load_all_pytorch_models(baseline_model_dirs)\n#print(f\"[INFO] Loaded {len(baseline_model_info)} Baseline models from {len(baseline_model_dirs)} directories.\")\n\n\ncustom_model_dirs = [Path(\"/kaggle/input/torchnewlossfunc\")] #test inference part\ncustom_model_info = load_all_pytorch_models(custom_model_dirs)\nprint(f\"[INFO] Loaded {len(custom_model_info)} Torch models from {len(custom_model_dirs)} directories.\")\n\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 》》》**Predict**","metadata":{}},{"cell_type":"code","source":"def predict_with_torch(df_seq, model_info_list, device = deviceGPU):\n \"\"\"\n 使用多个 PyTorch 模型进行推理，返回平均 softmax 概率。\n\n 参数:\n df_seq: 待处理单条 sample（Pandas DataFrame）\n model_info_list: 模型信息列表（包含 model, config, scaler, feature_cols）\n device: 推理设备（通常为 'cuda' 或 'cpu'）\n\n 返回:\n np.ndarray: 平均 softmax 概率向量\n \"\"\"\n list_probs = []\n\n for info in model_info_list:\n \n df_feat = apply_features_by_config(df_seq.copy(), info[\"config\"])\n df_feat = df_feat.loc[:, ~df_feat.columns.duplicated()]\n df_feat = df_feat[info[\"feature_cols\"]]\n missing = [col for col in info[\"feature_cols\"] if col not in df_feat.columns]\n if missing:\n raise ValueError(f\"col missing: {missing}\")\n \n \n trigger = 0\n for _, (a,b) in enumerate(zip(info[\"feature_cols\"], df_feat.columns)):\n if a != b:\n print(f\"{a} and {b} are different\")\n trigger += 1\n print(f\"{trigger} cols are different\")\n \n if not trigger:\n print(\"all cols are matching\")\n \n model = info[\"model\"]\n model.eval()\n df_seq.describe()\n X = df_feat.ffill().bfill().fillna(0).astype(\"float32\")\n X_scaled = info[\"scaler\"].transform(X)\n print(\"X mean before scale:\", np.round(X.mean(axis=0)[:5], 3))\n print(\"X std before scale:\", np.round(X.std(axis=0)[:5], 3))\n print(\"X mean after scale:\", np.round(X_scaled.mean(axis=0)[:5], 3))\n print(\"X std after scale:\", np.round(X_scaled.std(axis=0)[:5], 3))\n\n pad_len_from_config = info['pad_len']\n print(f\"pad_len is {pad_len_from_config}\")\n X_padded = pad_sequences([X_scaled], maxlen=pad_len_from_config, padding=\"post\", dtype=\"float32\")\n print(f\"paded shape{X_padded.shape}\")\n X_tensor = torch.tensor(X_padded, dtype=torch.float32).to(device)\n print(X_tensor.shape)\n \n with torch.no_grad():\n logits = model(X_tensor)\n prob = torch.softmax(logits,1)\n print(f\"logits tensor: {np.round(prob.cpu().numpy(),2)}\")\n list_probs.append(prob)\n \n print(f\"List of all probs{list_probs}\")\n avg_prob = torch.mean(torch.stack(list_probs), dim=0).squeeze(0).cpu().numpy()\n return avg_prob","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def build_base_features(df_seq):\n linear_accel = remove_gravity_from_acc(df_seq, df_seq)\n df_seq['linear_acc_x'], df_seq['linear_acc_y'], df_seq['linear_acc_z'] = linear_accel[:, 0], linear_accel[:, 1], linear_accel[:, 2]\n df_seq['linear_acc_mag'] = np.sqrt(df_seq['linear_acc_x']**2 + df_seq['linear_acc_y']**2 + df_seq['linear_acc_z']**2)\n df_seq['linear_acc_mag_jerk'] = df_seq['linear_acc_mag'].diff().fillna(0)\n angular_vel = calculate_angular_velocity_from_quat(df_seq)\n df_seq['angular_vel_x'], df_seq['angular_vel_y'], df_seq['angular_vel_z'] = angular_vel[:, 0], angular_vel[:, 1], angular_vel[:, 2]\n df_seq['angular_distance'] = calculate_angular_distance(df_seq)\n\n # Backward compatible aliases for older models\n #df_seq['acc_mag'] = df_seq['linear_acc_mag']\n #df_seq['acc_mag_jerk'] = df_seq['linear_acc_mag_jerk']\n #now implemented in load torch model method\n \n # rot_angle = √(rot_x² + rot_y² + rot_z²)\n df_seq['rot_angle'] = np.sqrt(df_seq['rot_x']**2 + df_seq['rot_y']**2 + df_seq['rot_z']**2)\n df_seq['rot_angle_vel'] = df_seq['rot_angle'].diff().fillna(0)\n\n\n \n \n for i in range(1, 6):\n pixel_cols = [f\"tof_{i}_v{p}\" for p in range(64)]; tof_data = df_seq[pixel_cols].replace(-1, np.nan)\n \n df_seq[f'tof_{i}_mean'], df_seq[f'tof_{i}_std'], df_seq[f'tof_{i}_min'], df_seq[f'tof_{i}_max'] = tof_data.mean(axis=1), tof_data.std(axis=1), tof_data.min(axis=1), tof_data.max(axis=1)\n df_seq.drop(columns = pixel_cols, inplace = True, errors = \"ignore\")\n return df_seq\ndef add_fft_mean_features(df_seq):\n linear_accel_group = df_seq[['linear_acc_x', 'linear_acc_y', 'linear_acc_z']].values\n fft_results = FFTforLinearAccel(linear_accel_group)\n df_seq['freq_mean_x'] = np.mean(np.abs(fft_results['x'][0]))\n df_seq['freq_mean_y'] = np.mean(np.abs(fft_results['y'][0]))\n df_seq['freq_mean_z'] = np.mean(np.abs(fft_results['z'][0]))\n return df_seq\ndef add_fft_max_features(df_seq):\n linear_accel_group = df_seq[['linear_acc_x', 'linear_acc_y', 'linear_acc_z']].values\n fft_results = FFTforLinearAccel(linear_accel_group)\n df_seq['freq_max_x'] = np.max(np.abs(fft_results['x'][0]))\n df_seq['freq_max_y'] = np.max(np.abs(fft_results['y'][0]))\n df_seq['freq_max_z'] = np.max(np.abs(fft_results['z'][0]))\n return df_seq\n\ndef add_fft_min_features(df_seq):\n linear_accel_group = df_seq[['linear_acc_x', 'linear_acc_y', 'linear_acc_z']].values\n fft_results = FFTforLinearAccel(linear_accel_group)\n df_seq['freq_mean_x'] = np.min(np.abs(fft_results['x'][0]))\n df_seq['freq_mean_y'] = np.min(np.abs(fft_results['y'][0]))\n df_seq['freq_mean_z'] = np.min(np.abs(fft_results['z'][0]))\n return df_seq\ndef add_fft_std_features(df_seq):\n linear_accel_group = df_seq[['linear_acc_x', 'linear_acc_y', 'linear_acc_z']].values\n fft_results = FFTforLinearAccel(linear_accel_group)\n df_seq['freq_std_x'] = np.std(np.abs(fft_results['x'][0]))\n df_seq['freq_std_y'] = np.std(np.abs(fft_results['y'][0]))\n df_seq['freq_std_z'] = np.std(np.abs(fft_results['z'][0]))\n df_seq = df_seq.ffill().bfill().fillna(0)\n return df_seq\n\ndef add_fft_energy_features(df_seq):\n \"\"\"\n 给 df_seq 添加 FFT 能量类特征（推理阶段，只处理一个序列）\n 要求 df_seq 已含有 linear_acc_x/y/z\n \"\"\"\n # 取线性加速度\n linear_accel_group = df_seq[['linear_acc_x', 'linear_acc_y', 'linear_acc_z']].values\n\n # 计算 FFT 结果\n fft_results = FFTforLinearAccel(linear_accel_group)\n\n # 从 FFT 中提取能量特征\n energy_features = getEnergyFeaturesFromLinearAccelFFT(fft_results)\n\n # 单个序列，所以直接整体赋值（每一行都是一样的值）\n for key, value in energy_features.items():\n df_seq[key] = value # 全部行赋值为 scalar\n df_seq = df_seq.ffill().bfill().fillna(0)\n\n return df_seq\ndef add_fft_energy_features_segmented(df_seq):\n \"\"\"\n 推理阶段：对整个 df_seq 计算分段 FFT 能量特征，并添加到每一行\n 要求：\n - df_seq 是一个完整的序列（只含一个 sequence_id）\n - 必须包含 linear_acc_x/y/z 列\n - 使用 segmented_fft_features_dict() 返回特征字典\n \"\"\"\n # 提取线性加速度\n linear_accel_group = df_seq[['linear_acc_x', 'linear_acc_y', 'linear_acc_z']].values\n\n # 调用训练时使用的函数，生成特征字典\n feats_dict = segmented_fft_features_dict(linear_accel_group)\n\n # 把每个特征都广播到整列（推理只有一个 sequence，可以这么做）\n for col, val in feats_dict.items():\n df_seq[col] = val\n df_seq = df_seq.ffill().bfill().fillna(0)\n\n return df_seq\n\ndef add_fft_flatness_features(df_seq):\n linear_accel = df_seq[['linear_acc_x', 'linear_acc_y', 'linear_acc_z']].values\n fft_results = FFTforLinearAccel(linear_accel)\n for axis in ['x', 'y', 'z']:\n amplitude = np.abs(fft_results[axis][0])\n flatness = np.exp(np.mean(np.log(amplitude + 1e-8))) / np.mean(amplitude)\n df_seq[f'spectral_flatness_{axis}'] = flatness\n df_seq = df_seq.ffill().bfill().fillna(0)\n\n return df_seq\n\ndef add_fft_entropy_features(df_seq):\n linear_accel = df_seq[['linear_acc_x', 'linear_acc_y', 'linear_acc_z']].values\n fft_results = FFTforLinearAccel(linear_accel)\n entropy_x, entropy_y, entropy_z = freqEntropyFromLinearAccelFFT(fft_results)\n df_seq['freq_entropy_x'] = entropy_x\n df_seq['freq_entropy_y'] = entropy_y\n df_seq['freq_entropy_z'] = entropy_z\n df_seq = df_seq.ffill().bfill().fillna(0)\n\n return df_seq\n\ndef add_curvature_path_length_features(df_seq):\n linear_accel = df_seq[['linear_acc_x', 'linear_acc_y', 'linear_acc_z']].values\n feats = computeTrajectoryLengthAndCurvature(linear_accel)\n df_seq[\"trajectory_length\"] = feats[\"trajectory_length\"]\n df_seq[\"curvature\"] = feats[\"curvature\"]\n df_seq = df_seq.ffill().bfill().fillna(0)\n\n return df_seq\n\ndef add_fft_energy_around_max_freq_features(df_seq):\n linear_accel = df_seq[['linear_acc_x', 'linear_acc_y', 'linear_acc_z']].values\n fft_results = FFTforLinearAccel(linear_accel)\n\n result = {}\n for axis in ['x', 'y', 'z']:\n amplitude = np.abs(fft_results[axis][0])\n freqs = fft_results[axis][1]\n freq_max = freqs[np.argmax(amplitude)]\n freq_std = np.std(amplitude)\n result[axis] = (freq_max, freq_std)\n\n energy_dict = freqEnergyAroundFreqMax(\n FreqMax={k: v[0] for k, v in result.items()},\n FreqSTD={k: v[1] for k, v in result.items()},\n fft_results=fft_results,\n k=1.0\n )\n\n df_seq['std_band_energy_x'] = energy_dict['std_band_energy_x']\n df_seq['std_band_energy_y'] = energy_dict['std_band_energy_y']\n df_seq['std_band_energy_z'] = energy_dict['std_band_energy_z']\n df_seq = df_seq.ffill().bfill().fillna(0)\n\n return df_seq\ndef add_nonoverlap_window_features(df_seq, fs=50):\n data = df_seq[['linear_acc_x', 'linear_acc_y', 'linear_acc_z']].values\n windows = nonOverlapWindowSlide(data, window_seconds=1, fs=fs)\n\n stats = {'mean': np.mean, 'std': np.std, 'min': np.min, 'max': np.max}\n for stat_name, func in stats.items():\n for i, axis in enumerate(['x', 'y', 'z']):\n col_name = f'nonoverlap_acc_{axis}_{stat_name}'\n value = np.mean([func(w[:, i]) for w in windows]) # 对所有窗口再平均\n df_seq[col_name] = value\n df_seq = df_seq.ffill().bfill().fillna(0)\n\n return df_seq\n\ndef add_overlap_window_features(df_seq, fs=fs, overlap=0.5):\n data = df_seq[['linear_acc_x', 'linear_acc_y', 'linear_acc_z']].values\n windows = OverlapWindowSlide(data, window_seconds=1, overlap=overlap, fs=fs)\n\n stats = {'mean': np.mean, 'std': np.std, 'min': np.min, 'max': np.max}\n for stat_name, func in stats.items():\n for i, axis in enumerate(['x', 'y', 'z']):\n col_name = f'overlap_acc_{axis}_{stat_name}'\n value = np.mean([func(w[:, i]) for w in windows]) # 对所有窗口再平均\n df_seq[col_name] = value\n df_seq = df_seq.ffill().bfill().fillna(0)\n\n return df_seq\n\ndef apply_features_by_config(df_seq, cfg, fs=fs):\n \"\"\"\n 根据模型的 config 字典决定提取哪些特征，适用于推理阶段单个样本。\n \"\"\"\n # FFT 基础特征\n if cfg.get(\"ifFFTMean\", False):\n df_seq = add_fft_mean_features(df_seq)\n if cfg.get(\"ifFFTStd\", False):\n df_seq = add_fft_std_features(df_seq)\n if cfg.get(\"ifFFTMax\", False):\n df_seq = add_fft_max_features(df_seq)\n if cfg.get(\"ifFFTMin\", False):\n df_seq = add_fft_min_features(df_seq)\n\n if cfg.get(\"ifFFTEnergy\", False):\n df_seq = add_fft_energy_features(df_seq)\n # 能量相关\n \n if cfg.get(\"ifFFTEnergySegmented\", False):\n df_seq = add_fft_energy_features_segmented(df_seq)\n \n # 熵与频谱宽度\n if cfg.get(\"ifFreqWidth\", False):\n df_seq = add_fft_flatness_features(df_seq)\n if cfg.get(\"ifFreqEntropy\", False):\n df_seq = add_fft_entropy_features(df_seq)\n\n # 曲率轨迹\n if cfg.get(\"ifCurvaturePathLength\", False):\n df_seq = add_curvature_path_length_features(df_seq)\n\n if cfg.get(\"ifFFTEnergyAroundMaxFreq\", False):\n df_seq = add_fft_energy_around_max_freq_features(df_seq)\n\n \n # 滑窗特征\n if cfg.get(\"ifNonOverlapWindow\", False):\n if cfg.get(\"ifNonOverlapWindowAcc\", False):\n df_seq = add_nonoverlap_window_features(df_seq, fs=fs)\n if cfg.get(\"ifNonOverlapWindowFreq\", False):\n df_seq = add_nonoverlap_window_freq_features(df_seq, fs=fs)\n\n if cfg.get(\"ifOverlapWindow\", False):\n if cfg.get(\"ifOverlapWindowAcc\", False):\n df_seq = add_overlap_window_features(df_seq, fs=fs)\n if cfg.get(\"ifOverlapWindowFreq\", False):\n df_seq = add_overlap_window_freq_features(df_seq, fs=fs)\n df_seq = df_seq.ffill().bfill().fillna(0)\n\n return df_seq\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 》》》**Submit Inference server**","metadata":{}},{"cell_type":"code","source":"\ndef predict(sequence, demographics=None):\n\n df_seq = sequence.to_pandas()\n df_seq = build_base_features(df_seq) # 提取基础传感器数据（共用）\n\n all_probs = []\n\n \n \n baseline_probs = []\n '''\n\n # ========== 1. Keras 模型集 ==========\n keras_probs = []\n for model_set in keras_model_info:\n for model in model_set[\"models\"]:\n df_feat = df_seq.copy() # keras 不需要特征工程（config=None）\n X = df_feat[model_set[\"feature_cols\"]].ffill().bfill().fillna(0).astype(\"float32\")\n X_scaled = model_set[\"scaler\"].transform(X)\n X_padded = pad_sequences([X_scaled], maxlen=model_set[\"pad_len\"], padding=\"post\", dtype=\"float32\")\n prob = model.predict(X_padded, verbose=0)[0]\n keras_probs.append(prob)\n if keras_probs:\n keras_avg = np.mean(keras_probs, axis=0)\n all_probs.append(keras_avg)\n print(\"keras works\")\n \n '''\n \n # ========== 2. PyTorch baseline 模型 ==========\n \n #baseline_avg = predict_with_torch(df_seq, baseline_model_info, deviceGPU)\n \n #print(\"torchBase works\")\n #print(f\"baselineProbs: {baseline_avg}\")\n \n \n # ========== 3. PyTorch custom 模型 ==========\n custom_avg = predict_with_torch(df_seq, custom_model_info, deviceGPU)\n print(\"torchCustom works\")\n print(f\"customeProbs: {np.round(custom_avg,2)}\")\n # ========== 集成输出 ==========\n ensemble_prob = custom_avg#* 0.5 + baseline_avg * 0.5\n print(f\"ensembleProb:{np.round(ensemble_prob,2)}\")\n # 使用任意模型的 gesture_classes（应一致）\n \n gesture_classes = np.load(\"/kaggle/input/cmi25-imu-thmtof-tf-bilstm-gru-attentionlb-xx/gesture_classes.npy\", allow_pickle = True)\n pred_label = gesture_classes[np.argmax(ensemble_prob)]\n \n print(\"argmax =\", np.argmax(ensemble_prob))\n print(\"gesture =\", gesture_classes[np.argmax(ensemble_prob)])\n \n\n return pred_label\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import kaggle_evaluation.cmi_inference_server\ninference_server = kaggle_evaluation.cmi_inference_server.CMIInferenceServer(predict)\n\nif os.getenv('KAGGLE_IS_COMPETITION_RERUN'):\n inference_server.serve()\nelse:\n inference_server.run_local_gateway(\n data_paths=(\n '/kaggle/input/cmi-detect-behavior-with-sensor-data/test.csv',\n '/kaggle/input/cmi-detect-behavior-with-sensor-data/test_demographics.csv',\n )\n )","metadata":{"trusted":true,"_kg_hide-output":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\n'''import shutil\nfrom pathlib import Path\n\nPRETRAINED_DIR = Path(\"/kaggle/input/cmi-d-111\")\n# 设置要打包的目录\nkaggleEvalDir = Path(\"/kaggle/input/cmi-detect-behavior-with-sensor-data/kaggle_evaluation\")\nlibDir = Path(\"/kaggle/usr/lib/cmi_2025_metric_copy_for_import/\")\n# 设置输出的 zip 文件路径\nzip_file_path = Path(\"/kaggle/working/libDir.zip\")\n\n# 打包整个目录为 .zip 文件\nshutil.make_archive(str(zip_file_path).replace('.zip', ''), 'zip', str(libDir))\n\nprint(f\"Directory has been zipped to {zip_file_path}\")\n\n'''","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"'''\n\nfrom IPython.display import FileLink\nFileLink(r'/kaggle/working/my_directory.zip')\n\n'''","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\n'''\nimport inspect\nimport cmi_2025_metric_copy_for_import\n\nprint(inspect.getfile(cmi_2025_metric_copy_for_import))'''","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}