{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"name":"python","version":"3.11.13"},"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":"# WARNING: Do not run to get high score, this is educationally testing bad assumptions! Check other posts for good scores.","metadata":{}},{"cell_type":"markdown","source":"# Simplified ECG Digitization\n\nSmall Challenge: remove the bad assumptions and fix the model! For fun.","metadata":{}},{"cell_type":"markdown","source":"## Import libraries\n\nLoading basic packages for image processing and data handling","metadata":{}},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## huge thanks to seowoohyeon's public code release! (https://www.kaggle.com/code/seowoohyeon/physionet-adjust)","metadata":{}},{"cell_type":"code","source":"import cv2\nimport numpy as np\nimport pandas as pd\nfrom tqdm import tqdm\nfrom collections import defaultdict","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Configuration\n\nSetting up constants for ECG leads and image processing","metadata":{}},{"cell_type":"code","source":"# ECG lead names\nLEADS = ['I', 'II', 'III', 'aVR', 'aVL', 'aVF', 'V1', 'V2', 'V3', 'V4', 'V5', 'V6']\n\n# Image parameters (using wrong values to hurt performance)\nCROP_TOP = 300  # Wrong crop value\nMV_PER_PIXEL = 100  # Wrong conversion factor","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Simple marker detection\n\nBasic template matching without sophisticated search","metadata":{}},{"cell_type":"code","source":"class SimpleMarkerFinder:\n    def __init__(self):\n        # Just using one reference image instead of averaging multiple\n        ref_img = cv2.imread('/kaggle/input/physionet-ecg-image-digitization/train/4292118763/4292118763-0001.png')\n        \n        # Fixed marker positions without proper extraction\n        self.markers = []\n        for i in range(3):\n            for j in range(5):\n                x = 700 + 280 * i\n                y = 100 + 500 * j\n                self.markers.append((y, x))\n        \n        # Add two more markers\n        self.markers.append((100, 1500))\n        self.markers.append((2100, 1500))\n    \n    def find_markers(self, img):\n        # Just return fixed positions without actually searching\n        return [np.array(m) if i < 15 else None for i, m in enumerate(self.markers)]","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Naive signal extraction\n\nExtract ECG trace by simple averaging without corrections","metadata":{}},{"cell_type":"code","source":"def get_trace_simple(img, row_idx, start_x, end_x):\n    # Crop image\n    cropped = img[CROP_TOP:, :]\n    \n    # Convert to grayscale badly\n    if len(cropped.shape) == 3:\n        gray = cropped[:, :, 0]  # Just use blue channel\n    else:\n        gray = cropped\n    \n    # Find row boundaries without proper algorithm\n    row_height = gray.shape[0] // 4\n    row_start = row_idx * row_height\n    row_end = (row_idx + 1) * row_height\n    row_data = gray[row_start:row_end, start_x:end_x]\n    \n    # Extract signal by finding darkest pixel in each column\n    signal = []\n    for col in range(row_data.shape[1]):\n        column = row_data[:, col]\n        # Just use argmin without any smoothing\n        dark_idx = np.argmin(column)\n        signal.append(dark_idx)\n    \n    return np.array(signal)","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Bad baseline correction\n\nRemove baseline using wrong assumption of linear drift","metadata":{}},{"cell_type":"code","source":"def remove_baseline_wrong(signal, start_y, end_y):\n    # Assume linear baseline (usually wrong)\n    baseline = np.linspace(start_y, end_y, len(signal))\n    \n    # Convert to millivolts using wrong scale\n    signal_mv = (baseline - signal) / MV_PER_PIXEL\n    \n    # Clip extreme values (loses important information)\n    signal_mv = np.clip(signal_mv, -0.5, 0.5)\n    \n    return signal_mv","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Process single lead\n\nExtract one ECG lead without sophisticated processing","metadata":{}},{"cell_type":"code","source":"def extract_lead_simple(img, lead_name, markers, num_samples):\n    # Hardcoded lead positions (wrong for many images)\n    lead_map = {\n        'I': (0, 0, 1), 'II': (0, 5, 6), 'III': (0, 10, 11),\n        'aVR': (1, 1, 2), 'aVL': (1, 6, 7), 'aVF': (1, 11, 12),\n        'V1': (2, 2, 3), 'V2': (2, 7, 8), 'V3': (2, 12, 13),\n        'V4': (3, 3, 4), 'V5': (3, 8, 9), 'V6': (3, 13, 14)\n    }\n    \n    if lead_name not in lead_map:\n        return np.zeros(num_samples)\n    \n    row, start_idx, end_idx = lead_map[lead_name]\n    \n    # Get marker positions (might be None)\n    start_marker = markers[start_idx] if start_idx < len(markers) and markers[start_idx] is not None else np.array([500, 700])\n    end_marker = markers[end_idx] if end_idx < len(markers) and markers[end_idx] is not None else np.array([500, 1000])\n    \n    # Extract trace\n    signal = get_trace_simple(img, row, int(start_marker[1]), int(end_marker[1]))\n    \n    # Remove baseline\n    signal_mv = remove_baseline_wrong(signal, start_marker[0], end_marker[0])\n    \n    # Resample to required length using simple linear interpolation\n    # No fancy alignment or signal processing\n    if len(signal_mv) > 0:\n        x_old = np.linspace(0, 1, len(signal_mv))\n        x_new = np.linspace(0, 1, num_samples)\n        resampled = np.interp(x_new, x_old, signal_mv)\n    else:\n        resampled = np.zeros(num_samples)\n    \n    return resampled.astype(np.float32)","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Build mean model\n\nCalculate average signal per lead from training data","metadata":{}},{"cell_type":"code","source":"def create_mean_predictions(train_df):\n    means = defaultdict(list)\n    \n    print(\"Building mean model...\")\n    for _, row in tqdm(train_df.iterrows(), total=len(train_df)):\n        labels_file = f'/kaggle/input/physionet-ecg-image-digitization/train/{row.id}/{row.id}.csv'\n        try:\n            labels = pd.read_csv(labels_file)\n            for lead in labels.columns:\n                vals = labels[lead].dropna().values\n                if len(vals) > 0:\n                    # Resample to fixed length\n                    resampled = np.interp(\n                        np.linspace(0, len(vals)-1, 10000),  # Using shorter length\n                        np.arange(len(vals)),\n                        vals\n                    )\n                    means[lead].append(resampled)\n        except:\n            pass\n    \n    # Average all signals\n    for lead in means:\n        means[lead] = np.mean(np.stack(means[lead]), axis=0)\n    \n    return means","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Process test images\n\nExtract predictions for test set with fallback to mean model","metadata":{}},{"cell_type":"code","source":"def process_test_image(img_path, lead_name, num_rows, mean_model, marker_finder):\n    # Try to read image\n    img = cv2.imread(img_path)\n    \n    if img is None:\n        # Fallback to mean\n        mean_signal = mean_model.get(lead_name, np.zeros(10000))\n        return np.interp(\n            np.linspace(0, 1, num_rows),\n            np.linspace(0, 1, len(mean_signal)),\n            mean_signal\n        )\n    \n    # Try to find markers (will often fail)\n    try:\n        markers = marker_finder.find_markers(img)\n        \n        # Extract lead\n        prediction = extract_lead_simple(img, lead_name, markers, num_rows)\n        \n        # If extraction failed, use mean\n        if np.all(prediction == 0) or np.any(np.isnan(prediction)):\n            mean_signal = mean_model.get(lead_name, np.zeros(10000))\n            prediction = np.interp(\n                np.linspace(0, 1, num_rows),\n                np.linspace(0, 1, len(mean_signal)),\n                mean_signal\n            )\n        \n        return prediction\n    except:\n        # On any error, use mean model\n        mean_signal = mean_model.get(lead_name, np.zeros(10000))\n        return np.interp(\n            np.linspace(0, 1, num_rows),\n            np.linspace(0, 1, len(mean_signal)),\n            mean_signal\n        )","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Main execution\n\nLoad data, process images, create submission","metadata":{}},{"cell_type":"code","source":"# Load data\nprint(\"Loading data...\")\ntrain_data = pd.read_csv('/kaggle/input/physionet-ecg-image-digitization/train.csv')\ntest_data = pd.read_csv('/kaggle/input/physionet-ecg-image-digitization/test.csv')\n\n# Create mean model\nmean_model = create_mean_predictions(train_data)\n\n# Initialize marker finder\nmarker_finder = SimpleMarkerFinder()\n\n# Process test images\nprint(\"Processing test images...\")\nsubmission_rows = []\ncurrent_image_id = None\ncurrent_predictions = {}\n\nfor _, test_row in tqdm(test_data.iterrows(), total=len(test_data)):\n    # Check if we need to process new image\n    if test_row.id != current_image_id:\n        img_path = f\"/kaggle/input/physionet-ecg-image-digitization/test/{test_row.id}.png\"\n        current_image_id = test_row.id\n        current_predictions = {}  # Clear cache\n    \n    # Get prediction for this lead\n    if test_row.lead not in current_predictions:\n        current_predictions[test_row.lead] = process_test_image(\n            img_path,\n            test_row.lead,\n            test_row.number_of_rows,\n            mean_model,\n            marker_finder\n        )\n    \n    prediction = current_predictions[test_row.lead]\n    \n    # Create submission rows\n    for timestep in range(test_row.number_of_rows):\n        submission_rows.append({\n            'id': f\"{test_row.id}_{timestep}_{test_row.lead}\",\n            'value': float(prediction[timestep])\n        })\n\n# Save submission\nprint(\"Saving submission...\")\nsubmission_df = pd.DataFrame(submission_rows)\nsubmission_df.to_csv('submission.csv', index=False)\nprint(\"Done!\")","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Spoilers\n\nChanges that hurt performance:\n- Removed sophisticated signal alignment algorithms\n- No occlusion detection or correction\n- No slope based interpolation for flat segments\n- No Einthoven's law enforcement\n- Wrong crop parameters and scaling factors\n- Fixed marker positions instead of template matching\n- Simple linear baseline assumption\n- No denoising or smoothing\n- Only using blue channel for grayscale\n- No proper top/bottom trace extraction\n- Clipping values loses important information\n- Heavy reliance on fallback mean model","metadata":{}}]}