{"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":31040,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# | CMI: Sequence Data Augmentation\n***\n\nIn this notebook I propose some data augmentations for this competition. Despite that our data is not sound sequences, some may still be applied if they are correctly adapted to our domain. Most of these functions are modified to work with our competition data which consists of short sequences (~100 steps) as arrays of length `L` and `C` channels.\n\nHere I propose the following augmentations, many more can be crafted, if you want to share your augmentations please leave a comment below:\n- GaussianNoise\n- PinkNoiseSNR\n- TimeStretch\n- TimeShift\n- ButterFilter\n- MixUp\n\nThis notebook was inspired by the great [notebook][1] of Hidehisa Arai @hidehisaarai1213 !\n\nI hope you learn something new from this work! 🙌🏼\n\n### Table of Contents \n
\n
  • Introduction
    • \n
  • Install libraries
    • \n
  • Import Libraries
    • \n
  • Configuration
    • \n
  • Utils
    • \n
  • Load Data
    • \n
  • Data Pre-processing
    • \n
  • Data Augmentation
    • \n
  • Gaussian Noise
    • \n
  • Pink Noise
    • \n
  • Time Stretch
    • \n
  • Time Shift
    • \n
  • Butter Filter
    • \n
  • Dataset
    • \n
  • MixUp
    • \n \n \n
    \n\n[1]: https://www.kaggle.com/code/hidehisaarai1213/rfcx-audio-data-augmentation-japanese-english","metadata":{}},{"cell_type":"markdown","source":"# | Install libraries [↑](#top) \n\n***","metadata":{}},{"cell_type":"code","source":"!pip install colorednoise","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-16T18:13:08.683699Z","iopub.execute_input":"2025-06-16T18:13:08.684017Z","iopub.status.idle":"2025-06-16T18:13:14.705978Z","shell.execute_reply.started":"2025-06-16T18:13:08.683992Z","shell.execute_reply":"2025-06-16T18:13:14.704893Z"},"_kg_hide-input":true,"_kg_hide-output":true,"scrolled":true,"jupyter":{"source_hidden":true}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | Import libraries [↑](#top) \n\n***","metadata":{}},{"cell_type":"code","source":"import colorednoise as cn\nimport gc\nimport librosa\nimport math\nimport matplotlib.pyplot as plt\nimport numpy as np\nimport os\nimport pandas as pd\nimport plotly.express as px\nimport plotly.graph_objects as go\nimport plotly.io as pio\nimport random\nimport time\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\n\n\nfrom copy import deepcopy\nfrom glob import glob\nfrom plotly.subplots import make_subplots\nfrom scipy.signal import butter, lfilter\nfrom torch.utils.data import DataLoader, Dataset, Subset\nfrom tqdm import tqdm\n\ndevice = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\npio.renderers.default = 'iframe'","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-06-16T19:18:32.470720Z","iopub.execute_input":"2025-06-16T19:18:32.471048Z","iopub.status.idle":"2025-06-16T19:18:32.479196Z","shell.execute_reply.started":"2025-06-16T19:18:32.471025Z","shell.execute_reply":"2025-06-16T19:18:32.477975Z"},"_kg_hide-input":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | Configuration [↑](#top) \n\n***","metadata":{}},{"cell_type":"code","source":"class config:\n MIXUP_PROB = 0.5\n\nclass paths:\n OUTPUT_DIR = \"/kaggle/working/output\"\n TEST_CSV = \"/kaggle/input/cmi-detect-behavior-with-sensor-data/test.csv\"\n TEST_DEMOGRAPHICS = \"/kaggle/input/cmi-detect-behavior-with-sensor-data/test_demographics.csv\"\n TRAIN_CSV = \"/kaggle/input/cmi-detect-behavior-with-sensor-data/train.csv\"\n TRAIN_DEMOGRAPHICS = \"/kaggle/input/cmi-detect-behavior-with-sensor-data/train_demographics.csv\"","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-16T18:13:57.494683Z","iopub.execute_input":"2025-06-16T18:13:57.494989Z","iopub.status.idle":"2025-06-16T18:13:57.501309Z","shell.execute_reply.started":"2025-06-16T18:13:57.494967Z","shell.execute_reply":"2025-06-16T18:13:57.500083Z"},"_kg_hide-input":true,"jupyter":{"source_hidden":true}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | Utils [↑](#top) \n\n***","metadata":{}},{"cell_type":"code","source":"def plot_dual_timeseries_px(x, x_aug, titles=['Original', 'Augmented']):\n \"\"\"\n Alternative approach using plotly express with melted dataframes\n \"\"\"\n L, N = x.shape\n cols = [\"acc_x\", \"acc_y\", \"acc_z\", \"rot_w\", \"rot_x\", \"rot_y\", \"rot_z\"]\n # Create dataframes for both arrays\n df1 = pd.DataFrame(x, columns=cols)\n df1['Time'] = np.arange(L)\n df1['Dataset'] = titles[0]\n df1_melted = df1.melt(id_vars=['Time', 'Dataset'], var_name='Channel', value_name='Value')\n\n L, N = x_aug.shape\n df2 = pd.DataFrame(x_aug, columns=cols)\n df2['Time'] = np.arange(L)\n df2['Dataset'] = titles[1]\n df2_melted = df2.melt(id_vars=['Time', 'Dataset'], var_name='Channel', value_name='Value')\n \n # Combine dataframes\n df_combined = pd.concat([df1_melted, df2_melted], ignore_index=True)\n \n # Create the plot\n fig = px.line(\n df_combined, \n x='Time', \n y='Value', \n color='Channel',\n facet_col='Dataset',\n title='Time Series Comparison'\n )\n \n # Update layout\n fig.update_layout(\n height=600, width=1000,\n template='plotly_dark',\n )\n \n return fig\n \n\ndef seed_everything(seed: int):\n random.seed(seed)\n np.random.seed(seed)\n torch.manual_seed(seed)\n os.environ['PYTHONHASHSEED'] = str(seed) \n \n\ndef sep():\n print(\"—\"*100)\n\n\nlabel_to_num = {\n 'Above ear - pull hair': 0, # < ------- TARGETS START\n 'Cheek - pinch skin': 1,\n 'Eyebrow - pull hair': 2,\n 'Eyelash - pull hair': 3,\n 'Forehead - pull hairline': 4,\n 'Forehead - scratch': 5,\n 'Neck - pinch skin': 6,\n 'Neck - scratch': 7, # < ------- TARGETS END\n 'Drink from bottle/cup': 8, # < ------- NON-TARGETS START\n 'Feel around in tray and pull out an object': 9,\n 'Glasses on/off': 10,\n 'Pinch knee/leg skin': 11,\n 'Pull air toward your face': 12,\n 'Scratch knee/leg skin': 13,\n 'Text on phone': 14,\n 'Wave hello': 15,\n 'Write name in air': 16,\n 'Write name on leg': 17 # < ------- NON-TARGETS END\n}\ntype_to_num = {\"Target\": 1, \"Non-Target\": 0}\nnum_to_label = {v: k for k, v in label_to_num.items()}\nnum_to_type = {v: k for k, v in type_to_num.items()}","metadata":{"execution":{"iopub.status.busy":"2025-06-16T19:36:20.230864Z","iopub.execute_input":"2025-06-16T19:36:20.231285Z","iopub.status.idle":"2025-06-16T19:36:20.244129Z","shell.execute_reply.started":"2025-06-16T19:36:20.231259Z","shell.execute_reply":"2025-06-16T19:36:20.242960Z"},"trusted":true,"_kg_hide-input":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | Load Data [↑](#top) \n\n***","metadata":{}},{"cell_type":"code","source":"df_train = pd.read_csv(paths.TRAIN_CSV)\ndf_test = pd.read_csv(paths.TEST_CSV)\ndf_train_demographics = pd.read_csv(paths.TRAIN_DEMOGRAPHICS)\ndf_test_demographics = pd.read_csv(paths.TEST_DEMOGRAPHICS)\n\nprint(f\"Train dataframe shape: {df_train.shape}\"), sep()\nprint(f\"Tesat dataframe shape: {df_test.shape}\"), sep()\nprint(f\"Train demographics dataframe shape: {df_train_demographics.shape}\"), sep()\nprint(f\"Test demographics dataframe shape: {df_test_demographics.shape}\")\n\ndf_train[\"target\"] = df_train[\"gesture\"].map(label_to_num)\ndf_train[\"sequence_type\"] = df_train[\"sequence_type\"].map(type_to_num)\n\ndisplay(df_train.head())\ndisplay(df_train_demographics.head())","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-16T18:16:34.915208Z","iopub.execute_input":"2025-06-16T18:16:34.916262Z","iopub.status.idle":"2025-06-16T18:17:03.023268Z","shell.execute_reply.started":"2025-06-16T18:16:34.916215Z","shell.execute_reply":"2025-06-16T18:17:03.022072Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | Data pre-processing [↑](#top) \n\n***","metadata":{}},{"cell_type":"code","source":"def standard_scale(arr: np.ndarray) -> np.ndarray:\n means = np.nanmean(arr, axis=0)\n stds = np.nanstd(arr, axis=0)\n stds = np.where(stds == 0, 1, stds) # Prevent division by zero for constant columns\n scaled = (arr - means) / stds\n return scaled\n\n\ndef pad_or_truncate(\n arr: np.ndarray,\n max_length: int = 200,\n pad_value: int = 0,\n mode: str = \"random\" # \"regular\" or \"random\"\n) -> np.ndarray:\n L, D = arr.shape\n\n if L > max_length:\n return arr[:max_length, :]\n\n elif L < max_length:\n if mode == \"regular\":\n padding = np.full((max_length - L, D), pad_value)\n return np.vstack((arr, padding))\n \n elif mode == \"random\":\n total_padding = max_length - L\n pad_start = np.random.randint(0, total_padding + 1)\n pad_end = total_padding - pad_start\n\n start_padding = np.full((pad_start, D), pad_value)\n end_padding = np.full((pad_end, D), pad_value)\n\n return np.vstack((start_padding, arr, end_padding))\n \n else:\n raise ValueError(f\"Unknown mode: {mode}. Use 'regular' or 'random'.\")\n\n else:\n return arr\n\n\nimu_cols = [\"acc_x\", \"acc_y\", \"acc_z\", \"rot_w\", \"rot_x\", \"rot_y\", \"rot_z\"]\nall_X, all_y, all_y_hard = [], [], []\nlengths = []\nsequence_ids = df_train.sequence_id.unique()[0:100]\n\nfor sequence_id in tqdm(sequence_ids):\n ds = df_train[df_train[\"sequence_id\"] == sequence_id]\n X = ds[imu_cols].values\n lengths.append(X.shape[0])\n y = ds.target.values[0]\n y_hard = ds.sequence_type.values[0]\n acc = standard_scale(X[:, 0:3])\n acc = np.where(np.isnan(acc), 0.0, acc)\n rot = X[:, 3:]\n X = np.concatenate([acc, rot], axis=1)\n X = np.where(np.isnan(X), 0.0, X) # fill NaNs\n all_X.append(X)\n all_y.append(y)\n all_y_hard.append(y_hard)\n\nall_y = np.array(all_y)\nall_y_hard = np.array(all_y_hard)\nprint(len(X))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-16T19:18:17.361191Z","iopub.execute_input":"2025-06-16T19:18:17.361610Z","iopub.status.idle":"2025-06-16T19:18:22.388521Z","shell.execute_reply.started":"2025-06-16T19:18:17.361583Z","shell.execute_reply":"2025-06-16T19:18:22.387451Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | Data Augmentation [↑](#top) \n\n***\n\nThe following classes were inspired by the [albumentations][1] library. These are not our data augmentations per se but helper classes that will help us build our data augmentation pipeline.\n\n- `SignalTransform`: A base class for all signal transformations (augmentations). It defines when and how a transform should be applied, using probabilistic control.\n- `Compose`: A pipeline class that applies multiple transforms in sequence.\n- `OneOf`: Randomly chooses one transform from a list and applies only that one.\n\n[1]: https://albumentations.ai/docs/","metadata":{}},{"cell_type":"code","source":"class SignalTransform:\n def __init__(self, always_apply: bool = False, p: float = 0.5):\n self.always_apply = always_apply\n self.p = p\n\n def __call__(self, y: np.ndarray):\n if self.always_apply:\n return self.apply(y)\n else:\n if np.random.rand() < self.p:\n return self.apply(y)\n else:\n return y\n\n def apply(self, y: np.ndarray):\n raise NotImplementedError\n\n\nclass Compose:\n def __init__(self, transforms: list):\n self.transforms = transforms\n\n def __call__(self, y: np.ndarray):\n for trns in self.transforms:\n y = trns(y)\n return y\n\n\nclass OneOf:\n def __init__(self, transforms: list):\n self.transforms = transforms\n\n def __call__(self, y: np.ndarray):\n n_trns = len(self.transforms)\n trns_idx = np.random.choice(n_trns)\n trns = self.transforms[trns_idx]\n return trns(y)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-16T18:25:36.256467Z","iopub.execute_input":"2025-06-16T18:25:36.256837Z","iopub.status.idle":"2025-06-16T18:25:36.265764Z","shell.execute_reply.started":"2025-06-16T18:25:36.256811Z","shell.execute_reply":"2025-06-16T18:25:36.264657Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | Gaussian Noise [↑](#top) \n\n***\n\nThe `AddGaussianNoise` class is a data augmentation transform for adding Gaussian (normally distributed) noise to a signal, often used in training models to improve robustness and generalization.\n\nThis class inherits from `SignalTransform`, meaning it benefits from the probabilistic behavior defined there (i.e., apply the transformation with a certain probability p).\n\nThis class randomly selects a noise scale between 0.0 and `max_noise_amplitude`. It then generates standard normal noise (mean 0, std 1) with the same shape as the input signal. Finally, it scales the noise by `noise_amplitude`, adds it to the signal, and casts the result back to the original data type.","metadata":{}},{"cell_type":"code","source":"class GaussianNoise(SignalTransform):\n def __init__(\n self, always_apply: bool = False, \n p: float = 0.5, max_noise_amplitude: float = 0.20, **kwargs\n ):\n super().__init__(always_apply, p)\n self.noise_amplitude = (0.0, max_noise_amplitude)\n\n def apply(self, x: np.ndarray, **params):\n noise_amplitude = np.random.uniform(*self.noise_amplitude)\n noise = np.random.randn(*x.shape) # shape (L, N)\n augmented = (x + noise * noise_amplitude).astype(x.dtype)\n return augmented\n\ni = np.random.choice(range(len(X)))\nx = all_X[i]\ntransforms = GaussianNoise(always_apply=True)\nx_aug = transforms(x)\n\nplot_dual_timeseries_px(x, x_aug)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-16T19:36:24.350356Z","iopub.execute_input":"2025-06-16T19:36:24.350683Z","iopub.status.idle":"2025-06-16T19:36:24.516318Z","shell.execute_reply.started":"2025-06-16T19:36:24.350664Z","shell.execute_reply":"2025-06-16T19:36:24.515329Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | Pink Noise [↑](#top) \n\n***\n\nThis class adds **pink noise** to a multichannel time-series signal while maintaining a controlled signal-to-noise ratio (SNR).\n\nIts purpose is to simulate realistic background noise in a signal by adding **pink noise** (a common type of noise in natural systems where power decreases with frequency) at a random SNR between `min_snr` and `max_snr`. The result is a noisy version of the input signal with pink noise at a controlled intensity.\n\nBrief Explanation\n\n1. Randomly selects an SNR from the specified range.\n2. Computes the amplitude of the signal.\n3. Calculates the noise amplitude needed to achieve the target SNR.\n4. Generates pink noise for each channel using `cn.powerlaw_psd_gaussian(1, len(y))`.\n5. Normalizes the pink noise to match the target amplitude.\n6. Adds the scaled pink noise to the original signal.","metadata":{}},{"cell_type":"code","source":"class PinkNoiseSNR(SignalTransform):\n def __init__(self, always_apply=False, p=0.5, min_snr=5.0, max_snr=20.0, **kwargs):\n super().__init__(always_apply, p)\n self.min_snr = min_snr\n self.max_snr = max_snr\n\n def apply(self, y: np.ndarray, **params):\n snr = np.random.uniform(self.min_snr, self.max_snr)\n a_signal = np.sqrt((y ** 2).max(axis=0)) # shape: (N,)\n a_noise = a_signal / (10 ** (snr / 20)) # shape: (N,)\n pink_noise = np.stack([cn.powerlaw_psd_gaussian(1, len(y)) for _ in range(y.shape[1])], axis=1)\n a_pink = np.sqrt((pink_noise ** 2).max(axis=0)) # shape: (N,)\n pink_noise_normalized = pink_noise * (a_noise / a_pink)\n augmented = (y + pink_noise_normalized).astype(y.dtype)\n return augmented\n\n\ni = np.random.choice(range(len(X)))\nx = all_X[i]\ntransforms = PinkNoiseSNR(always_apply=True)\nx_aug = transforms(x)\n\nplot_dual_timeseries_px(x, x_aug)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-16T18:39:23.170894Z","iopub.execute_input":"2025-06-16T18:39:23.171329Z","iopub.status.idle":"2025-06-16T18:39:23.369941Z","shell.execute_reply.started":"2025-06-16T18:39:23.171303Z","shell.execute_reply":"2025-06-16T18:39:23.369018Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | Time Stretch [↑](#top) \n\n***\n\nThis class changes the speed (duration) of a 1D or 2D signal by stretching or compressing it in the time dimension without changing the sampling rate.\n\nIts purpose is to simulate variations in signal duration by speeding up or slowing down the signal, which helps improve model robustness to different timing or tempo variations in signals like audio, sensor data, or time series.\n\nBrief Explanation:\n\n1. Randomly selects a stretch factor (`rate`) between `min_rate` and `max_rate`.\n2. Calculates the new length of the signal after stretching/compressing.\n3. Uses linear interpolation to resample the original signal over the new time indices.\n4. Supports both 1D (single channel) and 2D (multi-channel) input arrays.\n5. Returns the time-stretched signal with shape adjusted according to the rate.","metadata":{}},{"cell_type":"code","source":"class TimeStretch(SignalTransform):\n def __init__(self, max_rate=1.5, min_rate=0.5, always_apply=False, p=0.5):\n super().__init__(always_apply, p)\n self.max_rate = max_rate\n self.min_rate = min_rate\n self.always_apply = always_apply\n self.p = p\n\n def apply(self, x: np.ndarray):\n \"\"\"\n Stretch a 1D or 2D array in time using linear interpolation.\n - x: np.ndarray of shape (L,) or (L, N)\n - rate: float, e.g., 1.2 for 20% longer, 0.8 for 20% shorter\n \"\"\"\n rate = np.random.uniform(self.min_rate, self.max_rate)\n L = x.shape[0]\n L_new = int(L / rate)\n orig_idx = np.linspace(0, L - 1, num=L)\n new_idx = np.linspace(0, L - 1, num=L_new)\n\n if x.ndim == 1:\n return np.interp(new_idx, orig_idx, x)\n elif x.ndim == 2:\n return np.stack([\n np.interp(new_idx, orig_idx, x[:, i]) for i in range(x.shape[1])\n ], axis=1)\n else:\n raise ValueError(\"Only 1D or 2D arrays are supported.\")\n\n\ni = np.random.choice(range(len(X)))\nx = all_X[i]\ntransforms = TimeStretch(always_apply=True)\nx_aug = transforms(x)\n\nplot_dual_timeseries_px(x, x_aug)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-16T18:43:38.571360Z","iopub.execute_input":"2025-06-16T18:43:38.571658Z","iopub.status.idle":"2025-06-16T18:43:38.713575Z","shell.execute_reply.started":"2025-06-16T18:43:38.571639Z","shell.execute_reply":"2025-06-16T18:43:38.712616Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | Time Shift [↑](#top) \n\n***\n\n\nThis class randomly shifts a multichannel time-series signal along the time axis, simulating temporal misalignment or delay.\n\nIts purpose is to introduce random temporal shifts in the input signal to make models more robust to timing variations or misalignments, commonly useful in audio, sensor, or sequential data.\n\nBrief Explanation\n\n1. Confirms the input is a 2D array with shape `(L, N)` where L is time length.\n2. Computes the maximum allowed shift in samples based on `max_shift_pct`.\n3. Randomly chooses a shift value between `-max_shift` and `+max_shift`.\n4. Rolls (circularly shifts) the array along the time axis by the chosen amount.\n5. If `padding_mode` is `\"zero\"`, replaces the wrapped-around shifted section with zeros to avoid artificial wraparound artifacts.\n6. Returns the shifted signal.","metadata":{}},{"cell_type":"code","source":"class TimeShift(SignalTransform):\n def __init__(self, always_apply=False, p=0.5, max_shift_pct=0.25, padding_mode=\"replace\"):\n super().__init__(always_apply, p)\n \n assert 0 <= max_shift_pct <= 1.0, \"`max_shift_pct` must be between 0 and 1\"\n assert padding_mode in [\"replace\", \"zero\"], \"`padding_mode` must be either 'replace' or 'zero'\"\n \n self.max_shift_pct = max_shift_pct\n self.padding_mode = padding_mode\n\n def apply(self, x: np.ndarray, **params):\n assert x.ndim == 2, \"`x` must be a 2D array with shape (L, N)\"\n \n L = x.shape[0]\n max_shift = int(L * self.max_shift_pct)\n shift = np.random.randint(-max_shift, max_shift + 1)\n\n # Roll along time axis (axis=0)\n augmented = np.roll(x, shift, axis=0)\n\n if self.padding_mode == \"zero\":\n if shift > 0:\n augmented[:shift, :] = 0\n elif shift < 0:\n augmented[shift:, :] = 0\n\n return augmented\n\n\ni = np.random.choice(range(len(X)))\nx = all_X[i]\ntransforms = TimeShift(always_apply=True)\nx_aug = transforms(x)\n\nplot_dual_timeseries_px(x, x_aug)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-16T18:44:07.130934Z","iopub.execute_input":"2025-06-16T18:44:07.131338Z","iopub.status.idle":"2025-06-16T18:44:07.270971Z","shell.execute_reply.started":"2025-06-16T18:44:07.131314Z","shell.execute_reply":"2025-06-16T18:44:07.269780Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | Butter Filter [↑](#top) \n\n***\n\nThis class applies a low-pass Butterworth filter to multichannel time-series data.\n\nIts purpose is to remove high-frequency components from the signal by applying a smooth low-pass filter, which helps reduce noise or unwanted rapid fluctuations, often used in sensor data preprocessing or audio denoising.\n\nBrief Explanation:\n\n1. Checks input is 2D with shape `(L, N)` (time length and channels).\n2. Calls `butter_lowpass_filter` to process the signal.\n3. Calculates normalized cutoff frequency relative to Nyquist frequency.\n4. Designs Butterworth low-pass filter coefficients.\n5. Applies the filter independently on each channel using `lfilter`.\n6. Returns the filtered signal.","metadata":{}},{"cell_type":"code","source":"class ButterFilter(SignalTransform):\n def __init__(self, always_apply=False, p=0.5, cutoff_freq=20, sampling_rate=200, order=4):\n super().__init__(always_apply, p)\n \n self.cutoff_freq = cutoff_freq\n self.sampling_rate = sampling_rate\n self.order = order\n\n def apply(self, x: np.ndarray, **params):\n assert x.ndim == 2, \"`x` must be a 2D array with shape (L, N)\"\n return self.butter_lowpass_filter(x)\n\n def butter_lowpass_filter(self, data: np.ndarray):\n nyquist = 0.5 * self.sampling_rate\n normal_cutoff = self.cutoff_freq / nyquist\n b, a = butter(self.order, normal_cutoff, btype='low', analog=False)\n filtered_data = lfilter(b, a, data, axis=0) # filter each channel independently\n return filtered_data\n\n\ni = np.random.choice(range(len(X)))\nx = all_X[i]\ntransforms = ButterFilter(always_apply=True)\nx_aug = transforms(x)\n\nplot_dual_timeseries_px(x, x_aug)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-16T18:44:30.910308Z","iopub.execute_input":"2025-06-16T18:44:30.911437Z","iopub.status.idle":"2025-06-16T18:44:31.106794Z","shell.execute_reply.started":"2025-06-16T18:44:30.911400Z","shell.execute_reply":"2025-06-16T18:44:31.105729Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | Composed [↑](#top) \n\n***\n\nNow let's compose different augmentations into a single pipeline with different probabilities. In this example, only on of the three transformations inside the `OneOf` class is selected, if its probability is higher than `p` it is applied. Then the remaining transformations are applied if their probability is higher than `p`. Transformations are applied sequentially.","metadata":{}},{"cell_type":"code","source":"i = np.random.choice(range(len(X)))\nx = all_X[i]\ntransforms = Compose([\n OneOf([\n GaussianNoise(p=0.5, max_noise_amplitude=0.05),\n PinkNoiseSNR(p=0.5, min_snr=4.0, max_snr=20.0),\n ButterFilter(p=0.5)\n ]),\n TimeStretch(p=0.25),\n TimeShift(p=0.25)\n\n])\nx_aug = transforms(x)\n\nplot_dual_timeseries_px(x, x_aug)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-16T18:45:14.594519Z","iopub.execute_input":"2025-06-16T18:45:14.594856Z","iopub.status.idle":"2025-06-16T18:45:14.724369Z","shell.execute_reply.started":"2025-06-16T18:45:14.594833Z","shell.execute_reply":"2025-06-16T18:45:14.723169Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | Dataset [↑](#top) \n\n***\n\nInside this `Dataset` class we will select and structure our sequence data and controll which transformations will be applied. In addition, `MixUp` is implemented inside the `__getitem__` method, which mixes two sequences data linearly and combines their respective ground truths accordingly as a weighted average.","metadata":{}},{"cell_type":"code","source":"class CustomDataset(Dataset):\n def __init__(\n self, config, df: pd.DataFrame, X: list[np.ndarray], y: list[np.ndarray], y_hard: list[np.ndarray],\n transforms = None, mode: str = \"train\"\n ): \n self.config = config\n self.df = df\n self.X = X\n self.y = y\n self.y_hard = y_hard\n self.indexes = self.df.sequence_id.unique()[0:100]\n self.alpha = 0.3\n self.mode = mode\n self.transforms = transforms\n self.num_classes = 18\n \n def __len__(self):\n \"\"\"\n Length of dataset.\n \"\"\"\n return len(self.indexes)\n \n def __getitem__(self, i):\n \"\"\"\n Get one item.\n \"\"\"\n sequence_id = self.indexes[i]\n p = np.random.rand()\n if p <= 0.0:\n if self.mode == \"train\":\n X, y, y_hard = self.get_data(i)\n X = self.transforms(X)\n X = pad_or_truncate(X)\n else:\n X, y, y_hard = self.get_data(i)\n X = pad_or_truncate(X)\n else:\n if self.mode == \"train\":\n lam = np.random.beta(self.alpha, self.alpha)\n j = np.random.randint(0, len(self.indexes))\n X1, y1, y1_hard = self.get_data(i)\n X2, y2, y2_hard = self.get_data(j)\n X1, X2 = pad_or_truncate(X1), pad_or_truncate(X2) \n X = lam * X1 + (1 - lam) * X2\n y = lam * y1 + (1 - lam) * y2\n y_hard = lam * y1_hard + (1 - lam) * y2_hard\n else:\n X, y, y_hard = self.get_data(i)\n X = pad_or_truncate(X)\n \n output = {\n \"X\": torch.tensor(X, dtype=torch.float32),\n \"y\": torch.tensor(y, dtype=torch.float32),\n \"y_hard\": torch.tensor(y_hard, dtype=torch.float32),\n \"sequence_id\": sequence_id\n }\n return output\n\n def get_data(self, index):\n X = self.X[index]\n y = self.y[index]\n y = np.eye(self.num_classes, dtype=int)[y]\n y_hard = self.y_hard[index]\n return X, y, y_hard\n ","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-16T19:29:12.665121Z","iopub.execute_input":"2025-06-16T19:29:12.665555Z","iopub.status.idle":"2025-06-16T19:29:12.680241Z","shell.execute_reply.started":"2025-06-16T19:29:12.665525Z","shell.execute_reply":"2025-06-16T19:29:12.679235Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# | MixUp [↑](#top) \n\n***","metadata":{}},{"cell_type":"code","source":"train_dataset = CustomDataset(config, df_train, all_X, all_y, all_y_hard, transforms, mode=\"train\")\ntrain_loader = DataLoader(\n train_dataset,\n batch_size=2,\n shuffle=False,\n num_workers=0,\n pin_memory=True, drop_last=True\n)\nidx = np.random.choice(range(0, len(train_dataset)))\ncols = imu_cols\nsample = train_dataset[idx]\nX = sample[\"X\"]\ny = sample[\"y\"]\ny_hard = sample[\"y_hard\"]\nprint(f\"Mixup y_true: {y}\"), sep()\nprint(f\"Mixup y_true_hard: {y_hard}\"), sep()\nsequence_id = train_dataset[idx][\"sequence_id\"]\nN = X.shape[0]\ndf = pd.DataFrame(X, columns=cols)\ndf['step'] = range(N)\ndf_melted = df.melt(id_vars='step', var_name='sequence', value_name='value')\n\nfig = px.line(\n df_melted,\n x='step',\n y='value',\n color='sequence',\n title=f\"Sequences for {sequence_id}\",\n template='plotly_dark',\n width=800,\n height=600\n)\n\nfig.show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-16T19:29:21.268459Z","iopub.execute_input":"2025-06-16T19:29:21.268843Z","iopub.status.idle":"2025-06-16T19:29:21.733789Z","shell.execute_reply.started":"2025-06-16T19:29:21.268814Z","shell.execute_reply":"2025-06-16T19:29:21.732872Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}