{"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"},{"sourceId":623160,"sourceType":"modelInstanceVersion","modelInstanceId":468838,"modelId":484689},{"sourceId":623170,"sourceType":"modelInstanceVersion","modelInstanceId":468846,"modelId":484698}],"dockerImageVersionId":31154,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"## Acknowledgements & Additions\n\nThis notebook builds upon the open ECG digitizer pipeline released by Ángel Jacinto Sánchez Ruiz (https://www.kaggle.com/code/sacuscreed/open-ecg-digitizer-5e8bfc) and the Open-ECG-Digitizer team. Their models, training code, and shared weights make this solution possible.\n\nKey additions in this revision:\n- Dynamic resampling plus configurable post-processing tailored to the PhysioNet ECG Digitization task.\n- Lead-wise quality diagnostics with drift, amplitude, and stability metrics exported for offline inspection.\n- Optional Gaussian smoothing and baseline correction to denoise extracted waveforms while preserving morphology.\n- However... the score ended up much lower than the 7.32 score of Ángel Jacinto Sánchez Ruiz's code. :P\n","metadata":{}},{"cell_type":"code","source":"%cd /kaggle/input/open-ecg-digitizer/pytorch/default/1\nimport torch\nimport glob\nimport os.path\nimport yaml\nimport torch.nn.functional as F\nimport pandas as pd\nimport numpy as np\nfrom torchvision.io import read_image\nfrom torch import Tensor\n# Imports from https://github.com/Ahus-AIM/Open-ECG-Digitizer\nfrom src.model.unet import UNet\nfrom src.model.perspective_detector import PerspectiveDetector\nfrom src.model.cropper import Cropper\nfrom src.model.pixel_size_finder import PixelSizeFinder\nfrom src.model.signal_extractor import SignalExtractor\nfrom src.model.lead_identifier import LeadIdentifier\nfrom src.model.lead_identifier import LeadIdentifier","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-06T12:23:23.812513Z","iopub.execute_input":"2025-11-06T12:23:23.812709Z","iopub.status.idle":"2025-11-06T12:23:32.472099Z","shell.execute_reply.started":"2025-11-06T12:23:23.812691Z","shell.execute_reply":"2025-11-06T12:23:32.471257Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def add_noise_to_image(input_img, sigma=2, opacity=0.2):\n    noise = torch.sigmoid(torch.randn_like(input_img) * sigma)\n    input_img = (1-opacity)*input_img + opacity * noise\n    return input_img\n\ndef load_model(weights_path, **kwargs):\n    model = UNet(\n        num_in_channels=kwargs.get(\"num_in_channels\", 3),\n        num_out_channels=kwargs.get(\"num_out_channels\", 4),\n        dims=kwargs.get(\n            \"dims\",\n            [32, 64, 128, 256, 320, 320, 320, 320],\n        ),\n        depth=kwargs.get(\"depth\", 2),\n    )\n    state_dict = torch.load(weights_path, map_location=device)\n    # replace _orig_model. with nothing in all keys (model was trained with torch.compile)\n    state_dict = {k.replace(\"_orig_mod.\", \"\"): v for k, v in state_dict.items()}\n    model.load_state_dict(state_dict)\n    model.eval().to(device)\n    return model\n\n\ndef load_png_file(path):\n    img = read_image(path)\n    img = img.float() / 255.0  # Normalize to [0, 1]\n    img = img.unsqueeze(0)  # Add batch dimension\n    # ensure only 3 channels\n    if img.shape[1] > 3:\n        img = img[:, :3, :, :]\n    return img\n\n\ndef _crop_y(\n    image: Tensor,\n    signal_prob: Tensor,\n    grid_prob: Tensor,\n    text_prob: Tensor,\n) -> tuple[Tensor, Tensor, Tensor, Tensor]:\n    def get_bounds(tensor: Tensor) -> tuple[int, int]:\n        prob = torch.clamp(\n            tensor.squeeze().sum(dim=tensor.dim() - 3)\n            - tensor.squeeze().sum(dim=tensor.dim() - 3).mean(),\n            min=0,\n        )\n        non_zero = (prob > 0).nonzero(as_tuple=True)[0]\n        if non_zero.numel() == 0:\n            return 0, tensor.shape[2] - 1\n        return int(non_zero[0].item()), int(non_zero[-1].item())\n\n    y1, y2 = get_bounds(signal_prob + grid_prob)\n\n    slices = (slice(None), slice(None), slice(y1, y2 + 1), slice(None))\n    return (\n        image[slices],\n        signal_prob[slices],\n        grid_prob[slices],\n        text_prob[slices],\n    )\n\n\ndef _align_feature_maps(\n    cropper: Cropper,\n    image: Tensor,\n    signal_prob: Tensor,\n    grid_prob: Tensor,\n    text_prob: Tensor,\n    source_points: Tensor,\n) -> tuple[Tensor, Tensor, Tensor, Tensor]:\n    aligned_signal_prob = cropper.apply_perspective(\n        signal_prob,\n        source_points,\n        fill_value=0,\n    )\n    aligned_image = cropper.apply_perspective(\n        image,\n        source_points,\n        fill_value=0,\n    )\n    aligned_grid_prob = cropper.apply_perspective(\n        grid_prob,\n        source_points,\n        fill_value=0,\n    )\n    aligned_text_prob = cropper.apply_perspective(\n        text_prob,\n        source_points,\n        fill_value=0,\n    )\n    (\n        aligned_image,\n        aligned_signal_prob,\n        aligned_grid_prob,\n        aligned_text_prob,\n    ) = _crop_y(\n        aligned_image,\n        aligned_signal_prob,\n        aligned_grid_prob,\n        aligned_text_prob,\n    )\n\n    return (\n        aligned_image,\n        aligned_signal_prob,\n        aligned_grid_prob,\n        aligned_text_prob,\n    )\n\n\ndef plot_segmentation_and_image(\n    image,\n    segmentation,\n    aligned_signal,\n    aligned_grid,\n    lines,\n):\n    import matplotlib.pyplot as plt\n    import numpy as np\n\n    image_np = image.squeeze(0).permute(1, 2, 0).cpu().numpy()\n    probs = segmentation.squeeze(0).cpu()\n\n    # Make the black pixels white\n    show_featuremap = torch.ones(probs.shape[1], probs.shape[2], 3)\n    probs[2] /= probs[2].max()  # Normalize the third channel to [0, 1]\n    show_featuremap[:, :, [0, 1, 2]] -= 2 * probs[2].unsqueeze(-1)\n    show_featuremap[:, :, [1, 2]] -= probs[0].unsqueeze(-1)\n    show_featuremap = torch.clamp(show_featuremap, 0, 1).numpy()\n\n    straightened_featuremap = torch.ones(\n        aligned_signal.shape[2],\n        aligned_signal.shape[3],\n        3,\n        device=aligned_signal.device,\n    )\n    aligned_signal /= aligned_signal.max()\n    straightened_featuremap[:, :, [0, 1, 2]] -= 2 * aligned_signal[0, 0].unsqueeze(-1)\n    aligned_grid /= aligned_grid.max()\n    straightened_featuremap[:, :, [1, 2]] -= aligned_grid[0, 0].unsqueeze(-1)\n    straightened_featuremap = torch.clamp(straightened_featuremap, 0, 1)\n\n    fig, ax = plt.subplots(2, 2, figsize=(16, 12))\n    ax[0, 0].imshow(image_np)\n    ax[0, 0].axis(\"off\")\n\n    ax[0, 1].imshow(show_featuremap)\n    ax[0, 1].axis(\"off\")\n\n    ax[1, 0].imshow(straightened_featuremap.cpu())\n    ax[1, 0].axis(\"off\")\n\n    offsets = [-0, -9, -6, -0, -3, -6, -0, -3, -6, -0, -3, -6]\n    if lines.numel() > 0:\n        ax[1, 1].plot(lines.T.cpu().numpy() + offsets[: lines.shape[0]])\n    ax[1, 1].axis(\"off\")\n    plt.tight_layout()\n    plt.show()\n\n\ndef crop_image(image, probs):\n    perspective_detector = PerspectiveDetector(num_thetas=200)\n\n    cropper = Cropper(percentiles=(0.03, 0.97), alpha=0.99)\n\n    alignment_params = perspective_detector(probs[0, 0])\n\n    source_points = cropper(probs[0, 1], alignment_params)\n\n    signal_prob, grid_prob, text_prob = (\n        probs[:, [2]],\n        probs[:, [0]],\n        probs[:, [1]],\n    )\n\n    (\n        aligned_image,\n        aligned_signal_prob,\n        aligned_grid_prob,\n        aligned_text_prob,\n    ) = _align_feature_maps(\n        cropper,\n        image,\n        signal_prob,\n        grid_prob,\n        text_prob,\n        source_points,\n    )\n\n    return (\n        aligned_image,\n        aligned_signal_prob,\n        aligned_grid_prob,\n        aligned_text_prob,\n    )\n\n\ndef extract_signals(\n    aligned_signal_prob: Tensor,\n    aligned_grid_prob: Tensor,\n    aligned_text_prob: Tensor,\n    target_num_samples: int,\n) -> Tensor:\n    pixel_size_finder = PixelSizeFinder(\n        min_number_of_grid_lines=30,\n        max_number_of_grid_lines=100,\n        lower_grid_line_factor=0.1,\n    )\n    signal_extractor = SignalExtractor()\n\n    num_in_channels = 1\n    num_out_channels = 13\n    dims = [32, 64, 128, 256, 256]\n    depth = 2\n    layout_unet = load_model(\n        weights_path=\"/kaggle/input/open-ecg-digitizer-weights/pytorch/default/1/lead_name_unet_weights_07072025.pt\",\n        num_in_channels=num_in_channels,\n        num_out_channels=num_out_channels,\n        dims=dims,\n        depth=depth,\n        device=device,\n    )\n\n    layouts = yaml.safe_load(\n        open(\"src/config/lead_layouts_george-moody-2024.yml\", \"r\"),\n    )\n\n    identifier = LeadIdentifier(\n        layouts=layouts,\n        unet=layout_unet,\n        device=torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\"),\n        possibly_flipped=False,\n        target_num_samples=target_num_samples,\n        required_valid_samples=2,\n    )\n    mm_per_pixel_x, mm_per_pixel_y = pixel_size_finder(aligned_grid_prob)\n\n    avg_pixel_per_mm = (1 / mm_per_pixel_x + 1 / mm_per_pixel_y) / 2\n    signals = signal_extractor(aligned_signal_prob.squeeze())\n\n    signals = identifier(\n        signals,\n        aligned_text_prob,\n        avg_pixel_per_mm=avg_pixel_per_mm,\n    )\n\n    return signals\n\ndef resample_image(image: Tensor, resample_size: int) -> Tensor:\n    height, width = image.shape[2], image.shape[3]\n    min_dim = min(height, width)\n    max_dim = max(height, width)\n\n\n    if isinstance(resample_size, int):\n        if max_dim > resample_size:\n            scale = resample_size / max_dim\n            new_size = (int(height * scale), int(width * scale))\n            return F.interpolate(image, size=new_size, mode=\"bilinear\", align_corners=False, antialias=True)\n        return image\n\n    if isinstance(resample_size, tuple):\n        interpolated = F.interpolate(\n            image, size=resample_size, mode=\"bilinear\", align_corners=False, antialias=True\n        )\n        return interpolated\n\n    raise ValueError(f\"Invalid resample_size: {resample_size}. Expected int or tuple of (height, width).\")\n\nleads_names = ['I','II','III','aVR','aVL','aVF','V1','V2','V3','V4','V5','V6']\ndef get_slice(lead_name: str, number_of_rows: int):\n    assert lead_name in leads_names\n    if lead_name in (\"II\",):\n        return slice(0, number_of_rows)\n    if lead_name in ((\"I\", \"III\")):\n        return slice(0, number_of_rows)\n    if lead_name in ((\"aVR\", \"aVF\", \"aVL\")):\n        return slice(1*number_of_rows, 2*number_of_rows)\n    if lead_name in ((\"V1\", \"V2\", \"V3\")):\n        return slice(2*number_of_rows, 3*number_of_rows)\n    if lead_name in ((\"V4\", \"V5\", \"V6\")):\n        return slice(3*number_of_rows, 4*number_of_rows)\n        \n\n\ndef adaptive_resample_size(image: Tensor, max_dim: int, min_dim: int = 2048) -> int:\n    _, _, height, width = image.shape\n    largest = max(int(height), int(width))\n    if largest <= 0:\n        return max_dim\n    if min_dim and largest < min_dim:\n        return min_dim\n    if largest > max_dim:\n        return max_dim\n    return largest\n\ndef _ensure_odd(value: int) -> int:\n    value = int(value)\n    if value % 2 == 0:\n        value += 1\n    return max(value, 1)\n\ndef _rolling_mean(signal: np.ndarray, window: int) -> np.ndarray:\n    if window <= 1:\n        return np.zeros_like(signal, dtype=np.float32)\n    pad_left = window // 2\n    pad_right = window - 1 - pad_left\n    padded = np.pad(signal, (pad_left, pad_right), mode=\"edge\")\n    kernel = np.ones(window, dtype=np.float32) / window\n    return np.convolve(padded, kernel, mode=\"valid\").astype(np.float32)\n\ndef _gaussian_smooth(signal: np.ndarray, kernel_size: int, sigma: float) -> np.ndarray:\n    if kernel_size <= 1:\n        return signal.astype(np.float32)\n    offsets = np.arange(kernel_size, dtype=np.float32) - (kernel_size - 1) / 2.0\n    kernel = np.exp(-0.5 * (offsets / max(sigma, 1e-3)) ** 2)\n    kernel = (kernel / kernel.sum()).astype(np.float32)\n    pad_left = kernel_size // 2\n    pad_right = kernel_size - 1 - pad_left\n    padded = np.pad(signal, (pad_left, pad_right), mode=\"edge\")\n    smoothed = np.convolve(padded, kernel, mode=\"valid\")\n    return smoothed.astype(np.float32)\n\ndef postprocess_signals(signals, lead_names, sample_rate, config=None):\n    config = config or POSTPROCESS_CONFIG\n    baseline_cfg = config.get(\"baseline_correction\", {})\n    smoothing_cfg = config.get(\"smoothing\", {})\n    quality_cfg = config.get(\"quality\", {})\n    array = signals\n    if isinstance(array, torch.Tensor):\n        array = array.detach().cpu().numpy()\n    array = np.asarray(array, dtype=np.float32)\n    processed = []\n    metrics = []\n    for idx, lead_name in enumerate(lead_names):\n        lead_signal = array[idx].astype(np.float32).copy()\n        if lead_signal.size == 0:\n            processed.append(lead_signal)\n            metrics.append(\n                {\n                    \"lead\": lead_name,\n                    \"amplitude_range\": 0.0,\n                    \"baseline_drift\": 0.0,\n                    \"nan_fraction\": 1.0,\n                    \"rms\": 0.0,\n                    \"stability_score\": 0.0,\n                    \"saturation_ratio\": 0.0,\n                    \"quality_flag\": \"empty\",\n                    \"baseline_window\": 0,\n                    \"smoothing_kernel\": 0,\n                }\n            )\n            continue\n\n        finite_mask = np.isfinite(lead_signal)\n        nan_fraction = 1.0 - float(finite_mask.sum() / lead_signal.size)\n        if finite_mask.any():\n            fill_value = float(np.nanmedian(lead_signal[finite_mask]))\n            lead_signal = np.where(finite_mask, lead_signal, fill_value)\n        else:\n            lead_signal = np.zeros_like(lead_signal, dtype=np.float32)\n\n        baseline_window = 0\n        if baseline_cfg.get(\"enabled\", True):\n            baseline_window = max(\n                baseline_cfg.get(\"min_window\", 5),\n                int(sample_rate * baseline_cfg.get(\"seconds\", 0.6)),\n            )\n            baseline_window = min(baseline_window, lead_signal.size - 1)\n            baseline_window = _ensure_odd(max(baseline_window, 3))\n            if baseline_window < lead_signal.size:\n                baseline = _rolling_mean(lead_signal, baseline_window)\n                lead_signal = lead_signal - baseline\n\n        smoothing_kernel = 0\n        if smoothing_cfg.get(\"enabled\", True):\n            smoothing_kernel = max(\n                smoothing_cfg.get(\"min_kernel_size\", 5),\n                int(sample_rate * smoothing_cfg.get(\"seconds\", 0.08)),\n            )\n            smoothing_kernel = min(smoothing_kernel, smoothing_cfg.get(\"max_kernel_size\", 49))\n            smoothing_kernel = min(smoothing_kernel, lead_signal.size - 1)\n            smoothing_kernel = _ensure_odd(max(smoothing_kernel, 3))\n            if smoothing_kernel < lead_signal.size:\n                sigma = smoothing_cfg.get(\"sigma\", max(smoothing_kernel / 6.0, 1.0))\n                lead_signal = _gaussian_smooth(lead_signal, smoothing_kernel, sigma)\n\n        amplitude_range = float(np.max(lead_signal) - np.min(lead_signal))\n        eval_window = min(\n            lead_signal.size // 4,\n            max(5, int(sample_rate * quality_cfg.get(\"drift_seconds\", 0.4))),\n        )\n        if eval_window > 0:\n            start_mean = float(np.mean(lead_signal[:eval_window]))\n            end_mean = float(np.mean(lead_signal[-eval_window:]))\n            baseline_drift = end_mean - start_mean\n        else:\n            baseline_drift = 0.0\n        rms = float(np.sqrt(np.mean(np.square(lead_signal))))\n        derivative = np.diff(lead_signal, prepend=lead_signal[0])\n        diff_mean = float(np.mean(np.abs(derivative)))\n        stability_score = float(rms / (diff_mean + 1e-6))\n        threshold = quality_cfg.get(\"large_derivative_threshold\", 1.2)\n        saturation_ratio = float(np.mean(np.abs(derivative) > threshold))\n\n        quality_flag = \"ok\"\n        if amplitude_range < quality_cfg.get(\"min_amplitude\", 0.05):\n            quality_flag = \"low_amplitude\"\n        if nan_fraction > quality_cfg.get(\"nan_threshold\", 0.05):\n            quality_flag = \"nan_rich\"\n\n        processed.append(lead_signal.astype(np.float32))\n        metrics.append(\n            {\n                \"lead\": lead_name,\n                \"amplitude_range\": amplitude_range,\n                \"baseline_drift\": baseline_drift,\n                \"nan_fraction\": nan_fraction,\n                \"rms\": rms,\n                \"stability_score\": stability_score,\n                \"saturation_ratio\": saturation_ratio,\n                \"quality_flag\": quality_flag,\n                \"baseline_window\": baseline_window,\n                \"smoothing_kernel\": smoothing_kernel,\n            }\n        )\n\n    processed_array = np.stack(processed).astype(np.float32)\n    processed_tensor = torch.from_numpy(processed_array)\n    return processed_tensor, metrics\n\n\nPOSTPROCESS_CONFIG = {\n    \"baseline_correction\": {\"enabled\": True, \"seconds\": 0.6, \"min_window\": 15},\n    \"smoothing\": {\n        \"enabled\": True,\n        \"seconds\": 0.08,\n        \"min_kernel_size\": 5,\n        \"max_kernel_size\": 49,\n        \"sigma\": 2.5,\n    },\n    \"quality\": {\n        \"min_amplitude\": 0.05,\n        \"nan_threshold\": 0.05,\n        \"drift_seconds\": 0.4,\n        \"large_derivative_threshold\": 1.2,\n    },\n}\n\nINFERENCE_CONFIG = {\n    \"resample\": {\"max_dim\": 3200, \"min_dim\": 2400},\n    \"postprocess\": POSTPROCESS_CONFIG,\n    \"collect_quality_metrics\": True,\n}\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nnum_in_channels = 3\nnum_out_channels = 4\ndims = [32, 64, 128, 256, 320, 320, 320, 320]\ndepth = 2\nmodel = load_model(weights_path=\"/kaggle/input/open-ecg-digitizer-weights/pytorch/default/1/unet_weights_07072025.pt\", num_in_channels=num_in_channels, num_out_channels=num_out_channels, dims=dims, depth=depth, device=device)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-06T12:23:32.473430Z","iopub.execute_input":"2025-11-06T12:23:32.473831Z","iopub.status.idle":"2025-11-06T12:23:34.936264Z","shell.execute_reply.started":"2025-11-06T12:23:32.473805Z","shell.execute_reply":"2025-11-06T12:23:34.935379Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\ntest = pd.read_csv('/kaggle/input/physionet-ecg-image-digitization/test.csv')\noutput_path = '/kaggle/working/submission.csv'\nquality_records = []\n\nif os.path.exists(output_path):\n    os.remove(output_path)\npd.DataFrame(columns=[\"id\", \"value\"]).to_csv(output_path, index=False)\n\nold_id = None\nlines = None\n\nfor index, row in test.iterrows():\n    if row.id != old_id:\n        old_id = row.id\n\n        path = f\"/kaggle/input/physionet-ecg-image-digitization/test/{row.id}.png\"\n        target_num_samples = row.fs * 10  # Assuming 10 second signals.\n        input_img = load_png_file(path)\n\n        resample_target = adaptive_resample_size(\n            input_img,\n            max_dim=INFERENCE_CONFIG[\"resample\"][\"max_dim\"],\n            min_dim=INFERENCE_CONFIG[\"resample\"][\"min_dim\"],\n        )\n        input_img = resample_image(image=input_img, resample_size=resample_target)\n\n        with torch.no_grad():\n            logits = model(input_img.to(device))\n            output_probs = torch.softmax(logits, dim=1)\n            aligned_image, aligned_signal, aligned_grid, aligned_text = crop_image(\n                input_img,\n                output_probs,\n            )\n            extracted = extract_signals(\n                aligned_signal,\n                aligned_grid,\n                aligned_text,\n                target_num_samples=target_num_samples,\n            )\n            raw_lines = extracted[\"canonical_lines\"] * 1e-3  # microvolt to millivolt\n\n        processed_lines, per_lead_metrics = postprocess_signals(\n            raw_lines,\n            lead_names=leads_names,\n            sample_rate=row.fs,\n            config=POSTPROCESS_CONFIG,\n        )\n        for metric in per_lead_metrics:\n            metric.update(\n                {\n                    \"id\": row.id,\n                    \"fs\": row.fs,\n                    \"num_samples\": target_num_samples,\n                    \"resample_target\": resample_target,\n                }\n            )\n            quality_records.append(metric)\n        lines = processed_lines\n\n        if index == 0:\n            plot_segmentation_and_image(\n                input_img,\n                output_probs,\n                aligned_signal,\n                aligned_grid,\n                processed_lines,\n            )\n\n    file_id = row.id\n    lead_name = row.lead\n    number_of_rows_in_lead = row.number_of_rows\n\n    lead_index = leads_names.index(lead_name)\n\n    lead_data = lines[lead_index]\n    lead_data = lead_data[get_slice(lead_name, number_of_rows_in_lead)]\n\n    mean_val = np.nanmean(lead_data)\n    if np.isnan(mean_val):\n        mean_val = 0.0\n    lead_data = np.nan_to_num(lead_data, nan=mean_val)\n\n    assert len(lead_data) == number_of_rows_in_lead\n\n    chunk = []\n    for t in range(number_of_rows_in_lead):\n        chunk.append({\"id\": f\"{file_id}_{t}_{lead_name}\", \"value\": float(lead_data[t])})\n\n    if chunk:\n        pd.DataFrame(chunk).to_csv(output_path, mode='a', index=False, header=False)\n\nif INFERENCE_CONFIG.get(\"collect_quality_metrics\", True):\n    if quality_records:\n        quality_diagnostics = pd.DataFrame(quality_records)\n        quality_diagnostics.to_csv('/kaggle/working/quality_diagnostics.csv', index=False)\n    else:\n        quality_diagnostics = pd.DataFrame(\n            columns=[\n                \"id\",\n                \"lead\",\n                \"amplitude_range\",\n                \"baseline_drift\",\n                \"nan_fraction\",\n                \"rms\",\n                \"stability_score\",\n                \"saturation_ratio\",\n                \"quality_flag\",\n                \"baseline_window\",\n                \"smoothing_kernel\",\n                \"fs\",\n                \"num_samples\",\n                \"resample_target\",\n            ]\n        )\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-06T12:23:34.937250Z","iopub.execute_input":"2025-11-06T12:23:34.937485Z","iopub.status.idle":"2025-11-06T12:24:07.086807Z","shell.execute_reply.started":"2025-11-06T12:23:34.937459Z","shell.execute_reply":"2025-11-06T12:24:07.085941Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"if 'quality_diagnostics' in globals() and not quality_diagnostics.empty:\n    print(\"Quality diagnostics preview (first 12 rows):\")\n    print(quality_diagnostics.head(12).to_string(index=False))\n    summary = quality_diagnostics.groupby(\"lead\").agg(\n        amplitude_mean=(\"amplitude_range\", \"mean\"),\n        amplitude_std=(\"amplitude_range\", \"std\"),\n        drift_mean=(\"baseline_drift\", \"mean\"),\n        drift_std=(\"baseline_drift\", \"std\"),\n        nan_fraction_mean=(\"nan_fraction\", \"mean\"),\n        stability_mean=(\"stability_score\", \"mean\"),\n    ).round(6)\n    print(\"Lead-level diagnostic summary:\")\n    print(summary.to_string())\nelse:\n    print(\"No quality diagnostics were generated. Check configuration or input files.\")\nprint(f\"Submission saved to {output_path}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-06T12:24:36.853755Z","iopub.execute_input":"2025-11-06T12:24:36.854057Z","iopub.status.idle":"2025-11-06T12:24:36.888727Z","shell.execute_reply.started":"2025-11-06T12:24:36.854038Z","shell.execute_reply":"2025-11-06T12:24:36.888109Z"}},"outputs":[],"execution_count":null}]}