{"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":"gpu","dataSources":[{"sourceId":97984,"databundleVersionId":14096757,"sourceType":"competition"}],"dockerImageVersionId":31154,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"## Training data overview\n\nThere are 977 electrocardiograms (ids) in train, accounting for 84 GByte. Every electrocardiogram has 9 PNG files and one CSV file.\n\nThere are nine image types per ECG:\n- 0001 Original color ECG image generated by ECG-image-kit.\n- 0003 Image printed in color and scanned in color. → processed by `convert_scanned_color()`\n- 0004 Image printed in color and scanned in black and white. → processed by `convert_scanned_grayscale()`\n- 0005 Mobile photos of color printed images.\n- 0006 Mobile photos of ECGs on the screen of laptop.\n- 0009 Mobile photos of stained and soaked printed ECGs.\n- 0010 Mobile photos of printed ECGs with extensive damage.\n- 0011 Scans of printed ECG images with mold in color. → processed by `convert_scanned_color()`\n- 0012 Scans of printed ECG images with mold in black and white. → processed by `convert_scanned_grayscale()`\n\nThe sampling frequencies in train are 250, 256, 500, 512, 1000, 1025 per second.","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19"}},{"cell_type":"code","source":"import pandas as pd\nimport numpy as np\nimport cv2\nfrom glob import glob\nimport matplotlib.pyplot as plt\nfrom collections import defaultdict\nfrom tqdm import tqdm\nfrom scipy.signal import medfilt\n\nfrom sklearn.model_selection import KFold\nfrom sklearn.metrics import r2_score\n\nfrom tensorflow.keras.layers import Input, Dense, Activation, Reshape, GaussianNoise\nfrom tensorflow.keras.initializers import Constant\nfrom tensorflow.keras.models import Model\nfrom tensorflow.keras.optimizers import Adam\nfrom tensorflow.keras.losses import MeanSquaredError\nfrom tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau, TerminateOnNaN","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-19T06:59:22.797973Z","iopub.execute_input":"2025-11-19T06:59:22.798235Z","iopub.status.idle":"2025-11-19T06:59:38.508772Z","shell.execute_reply.started":"2025-11-19T06:59:22.798183Z","shell.execute_reply":"2025-11-19T06:59:38.507922Z"},"_kg_hide-input":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Auxiliary functions","metadata":{}},{"cell_type":"code","source":"# Competition metric\n# From https://www.kaggle.com/code/metric/physionet-ecg-signal-extraction-metric\nfrom typing import Tuple\n\nimport numpy as np\nimport pandas as pd\n\nimport scipy.optimize\nimport scipy.signal\n\n\nLEADS = ['I', 'II', 'III', 'aVR', 'aVL', 'aVF', 'V1', 'V2', 'V3', 'V4', 'V5', 'V6']\nMAX_TIME_SHIFT = 0.2\nPERFECT_SCORE = 384\n\n\nclass ParticipantVisibleError(Exception):\n    pass\n\n\ndef compute_power(label: np.ndarray, prediction: np.ndarray) -> Tuple[float, float]:\n    if label.ndim != 1 or prediction.ndim != 1:\n        raise ParticipantVisibleError('Inputs must be 1-dimensional arrays.')\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    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    if noise == 0:\n        # Perfect reconstruction\n        snr = PERFECT_SCORE\n    elif signal == 0:\n        snr = 0\n    else:\n        snr = min((signal / noise), PERFECT_SCORE)\n    return snr\n\n\ndef align_signals(label: np.ndarray, pred: np.ndarray, max_shift: float = float('inf')) -> np.ndarray:\n    if np.any(~np.isfinite(label)):\n        raise ParticipantVisibleError('values in label should all be finite')\n    if np.sum(np.isfinite(pred)) == 0:\n        raise ParticipantVisibleError('prediction can not all be infinite')\n\n    # Initialize the reference and digitized signals.\n    label_arr = np.asarray(label, dtype=np.float64)\n    pred_arr = np.asarray(pred, dtype=np.float64)\n\n    label_mean = np.mean(label_arr)\n    pred_mean = np.mean(pred_arr)\n\n    label_arr_centered = label_arr - label_mean\n    pred_arr_centered = pred_arr - pred_mean\n\n    # Compute the correlation between the reference and digitized signals and locate the maximum correlation.\n    correlation = scipy.signal.correlate(label_arr_centered, pred_arr_centered, mode='full')\n\n    n_label = np.size(label_arr)\n    n_pred = np.size(pred_arr)\n\n    lags = scipy.signal.correlation_lags(n_label, n_pred, mode='full')\n    valid_lags_mask = (lags >= -max_shift) & (lags <= max_shift)\n\n    max_correlation = np.nanmax(correlation[valid_lags_mask])\n    all_max_indices = np.flatnonzero(correlation == max_correlation)\n    best_idx = min(all_max_indices, key=lambda i: abs(lags[i]))\n    time_shift = lags[best_idx]\n    start_padding_len = max(time_shift, 0)\n    pred_slice_start = max(-time_shift, 0)\n    pred_slice_end = min(n_label - time_shift, n_pred)\n    end_padding_len = max(n_label - n_pred - time_shift, 0)\n    aligned_pred = np.concatenate((np.full(start_padding_len, np.nan), pred_arr[pred_slice_start:pred_slice_end], np.full(end_padding_len, np.nan)))\n\n    def objective_func(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        results = scipy.optimize.minimize_scalar(objective_func, method='Brent')\n        vertical_shift = results.x\n        aligned_pred -= vertical_shift\n    return aligned_pred\n\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    if sampling_frequency != int(len(group[group['lead'] == 'II']) / 10):\n        raise ParticipantVisibleError('The sequence_length should be sampling frequency * 10s')\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    Examples\n    --------\n    >>> import pandas as pd\n    >>> import numpy as np\n    >>> row_id_column_name = \"id\"\n    >>> solution = pd.DataFrame({'id': ['343_0_I', '343_1_I', '343_2_I', '343_0_III', '343_1_III','343_2_III','343_0_aVR', '343_1_aVR','343_2_aVR',\\\n    '343_0_aVL', '343_1_aVL', '343_2_aVL', '343_0_aVF', '343_1_aVF','343_2_aVF','343_0_V1', '343_1_V1', '343_2_V1','343_0_V2', '343_1_V2','343_2_V2',\\\n    '343_0_V3', '343_1_V3', '343_2_V3','343_0_V4', '343_1_V4', '343_2_V4', '343_0_V5', '343_1_V5','343_2_V5','343_0_V6', '343_1_V6','343_2_V6',\\\n    '343_0_II', '343_1_II','343_2_II', '343_3_II', '343_4_II', '343_5_II','343_6_II', '343_7_II','343_8_II','343_9_II','343_10_II','343_11_II'],\\\n    'fs': [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],\\\n    'value':[0.1,0.3,0.4,0.6,0.6,0.4,0.2,0.3,0.4,0.5,0.2,0.7,0.2,0.3,0.4,0.8,0.6,0.7, 0.2,0.3,-0.1,0.5,0.6,0.7,0.2,0.9,0.4,0.5,0.6,0.7,0.1,0.3,0.4,\\\n    0.6,0.6,0.4,0.2,0.3,0.4,0.5,0.2,0.7,0.2,0.3,0.4]})\n    >>> submission = solution.copy()\n    >>> round(score(solution, submission, row_id_column_name), 4)\n    25.8433\n    >>> submission.loc[0, 'value'] = 0.9 # Introduce some noise\n    >>> round(score(solution, submission, row_id_column_name), 4)\n    13.6291\n    >>> submission.loc[4, 'value'] = 0.3 # Introduce some noise\n    >>> round(score(solution, submission, row_id_column_name), 4)\n    13.0576\n\n    >>> solution = pd.DataFrame({'id': ['343_0_I', '343_1_I', '343_2_I', '343_0_III', '343_1_III','343_2_III','343_0_aVR', '343_1_aVR','343_2_aVR',\\\n    '343_0_aVL', '343_1_aVL', '343_2_aVL', '343_0_aVF', '343_1_aVF','343_2_aVF','343_0_V1', '343_1_V1', '343_2_V1','343_0_V2', '343_1_V2','343_2_V2',\\\n    '343_0_V3', '343_1_V3', '343_2_V3','343_0_V4', '343_1_V4', '343_2_V4', '343_0_V5', '343_1_V5','343_2_V5','343_0_V6', '343_1_V6','343_2_V6',\\\n    '343_0_II', '343_1_II','343_2_II', '343_3_II', '343_4_II', '343_5_II','343_6_II', '343_7_II','343_8_II','343_9_II','343_10_II','343_11_II'],\\\n    'fs': [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],\\\n    'value':[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]})\n    >>> round(score(solution, submission, row_id_column_name), 4)\n    -384\n    >>> submission = solution.copy()\n    >>> round(score(solution, submission, row_id_column_name), 4)\n    25.8433\n\n    >>> # test alignment\n    >>> label = np.array([0, 1, 2, 1, 0])\n    >>> pred = np.array([0, 1, 2, 1, 0])\n    >>> aligned = align_signals(label, pred)\n    >>> expected_array = np.array([0, 1, 2, 1, 0])\n    >>> np.allclose(aligned, expected_array, equal_nan=True)\n    True\n\n    >>> # Test 2: Vertical shift (DC offset) should be removed\n    >>> label = np.array([0, 1, 2, 1, 0])\n    >>> pred = np.array([10, 11, 12, 11, 10])\n    >>> aligned = align_signals(label, pred)\n    >>> expected_array = np.array([0, 1, 2, 1, 0])\n    >>> np.allclose(aligned, expected_array, equal_nan=True)\n    True\n\n    >>> # Test 3: Time shift should be corrected\n    >>> label = np.array([0, 0, 1, 2, 1, 0., 0.])\n    >>> pred = np.array([1, 2, 1, 0, 0, 0, 0])\n    >>> aligned = align_signals(label, pred)\n    >>> expected_array = np.array([np.nan, np.nan, 1, 2, 1, 0, 0])\n    >>> np.allclose(aligned, expected_array, equal_nan=True)\n    True\n    \n    >>> # Test 4: max_shift constraint prevents optimal alignment\n    >>> label = np.array([0, 0, 0, 0, 1, 2, 1]) # Peak is far\n    >>> pred = np.array([1, 2, 1, 0, 0, 0, 0])\n    >>> aligned = align_signals(label, pred, max_shift=10)\n    >>> expected_array = np.array([ np.nan, np.nan, np.nan, np.nan, 1, 2, 1])\n    >>> np.allclose(aligned, expected_array, equal_nan=True)\n    True\n\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    merged_df = pd.merge(solution, submission, on=row_id_column_name, suffixes=('_true', '_pred'))\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.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    return max(float(10 * np.log10(image_scores.mean())), -PERFECT_SCORE)","metadata":{"trusted":true,"_kg_hide-input":true,"execution":{"iopub.status.busy":"2025-11-19T06:59:38.510144Z","iopub.execute_input":"2025-11-19T06:59:38.510641Z","iopub.status.idle":"2025-11-19T06:59:38.527955Z","shell.execute_reply.started":"2025-11-19T06:59:38.510621Z","shell.execute_reply":"2025-11-19T06:59:38.527301Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def plot_training_history(history):\n    \"\"\"Plot a Keras training history.\"\"\"\n    if len(history['loss']) >= 2:\n        _, axs = plt.subplots(1, 1, figsize=(6, 3), squeeze=False)\n        axs = axs.ravel()\n        axs[0].plot(np.arange(len(history['loss'])) + 1, history['loss'], ':', label='train_loss')\n        axs[0].plot(np.arange(len(history['val_loss'])) + 1, history['val_loss'], label='val_loss')\n        axs[0].legend()\n        axs[0].set_title('Training history')\n        plt.show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-19T06:59:38.528917Z","iopub.execute_input":"2025-11-19T06:59:38.529189Z","iopub.status.idle":"2025-11-19T06:59:38.557141Z","shell.execute_reply.started":"2025-11-19T06:59:38.529148Z","shell.execute_reply":"2025-11-19T06:59:38.556311Z"},"_kg_hide-input":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Reading the metadata and the labels","metadata":{}},{"cell_type":"code","source":"train = pd.read_csv('/kaggle/input/physionet-ecg-image-digitization/train.csv')\ntest = pd.read_csv('/kaggle/input/physionet-ecg-image-digitization/test.csv')\nlabel_dict = {}\nfor idx, row in tqdm(train.iterrows(), total=len(train)):\n    label_dict[idx] = pd.read_csv(f'/kaggle/input/physionet-ecg-image-digitization/train/{row.id}/{row.id}.csv')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-19T06:59:38.558041Z","iopub.execute_input":"2025-11-19T06:59:38.558469Z","iopub.status.idle":"2025-11-19T06:59:53.442722Z","shell.execute_reply.started":"2025-11-19T06:59:38.558451Z","shell.execute_reply":"2025-11-19T06:59:53.441978Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Looping through the training dataset\n\nWe define a generator function `train_images_and_labels`, which loops through a subset of the training images.","metadata":{}},{"cell_type":"code","source":"def train_images_and_labels(start=None, end=None, image_types=None, use_tqdm=True):\n    \"\"\"Generator function which yields a subset of the training images.\n    \n    Parameters\n    start: start index of slice\n    end: end index of slice\n    image_types: list of image types to select\n    \"\"\"\n    t = train.iloc[start:end]\n    iterable = t.iterrows()\n    if use_tqdm:\n        iterable = tqdm(iterable, total=len(t))\n    for idx, row in iterable:\n        png_paths = sorted(glob(f'/kaggle/input/physionet-ecg-image-digitization/train/{row.id}/{row.id}-*.png'))\n        labels = label_dict[idx]\n        for path in png_paths:\n            img_type = int(path[-8:-4])\n            if image_types is None or img_type in image_types:\n                ima = cv2.imread(path)\n\n                # The following lines document the possible shapes for every image type in train\n                # Test files may be different\n                shape = ima.shape\n                assert len(shape) == 3\n                assert (img_type == 1) <= (shape == (1700, 2200, 3)) # 200 pixels per inch on Letter paper\n                assert (img_type == 3) <= (shape[0] == 1652)\n                assert (img_type == 4) <= (shape[0] == 1652)\n                assert (img_type == 5) <= (shape in {(3024, 4032, 3), (1344, 1008, 3), (4032, 3024, 3)})\n                assert (img_type == 6) <= ((shape == (4000, 3000, 3) or (shape == (3000, 4000, 3))))\n                assert (img_type == 9) <= ((shape == (3024, 4032, 3) or (shape == (4032, 3024, 3))))\n                assert (img_type == 10) <= ((shape == (3024, 4032, 3) or (shape == (4032, 3024, 3))))\n                assert (img_type == 11) <= (shape[0] == 1652)\n                assert (img_type == 12) <= (shape[0] == 1652)\n\n                yield idx, ima, img_type, labels\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-19T06:59:53.443619Z","iopub.execute_input":"2025-11-19T06:59:53.443886Z","iopub.status.idle":"2025-11-19T06:59:53.455685Z","shell.execute_reply.started":"2025-11-19T06:59:53.443863Z","shell.execute_reply":"2025-11-19T06:59:53.454907Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## The average ECG\n\nFor regression tasks, the average of the true labels is often a good baseline. \n\nWe compute the mean time series per lead so that we can predict this mean for those image types which our model cannot handle.\n\n(리드 별 평균 시계열을 구함.)","metadata":{}},{"cell_type":"code","source":"def fit_mean_model(train, verbose=False):\n    \"\"\"Compute minima, maxima and means of the time series\"\"\"\n    mean_dict = defaultdict(list)\n    for idx, row in tqdm(train.iterrows(), total=len(train)):\n        labels = label_dict[idx]\n        for lead in labels.columns:\n            values = labels[lead]\n            values = values[~values.isna()]\n            mean_dict[lead].append(values)\n    \n    for lead in mean_dict.keys():\n        # Upsample every time series to 20000 samples\n        mean_dict[lead] = [\n            np.interp(np.linspace(0, len(values)-1, 20000), np.arange(len(values)), values)\n            for values in mean_dict[lead]\n        ]\n\n        # Stack all ECGs\n        mean_dict[lead] = np.stack(mean_dict[lead])\n\n        # Plot the mean ECG\n        if verbose:\n            m = mean_dict[lead].mean(axis=0)\n            # s = mean_dict[lead].std(axis=0)\n            plt.figure(figsize=(6, 1.5))\n            plt.title(f\"Mean curve for {lead}\")\n            plt.plot(m)\n            # plt.plot(m-s/30)\n            # plt.plot(m+s/30)\n            plt.axhline(0, color='gray')\n            plt.ylabel('mV')\n            plt.gca().get_xaxis().set_visible(False)\n            plt.show()\n\n    return mean_dict\n\ndef validate_mean_model(val, mean_dict):\n    snr_list = []\n    for idx, row in tqdm(val.iterrows(), total=len(val)):\n        labels = label_dict[idx]\n        # Evaluate the signal-to-noise ratio\n        sum_signal = 0\n        sum_noise = 0\n        for lead in labels.columns:\n            label = labels[lead]\n            label = label[~ label.isna()]\n            pred = mean_dict[lead].mean(axis=0)\n            pred = np.interp(np.linspace(0, 1, len(label)), np.linspace(0, 1, len(pred)), pred)\n            assert len(label) == len(pred)\n    \n            aligned_pred = align_signals(label, pred, int(row.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    \n        snr = compute_snr(sum_signal, sum_noise)\n        snr_list.append(snr)\n    \n    snr = np.array(snr_list).mean()\n    val_score = max(float(10 * np.log10(snr)), -PERFECT_SCORE)\n    print(f\"# Validation SNR for mean prediction: {snr:.2f} {val_score=:.2f}\")\n\n# # Validate the mean model\n# train_test_split_loc = 780\n# mean_dict = fit_mean_model(train.iloc[:train_test_split_loc], verbose=True)\n# validate_mean_model(train.iloc[train_test_split_loc:], mean_dict)\n\n# Refit the mean model to the full dataset\nmean_dict = fit_mean_model(train, verbose=True)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-19T06:59:53.457979Z","iopub.execute_input":"2025-11-19T06:59:53.458207Z","iopub.status.idle":"2025-11-19T07:00:01.833647Z","shell.execute_reply.started":"2025-11-19T06:59:53.458189Z","shell.execute_reply":"2025-11-19T07:00:01.832898Z"},"_kg_hide-output":true,"_kg_hide-input":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Finding lead endpoints with MarkerFinder\n\nBefore decoding an image, it's good to know the coordinates of the 17 lead endpoints in the ECG. The following cell defines the class `MarkerFinder`, which determines these points. 13 points are found by the pattern matching function `cv2.matchTemplate()`; the right endpoints of the four lines are inferred as linear combinations of the other vectors.","metadata":{}},{"cell_type":"code","source":"class MarkerFinder:\n    \"\"\"This class finds the 13 markers in scanned ecg images and guesses the 4 line ends.\"\"\"\n    # From https://www.kaggle.com/code/ambrosm/ecg-original-explained-baseline\n    \n    def __init__(self, show_templates=False):\n        # Derive the templates from type 1 images\n        # np.max keeps the gridlines and markers and removes the ecg lines\n        # 주 : 세 개의 이미지를 가져와서 하나로 합치고, 가장 뚜렷한 부분을 찾으면 그게 grid가 됨.\n        ima = np.max([\n            cv2.imread('/kaggle/input/physionet-ecg-image-digitization/train/4292118763/4292118763-0001.png'),\n            cv2.imread('/kaggle/input/physionet-ecg-image-digitization/train/4289880010/4289880010-0001.png'),\n            cv2.imread('/kaggle/input/physionet-ecg-image-digitization/train/4284351157/4284351157-0001.png'),\n        ], axis=0)\n\n        # Template points in global coordinates of type 1 images\n        # 주 : Type1에 대해서 미리 좌표를 하드코딩\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]) # y, x\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        # Top left corner of template rectangle\n        # 17개의 절대 좌표(마커 중심)을 둘러싼 템플릿 영역의 좌측 상단 픽셀 위치를 계산.\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) # y, x\n                else:\n                    template_positions[i] = (absolute_points[i][0] - 37, absolute_points[i][1] - 13)\n\n        # Height and width of the templates\n        # 템플릿 매칭에 사용할 템플릿의 크기를 정의.\n        template_sizes = np.array([(105, 60)] * 17) # height, width\n\n        # Transform the points to relative coordinates (inside the template)\n        # 템플릿 영역의 좌측 상단을 (0,0)으로 했을 때, 마커 중심이 템플릿 내부에서 어디에 위치하는지 상대 좌표를 계산.\n        template_points = [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\n                           else None\n                           for i in range(len(absolute_points))]\n\n        # Save the template matrices\n        templates = [None] * 17\n        for i in range(len(template_positions)):\n            if template_points[i] is not None:\n                template = (ima[template_positions[i][0]:template_positions[i][0]+template_sizes[i][0],\n                            template_positions[i][1]:template_positions[i][1]+template_sizes[i][1]])\n                templates[i] = template\n\n        # Plot the template matrices\n        if show_templates:\n            _, axs = plt.subplots(4, 4, figsize=(5, 7))\n            for i in range(len(template_positions)):\n                if template_points[i] is not None:\n                    template = templates[i].copy()\n                    cv2.rectangle(template,\n                                  (template_points[i][1]-1, template_points[i][0]-1),\n                                  (template_points[i][1]+1, template_points[i][0]+1), \n                                  [255, 0, 0], 2)\n                    axs[i // 5, i % 5].imshow(template)\n            for i in range(13, len(axs.ravel())):\n                axs.ravel()[i].axis('off')\n            plt.tight_layout()\n            plt.suptitle('The templates for the 13 markers', y=1.01)\n            plt.show()\n\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    def find_markers(self, ima, warn=False, plot=False, title=''):\n        \"\"\"Return 17 markers as list of size-2 integer arrays (row, column)\n\n        Parameters:\n        ima: array of shape (1652, height, 3)\n        \"\"\"\n\n        # 입력 이미지 높이 검증        \n        if ima.shape[0] != 1652:\n            raise ValueError(\"Implemented only for scanned images (image types 3, 4, 11, 12)\")\n\n        markers = np.full((17, 2), -1)\n\n        # Find 13 template-based markers\n        for j in range(len(self._templates)):\n            if self._template_points[j] is not None:\n                # t, l : 원본 이미지 기준 검색 영역 좌측 상단 좌표.\n                t = self._template_positions[j][0]-100\n                l = max(self._template_positions[j][1]-100, 0)\n                # search_range : 마커가 예상되는 위치 주변의 확대 검색 영역\n                search_range = (ima[t:self._template_positions[j][0]+100+self._template_sizes[j][0],\n                                l:self._template_positions[j][1]+250+self._template_sizes[j][0]])\n                # templates[j]와 가장 높은 상관관계 보이는 영역 찾기\n                res = cv2.matchTemplate(search_range, self._templates[j], cv2.TM_CCOEFF)\n                min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(res)\n    \n                top_left = max_loc\n                if warn and max_val < 3e7:\n                    bottom_right = (top_left[0] + self._templates[j].shape[1],\n                                    top_left[1] + self._templates[j].shape[0])\n                    print(j, top_left, max_val)\n                    search_range = search_range.copy()\n                    cv2.rectangle(search_range, top_left, bottom_right, 0, 2)\n                    plt.imshow(search_range)\n                    plt.show()\n                # 최종 마커 좌표 계산\n                markers[j] = np.array((t + top_left[1] + self._template_points[j][0],\n                                       l + top_left[0] + self._template_points[j][1]))\n\n        # Guess the ends of the first three lines (can be outside the bounding box of the image)\n        for i in range(3):\n            # 1~3번째 줄 : 마지막 두 마커의 간격을 동일하다고 가정해 다음 마커 위치 추정.\n            m = markers[5 * i + 3] * 2 - markers[5 * i + 2]\n            markers[5 * i + 4] = m\n\n        # Guess the end of the fourth line (can be outside the bounding box of the image)\n        # 네 번째 라인의 끝점을 다른 마커들의 위치 관계를 이용해 추정.\n        markers[16] = ((markers[14] * (284 + 260) - markers[9] * 260) / 284).astype(int)\n\n        if plot:\n            ima = ima.copy()\n            for m in markers:\n                if m is not None:\n                    cv2.rectangle(ima, (m[1]-40, m[0]-40), (m[1]+40, m[0]+40), (255, 0, 0), 2)\n            # plt.figure(figsize=(12, 8))\n            plt.imshow(ima)\n            plt.title(title)\n            plt.show()\n\n        return markers\n\n    # def baseline(self, i):\n    #     \"\"\"y coordinate of ith baseline in type 1 images\"\"\"\n    #     if i not in [0, 1, 2, 3]:\n    #         raise ValueError(\"i must be in [0, 1, 2, 3]\")\n    #     return self._absolute_points[5 * i][0]\n        \n    @staticmethod\n    def lead_info(lead):\n        \"\"\"Specify which markers mark the begin and the end of a lead.\"\"\"\n        begin, end = {\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        }[lead]\n        return begin // 5, begin, end\n\n    def demo(self, ima, warn=False, title=''):\n        \"\"\"Plot the image with red markers\"\"\"\n        markers = self.find_markers(ima, warn, plot=True, title=title)\n\nmf = MarkerFinder(show_templates=False)\n\nima = cv2.imread('/kaggle/input/physionet-ecg-image-digitization/train/1026034238/1026034238-0011.png') # correct\nmf.demo(ima, warn=False, title='Scanned ECG with 17 line endpoints')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-19T07:00:01.834467Z","iopub.execute_input":"2025-11-19T07:00:01.834730Z","iopub.status.idle":"2025-11-19T07:00:03.244895Z","shell.execute_reply.started":"2025-11-19T07:00:01.834703Z","shell.execute_reply":"2025-11-19T07:00:03.244029Z"},"_kg_hide-input":true},"outputs":[],"execution_count":null}]}