{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":97984,"databundleVersionId":14096757,"sourceType":"competition"}],"dockerImageVersionId":31153,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"
\n
\n

Published on October 31, 2025

\n

Author: Jocelyn C. Dumlao

\n

\n LinkedIn\n GitHub\n YouTube\n Kaggle\n
\n
","metadata":{}},{"cell_type":"markdown","source":"#

Import Libraries

","metadata":{}},{"cell_type":"code","source":"import os\nfrom glob import glob\nfrom collections import defaultdict\nfrom typing import Tuple\nimport numpy as np\nimport pandas as pd\nimport cv2\nimport matplotlib.pyplot as plt\nfrom tqdm import tqdm\nimport scipy.optimize\nimport scipy.signal","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-31T07:08:08.463536Z","iopub.execute_input":"2025-10-31T07:08:08.464119Z","iopub.status.idle":"2025-10-31T07:08:09.357174Z","shell.execute_reply.started":"2025-10-31T07:08:08.464091Z","shell.execute_reply":"2025-10-31T07:08:09.356482Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"#

Constants & Exceptions

","metadata":{}},{"cell_type":"code","source":"# Constants & Exceptions\n\nLEADS = ['I', 'II', 'III', 'aVR', 'aVL', 'aVF', 'V1', 'V2', 'V3', 'V4', 'V5', 'V6']\nMAX_TIME_SHIFT = 0.2 # seconds\nPERFECT_SCORE = 384\n\n\nclass ParticipantVisibleError(Exception):\n pass","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-31T07:08:13.697475Z","iopub.execute_input":"2025-10-31T07:08:13.698346Z","iopub.status.idle":"2025-10-31T07:08:13.703289Z","shell.execute_reply.started":"2025-10-31T07:08:13.698318Z","shell.execute_reply":"2025-10-31T07:08:13.702278Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"#

Utility Helpers

","metadata":{}},{"cell_type":"code","source":"# Utility helpers\n\ndef safe_read_image(path: str) -> np.ndarray:\n \"\"\"Read image and raise clear error if not found.\"\"\"\n img = cv2.imread(path)\n if img is None:\n raise FileNotFoundError(f\"Image not found or unreadable: {path}\")\n return img\n\n\ndef is_color_image(ima: np.ndarray) -> bool:\n \"\"\"Test if a 3-channel image has colors (non-zero std across channels).\"\"\"\n if ima is None:\n return False\n if ima.ndim < 3:\n return False\n return ima.std(axis=2).mean() != 0","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-31T07:08:16.665717Z","iopub.execute_input":"2025-10-31T07:08:16.666084Z","iopub.status.idle":"2025-10-31T07:08:16.672691Z","shell.execute_reply.started":"2025-10-31T07:08:16.666058Z","shell.execute_reply":"2025-10-31T07:08:16.671612Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"#

Scoring Utilities

","metadata":{}},{"cell_type":"code","source":"# Scoring utilities (SNR metric)\n\ndef compute_power(label: np.ndarray, prediction: np.ndarray) -> Tuple[float, float]:\n \"\"\"Compute signal and noise power. Inputs must be 1-D arrays.\"\"\"\n label = np.asarray(label, dtype=np.float64)\n prediction = np.asarray(prediction, dtype=np.float64)\n\n if label.ndim != 1 or prediction.ndim != 1:\n raise ParticipantVisibleError('Inputs must be 1-dimensional arrays.')\n\n finite_mask = np.isfinite(prediction)\n if not np.any(finite_mask):\n raise ParticipantVisibleError(\"The 'prediction' array contains no finite values (all NaN or inf).\")\n\n # Replace non-finite values in prediction with 0 for noise calc\n prediction = prediction.copy()\n prediction[~np.isfinite(prediction)] = 0\n noise = label - prediction\n p_signal = np.sum(label ** 2)\n p_noise = np.sum(noise ** 2)\n return p_signal, p_noise\n\n\ndef compute_snr(signal: float, noise: float) -> float:\n \"\"\"Return SNR or capped PERFECT_SCORE for perfect reconstructions.\"\"\"\n if noise == 0:\n return PERFECT_SCORE\n if signal == 0:\n return 0.0\n return min((signal / noise), PERFECT_SCORE)\n\n\ndef align_signals(label: np.ndarray, pred: np.ndarray, max_shift: float = float('inf')) -> np.ndarray:\n \"\"\"\n Align pred to label by cross-correlation (time shift) and vertical offset (minimize squared error).\n max_shift is given in samples (not seconds) in this implementation when provided as int.\n \"\"\"\n label_arr = np.asarray(label, dtype=np.float64)\n pred_arr = np.asarray(pred, dtype=np.float64)\n\n if np.any(~np.isfinite(label_arr)):\n raise ParticipantVisibleError('values in label should all be finite')\n if np.sum(np.isfinite(pred_arr)) == 0:\n raise ParticipantVisibleError('prediction can not all be infinite')\n\n label_mean = np.mean(label_arr)\n pred_mean = np.mean(pred_arr)\n\n label_centered = label_arr - label_mean\n pred_centered = pred_arr - pred_mean\n\n correlation = scipy.signal.correlate(label_centered, pred_centered, mode='full')\n n_label = label_arr.size\n n_pred = pred_arr.size\n lags = scipy.signal.correlation_lags(n_label, n_pred, mode='full')\n\n # If max_shift is a float (seconds) caller must convert to samples; here accept int as pre-converted too.\n if np.isfinite(max_shift) and max_shift < np.inf:\n # ensure mask length matches lags\n valid_lags_mask = (lags >= -int(max_shift)) & (lags <= int(max_shift))\n else:\n valid_lags_mask = np.ones_like(lags, dtype=bool)\n\n # restrict correlation to valid lags and get best lag\n corr_valid = correlation.copy()\n corr_valid[~valid_lags_mask] = -np.inf\n best_idx = int(np.nanargmax(corr_valid))\n time_shift = lags[best_idx]\n\n # Build aligned_pred by padding with nan where out-of-range\n start_padding_len = max(time_shift, 0)\n pred_slice_start = max(-time_shift, 0)\n pred_slice_end = min(n_pred, n_label - time_shift)\n end_padding_len = max(n_label - n_pred - time_shift, 0)\n middle = pred_arr[pred_slice_start:pred_slice_end]\n aligned_pred = np.concatenate((np.full(start_padding_len, np.nan), middle, np.full(end_padding_len, np.nan)))\n\n # Optimize vertical offset (v_shift) to minimize squared error\n def objective(v_shift):\n return np.nansum((label_arr - (aligned_pred - v_shift)) ** 2)\n\n if np.any(np.isfinite(label_arr) & np.isfinite(aligned_pred)):\n res = scipy.optimize.minimize_scalar(objective, method='Brent')\n v_shift = float(res.x)\n aligned_pred = aligned_pred - v_shift\n\n return aligned_pred\n\ndef _calculate_image_score(group: pd.DataFrame) -> float:\n \"\"\"Helper function to calculate the total SNR score for a single image group.\"\"\"\n\n unique_fs_values = group['fs'].unique()\n if len(unique_fs_values) != 1:\n raise ParticipantVisibleError('Sampling frequency should be consistent across each ecg')\n sampling_frequency = unique_fs_values[0]\n \n sum_signal = 0\n sum_noise = 0\n for lead in LEADS:\n sub = group[group['lead'] == lead]\n label = sub['value_true'].values\n pred = sub['value_pred'].values\n\n aligned_pred = align_signals(label, pred, int(sampling_frequency * MAX_TIME_SHIFT))\n p_signal, p_noise = compute_power(label, aligned_pred)\n sum_signal += p_signal\n sum_noise += p_noise\n return compute_snr(sum_signal, sum_noise)\n\n\ndef score(solution: pd.DataFrame, submission: pd.DataFrame, row_id_column_name: str) -> float:\n \"\"\"\n Compute the mean Signal-to-Noise Ratio (SNR) across multiple ECG leads and images for the PhysioNet 2025 competition.\n The final score is the average of the sum of SNRs over different lines, averaged over all unique images.\n Args:\n solution: DataFrame with ground truth values. Expected columns: 'id' and one for each lead.\n submission: DataFrame with predicted values. Expected columns: 'id' and one for each lead.\n row_id_column_name: The name of the unique identifier column, typically 'id'.\n Returns:\n The final competition score.\n \"\"\"\n for df in [solution, submission]:\n if row_id_column_name not in df.columns:\n raise ParticipantVisibleError(f\"'{row_id_column_name}' column not found in DataFrame.\")\n if df['value'].isna().any():\n raise ParticipantVisibleError('NaN exists in solution/submission')\n if not np.isfinite(df['value']).all():\n raise ParticipantVisibleError('Infinity exists in solution/submission')\n\n submission = submission[['id', 'value']]\n solution.rename(columns={'value': 'value_true'}, inplace=True)\n submission.rename(columns={'value': 'value_pred'}, inplace=True)\n\n merged_df = pd.merge(solution, submission, on=row_id_column_name)\n merged_df['image_id'] = merged_df[row_id_column_name].str.split('_').str[0]\n merged_df['row_id'] = merged_df[row_id_column_name].str.split('_').str[1].astype('int64')\n merged_df['lead'] = merged_df[row_id_column_name].str.split('_').str[2]\n merged_df['fs'] = 500 # Assuming a default fs of 500, add fs to your solution/submission or load it dynamically if it varies\n\n merged_df.sort_values(by=['image_id', 'row_id', 'lead'], inplace=True)\n image_scores = merged_df.groupby('image_id').apply(_calculate_image_score, include_groups=False)\n mean_snr = image_scores.mean()\n if np.isnan(mean_snr):\n return -PERFECT_SCORE\n return max(float(10 * np.log10(mean_snr)), -PERFECT_SCORE)\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-31T07:08:19.747152Z","iopub.execute_input":"2025-10-31T07:08:19.747973Z","iopub.status.idle":"2025-10-31T07:08:19.767624Z","shell.execute_reply.started":"2025-10-31T07:08:19.747946Z","shell.execute_reply":"2025-10-31T07:08:19.766625Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"#

Mean Model / Data Loaders

","metadata":{}},{"cell_type":"code","source":"# Mean model / data loaders\n\ndef fit_mean_model(train_df, base_path):\n \"\"\"\n Build mean time-series per lead across training CSVs.\n Resample each series to 20000 samples for stacking.\n \"\"\"\n mean_dict = defaultdict(list)\n for idx, row in tqdm(train_df.iterrows(), total=len(train_df), desc=\"Calculating Mean ECG\"):\n csv_path = os.path.join(base_path, str(row.id), f\"{row.id}.csv\")\n if not os.path.exists(csv_path):\n # skip missing\n continue\n try:\n labels = pd.read_csv(csv_path)\n except FileNotFoundError:\n print(f\"Warning: CSV not found at {csv_path}. Skipping.\")\n continue\n\n for lead in labels.columns:\n values = labels[lead].dropna().values.astype(float)\n if values.size == 0:\n continue\n resampled = np.interp(np.linspace(0, values.size - 1, 20000), np.arange(values.size), values)\n mean_dict[lead].append(resampled)\n # Stack\n for lead in list(mean_dict.keys()):\n mean_dict[lead] = np.stack(mean_dict[lead], axis=0)\n return mean_dict\n\n\ndef validate_mean_model(val_df, mean_dict, base_path):\n snr_list = []\n for idx, row in tqdm(val_df.iterrows(), total=len(val_df), desc=\"Validating Mean Model\"):\n csv_path = os.path.join(base_path, str(row.id), f\"{row.id}.csv\")\n if not os.path.exists(csv_path):\n continue\n try:\n labels = pd.read_csv(csv_path)\n except FileNotFoundError:\n print(f\"Warning: CSV not found at {csv_path}. Skipping.\")\n continue\n\n sum_signal = 0.0\n sum_noise = 0.0\n fs = int(getattr(row, 'fs', 500))\n for lead in labels.columns:\n label = labels[lead].dropna().values.astype(float)\n if label.size == 0:\n continue\n # get mean pred and resample to label length\n if lead not in mean_dict:\n # fallback: zeros\n pred = np.zeros(label.size)\n else:\n pred_mean = mean_dict[lead].mean(axis=0)\n pred = np.interp(np.linspace(0, 1, label.size), np.linspace(0, 1, pred_mean.size), pred_mean)\n aligned_pred = align_signals(label, pred, int(fs * MAX_TIME_SHIFT))\n p_signal, p_noise = compute_power(label, aligned_pred)\n sum_signal += p_signal\n sum_noise += p_noise\n snr = compute_snr(sum_signal, sum_noise)\n snr_list.append(snr)\n if len(snr_list) == 0:\n return -PERFECT_SCORE\n mean_snr = np.mean(snr_list)\n val_score = max(float(10 * np.log10(mean_snr)), -PERFECT_SCORE)\n print(f\"# Validation SNR for mean prediction: {mean_snr:.2f} val_score={val_score:.2f}\")\n return val_score\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-31T07:08:26.701487Z","iopub.execute_input":"2025-10-31T07:08:26.702419Z","iopub.status.idle":"2025-10-31T07:08:26.714181Z","shell.execute_reply.started":"2025-10-31T07:08:26.702392Z","shell.execute_reply":"2025-10-31T07:08:26.713333Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"#

Marker Finder

","metadata":{}},{"cell_type":"code","source":"# Marker finder - robust version\n\nclass MarkerFinder:\n def __init__(self, show_templates=False, base_path='/kaggle/input/physionet-ecg-image-digitization/train'):\n # load sample images (use try/except if missing)\n candidate_paths = [\n os.path.join(base_path, '4292118763', '4292118763-0001.png'),\n os.path.join(base_path, '4289880010', '4289880010-0001.png'),\n os.path.join(base_path, '4284351157', '4284351157-0001.png'),\n ]\n imgs = []\n for p in candidate_paths:\n try:\n imgs.append(safe_read_image(p))\n except FileNotFoundError:\n # skip missing examples\n pass\n if len(imgs) == 0:\n raise FileNotFoundError(\"No template images found in candidate paths. Please check dataset mount.\")\n ima = np.maximum.reduce(imgs) # element-wise max across available templates\n\n # Absolute points originally computed for the dataset templates\n absolute_points = np.zeros((17, 2), dtype=int)\n for i in range(3):\n absolute_points[5 * i] = np.array([707 + 284 * i, 118])\n for j in range(1, 5):\n absolute_points[5 * i + j] = np.array([707 + 284 * i, 118 + 492 * j])\n absolute_points[5 * 3] = np.array([1535, 118])\n absolute_points[5 * 3 + 1] = np.array([1535, 118 + 492 * 4])\n\n # template_positions stored as (x, y) top-left\n template_positions = [None] * 17\n for i in range(len(absolute_points)):\n if absolute_points[i][1] < 118 + 492 * 4:\n if i % 5 == 0:\n template_positions[i] = (absolute_points[i][0] - 87, absolute_points[i][1] - 50)\n else:\n template_positions[i] = (absolute_points[i][0] - 37, absolute_points[i][1] - 13)\n\n template_sizes = np.array([(105, 60)] * 17)\n\n # compute template_points as offsets inside template (x_off, y_off)\n template_points = [\n np.array([absolute_points[i][0] - template_positions[i][0],\n absolute_points[i][1] - template_positions[i][1]])\n if template_positions[i] is not None else None\n for i in range(len(absolute_points))\n ]\n\n # Extract templates safely: note image indexing is [y:y+h, x:x+w]\n templates = [None] * 17\n for i in range(len(template_positions)):\n if template_positions[i] is not None:\n tx, ty = template_positions[i] # x, y\n w, h = template_sizes[i] # width, height\n # clamp to image bounds\n x0 = max(0, tx)\n y0 = max(0, ty)\n x1 = min(ima.shape[1], tx + w)\n y1 = min(ima.shape[0], ty + h)\n if x1 <= x0 or y1 <= y0:\n templates[i] = None\n else:\n templates[i] = ima[y0:y1, x0:x1].copy()\n\n # Save attributes\n self._absolute_points = absolute_points\n self._template_positions = template_positions\n self._template_sizes = template_sizes\n self._template_points = template_points\n self._templates = templates\n\n if show_templates:\n fig, axs = plt.subplots(4, 4, figsize=(8, 10))\n for i in range(len(templates)):\n ax = axs.ravel()[i]\n tpl = templates[i]\n if tpl is not None:\n # convert BGR->RGB for plotting\n ax.imshow(cv2.cvtColor(tpl, cv2.COLOR_BGR2RGB))\n ax.set_title(f\"Tpl {i}\")\n ax.axis('off')\n plt.tight_layout()\n plt.show()\n\n @staticmethod\n def lead_info(lead: str):\n mapping = {\n 'I': (0, 1),\n 'II-subset': (5, 6),\n 'III': (10, 11),\n 'aVR': (1, 2),\n 'aVL': (6, 7),\n 'aVF': (11, 12),\n 'V1': (2, 3),\n 'V2': (7, 8),\n 'V3': (12, 13),\n 'V4': (3, 4),\n 'V5': (8, 9),\n 'V6': (13, 14),\n 'II': (15, 16),\n }\n begin, end = mapping[lead]\n return begin // 5, begin, end\n\n def find_markers(self, ima, warn=False, plot=False, title=''):\n \"\"\"\n Find 13 template-based markers and estimate the guessed markers.\n Returns list of 17 markers as (row, col) or None.\n \"\"\"\n if ima is None:\n raise ValueError(\"Input image is None\")\n if ima.shape[0] != 1652:\n # Not fatal; we allow other heights in this robust script but warn\n # raise ValueError(\"Implemented only for scanned images (image types 3, 4, 11, 12)\")\n pass\n\n markers = [None] * 17\n for j in range(len(self._templates)):\n tpl = self._templates[j]\n tpl_point = self._template_points[j]\n tpl_pos = self._template_positions[j]\n tpl_size = self._template_sizes[j]\n if tpl is None or tpl_point is None or tpl_pos is None:\n continue\n # define search window around template position (x,y)\n tx, ty = tpl_pos # x, y\n w, h = tpl_size\n # safe margins\n search_x0 = max(0, tx - 100)\n search_y0 = max(0, ty - 100)\n search_x1 = min(ima.shape[1], tx + w + 250)\n search_y1 = min(ima.shape[0], ty + h + 100)\n search_range = ima[search_y0:search_y1, search_x0:search_x1]\n if search_range.size == 0:\n continue\n try:\n res = cv2.matchTemplate(search_range, tpl, cv2.TM_CCOEFF)\n except Exception as e:\n if warn:\n print(f\"matchTemplate failed for tpl {j}: {e}\")\n continue\n _, max_val, _, max_loc = cv2.minMaxLoc(res)\n top_left = max_loc # (x, y) in search_range coords\n # compute absolute marker coord (row, col) (i.e., y, x)\n abs_row = search_y0 + top_left[1] + int(tpl_point[1])\n abs_col = search_x0 + top_left[0] + int(tpl_point[0])\n markers[j] = np.array((abs_row, abs_col))\n if warn and max_val < 3e7:\n print(j, (abs_row, abs_col), max_val)\n\n # Guess the missing ends (safe guards if markers missing)\n for i in range(3):\n idx2 = 5 * i + 2\n idx3 = 5 * i + 3\n idx4 = 5 * i + 4\n if markers[idx2] is not None and markers[idx3] is not None:\n m = markers[idx3] * 2 - markers[idx2]\n markers[idx4] = m.astype(int)\n\n # guess marker 16 similarly if required\n if markers[14] is not None and markers[9] is not None:\n markers[16] = ((markers[14] * (284 + 260) - markers[9] * 260) / 284).astype(int)\n\n if plot:\n disp = ima.copy()\n # ensure display image is color for drawing\n if disp.ndim == 2:\n disp = cv2.cvtColor(disp, cv2.COLOR_GRAY2BGR)\n for m in markers:\n if m is not None and m.size == 2:\n top_left = (int(m[1]) - 40, int(m[0]) - 40)\n bottom_right = (int(m[1]) + 40, int(m[0]) + 40)\n # clamp coords\n tl = (max(0, top_left[0]), max(0, top_left[1]))\n br = (min(disp.shape[1] - 1, bottom_right[0]), min(disp.shape[0] - 1, bottom_right[1]))\n cv2.rectangle(disp, tl, br, (255, 0, 0), 2)\n plt.figure(figsize=(10, 6))\n # convert BGR->RGB\n try:\n plt.imshow(cv2.cvtColor(disp, cv2.COLOR_BGR2RGB))\n except Exception:\n plt.imshow(disp)\n plt.title(title)\n plt.axis('off')\n plt.show()\n\n return markers\n\n\n def demo(self, ima, warn=False, title=''):\n markers = self.find_markers(ima, warn=warn, plot=True, title=title)\n return markers\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-31T07:08:31.181905Z","iopub.execute_input":"2025-10-31T07:08:31.182297Z","iopub.status.idle":"2025-10-31T07:08:31.282878Z","shell.execute_reply.started":"2025-10-31T07:08:31.182273Z","shell.execute_reply":"2025-10-31T07:08:31.282182Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"#

Image

","metadata":{}},{"cell_type":"code","source":"# Image -> (scanned color)\n\ndef find_line_by_topdown_sweep(ima_bool: np.ndarray):\n \"\"\"\n Given a 2D boolean image (True=white, False=black) returns topmost black pixels per column (top)\n and topmost white pixels per column (bottom). Modifies ima_bool in-place to remove the found line.\n \"\"\"\n # Ensure boolean\n ima = ima_bool.astype(bool)\n # topmost black pixel: find first False (i.e., ~ima)\n top = np.argmax(~ima, axis=0) # returns index of first True in ~ima; if no False found, yields 0\n # Paint black everything above top\n rows = np.arange(ima.shape[0]).reshape(-1, 1)\n mask = rows >= top # True for rows below or equal top\n ima &= mask\n # Find topmost white pixel in modified image\n bottom = np.argmax(ima, axis=0)\n bottomx = np.maximum(bottom, int(np.median(top)) + 100)\n mask2 = rows < bottomx # True for rows above bottomx\n ima |= mask2\n # slightly expand to cover near columns\n ima[:, :-1] |= mask2[:, 1:]\n ima[:, 1:] |= mask2[:, :-1]\n return top, bottom\n\n\ndef get_lead_from_top_bottom(tops, bottoms, lead, number_of_rows, markers, mf: MarkerFinder):\n \"\"\"\n Extract time series for one lead given top/bottom arrays and markers.\n tops/bottoms are lists of 4 arrays (one per line) where each array length == image width.\n markers is list of 17 (row, col) coordinates.\n \"\"\"\n line, begin_idx, end_idx = mf.lead_info(lead)\n top = tops[line]\n bottom = bottoms[line]\n begin = markers[begin_idx]\n end = markers[end_idx]\n if begin is None or end is None:\n # fallback to zeros if markers missing\n return np.zeros(number_of_rows, dtype=float)\n\n # compute column range clamp\n col0 = int(max(0, begin[1]))\n col1 = int(min(len(top), end[1]))\n if col1 <= col0:\n # degenerate case\n return np.zeros(number_of_rows, dtype=float)\n\n pred0 = (top[col0:col1] + bottom[col0:col1]) / 2.0\n # baseline: linear interpolation between begin row and end row across columns\n baseline = np.linspace(begin[0], end[0], col1 - col0)\n\n # ensure same length\n if pred0.size == 0:\n return np.zeros(number_of_rows, dtype=float)\n\n pred = baseline[:pred0.size] - pred0\n\n # scale (80 pixels = 1 mV)\n pred = pred / 80.0\n\n # small fixes for marker-obscured pixels\n if lead in ['aVR', 'aVL', 'aVF', 'V1', 'V2', 'V3', 'V4', 'V5', 'V6']:\n if pred.size >= 5:\n pred[:4][pred[:4] > 0.2] = pred[4]\n if lead in ['I', 'II-subset', 'III', 'aVR', 'aVL', 'aVF', 'V1', 'V2', 'V3']:\n if pred.size >= 6:\n last5 = pred[-5:]\n mask = last5 > 0.2\n if np.any(mask):\n pred[-5:][mask] = pred[-6]\n if lead in ['I', 'II-subset', 'III', 'II']:\n if pred.size >= 2 and 0.9 < pred[0] < 1.1 and pred[1] < 0.5:\n pred[0] = pred[1]\n\n # upsample / resample to required number_of_rows\n pred_up = np.interp(np.linspace(0, 1, number_of_rows),\n np.linspace(0, 1, pred.size),\n pred)\n\n # clamp implausible values\n pred_up = np.where(np.abs(pred_up) <= 0.9, pred_up, 0.0)\n\n return pred_up\n\n\ndef convert_scanned_color(ima: np.ndarray, markers, n_timesteps: dict, verbose=False, mf: MarkerFinder = None):\n \"\"\"\n Convert scanned color ECG image to 12 leads.\n ima: BGR image (height ~1652).\n markers: list of 17 markers from MarkerFinder.find_markers.\n n_timesteps: dict mapping lead->num_samples required.\n \"\"\"\n if mf is None:\n raise ValueError(\"MarkerFinder instance (mf) must be provided.\")\n\n crop_top = 400\n if ima.shape[0] <= crop_top:\n raise ValueError(\"Image too small to crop top. Got height {}\".format(ima.shape[0]))\n\n # Use red channel for better contrast of ECG ink on many scans\n red_channel = ima[crop_top:, :, 2]\n bw_image = red_channel > 160 # boolean: True = white background\n\n # denoise by 3x3 majority filter (keep if >=7 neighbors white)\n iima = bw_image.astype(np.uint8)\n denoised_image = (iima[:-2, :-2] + iima[:-2, 1:-1] + iima[:-2, 2:]\n + iima[1:-1, :-2] + iima[1:-1, 1:-1] + iima[1:-1, 2:]\n + iima[2:, :-2] + iima[2:, 1:-1] + iima[2:, 2:]) >= 7\n # pad to original width\n denoised = np.pad(denoised_image, ((1, 1), (1, 1)), mode='constant', constant_values=True)\n\n if verbose:\n plt.figure(figsize=(8, 3))\n plt.imshow(denoised, cmap='gray')\n plt.title('Denoised BW')\n plt.axis('off')\n plt.show()\n\n # find four lines by iteratively removing topmost line\n tops = []\n bottoms = []\n working_image = denoised.copy()\n for i in range(4):\n top, bottom = find_line_by_topdown_sweep(working_image)\n tops.append(top + crop_top)\n bottoms.append(bottom + crop_top)\n\n # Extract leads: include II-subset artificially as required\n n_timesteps_local = dict(n_timesteps) # copy\n n_timesteps_local['II-subset'] = n_timesteps_local.get('I', n_timesteps_local.get('II', 5000))\n preds = {}\n for lead in LEADS + ['II-subset']:\n required = int(n_timesteps_local.get(lead, 2000))\n preds[lead] = get_lead_from_top_bottom(tops, bottoms, lead, required, markers, mf)\n\n # combine II-subset into II\n if 'II' in preds and 'II-subset' in preds:\n subset_len = min(len(preds['II-subset']), len(preds['II']))\n preds['II'][:subset_len] = (preds['II'][:subset_len] + preds['II-subset'][:subset_len]) / 2.0\n del preds['II-subset']\n\n # Apply Einthoven's and augmented limb leads corrections\n apply_einthoven(preds)\n return preds\n\n\ndef apply_einthoven(preds: dict):\n \"\"\"Apply Einthoven relationships to improve consistency.\"\"\"\n # safety checks\n if 'I' in preds and 'III' in preds and 'II' in preds:\n L = min(len(preds['I']), len(preds['III']), len(preds['II']))\n residual = preds['I'][:L] + preds['III'][:L] - preds['II'][:L]\n correction = residual / 3.0\n preds['I'][:L] = preds['I'][:L] - correction\n preds['III'][:L] = preds['III'][:L] - correction\n preds['II'][:L] = preds['II'][:L] + correction\n if 'aVR' in preds and 'aVL' in preds and 'aVF' in preds:\n L2 = min(len(preds['aVR']), len(preds['aVL']), len(preds['aVF']))\n residual2 = preds['aVR'][:L2] + preds['aVL'][:L2] + preds['aVF'][:L2]\n correction2 = residual2 / 3.0\n preds['aVR'][:L2] = preds['aVR'][:L2] - correction2\n preds['aVL'][:L2] = preds['aVL'][:L2] - correction2\n preds['aVF'][:L2] = preds['aVF'][:L2] - correction2\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-31T07:08:38.524539Z","iopub.execute_input":"2025-10-31T07:08:38.525226Z","iopub.status.idle":"2025-10-31T07:08:38.545872Z","shell.execute_reply.started":"2025-10-31T07:08:38.525198Z","shell.execute_reply":"2025-10-31T07:08:38.544970Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"#

Validation Helper

","metadata":{}},{"cell_type":"code","source":"# Validation helper (visual)\n\ndef validate_algorithm(train_df, image_types, convert_func, mf: MarkerFinder, base_path):\n for idx, row in train_df.iterrows():\n csv_path = os.path.join(base_path, str(row.id), f\"{row.id}.csv\")\n if not os.path.exists(csv_path):\n continue\n try:\n labels = pd.read_csv(csv_path)\n except FileNotFoundError:\n print(f\"Warning: CSV not found at {csv_path}. Skipping.\")\n continue\n png_paths = sorted(glob(os.path.join(base_path, str(row.id), f\"{row.id}-*.png\")))\n if len(png_paths) == 0:\n continue\n for path in png_paths:\n try:\n img_type = int(path[-8:-4])\n except Exception:\n img_type = None\n if img_type not in image_types:\n continue\n try:\n ima = safe_read_image(path)\n except FileNotFoundError:\n continue\n markers = mf.find_markers(ima, plot=True, title=f\"Markers: {row.id}\")\n n_timesteps = {lead: int((~labels[lead].isna()).sum()) for lead in LEADS if lead in labels.columns}\n # Provide reasonable defaults if missing\n for lead in LEADS:\n n_timesteps.setdefault(lead, 5000)\n preds = convert_func(ima, markers, n_timesteps, verbose=True, mf=mf)\n # plot one lead (I)\n lead_to_plot = 'I'\n plt.figure(figsize=(10, 3))\n true_signal = labels[lead_to_plot].dropna().values if lead_to_plot in labels.columns else np.zeros(len(preds[lead_to_plot]))\n plt.plot(true_signal, label='True')\n plt.plot(preds[lead_to_plot], label='Predicted')\n plt.title(f'Lead {lead_to_plot} for {row.id}')\n plt.legend()\n plt.show()\n\n # compute simple SNR for lead I\n fs = int(getattr(row, 'fs', 500))\n aligned = align_signals(true_signal, preds[lead_to_plot], int(fs * MAX_TIME_SHIFT))\n p_s, p_n = compute_power(true_signal, aligned)\n snr = compute_snr(p_s, p_n)\n print(f\"SNR for {row.id} lead {lead_to_plot}: {snr:.2f}\")\n break\n break\n\n\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-31T07:08:45.458452Z","iopub.execute_input":"2025-10-31T07:08:45.458780Z","iopub.status.idle":"2025-10-31T07:08:45.469893Z","shell.execute_reply.started":"2025-10-31T07:08:45.458753Z","shell.execute_reply":"2025-10-31T07:08:45.468998Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"#

Submission

","metadata":{}},{"cell_type":"code","source":"# Submission \n\ndef create_submission(test_df, mean_dict, convert_func, mf: MarkerFinder, base_path):\n submission_rows = []\n old_id = None\n preds = None\n for idx, row in tqdm(test_df.iterrows(), total=len(test_df), desc=\"Creating submission\"):\n # row expected to have attributes: id, lead, number_of_rows, fs\n fs = int(getattr(row, 'fs', 500))\n if row.id != old_id:\n # read image (some test IDs are single .png, some multi-page; adapt path)\n png_candidates = glob(os.path.join(base_path, f\"{row.id}*.png\"))\n if len(png_candidates) == 0:\n # fallback: use mean model\n preds = None\n old_id = row.id\n else:\n # choose first image\n path = png_candidates[0]\n try:\n ima = safe_read_image(path)\n except FileNotFoundError:\n preds = None\n old_id = row.id\n continue\n good_shape = (ima.shape[0] == 1652)\n if good_shape and is_color_image(ima):\n markers = mf.find_markers(ima)\n n_timesteps = {lead: fs * 10 if lead == 'II' else fs * 10 // 4 for lead in LEADS}\n preds = convert_func(ima, markers, n_timesteps, verbose=False, mf=mf)\n else:\n preds = None\n old_id = row.id\n\n # determine pred for this row\n if preds is None:\n # use mean model fallback\n pred_mean = mean_dict.get(row.lead)\n if pred_mean is None:\n # zeros fallback\n pred = np.zeros(int(row.number_of_rows))\n print(f\"Warning: Lead {row.lead} not found in mean_dict. Using zeros.\") # Add a warning\n else:\n mean_series = pred_mean.mean(axis=0)\n pred = np.interp(np.linspace(0, 1, int(row.number_of_rows)),\n np.linspace(0, 1, mean_series.size),\n mean_series)\n else:\n pred = preds.get(row.lead)\n if pred is None:\n pred = np.zeros(int(row.number_of_rows))\n print(f\"Warning: Lead {row.lead} not found in image predictions. Using zeros.\")# Add a warning\n\n # ensure correct length\n pred_len = len(pred)\n expected_len = int(row.number_of_rows)\n\n if pred_len != expected_len:\n print(f\"ERROR: Prediction length mismatch for id={row.id}, lead={row.lead}.\")\n print(f\" Expected length: {expected_len}, Actual length: {pred_len}\")\n # Handle the error: either truncate, pad, or use a fallback\n if pred_len < expected_len:\n # Pad with zeros (or a more appropriate value)\n padding = np.zeros(expected_len - pred_len)\n pred = np.concatenate([pred, padding])\n else:\n # Truncate the prediction\n pred = pred[:expected_len]\n\n\n assert len(pred) == int(row.number_of_rows), f\"Length mismatch after correction for id={row.id}, lead={row.lead}\"\n\n\n for t in range(int(row.number_of_rows)):\n submission_rows.append({'id': f\"{row.id}_{t}_{row.lead}\", 'value': float(pred[t])}) # Ensure 't' is int\n\n submission_df = pd.DataFrame(submission_rows)\n submission_df['value'] = submission_df['value'].astype(float)\n submission_df.fillna(0, inplace=True) # Replace NaN with 0\n submission_df.replace([float('inf'), float('-inf')], 0, inplace=True) # Replace infinite values with 0\n\n print(f\"Length: {len(submission_df)}\")\n return submission_df\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-31T07:08:49.612254Z","iopub.execute_input":"2025-10-31T07:08:49.612602Z","iopub.status.idle":"2025-10-31T07:08:49.624805Z","shell.execute_reply.started":"2025-10-31T07:08:49.612578Z","shell.execute_reply":"2025-10-31T07:08:49.624004Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Usage guarded\n\nif __name__ == '__main__':\n \n TRAIN_CSV = '/kaggle/input/physionet-ecg-image-digitization/train.csv'\n TEST_CSV = '/kaggle/input/physionet-ecg-image-digitization/test.csv'\n TRAIN_BASE = '/kaggle/input/physionet-ecg-image-digitization/train'\n TEST_BASE = '/kaggle/input/physionet-ecg-image-digitization/test'\n\n # load small portions to demo (guard with exists)\n if not os.path.exists(TRAIN_CSV):\n print(\"Train CSV not found at\", TRAIN_CSV)\n else:\n train = pd.read_csv(TRAIN_CSV)\n\n # build mean model on a small subset for speed if dataset present\n subset = train.iloc[:min(200, len(train))]\n mean_dict = fit_mean_model(subset, base_path=TRAIN_BASE)\n # validate on next chunk\n if len(train) > 200:\n validate_mean_model(train.iloc[200:300], mean_dict, base_path=TRAIN_BASE)\n\n # Initialize MarkerFinder (will raise if templates missing)\n try:\n mf = MarkerFinder(show_templates=False, base_path=TRAIN_BASE)\n except FileNotFoundError as e:\n print(\"MarkerFinder initialization failed:\", e)\n mf = None\n\n # Demo marker detection if we have an example image\n if mf is not None:\n example_path = os.path.join(TRAIN_BASE, str(train.iloc[0].id), f\"{train.iloc[0].id}-0001.png\")\n if os.path.exists(example_path):\n try:\n ex_img = safe_read_image(example_path)\n mf.demo(ex_img, warn=False, title=f\"Demo {train.iloc[0].id}\")\n except FileNotFoundError:\n pass\n\n # run a single validation example (visual)\n validate_algorithm(train.iloc[:5], image_types=[3, 11], convert_func=convert_scanned_color, mf=mf, base_path=TRAIN_BASE)\n\n # If test CSV exists, create (small) submission\n if os.path.exists(TEST_CSV) and mf is not None:\n test = pd.read_csv(TEST_CSV)\n small_test = test.iloc[:min(200, len(test))]\n\n # Create a dummy solution.csv for testing the `score` function\n # Replace with actual ground truth if available\n solution = small_test[['id', 'fs']].copy()\n solution['value'] = 0 # Replace with ground truth values if available\n\n submission_df = create_submission(small_test, mean_dict, convert_scanned_color, mf, base_path=TEST_BASE)\n\n submission_df.to_csv('submission.csv', index=False)\n print(\"submission.csv created (first 200 rows).\")\n else:\n print(\"Test CSV not found or MarkerFinder unavailable; skipping submission creation.\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-31T07:09:02.126161Z","iopub.execute_input":"2025-10-31T07:09:02.126479Z","iopub.status.idle":"2025-10-31T07:09:17.034341Z","shell.execute_reply.started":"2025-10-31T07:09:02.126457Z","shell.execute_reply":"2025-10-31T07:09:17.033381Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!head submission.csv","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-31T07:09:22.556459Z","iopub.execute_input":"2025-10-31T07:09:22.557141Z","iopub.status.idle":"2025-10-31T07:09:22.701364Z","shell.execute_reply.started":"2025-10-31T07:09:22.557114Z","shell.execute_reply":"2025-10-31T07:09:22.699949Z"}},"outputs":[],"execution_count":null}]}