deepecgkit.explainability

Attribution for trained ECG models. Each method assigns every sample of every lead a contribution to one output class, and the surrounding helpers reduce that map to a distribution over leads and to a beat-aligned profile with its P, QRS and T shares.

The package re-exports every symbol below, so from deepecgkit.explainability import ECGExplainer works whichever submodule a name is defined in. The plotting helpers are imported on first use, so importing the package does not pull matplotlib into the path of every command.

Explainer

explain

explain(
    data: ExplainInput,
    model: Module | None = None,
    checkpoint: str | Path | None = None,
    weights: str | Path | None = None,
    model_name: str | None = None,
    input_channels: int | None = None,
    output_size: int | None = None,
    target: TargetSpec = None,
    split: str = "test",
    **kwargs: Any,
) -> Attribution

Attribute a model's predictions in one call.

Exactly one weight source must be given: an already-built model, a checkpoint path, or pretrained weights.

Parameters:

Name	Type	Description	Default
data	ExplainInput	An array, tensor, file path, Dataset, DataLoader, or ECGDataModule	required
model	Module | None	An already-built model	None
checkpoint	str | Path | None	Path to a checkpoint	None
weights	str | Path | None	Registered pretrained weight name, or a path to a weights file	None
model_name	str | None	Registered model name; optional for a checkpoint that records its own	None
input_channels	int | None	Lead count, when it cannot be recovered from the weights	None
output_size	int | None	Classifier width, when it cannot be recovered	None
target	TargetSpec	None for each sample's own predicted class, an integer, or one index per sample	None
split	str	Which split to use when data is an ECGDataModule	'test'
**kwargs	Any	Forwarded to ECGExplainer	{}

Returns:

Type	Description
Attribution	The attribution, with its lead distribution and beat profile.

Raises:

Type	Description
ValueError	If not exactly one weight source is given, or a checkpoint neither records its architecture nor is given a model name.

Example

result = explain(signals, checkpoint="best.ckpt", model_name="kanres") result.pqrst_fractions["P"] 0.31

Source code in deepecgkit/explainability/attribution.py

def explain(
    data: ExplainInput,
    model: nn.Module | None = None,
    checkpoint: str | Path | None = None,
    weights: str | Path | None = None,
    model_name: str | None = None,
    input_channels: int | None = None,
    output_size: int | None = None,
    target: TargetSpec = None,
    split: str = "test",
    **kwargs: Any,
) -> Attribution:
    """Attribute a model's predictions in one call.

    Exactly one weight source must be given: an already-built model, a
    checkpoint path, or pretrained weights.

    Args:
        data: An array, tensor, file path, Dataset, DataLoader, or ECGDataModule
        model: An already-built model
        checkpoint: Path to a checkpoint
        weights: Registered pretrained weight name, or a path to a weights file
        model_name: Registered model name; optional for a checkpoint that
            records its own
        input_channels: Lead count, when it cannot be recovered from the weights
        output_size: Classifier width, when it cannot be recovered
        target: None for each sample's own predicted class, an integer, or one
            index per sample
        split: Which split to use when data is an ECGDataModule
        **kwargs: Forwarded to ECGExplainer

    Returns:
        The attribution, with its lead distribution and beat profile.

    Raises:
        ValueError: If not exactly one weight source is given, or a checkpoint
            neither records its architecture nor is given a model name.

    Example:
        >>> result = explain(signals, checkpoint="best.ckpt", model_name="kanres")
        >>> result.pqrst_fractions["P"]
        0.31
    """
    supplied = [
        name
        for name, value in (("model", model), ("checkpoint", checkpoint), ("weights", weights))
        if value is not None
    ]
    if len(supplied) != 1:
        raise ValueError(
            f"Pass exactly one of model, checkpoint, or weights; got "
            f"{', '.join(supplied) or 'none'}."
        )

    if model is not None:
        explainer = ECGExplainer(model, model_name=model_name, **kwargs)
    elif checkpoint is not None:
        explainer = ECGExplainer.from_checkpoint(
            checkpoint,
            model_name,
            input_channels=input_channels,
            output_size=output_size,
            **kwargs,
        )
    else:
        explainer = ECGExplainer.from_pretrained(weights, model_name=model_name, **kwargs)

    return explainer.explain(data, target=target, split=split)

ECGExplainer

Runs an attribution method over data and summarises it for ECG.

The explainer holds the same responsibilities as FeatureExtractor, with one inversion: it must run with gradients enabled, including when a caller invokes it from inside an evaluation loop that has already entered torch.no_grad().

Parameters:

Name	Type	Description	Default
model	Module	Any registered model	required
model_name	str | None	Registry name recorded on the results, inferred from the model when omitted	None
method	str	One of METHODS	'saliency'
batch_size	int	Recordings per attribution pass	8
device	str	Device name, or auto to prefer CUDA then MPS then CPU	'auto'
sampling_rate	float	Samples per second, used for beat alignment and plots	500.0
lead_names	list[str] | None	Display names for the leads, defaulting to the standard twelve-lead order when twelve leads are present	None
max_samples	int | None	Stop after this many recordings, or None for all of them	None
ig_steps	int	Path length for integrated gradients	64
occlusion_window	int	Span width in samples for occlusion	100
occlusion_stride	int	Step between spans for occlusion	50
align_beats	bool	Detect R-peaks and build the beat profile	True
detect_lead	int | None	Lead index R-peaks are detected on, defaulting to the second lead when there is more than one	None
progress	bool	Show a progress bar when tqdm is installed	False

Example

explainer = ECGExplainer.from_checkpoint("best.ckpt", "kanres") result = explainer.explain(signals) result.lead_importance.shape (100, 12)

Source code in deepecgkit/explainability/attribution.py

class ECGExplainer:
    """Runs an attribution method over data and summarises it for ECG.

    The explainer holds the same responsibilities as FeatureExtractor, with one
    inversion: it must run with gradients enabled, including when a caller
    invokes it from inside an evaluation loop that has already entered
    ``torch.no_grad()``.

    Args:
        model: Any registered model
        model_name: Registry name recorded on the results, inferred from the
            model when omitted
        method: One of METHODS
        batch_size: Recordings per attribution pass
        device: Device name, or auto to prefer CUDA then MPS then CPU
        sampling_rate: Samples per second, used for beat alignment and plots
        lead_names: Display names for the leads, defaulting to the standard
            twelve-lead order when twelve leads are present
        max_samples: Stop after this many recordings, or None for all of them
        ig_steps: Path length for integrated gradients
        occlusion_window: Span width in samples for occlusion
        occlusion_stride: Step between spans for occlusion
        align_beats: Detect R-peaks and build the beat profile
        detect_lead: Lead index R-peaks are detected on, defaulting to the
            second lead when there is more than one
        progress: Show a progress bar when tqdm is installed

    Example:
        >>> explainer = ECGExplainer.from_checkpoint("best.ckpt", "kanres")
        >>> result = explainer.explain(signals)
        >>> result.lead_importance.shape
        (100, 12)
    """

    def __init__(
        self,
        model: nn.Module,
        model_name: str | None = None,
        method: str = "saliency",
        batch_size: int = 8,
        device: str = "auto",
        sampling_rate: float = 500.0,
        lead_names: list[str] | None = None,
        max_samples: int | None = None,
        ig_steps: int = 64,
        occlusion_window: int = 100,
        occlusion_stride: int = 50,
        align_beats: bool = True,
        detect_lead: int | None = None,
        num_workers: int = 0,
        progress: bool = False,
    ) -> None:
        self.model = model
        self.model_name = model_name or resolve_model_name(model)
        self.method = method
        self.batch_size = batch_size
        self.device = device
        self.sampling_rate = sampling_rate
        self.lead_names = lead_names
        self.max_samples = max_samples
        self.ig_steps = ig_steps
        self.occlusion_window = occlusion_window
        self.occlusion_stride = occlusion_stride
        self.align_beats = align_beats
        self.detect_lead = detect_lead
        self.num_workers = num_workers
        self.progress = progress

    @classmethod
    def from_checkpoint(
        cls,
        checkpoint: str | Path,
        model_name: str | None = None,
        input_channels: int | None = None,
        output_size: int | None = None,
        **kwargs: Any,
    ) -> "ECGExplainer":
        """Build an explainer from a saved checkpoint.

        Args:
            checkpoint: Path to an ECGTrainer checkpoint or a bare state dict
            model_name: Registered model name matching the weights; optional
                when the checkpoint records it
            input_channels: Lead count, required for architectures whose weights
                do not encode it
            output_size: Classifier width, normally recovered from the weights
            **kwargs: Forwarded to the constructor

        Returns:
            An explainer wrapping the loaded model.
        """
        model, _ = build_model_from_checkpoint(
            checkpoint,
            name=model_name,
            input_channels=input_channels,
            output_size=output_size,
        )
        return cls(model, model_name=model_name, **kwargs)

    @classmethod
    def from_pretrained(
        cls,
        weights: str | Path,
        model_name: str | None = None,
        **kwargs: Any,
    ) -> "ECGExplainer":
        """Build an explainer from registered pretrained weights or a file.

        Args:
            weights: Name of registered pretrained weights, or a path
            model_name: Registered model name, inferred from the weight registry
                when the weights are registered
            **kwargs: Forwarded to the constructor, except that unrecognised
                arguments are passed to the model instead

        Returns:
            An explainer wrapping the loaded model.

        Raises:
            ValueError: If the model name is neither given nor inferable.
        """
        explainer_kwargs, model_kwargs = split_wrapper_kwargs(cls, kwargs)
        model, resolved = load_pretrained_resolving_name(weights, model_name, **model_kwargs)
        return cls(model, model_name=resolved, **explainer_kwargs)

    @property
    def input_channels(self) -> int | None:
        """Lead count recovered from the model, or None when unrecoverable."""
        return model_input_channels(self.model)

    def explain(
        self, data: ExplainInput, target: TargetSpec = None, split: str = "test"
    ) -> Attribution:
        """Attribute a model's predictions over the data.

        Args:
            data: An array, tensor, file path, Dataset, DataLoader, or
                ECGDataModule
            target: None for each sample's own predicted class, an integer for
                one class throughout, or one index per sample
            split: Which split to use when data is an ECGDataModule

        Returns:
            The attribution, with the lead distribution always present and the
            beat profile present when align_beats is set and beats were found.

        Raises:
            ValueError: If the split or method is unknown.
        """
        if split not in SPLITS:
            raise ValueError(f"Unknown split: {split}. Choose one of {', '.join(SPLITS)}.")
        if self.method not in _METHODS:
            raise ValueError(f"Unknown method: {self.method}. Choose one of {', '.join(METHODS)}.")

        device = _resolve_device(self.device)
        loader = self._as_dataloader(data, split)

        value_chunks: list[np.ndarray] = []
        signal_chunks: list[np.ndarray] = []
        target_chunks: list[np.ndarray] = []
        crop_offset = 0
        seen = 0

        with _gradient_ready(self.model, device):
            iterator: Any = loader
            if self.progress and tqdm is not None:
                iterator = tqdm(loader, desc=f"Attributing ({self.method})")

            for batch in iterator:
                signals = batch[0] if isinstance(batch, list | tuple) else batch
                signals = signals.to(device=device, dtype=torch.float32)

                if self.max_samples is not None:
                    remaining = self.max_samples - seen
                    if remaining <= 0:
                        break
                    signals = signals[:remaining]

                signals, crop_offset = fit_to_constraints(signals, self.model, self.model_name)
                batch_target = self._batch_target(target, seen, signals.shape[0])

                with torch.no_grad():
                    index = _resolve_target(_forward(self.model, signals), batch_target)

                values = self._run_method(signals, index)
                value_chunks.append(values.detach().cpu().numpy().astype(np.float32))
                signal_chunks.append(signals.detach().cpu().numpy().astype(np.float32))
                target_chunks.append(index.detach().cpu().numpy().astype(np.int64))
                seen += signals.shape[0]

        if not value_chunks:
            raise ValueError("The supplied data yielded no recordings to explain.")

        values = np.concatenate(value_chunks, axis=0)
        signal = np.concatenate(signal_chunks, axis=0)
        targets = np.concatenate(target_chunks, axis=0)
        names = self.lead_names or default_lead_names(values.shape[1])

        result = Attribution(
            values=values,
            signal=signal,
            lead_importance=lead_importance(values),
            target=targets,
            method=self.method,
            model_name=self.model_name,
            sampling_rate=self.sampling_rate,
            crop_offset=crop_offset,
            lead_names=names,
        )

        if self.align_beats:
            self._attach_beat_profile(result, values, signal)

        return result

    def _run_method(self, signals: torch.Tensor, index: torch.Tensor) -> torch.Tensor:
        """Dispatch to the configured attribution method.

        Args:
            signals: Batch of shape (batch, leads, time) on the target device
            index: Resolved class index per sample

        Returns:
            Attribution of shape (batch, leads, time).
        """
        if self.method == "integrated-gradients":
            return integrated_gradients(self.model, signals, index, steps=self.ig_steps)
        if self.method == "occlusion":
            return occlusion(
                self.model,
                signals,
                index,
                window=self.occlusion_window,
                stride=self.occlusion_stride,
            )
        return saliency(self.model, signals, index)

    def _batch_target(self, target: TargetSpec, seen: int, size: int) -> TargetSpec:
        """Slice a per-sample target specification down to the current batch.

        Args:
            target: The caller's target specification
            seen: How many recordings have already been explained
            size: Size of the current batch

        Returns:
            The target specification that applies to this batch.
        """
        if target is None or isinstance(target, int):
            return target
        indices = np.asarray(target).reshape(-1)
        if indices.size == 1:
            return int(indices[0])
        return indices[seen : seen + size]

    def _attach_beat_profile(
        self, result: Attribution, values: np.ndarray, signal: np.ndarray
    ) -> None:
        """Detect beats, average attribution over them, and record the shares.

        R-peaks are detected on the ECG, never on the attribution. The cardiac
        cycle is a property of the recording, so anchoring to the attribution
        would make the window positions depend on the method under test and the
        beat counts would differ between saliency and occlusion on the same
        recording.

        Profiles are pooled across the batch weighted by how many beats each
        recording contributed, so the result answers a question about the model
        rather than about one recording.

        Args:
            result: The attribution to annotate in place
            values: Attribution of shape (batch, leads, time)
            signal: The signal the model saw, of the same shape
        """
        lead = self.detect_lead
        if lead is None:
            lead = 1 if values.shape[1] > 1 else 0

        profiles: list[np.ndarray] = []
        weights: list[int] = []
        times: np.ndarray | None = None

        for attribution, trace in zip(values, signal, strict=False):
            peaks = detect_r_peaks(trace[lead], self.sampling_rate)
            profile, axis, count = beat_aligned_attribution(
                attribution, peaks, self.sampling_rate, BEAT_WINDOW_MS
            )
            times = axis
            if count:
                profiles.append(profile)
                weights.append(count)

        if not profiles or times is None:
            warnings.warn(
                "No complete beats were found, so no beat profile was produced. Check the "
                "sampling rate, or widen the detector's band with detect_r_peaks.",
                RuntimeWarning,
                stacklevel=3,
            )
            return

        total = float(sum(weights))
        pooled = np.tensordot(np.asarray(weights, dtype=np.float64) / total, profiles, axes=(0, 0))

        result.beat_profile = pooled.astype(np.float32)
        result.beat_times_ms = times
        result.beat_count = int(total)
        result.pqrst_fractions = pqrst_fractions(pooled, times, PQRST_WINDOWS_MS)
        result.pqrst_enrichment = pqrst_enrichment(result.pqrst_fractions, times, PQRST_WINDOWS_MS)

    def _as_dataloader(self, data: ExplainInput, split: str) -> DataLoader:
        """Normalise any supported input to a non-shuffling DataLoader."""
        return as_inference_dataloader(
            data,
            split=split,
            batch_size=self.batch_size,
            num_workers=self.num_workers,
            input_channels=self.input_channels,
            verb="explained",
        )

input_channels property

input_channels: int | None

Lead count recovered from the model, or None when unrecoverable.

from_checkpoint classmethod

from_checkpoint(
    checkpoint: str | Path,
    model_name: str | None = None,
    input_channels: int | None = None,
    output_size: int | None = None,
    **kwargs: Any,
) -> ECGExplainer

Build an explainer from a saved checkpoint.

Parameters:

Name	Type	Description	Default
checkpoint	str | Path	Path to an ECGTrainer checkpoint or a bare state dict	required
model_name	str | None	Registered model name matching the weights; optional when the checkpoint records it	None
input_channels	int | None	Lead count, required for architectures whose weights do not encode it	None
output_size	int | None	Classifier width, normally recovered from the weights	None
**kwargs	Any	Forwarded to the constructor	{}

Returns:

Type	Description
ECGExplainer	An explainer wrapping the loaded model.

Source code in deepecgkit/explainability/attribution.py

@classmethod
def from_checkpoint(
    cls,
    checkpoint: str | Path,
    model_name: str | None = None,
    input_channels: int | None = None,
    output_size: int | None = None,
    **kwargs: Any,
) -> "ECGExplainer":
    """Build an explainer from a saved checkpoint.

    Args:
        checkpoint: Path to an ECGTrainer checkpoint or a bare state dict
        model_name: Registered model name matching the weights; optional
            when the checkpoint records it
        input_channels: Lead count, required for architectures whose weights
            do not encode it
        output_size: Classifier width, normally recovered from the weights
        **kwargs: Forwarded to the constructor

    Returns:
        An explainer wrapping the loaded model.
    """
    model, _ = build_model_from_checkpoint(
        checkpoint,
        name=model_name,
        input_channels=input_channels,
        output_size=output_size,
    )
    return cls(model, model_name=model_name, **kwargs)

from_pretrained classmethod

from_pretrained(
    weights: str | Path,
    model_name: str | None = None,
    **kwargs: Any,
) -> ECGExplainer

Build an explainer from registered pretrained weights or a file.

Parameters:

Name	Type	Description	Default
weights	str | Path	Name of registered pretrained weights, or a path	required
model_name	str | None	Registered model name, inferred from the weight registry when the weights are registered	None
**kwargs	Any	Forwarded to the constructor, except that unrecognised arguments are passed to the model instead	{}

Returns:

Type	Description
ECGExplainer	An explainer wrapping the loaded model.

Raises:

Type	Description
ValueError	If the model name is neither given nor inferable.

Source code in deepecgkit/explainability/attribution.py

@classmethod
def from_pretrained(
    cls,
    weights: str | Path,
    model_name: str | None = None,
    **kwargs: Any,
) -> "ECGExplainer":
    """Build an explainer from registered pretrained weights or a file.

    Args:
        weights: Name of registered pretrained weights, or a path
        model_name: Registered model name, inferred from the weight registry
            when the weights are registered
        **kwargs: Forwarded to the constructor, except that unrecognised
            arguments are passed to the model instead

    Returns:
        An explainer wrapping the loaded model.

    Raises:
        ValueError: If the model name is neither given nor inferable.
    """
    explainer_kwargs, model_kwargs = split_wrapper_kwargs(cls, kwargs)
    model, resolved = load_pretrained_resolving_name(weights, model_name, **model_kwargs)
    return cls(model, model_name=resolved, **explainer_kwargs)

explain

explain(
    data: ExplainInput,
    target: TargetSpec = None,
    split: str = "test",
) -> Attribution

Attribute a model's predictions over the data.

Parameters:

Name	Type	Description	Default
data	ExplainInput	An array, tensor, file path, Dataset, DataLoader, or ECGDataModule	required
target	TargetSpec	None for each sample's own predicted class, an integer for one class throughout, or one index per sample	None
split	str	Which split to use when data is an ECGDataModule	'test'

Returns:

Type	Description
Attribution	The attribution, with the lead distribution always present and the
Attribution	beat profile present when align_beats is set and beats were found.

Raises:

Type	Description
ValueError	If the split or method is unknown.

Source code in deepecgkit/explainability/attribution.py

def explain(
    self, data: ExplainInput, target: TargetSpec = None, split: str = "test"
) -> Attribution:
    """Attribute a model's predictions over the data.

    Args:
        data: An array, tensor, file path, Dataset, DataLoader, or
            ECGDataModule
        target: None for each sample's own predicted class, an integer for
            one class throughout, or one index per sample
        split: Which split to use when data is an ECGDataModule

    Returns:
        The attribution, with the lead distribution always present and the
        beat profile present when align_beats is set and beats were found.

    Raises:
        ValueError: If the split or method is unknown.
    """
    if split not in SPLITS:
        raise ValueError(f"Unknown split: {split}. Choose one of {', '.join(SPLITS)}.")
    if self.method not in _METHODS:
        raise ValueError(f"Unknown method: {self.method}. Choose one of {', '.join(METHODS)}.")

    device = _resolve_device(self.device)
    loader = self._as_dataloader(data, split)

    value_chunks: list[np.ndarray] = []
    signal_chunks: list[np.ndarray] = []
    target_chunks: list[np.ndarray] = []
    crop_offset = 0
    seen = 0

    with _gradient_ready(self.model, device):
        iterator: Any = loader
        if self.progress and tqdm is not None:
            iterator = tqdm(loader, desc=f"Attributing ({self.method})")

        for batch in iterator:
            signals = batch[0] if isinstance(batch, list | tuple) else batch
            signals = signals.to(device=device, dtype=torch.float32)

            if self.max_samples is not None:
                remaining = self.max_samples - seen
                if remaining <= 0:
                    break
                signals = signals[:remaining]

            signals, crop_offset = fit_to_constraints(signals, self.model, self.model_name)
            batch_target = self._batch_target(target, seen, signals.shape[0])

            with torch.no_grad():
                index = _resolve_target(_forward(self.model, signals), batch_target)

            values = self._run_method(signals, index)
            value_chunks.append(values.detach().cpu().numpy().astype(np.float32))
            signal_chunks.append(signals.detach().cpu().numpy().astype(np.float32))
            target_chunks.append(index.detach().cpu().numpy().astype(np.int64))
            seen += signals.shape[0]

    if not value_chunks:
        raise ValueError("The supplied data yielded no recordings to explain.")

    values = np.concatenate(value_chunks, axis=0)
    signal = np.concatenate(signal_chunks, axis=0)
    targets = np.concatenate(target_chunks, axis=0)
    names = self.lead_names or default_lead_names(values.shape[1])

    result = Attribution(
        values=values,
        signal=signal,
        lead_importance=lead_importance(values),
        target=targets,
        method=self.method,
        model_name=self.model_name,
        sampling_rate=self.sampling_rate,
        crop_offset=crop_offset,
        lead_names=names,
    )

    if self.align_beats:
        self._attach_beat_profile(result, values, signal)

    return result

Attribution dataclass

Attribution for a batch of recordings, with its ECG-native summaries.

Attributes:

Name	Type	Description
values	ndarray	Per-lead per-timestep attribution of shape (batch, leads, time)
signal	ndarray	The signal the model actually saw, of the same shape, kept so attribution and trace can never fall out of alignment
lead_importance	ndarray	Distribution over leads of shape (batch, leads), rows summing to one
target	ndarray	Class index explained for each sample, of shape (batch,)
method	str	Which of METHODS produced the values
model_name	str | None	Registry name of the model, for provenance
sampling_rate	float	Samples per second the time axis assumes
crop_offset	int	Index in the original recording where these values begin, non-zero when the signal was centre-cropped to fit the model
lead_names	list[str] | None	Display names aligned with the lead axis
beat_profile	ndarray | None	Mean attribution over R-peak-aligned beats, of shape (leads, window), or None when beat alignment was not run
beat_times_ms	ndarray | None	Time axis of the beat window in milliseconds relative to the R-peak, or None
pqrst_fractions	dict[str, float] | None	Share of beat-profile mass in each cardiac region, plus the baseline remainder, or None
pqrst_enrichment	dict[str, float] | None	Each region's share divided by its share of the beat window's duration, so one means chance, or None
beat_count	int	Number of beats averaged into the profile

Source code in deepecgkit/explainability/attribution.py

@dataclass
class Attribution:
    """Attribution for a batch of recordings, with its ECG-native summaries.

    Attributes:
        values: Per-lead per-timestep attribution of shape (batch, leads, time)
        signal: The signal the model actually saw, of the same shape, kept so
            attribution and trace can never fall out of alignment
        lead_importance: Distribution over leads of shape (batch, leads), rows
            summing to one
        target: Class index explained for each sample, of shape (batch,)
        method: Which of METHODS produced the values
        model_name: Registry name of the model, for provenance
        sampling_rate: Samples per second the time axis assumes
        crop_offset: Index in the original recording where these values begin,
            non-zero when the signal was centre-cropped to fit the model
        lead_names: Display names aligned with the lead axis
        beat_profile: Mean attribution over R-peak-aligned beats, of shape
            (leads, window), or None when beat alignment was not run
        beat_times_ms: Time axis of the beat window in milliseconds relative to
            the R-peak, or None
        pqrst_fractions: Share of beat-profile mass in each cardiac region, plus
            the baseline remainder, or None
        pqrst_enrichment: Each region's share divided by its share of the beat
            window's duration, so one means chance, or None
        beat_count: Number of beats averaged into the profile
    """

    values: np.ndarray
    signal: np.ndarray
    lead_importance: np.ndarray
    target: np.ndarray
    method: str
    model_name: str | None = None
    sampling_rate: float = 500.0
    crop_offset: int = 0
    lead_names: list[str] | None = None
    beat_profile: np.ndarray | None = None
    beat_times_ms: np.ndarray | None = None
    pqrst_fractions: dict[str, float] | None = field(default=None)
    pqrst_enrichment: dict[str, float] | None = field(default=None)
    beat_count: int = 0

    @property
    def num_leads(self) -> int:
        """Number of leads the attribution covers."""
        return int(self.values.shape[1])

    def __len__(self) -> int:
        """Number of recordings explained."""
        return int(self.values.shape[0])

    def mean_lead_importance(self) -> np.ndarray:
        """Average the per-sample lead distribution over the batch.

        Returns:
            Array of shape (leads,) summing to one.
        """
        return self.lead_importance.mean(axis=0).astype(np.float32)

    def summary(self) -> dict[str, Any]:
        """Describe the attribution in JSON-serialisable form.

        Returns:
            A dictionary holding the provenance, the mean lead distribution
            keyed by lead name, and the cardiac-region breakdown.
        """
        names = self.lead_names or default_lead_names(self.num_leads)
        mean_importance = self.mean_lead_importance()
        return {
            "method": self.method,
            "model_name": self.model_name,
            "samples": len(self),
            "leads": self.num_leads,
            "timesteps": int(self.values.shape[2]),
            "sampling_rate": float(self.sampling_rate),
            "crop_offset": int(self.crop_offset),
            "targets": [int(value) for value in self.target],
            "lead_importance": {
                name: float(value) for name, value in zip(names, mean_importance, strict=False)
            },
            "beat_count": int(self.beat_count),
            "pqrst_fractions": self.pqrst_fractions,
            "pqrst_enrichment": self.pqrst_enrichment,
        }

    def save(self, path: str | Path) -> None:
        """Write the attribution arrays to disk.

        Args:
            path: Destination ending in .npz

        Raises:
            ValueError: If the suffix is not .npz. The arrays are per-lead and
                per-timestep, so the flat formats FeatureSet supports would lose
                the axis that makes them worth keeping.
        """
        out_path = Path(path)
        if out_path.suffix != ".npz":
            raise ValueError(
                f"Unsupported output format: {out_path.suffix}. Attribution is saved as .npz."
            )

        out_path.parent.mkdir(parents=True, exist_ok=True)
        payload: dict[str, Any] = {
            "values": self.values,
            "signal": self.signal,
            "lead_importance": self.lead_importance,
            "target": self.target,
            "method": np.asarray(self.method),
            "sampling_rate": np.asarray(self.sampling_rate),
            "crop_offset": np.asarray(self.crop_offset),
            "beat_count": np.asarray(self.beat_count),
        }
        if self.model_name is not None:
            payload["model_name"] = np.asarray(self.model_name)
        if self.lead_names is not None:
            payload["lead_names"] = np.asarray(self.lead_names)
        if self.beat_profile is not None:
            payload["beat_profile"] = self.beat_profile
        if self.beat_times_ms is not None:
            payload["beat_times_ms"] = self.beat_times_ms
        if self.pqrst_fractions is not None:
            payload["pqrst_fractions"] = np.asarray(json.dumps(self.pqrst_fractions))
        if self.pqrst_enrichment is not None:
            payload["pqrst_enrichment"] = np.asarray(json.dumps(self.pqrst_enrichment))
        np.savez(str(out_path), **payload)

num_leads property

num_leads: int

Number of leads the attribution covers.

mean_lead_importance

mean_lead_importance() -> np.ndarray

Average the per-sample lead distribution over the batch.

Returns:

Type	Description
ndarray	Array of shape (leads,) summing to one.

Source code in deepecgkit/explainability/attribution.py

def mean_lead_importance(self) -> np.ndarray:
    """Average the per-sample lead distribution over the batch.

    Returns:
        Array of shape (leads,) summing to one.
    """
    return self.lead_importance.mean(axis=0).astype(np.float32)

summary

summary() -> dict[str, Any]

Describe the attribution in JSON-serialisable form.

Returns:

Type	Description
dict[str, Any]	A dictionary holding the provenance, the mean lead distribution
dict[str, Any]	keyed by lead name, and the cardiac-region breakdown.

Source code in deepecgkit/explainability/attribution.py

def summary(self) -> dict[str, Any]:
    """Describe the attribution in JSON-serialisable form.

    Returns:
        A dictionary holding the provenance, the mean lead distribution
        keyed by lead name, and the cardiac-region breakdown.
    """
    names = self.lead_names or default_lead_names(self.num_leads)
    mean_importance = self.mean_lead_importance()
    return {
        "method": self.method,
        "model_name": self.model_name,
        "samples": len(self),
        "leads": self.num_leads,
        "timesteps": int(self.values.shape[2]),
        "sampling_rate": float(self.sampling_rate),
        "crop_offset": int(self.crop_offset),
        "targets": [int(value) for value in self.target],
        "lead_importance": {
            name: float(value) for name, value in zip(names, mean_importance, strict=False)
        },
        "beat_count": int(self.beat_count),
        "pqrst_fractions": self.pqrst_fractions,
        "pqrst_enrichment": self.pqrst_enrichment,
    }

save

save(path: str | Path) -> None

Write the attribution arrays to disk.

Parameters:

Name	Type	Description	Default
path	str | Path	Destination ending in .npz	required

Raises:

Type	Description
ValueError	If the suffix is not .npz. The arrays are per-lead and per-timestep, so the flat formats FeatureSet supports would lose the axis that makes them worth keeping.

Source code in deepecgkit/explainability/attribution.py

def save(self, path: str | Path) -> None:
    """Write the attribution arrays to disk.

    Args:
        path: Destination ending in .npz

    Raises:
        ValueError: If the suffix is not .npz. The arrays are per-lead and
            per-timestep, so the flat formats FeatureSet supports would lose
            the axis that makes them worth keeping.
    """
    out_path = Path(path)
    if out_path.suffix != ".npz":
        raise ValueError(
            f"Unsupported output format: {out_path.suffix}. Attribution is saved as .npz."
        )

    out_path.parent.mkdir(parents=True, exist_ok=True)
    payload: dict[str, Any] = {
        "values": self.values,
        "signal": self.signal,
        "lead_importance": self.lead_importance,
        "target": self.target,
        "method": np.asarray(self.method),
        "sampling_rate": np.asarray(self.sampling_rate),
        "crop_offset": np.asarray(self.crop_offset),
        "beat_count": np.asarray(self.beat_count),
    }
    if self.model_name is not None:
        payload["model_name"] = np.asarray(self.model_name)
    if self.lead_names is not None:
        payload["lead_names"] = np.asarray(self.lead_names)
    if self.beat_profile is not None:
        payload["beat_profile"] = self.beat_profile
    if self.beat_times_ms is not None:
        payload["beat_times_ms"] = self.beat_times_ms
    if self.pqrst_fractions is not None:
        payload["pqrst_fractions"] = np.asarray(json.dumps(self.pqrst_fractions))
    if self.pqrst_enrichment is not None:
        payload["pqrst_enrichment"] = np.asarray(json.dumps(self.pqrst_enrichment))
    np.savez(str(out_path), **payload)

Attribution methods

saliency

saliency(
    model: Module,
    signals: Tensor,
    target: TargetSpec = None,
    absolute: bool = True,
) -> torch.Tensor

Attribute by the gradient of the target logit with respect to the input.

The cheapest of the three methods, at one forward and one backward pass, and the noisiest: the gradient measures sensitivity at a single point, so it reacts to sharp local structure such as the QRS complex whether or not the model's decision depends on it.

Parameters:

Name	Type	Description	Default
model	Module	Any registered model, already on the right device	required
signals	Tensor	Batch of shape (batch, leads, time)	required
target	TargetSpec	None for the model's own prediction, an integer, or one index per sample	None
absolute	bool	Return gradient magnitude rather than signed gradient	True

Returns:

Type	Description
Tensor	Attribution of shape (batch, leads, time), matching the input.

Source code in deepecgkit/explainability/attribution.py

def saliency(
    model: nn.Module,
    signals: torch.Tensor,
    target: TargetSpec = None,
    absolute: bool = True,
) -> torch.Tensor:
    """Attribute by the gradient of the target logit with respect to the input.

    The cheapest of the three methods, at one forward and one backward pass, and
    the noisiest: the gradient measures sensitivity at a single point, so it
    reacts to sharp local structure such as the QRS complex whether or not the
    model's decision depends on it.

    Args:
        model: Any registered model, already on the right device
        signals: Batch of shape (batch, leads, time)
        target: None for the model's own prediction, an integer, or one index
            per sample
        absolute: Return gradient magnitude rather than signed gradient

    Returns:
        Attribution of shape (batch, leads, time), matching the input.
    """
    inputs = signals.clone().detach().requires_grad_(True)
    with torch.enable_grad():
        logits = _forward(model, inputs)
        index = _resolve_target(logits, target)
        gradients = torch.autograd.grad(_target_scores(logits, index).sum(), inputs)[0]

    return gradients.abs() if absolute else gradients

integrated_gradients

integrated_gradients(
    model: Module,
    signals: Tensor,
    target: TargetSpec = None,
    steps: int = 64,
    baseline: Tensor | None = None,
) -> torch.Tensor

Attribute by integrating gradients along a path from a baseline.

Gradients are sampled at step midpoints, which makes the Riemann sum symmetric and lets the result satisfy the completeness axiom closely: the attribution sums to the difference between the model's score on the signal and on the baseline. Attribution is signed, because that identity is the method's value and taking magnitudes would destroy it.

The default baseline of zeros is the sensible one for z-scored ECG, where zero is the per-lead mean, and a poor one for unnormalised millivolts.

Parameters:

Name	Type	Description	Default
model	Module	Any registered model, already on the right device	required
signals	Tensor	Batch of shape (batch, leads, time)	required
target	TargetSpec	None for the model's own prediction, an integer, or one index per sample	None
steps	int	Number of points along the path	64
baseline	Tensor | None	Reference input of the same shape, defaulting to zeros	None

Returns:

Type	Description
Tensor	Signed attribution of shape (batch, leads, time), matching the input.

Raises:

Type	Description
ValueError	If steps is not positive, or the baseline shape differs.

Source code in deepecgkit/explainability/attribution.py

def integrated_gradients(
    model: nn.Module,
    signals: torch.Tensor,
    target: TargetSpec = None,
    steps: int = 64,
    baseline: torch.Tensor | None = None,
) -> torch.Tensor:
    """Attribute by integrating gradients along a path from a baseline.

    Gradients are sampled at step midpoints, which makes the Riemann sum
    symmetric and lets the result satisfy the completeness axiom closely: the
    attribution sums to the difference between the model's score on the signal
    and on the baseline. Attribution is signed, because that identity is the
    method's value and taking magnitudes would destroy it.

    The default baseline of zeros is the sensible one for z-scored ECG, where
    zero is the per-lead mean, and a poor one for unnormalised millivolts.

    Args:
        model: Any registered model, already on the right device
        signals: Batch of shape (batch, leads, time)
        target: None for the model's own prediction, an integer, or one index
            per sample
        steps: Number of points along the path
        baseline: Reference input of the same shape, defaulting to zeros

    Returns:
        Signed attribution of shape (batch, leads, time), matching the input.

    Raises:
        ValueError: If steps is not positive, or the baseline shape differs.
    """
    if steps < 1:
        raise ValueError(f"steps must be at least 1, got {steps}.")

    reference = torch.zeros_like(signals) if baseline is None else baseline.to(signals.device)
    if reference.shape != signals.shape:
        raise ValueError(
            f"baseline of shape {tuple(reference.shape)} does not match the signals' shape "
            f"{tuple(signals.shape)}."
        )

    with torch.no_grad():
        index = _resolve_target(_forward(model, signals), target)

    difference = signals - reference
    total = torch.zeros_like(signals)
    for step in range(steps):
        alpha = (step + 0.5) / steps
        point = (reference + alpha * difference).detach().requires_grad_(True)
        with torch.enable_grad():
            scores = _target_scores(_forward(model, point), index)
            total = total + torch.autograd.grad(scores.sum(), point)[0]

    return difference * total / steps

occlusion

occlusion(
    model: Module,
    signals: Tensor,
    target: TargetSpec = None,
    window: int = 100,
    stride: int = 50,
    baseline: float = 0.0,
) -> torch.Tensor

Attribute by how much the score drops when a span is blanked out.

One lead-and-span at a time is replaced by the baseline and the drop in the target logit is spread over that span, averaging where windows overlap. The result is signed: a positive value means blanking the span lowered the score, so the span supported the prediction.

Attribution bleeds one window width past any sharp boundary in the model's receptive field, because a window straddling the boundary perturbs samples on both sides and the single resulting score change is smeared across all of it. Narrower windows sharpen the result and cost proportionally more forward passes.

Parameters:

Name	Type	Description	Default
model	Module	Any registered model, already on the right device	required
signals	Tensor	Batch of shape (batch, leads, time)	required
target	TargetSpec	None for the model's own prediction, an integer, or one index per sample	None
window	int	Width in samples of the occluded span	100
stride	int	Step in samples between successive spans	50
baseline	float	Value written into the occluded span	0.0

Returns:

Type	Description
Tensor	Signed attribution of shape (batch, leads, time), matching the input.

Raises:

Type	Description
ValueError	If window or stride is not positive.

Source code in deepecgkit/explainability/attribution.py

def occlusion(
    model: nn.Module,
    signals: torch.Tensor,
    target: TargetSpec = None,
    window: int = 100,
    stride: int = 50,
    baseline: float = 0.0,
) -> torch.Tensor:
    """Attribute by how much the score drops when a span is blanked out.

    One lead-and-span at a time is replaced by the baseline and the drop in the
    target logit is spread over that span, averaging where windows overlap. The
    result is signed: a positive value means blanking the span lowered the
    score, so the span supported the prediction.

    Attribution bleeds one window width past any sharp boundary in the model's
    receptive field, because a window straddling the boundary perturbs samples
    on both sides and the single resulting score change is smeared across all of
    it. Narrower windows sharpen the result and cost proportionally more
    forward passes.

    Args:
        model: Any registered model, already on the right device
        signals: Batch of shape (batch, leads, time)
        target: None for the model's own prediction, an integer, or one index
            per sample
        window: Width in samples of the occluded span
        stride: Step in samples between successive spans
        baseline: Value written into the occluded span

    Returns:
        Signed attribution of shape (batch, leads, time), matching the input.

    Raises:
        ValueError: If window or stride is not positive.
    """
    if window < 1:
        raise ValueError(f"window must be at least 1 sample, got {window}.")
    if stride < 1:
        raise ValueError(f"stride must be at least 1 sample, got {stride}.")

    _, leads, length = signals.shape
    span = min(window, length)

    starts = list(range(0, max(length - span, 0) + 1, stride))
    if starts[-1] + span < length:
        starts.append(length - span)

    total = torch.zeros_like(signals)
    counts = torch.zeros_like(signals)

    with torch.no_grad():
        logits = _forward(model, signals)
        index = _resolve_target(logits, target)
        reference = _target_scores(logits, index)

        for lead in range(leads):
            for start in starts:
                stop = start + span
                perturbed = signals.clone()
                perturbed[:, lead, start:stop] = baseline
                scores = _target_scores(_forward(model, perturbed), index)
                total[:, lead, start:stop] += (reference - scores).view(-1, 1)
                counts[:, lead, start:stop] += 1.0

    return total / counts.clamp(min=1.0)

lead_importance

lead_importance(values: ndarray) -> np.ndarray

Reduce per-timestep attribution to a distribution over leads.

Each lead is scored by its mean absolute attribution, then the scores are normalised to sum to one. Magnitude rather than sign, because the question is how much a lead mattered rather than which way it pushed; mean rather than sum, so the scores stay comparable if leads ever carry different lengths.

A single-lead model yields [1.0]. That is correct rather than degenerate: with one lead there is nowhere else for the evidence to be.

Parameters:

Name	Type	Description	Default
values	ndarray	Attribution of shape (batch, leads, time)	required

Returns:

Type	Description
ndarray	Array of shape (batch, leads) whose rows sum to one. A row whose
ndarray	attribution is entirely zero becomes uniform, since a model that reacted
ndarray	to nothing gives no reason to prefer any lead.

Raises:

Type	Description
ValueError	If the attribution is not three-dimensional.

Source code in deepecgkit/explainability/attribution.py

def lead_importance(values: np.ndarray) -> np.ndarray:
    """Reduce per-timestep attribution to a distribution over leads.

    Each lead is scored by its mean absolute attribution, then the scores are
    normalised to sum to one. Magnitude rather than sign, because the question
    is how much a lead mattered rather than which way it pushed; mean rather
    than sum, so the scores stay comparable if leads ever carry different
    lengths.

    A single-lead model yields ``[1.0]``. That is correct rather than
    degenerate: with one lead there is nowhere else for the evidence to be.

    Args:
        values: Attribution of shape (batch, leads, time)

    Returns:
        Array of shape (batch, leads) whose rows sum to one. A row whose
        attribution is entirely zero becomes uniform, since a model that reacted
        to nothing gives no reason to prefer any lead.

    Raises:
        ValueError: If the attribution is not three-dimensional.
    """
    attribution = np.asarray(values, dtype=np.float64)
    if attribution.ndim != 3:
        raise ValueError(
            f"lead_importance expects attribution of shape (batch, leads, time), got shape "
            f"{attribution.shape}."
        )

    magnitude = np.abs(attribution).mean(axis=-1)
    total = magnitude.sum(axis=-1, keepdims=True)
    uniform = np.full_like(magnitude, 1.0 / magnitude.shape[-1])
    return np.where(total > 0.0, magnitude / np.where(total > 0.0, total, 1.0), uniform).astype(
        np.float32
    )

Beat alignment

detect_r_peaks

detect_r_peaks(
    signal: ndarray,
    sampling_rate: float = 500.0,
    bandpass: tuple[float, float] = _QRS_BANDPASS_HZ,
    min_rr_seconds: float = _MINIMUM_RR_SECONDS,
    threshold_factor: float = _THRESHOLD_FACTOR,
    integration_seconds: float = _INTEGRATION_SECONDS,
) -> np.ndarray

Locate R-peaks in a single-lead signal.

The pipeline is the classical Pan-Tompkins shape: band-pass to the QRS band, differentiate, square, integrate over a moving window, then take peaks above a threshold with a refractory distance between them. On synthetic ECG with known R positions this recovers every peak to the exact sample.

The band is a parameter rather than a constant because it encodes an assumption about where QRS energy lives. Real ECG puts it near 5 to 15 Hz, but synthetic test signals often do not, and a detector that silently returns the wrong beats is worse than one that is asked to be told.

Parameters:

Name	Type	Description	Default
signal	ndarray	One-dimensional signal of shape (time,)	required
sampling_rate	float	Samples per second	500.0
bandpass	tuple[float, float]	Low and high cutoff in Hz for the QRS band	_QRS_BANDPASS_HZ
min_rr_seconds	float	Refractory period, the closest two peaks may be	_MINIMUM_RR_SECONDS
threshold_factor	float	Peaks must exceed mean plus this many standard deviations of the integrated energy	_THRESHOLD_FACTOR
integration_seconds	float	Width of the moving-average window	_INTEGRATION_SECONDS

Returns:

Type	Description
ndarray	Sorted sample indices of the detected R-peaks, empty when the signal is
ndarray	too short or carries no energy.

Raises:

Type	Description
ValueError	If the signal is not one-dimensional, or the band is not a rising pair below the Nyquist frequency.

Source code in deepecgkit/explainability/beats.py

def detect_r_peaks(
    signal: np.ndarray,
    sampling_rate: float = 500.0,
    bandpass: tuple[float, float] = _QRS_BANDPASS_HZ,
    min_rr_seconds: float = _MINIMUM_RR_SECONDS,
    threshold_factor: float = _THRESHOLD_FACTOR,
    integration_seconds: float = _INTEGRATION_SECONDS,
) -> np.ndarray:
    """Locate R-peaks in a single-lead signal.

    The pipeline is the classical Pan-Tompkins shape: band-pass to the QRS
    band, differentiate, square, integrate over a moving window, then take
    peaks above a threshold with a refractory distance between them. On
    synthetic ECG with known R positions this recovers every peak to the exact
    sample.

    The band is a parameter rather than a constant because it encodes an
    assumption about where QRS energy lives. Real ECG puts it near 5 to 15 Hz,
    but synthetic test signals often do not, and a detector that silently
    returns the wrong beats is worse than one that is asked to be told.

    Args:
        signal: One-dimensional signal of shape (time,)
        sampling_rate: Samples per second
        bandpass: Low and high cutoff in Hz for the QRS band
        min_rr_seconds: Refractory period, the closest two peaks may be
        threshold_factor: Peaks must exceed mean plus this many standard
            deviations of the integrated energy
        integration_seconds: Width of the moving-average window

    Returns:
        Sorted sample indices of the detected R-peaks, empty when the signal is
        too short or carries no energy.

    Raises:
        ValueError: If the signal is not one-dimensional, or the band is not a
            rising pair below the Nyquist frequency.
    """
    values = np.asarray(signal, dtype=np.float64)
    if values.ndim != 1:
        raise ValueError(
            f"detect_r_peaks expects a one-dimensional signal, got shape {values.shape}."
        )

    nyquist = 0.5 * sampling_rate
    low, high = bandpass
    if not 0.0 < low < high:
        raise ValueError(f"bandpass must be a rising pair of positive cutoffs, got {bandpass}.")
    if low >= nyquist:
        raise ValueError(
            f"bandpass low cutoff {low} Hz is at or above the Nyquist frequency {nyquist} Hz."
        )

    window = max(1, int(integration_seconds * sampling_rate))
    if values.size <= 3 * window:
        return np.empty(0, dtype=np.int64)

    numerator, denominator = scipy_signal.butter(
        2, [low / nyquist, min(high / nyquist, 0.99)], btype="band"
    )
    filtered = scipy_signal.filtfilt(numerator, denominator, values)
    energy = np.gradient(filtered) ** 2
    integrated = np.convolve(energy, np.ones(window) / window, mode="same")

    spread = float(integrated.std())
    if spread <= 0.0:
        return np.empty(0, dtype=np.int64)

    peaks, _ = scipy_signal.find_peaks(
        integrated,
        height=float(integrated.mean()) + threshold_factor * spread,
        distance=max(1, int(min_rr_seconds * sampling_rate)),
    )
    return peaks.astype(np.int64)

beat_aligned_attribution

beat_aligned_attribution(
    values: ndarray,
    r_peaks: ndarray,
    sampling_rate: float = 500.0,
    window_ms: tuple[int, int] = BEAT_WINDOW_MS,
) -> tuple[np.ndarray, np.ndarray, int]

Average attribution over R-peak-aligned windows.

Beats whose window would run past either end of the recording are dropped rather than padded or clipped, and the surviving count is returned so a caller can report it. A truncated beat averaged in alongside whole ones would pull the profile toward the edges of the recording, which is an artefact of where the recording happened to start.

Parameters:

Name	Type	Description	Default
values	ndarray	Attribution of shape (leads, time)	required
r_peaks	ndarray	Sample indices of the R-peaks	required
sampling_rate	float	Samples per second	500.0
window_ms	tuple[int, int]	Start and end of the beat window in milliseconds relative to the R-peak, negative before it	BEAT_WINDOW_MS

Returns:

Type	Description
ndarray	Tuple of the mean profile of shape (leads, window), the window's time
ndarray	axis in milliseconds, and the number of beats averaged. When no beat
int	fits, the profile is all zeros and the count is zero.

Raises:

Type	Description
ValueError	If the attribution is not two-dimensional, or the window does not start before it ends.

Source code in deepecgkit/explainability/beats.py

def beat_aligned_attribution(
    values: np.ndarray,
    r_peaks: np.ndarray,
    sampling_rate: float = 500.0,
    window_ms: tuple[int, int] = BEAT_WINDOW_MS,
) -> tuple[np.ndarray, np.ndarray, int]:
    """Average attribution over R-peak-aligned windows.

    Beats whose window would run past either end of the recording are dropped
    rather than padded or clipped, and the surviving count is returned so a
    caller can report it. A truncated beat averaged in alongside whole ones
    would pull the profile toward the edges of the recording, which is an
    artefact of where the recording happened to start.

    Args:
        values: Attribution of shape (leads, time)
        r_peaks: Sample indices of the R-peaks
        sampling_rate: Samples per second
        window_ms: Start and end of the beat window in milliseconds relative to
            the R-peak, negative before it

    Returns:
        Tuple of the mean profile of shape (leads, window), the window's time
        axis in milliseconds, and the number of beats averaged. When no beat
        fits, the profile is all zeros and the count is zero.

    Raises:
        ValueError: If the attribution is not two-dimensional, or the window
            does not start before it ends.
    """
    attribution = np.asarray(values, dtype=np.float64)
    if attribution.ndim != 2:
        raise ValueError(
            f"beat_aligned_attribution expects attribution of shape (leads, time), got "
            f"shape {attribution.shape}."
        )

    start_ms, end_ms = window_ms
    if start_ms >= end_ms:
        raise ValueError(f"window_ms must start before it ends, got {window_ms}.")

    rate = float(sampling_rate)
    pre = round(abs(start_ms) / 1000.0 * rate)
    post = round(end_ms / 1000.0 * rate)
    span = pre + post
    times_ms = (np.arange(span) - pre) / rate * 1000.0

    leads, length = attribution.shape
    windows = [
        attribution[:, int(peak) - pre : int(peak) + post]
        for peak in r_peaks
        if int(peak) - pre >= 0 and int(peak) + post <= length
    ]

    if not windows:
        return np.zeros((leads, span), dtype=np.float32), times_ms, 0

    profile = np.mean(np.stack(windows, axis=0), axis=0)
    return profile.astype(np.float32), times_ms, len(windows)

pqrst_fractions

pqrst_fractions(
    profile: ndarray,
    times_ms: ndarray,
    windows_ms: dict[str, tuple[int, int]] | None = None,
) -> dict[str, float]

Split a beat profile into its P, QRS and T shares.

Shares are taken against the total mass of the profile, not against the mass inside the three windows, so they do not sum to one. The remainder is the baseline between the named regions, and hiding it by renormalising would make a model that attends mostly to the isoelectric line look like one that attends to the P wave.

Parameters:

Name	Type	Description	Default
profile	ndarray	Mean beat profile of shape (leads, window) or (window,)	required
times_ms	ndarray	Time axis of the window in milliseconds relative to the R-peak	required
windows_ms	dict[str, tuple[int, int]] | None	Region boundaries in milliseconds, defaulting to PQRST_WINDOWS_MS	None

Returns:

Type	Description
dict[str, float]	Fraction of total profile mass falling in each named region, plus a
dict[str, float]	"baseline" entry holding the remainder. All zeros when the profile
dict[str, float]	carries no mass.

Raises:

Type	Description
ValueError	If the time axis does not match the profile's last axis.

Source code in deepecgkit/explainability/beats.py

def pqrst_fractions(
    profile: np.ndarray,
    times_ms: np.ndarray,
    windows_ms: dict[str, tuple[int, int]] | None = None,
) -> dict[str, float]:
    """Split a beat profile into its P, QRS and T shares.

    Shares are taken against the total mass of the profile, not against the
    mass inside the three windows, so they do not sum to one. The remainder is
    the baseline between the named regions, and hiding it by renormalising
    would make a model that attends mostly to the isoelectric line look like one
    that attends to the P wave.

    Args:
        profile: Mean beat profile of shape (leads, window) or (window,)
        times_ms: Time axis of the window in milliseconds relative to the R-peak
        windows_ms: Region boundaries in milliseconds, defaulting to
            PQRST_WINDOWS_MS

    Returns:
        Fraction of total profile mass falling in each named region, plus a
        "baseline" entry holding the remainder. All zeros when the profile
        carries no mass.

    Raises:
        ValueError: If the time axis does not match the profile's last axis.
    """
    values = np.abs(np.asarray(profile, dtype=np.float64))
    if values.ndim == 2:
        values = values.mean(axis=0)

    axis = np.asarray(times_ms, dtype=np.float64)
    if axis.shape != values.shape:
        raise ValueError(
            f"times_ms of shape {axis.shape} does not match the profile window of shape "
            f"{values.shape}."
        )

    regions = windows_ms if windows_ms is not None else PQRST_WINDOWS_MS
    total = float(values.sum())
    if total <= 0.0:
        fractions = {name: 0.0 for name in regions}
        fractions["baseline"] = 0.0
        return fractions

    fractions = {}
    for name, (start_ms, end_ms) in regions.items():
        selected = (axis >= start_ms) & (axis < end_ms)
        fractions[name] = float(values[selected].sum() / total)

    fractions["baseline"] = float(max(0.0, 1.0 - sum(fractions.values())))
    return fractions

pqrst_enrichment

pqrst_enrichment(
    fractions: dict[str, float],
    times_ms: ndarray,
    windows_ms: dict[str, tuple[int, int]] | None = None,
) -> dict[str, float]

Compare each region's attribution share against its share of the beat.

A value of one means the region received exactly the attention its width would earn by chance, above one means the model concentrates there, and below one means it avoids the region. This is the number the P-wave claim turns on: a P share of 0.09 sounds small until it is read against the P window's 0.25 share of the beat, at which point it says the model attends to the P-wave region at roughly a third of chance.

Parameters:

Name	Type	Description	Default
fractions	dict[str, float]	Output of pqrst_fractions	required
times_ms	ndarray	Time axis of the beat window in milliseconds	required
windows_ms	dict[str, tuple[int, int]] | None	Region boundaries in milliseconds, defaulting to PQRST_WINDOWS_MS	None

Returns:

Type	Description
dict[str, float]	Ratio of attribution share to duration share for each region. A region
dict[str, float]	occupying no samples yields 0.0 rather than an infinity.

Source code in deepecgkit/explainability/beats.py

def pqrst_enrichment(
    fractions: dict[str, float],
    times_ms: np.ndarray,
    windows_ms: dict[str, tuple[int, int]] | None = None,
) -> dict[str, float]:
    """Compare each region's attribution share against its share of the beat.

    A value of one means the region received exactly the attention its width
    would earn by chance, above one means the model concentrates there, and
    below one means it avoids the region. This is the number the P-wave claim
    turns on: a P share of 0.09 sounds small until it is read against the
    P window's 0.25 share of the beat, at which point it says the model attends
    to the P-wave region at roughly a third of chance.

    Args:
        fractions: Output of pqrst_fractions
        times_ms: Time axis of the beat window in milliseconds
        windows_ms: Region boundaries in milliseconds, defaulting to
            PQRST_WINDOWS_MS

    Returns:
        Ratio of attribution share to duration share for each region. A region
        occupying no samples yields 0.0 rather than an infinity.
    """
    shares = region_duration_fractions(times_ms, windows_ms)
    return {
        name: float(fractions.get(name, 0.0) / share) if share > 0.0 else 0.0
        for name, share in shares.items()
    }

region_duration_fractions

region_duration_fractions(
    times_ms: ndarray,
    windows_ms: dict[str, tuple[int, int]] | None = None,
) -> dict[str, float]

Return each region's share of the beat window's duration.

This is the profile a model spreading its attention evenly over the cardiac cycle would produce, and it is what makes a raw share interpretable. The P window is 200 ms of an 800 ms beat, so 25 per cent of the attribution falling in it is not evidence of anything.

Parameters:

Name	Type	Description	Default
times_ms	ndarray	Time axis of the beat window in milliseconds relative to the R-peak	required
windows_ms	dict[str, tuple[int, int]] | None	Region boundaries in milliseconds, defaulting to PQRST_WINDOWS_MS	None

Returns:

Type	Description
dict[str, float]	Fraction of the window's samples falling in each named region, plus a
dict[str, float]	"baseline" entry holding the remainder.

Source code in deepecgkit/explainability/beats.py

def region_duration_fractions(
    times_ms: np.ndarray,
    windows_ms: dict[str, tuple[int, int]] | None = None,
) -> dict[str, float]:
    """Return each region's share of the beat window's duration.

    This is the profile a model spreading its attention evenly over the cardiac
    cycle would produce, and it is what makes a raw share interpretable. The
    P window is 200 ms of an 800 ms beat, so 25 per cent of the attribution
    falling in it is not evidence of anything.

    Args:
        times_ms: Time axis of the beat window in milliseconds relative to the
            R-peak
        windows_ms: Region boundaries in milliseconds, defaulting to
            PQRST_WINDOWS_MS

    Returns:
        Fraction of the window's samples falling in each named region, plus a
        "baseline" entry holding the remainder.
    """
    axis = np.asarray(times_ms, dtype=np.float64)
    regions = windows_ms if windows_ms is not None else PQRST_WINDOWS_MS
    total = float(axis.size)

    shares = {}
    for name, (start_ms, end_ms) in regions.items():
        selected = (axis >= start_ms) & (axis < end_ms)
        shares[name] = float(selected.sum() / total) if total else 0.0

    shares["baseline"] = float(max(0.0, 1.0 - sum(shares.values())))
    return shares

Model constraints

length_bounds

length_bounds(
    model: Module, model_name: str | None = None
) -> tuple[int, int | None]

Return the smallest and largest signal length a model accepts.

A live min_input_length attribute wins over the table, because AFModel's floor is derived from its recording_length and is 7971 at 30 seconds but 1393 at 6 or 10. The table cannot know which was built.

Parameters:

Name	Type	Description	Default
model	Module	The model the attribution will run against	required
model_name	str | None	Registry name, inferred from the model when omitted	None

Returns:

Type	Description
int	Tuple of the minimum length and the maximum length, where the maximum is
int | None	None for the seventeen architectures that have no ceiling.

Source code in deepecgkit/explainability/constraints.py

def length_bounds(model: nn.Module, model_name: str | None = None) -> tuple[int, int | None]:
    """Return the smallest and largest signal length a model accepts.

    A live ``min_input_length`` attribute wins over the table, because AFModel's
    floor is derived from its ``recording_length`` and is 7971 at 30 seconds but
    1393 at 6 or 10. The table cannot know which was built.

    Args:
        model: The model the attribution will run against
        model_name: Registry name, inferred from the model when omitted

    Returns:
        Tuple of the minimum length and the maximum length, where the maximum is
        None for the seventeen architectures that have no ceiling.
    """
    resolved = model_name or resolve_model_name(model)
    info = get_model_info(resolved) if resolved and resolved in get_model_names() else {}

    declared = getattr(model, "min_input_length", None)
    if isinstance(declared, int):
        minimum = declared
    else:
        minimum = info.get("min_input_length") or 1

    return minimum, info.get("max_input_length")

fit_to_constraints

fit_to_constraints(
    signals: Tensor,
    model: Module,
    model_name: str | None = None,
) -> tuple[torch.Tensor, int]

Centre-crop a batch so the model can consume it, and report the offset.

Cropping is centred to match ECGStandardizer(clip_method="center"), so a recording trimmed here covers the same span it would have during training. The returned offset lets a plot label the time axis against the original recording rather than against the crop.

A signal that is too short cannot be fixed by cropping, so that case raises instead of padding: padding would feed the model fabricated samples and the resulting attribution would describe the padding as much as the ECG.

Parameters:

Name	Type	Description	Default
signals	Tensor	Batch of shape (batch, leads, time)	required
model	Module	The model the attribution will run against	required
model_name	str | None	Registry name, inferred from the model when omitted	None

Returns:

Type	Description
Tensor	Tuple of the possibly cropped batch and the index in the original signal
int	where the returned batch begins.

Raises:

Type	Description
ValueError	If the batch is not three-dimensional, if its lead count does not match the model, or if it is shorter than the model's minimum.

Source code in deepecgkit/explainability/constraints.py

def fit_to_constraints(
    signals: torch.Tensor, model: nn.Module, model_name: str | None = None
) -> tuple[torch.Tensor, int]:
    """Centre-crop a batch so the model can consume it, and report the offset.

    Cropping is centred to match ``ECGStandardizer(clip_method="center")``, so a
    recording trimmed here covers the same span it would have during training.
    The returned offset lets a plot label the time axis against the original
    recording rather than against the crop.

    A signal that is too short cannot be fixed by cropping, so that case raises
    instead of padding: padding would feed the model fabricated samples and the
    resulting attribution would describe the padding as much as the ECG.

    Args:
        signals: Batch of shape (batch, leads, time)
        model: The model the attribution will run against
        model_name: Registry name, inferred from the model when omitted

    Returns:
        Tuple of the possibly cropped batch and the index in the original signal
        where the returned batch begins.

    Raises:
        ValueError: If the batch is not three-dimensional, if its lead count
            does not match the model, or if it is shorter than the model's
            minimum.
    """
    if signals.dim() != 3:
        raise ValueError(
            f"Attribution expects input of shape (batch, leads, time), got a "
            f"{signals.dim()}-D tensor of shape {tuple(signals.shape)}."
        )

    resolved = model_name or resolve_model_name(model)
    label = resolved or type(model).__name__
    minimum, maximum = length_bounds(model, resolved)

    expected_leads = getattr(model, "input_channels", None)
    if isinstance(expected_leads, int) and signals.shape[1] != expected_leads:
        raise ValueError(
            f"{label} was built for {expected_leads} leads but received "
            f"{signals.shape[1]}. Rebuild the model with "
            f"input_channels={signals.shape[1]} or pass matching data."
        )

    length = signals.shape[-1]
    if length < minimum:
        raise ValueError(
            f"{label} needs at least {minimum} samples per lead but received {length}. "
            f"Pass a longer recording, or resample to a rate that reaches that length."
        )

    if maximum is None or length <= maximum:
        return signals, 0

    offset = (length - maximum) // 2
    warnings.warn(
        f"{label} accepts at most {maximum} samples per lead; centre-cropping {length} "
        f"to {maximum} (offset {offset}). Attribution describes only the cropped span.",
        RuntimeWarning,
        stacklevel=3,
    )
    return signals[..., offset : offset + maximum], offset

Visualisation

plot_attribution

plot_attribution(
    signal: ndarray,
    values: ndarray,
    sampling_rate: float = 500.0,
    leads: list[str] | None = None,
    time_offset: float = 0.0,
    title: str = "ECG Attribution",
    save_path: str | None = None,
    unit: str | None = None,
) -> None

Plot each lead with its attribution shaded behind the trace.

Attribution magnitude is drawn as a background band on one colour scale shared by every lead, so a lead the model ignored looks empty next to one it relied on. Scaling each lead independently would render those two identically, which is the misreading the lead-importance plot exists to prevent.

Parameters:

Name	Type	Description	Default
signal	ndarray	ECG of shape (leads, time)	required
values	ndarray	Attribution of the same shape	required
sampling_rate	float	Samples per second	500.0
leads	list[str] | None	Display names, one per lead	None
time_offset	float	Seconds into the original recording where this window starts, so a centre-cropped attribution still reads correctly	0.0
title	str	Figure title	'ECG Attribution'
save_path	str | None	Where to write the figure, or None to show it	None
unit	str | None	Unit of the amplitude axis, such as "mV" for a raw recording. The signal a model saw is normalised and has none, so none is shown by default.	None

Raises:

Type	Description
ValueError	If the signal is not two-dimensional or the shapes differ.

Source code in deepecgkit/explainability/visualization.py

def plot_attribution(
    signal: np.ndarray,
    values: np.ndarray,
    sampling_rate: float = 500.0,
    leads: list[str] | None = None,
    time_offset: float = 0.0,
    title: str = "ECG Attribution",
    save_path: str | None = None,
    unit: str | None = None,
) -> None:
    """Plot each lead with its attribution shaded behind the trace.

    Attribution magnitude is drawn as a background band on one colour scale
    shared by every lead, so a lead the model ignored looks empty next to one it
    relied on. Scaling each lead independently would render those two
    identically, which is the misreading the lead-importance plot exists to
    prevent.

    Args:
        signal: ECG of shape (leads, time)
        values: Attribution of the same shape
        sampling_rate: Samples per second
        leads: Display names, one per lead
        time_offset: Seconds into the original recording where this window
            starts, so a centre-cropped attribution still reads correctly
        title: Figure title
        save_path: Where to write the figure, or None to show it
        unit: Unit of the amplitude axis, such as "mV" for a raw recording.
            The signal a model saw is normalised and has none, so none is
            shown by default.

    Raises:
        ValueError: If the signal is not two-dimensional or the shapes differ.
    """
    trace = np.asarray(signal, dtype=np.float64)
    attribution = np.abs(np.asarray(values, dtype=np.float64))

    if trace.ndim != 2:
        raise ValueError(
            f"plot_attribution expects a signal of shape (leads, time), got shape {trace.shape}."
        )
    if trace.shape != attribution.shape:
        raise ValueError(
            f"Signal of shape {trace.shape} and attribution of shape {attribution.shape} "
            f"must match. Both are (leads, time)."
        )

    n_leads, n_samples = trace.shape
    names = leads or [f"Lead {i + 1}" for i in range(n_leads)]
    time = time_offset + np.arange(n_samples) / sampling_rate
    ceiling = float(attribution.max())

    _fig, axes = plt.subplots(n_leads, 1, figsize=(12, 2 * n_leads), sharex=True)
    if n_leads == 1:
        axes = [axes]

    mesh = None
    for i, ax in enumerate(axes):
        mesh = ax.imshow(
            attribution[i][np.newaxis, :],
            aspect="auto",
            cmap=_ATTRIBUTION_CMAP,
            vmin=0.0,
            vmax=ceiling if ceiling > 0.0 else 1.0,
            extent=(time[0], time[-1], float(trace[i].min()), float(trace[i].max())),
        )
        ax.plot(time, trace[i], color="black", linewidth=0.8)
        ax.set_ylabel(f"{names[i]} ({unit})" if unit else names[i])
        ax.grid(True, alpha=0.3)

    axes[-1].set_xlabel("Time (s)")
    plt.suptitle(title)
    plt.tight_layout()
    if mesh is not None:
        plt.colorbar(mesh, ax=axes, label="|attribution|", fraction=0.02, pad=0.01)

    finalize_figure(save_path)

plot_lead_importance

plot_lead_importance(
    importance: ndarray,
    leads: list[str] | None = None,
    title: str = "Lead Importance",
    save_path: str | None = None,
) -> None

Plot the share of attribution each lead carries.

Parameters:

Name	Type	Description	Default
importance	ndarray	Distribution of shape (leads,), or (batch, leads) to be averaged over the batch	required
leads	list[str] | None	Display names, one per lead	None
title	str	Figure title	'Lead Importance'
save_path	str | None	Where to write the figure, or None to show it	None

Raises:

Type	Description
ValueError	If the distribution has more than two dimensions.

Source code in deepecgkit/explainability/visualization.py

def plot_lead_importance(
    importance: np.ndarray,
    leads: list[str] | None = None,
    title: str = "Lead Importance",
    save_path: str | None = None,
) -> None:
    """Plot the share of attribution each lead carries.

    Args:
        importance: Distribution of shape (leads,), or (batch, leads) to be
            averaged over the batch
        leads: Display names, one per lead
        title: Figure title
        save_path: Where to write the figure, or None to show it

    Raises:
        ValueError: If the distribution has more than two dimensions.
    """
    shares = np.asarray(importance, dtype=np.float64)
    if shares.ndim == 2:
        shares = shares.mean(axis=0)
    if shares.ndim != 1:
        raise ValueError(
            f"plot_lead_importance expects shape (leads,) or (batch, leads), got shape "
            f"{np.asarray(importance).shape}."
        )

    names = leads or [f"Lead {i + 1}" for i in range(shares.size)]
    positions = np.arange(shares.size)

    plt.figure(figsize=(8, max(4, 0.4 * shares.size)))
    plt.barh(positions, shares, color="tab:blue", alpha=0.8)
    plt.yticks(positions, names)
    plt.gca().invert_yaxis()
    plt.xlabel("Share of total attribution")
    plt.title(title)
    plt.grid(True, alpha=0.3, axis="x")
    plt.tight_layout()

    finalize_figure(save_path)

plot_beat_profile

plot_beat_profile(
    profile: ndarray,
    times_ms: ndarray,
    leads: list[str] | None = None,
    windows_ms: dict[str, tuple[int, int]] | None = None,
    title: str = "Beat-Aligned Attribution",
    save_path: str | None = None,
) -> None

Plot mean attribution across the cardiac cycle with the regions shaded.

The P, QRS and T windows are drawn behind the curve so the reported shares can be checked against the shape they came from, rather than being taken on trust from a table.

Parameters:

Name	Type	Description	Default
profile	ndarray	Mean beat profile of shape (leads, window) or (window,)	required
times_ms	ndarray	Time axis of the window in milliseconds relative to the R-peak	required
leads	list[str] | None	Display names, one per lead, used for the per-lead traces	None
windows_ms	dict[str, tuple[int, int]] | None	Region boundaries in milliseconds, defaulting to PQRST_WINDOWS_MS	None
title	str	Figure title	'Beat-Aligned Attribution'
save_path	str | None	Where to write the figure, or None to show it	None

Raises:

Type	Description
ValueError	If the time axis does not match the profile window.

Source code in deepecgkit/explainability/visualization.py

def plot_beat_profile(
    profile: np.ndarray,
    times_ms: np.ndarray,
    leads: list[str] | None = None,
    windows_ms: dict[str, tuple[int, int]] | None = None,
    title: str = "Beat-Aligned Attribution",
    save_path: str | None = None,
) -> None:
    """Plot mean attribution across the cardiac cycle with the regions shaded.

    The P, QRS and T windows are drawn behind the curve so the reported shares
    can be checked against the shape they came from, rather than being taken on
    trust from a table.

    Args:
        profile: Mean beat profile of shape (leads, window) or (window,)
        times_ms: Time axis of the window in milliseconds relative to the R-peak
        leads: Display names, one per lead, used for the per-lead traces
        windows_ms: Region boundaries in milliseconds, defaulting to
            PQRST_WINDOWS_MS
        title: Figure title
        save_path: Where to write the figure, or None to show it

    Raises:
        ValueError: If the time axis does not match the profile window.
    """
    values = np.asarray(profile, dtype=np.float64)
    axis = np.asarray(times_ms, dtype=np.float64)
    if values.ndim == 1:
        values = values[np.newaxis, :]
    if values.ndim != 2 or values.shape[1] != axis.size:
        raise ValueError(
            f"Profile of shape {np.asarray(profile).shape} does not match a time axis of "
            f"{axis.size} samples."
        )

    names = leads or [f"Lead {i + 1}" for i in range(values.shape[0])]
    regions = windows_ms if windows_ms is not None else PQRST_WINDOWS_MS

    plt.figure(figsize=(10, 5))
    for name, (start_ms, end_ms) in regions.items():
        plt.axvspan(
            start_ms,
            end_ms,
            color=_REGION_COLOURS.get(name, "tab:grey"),
            alpha=0.15,
            label=name,
        )

    for i in range(values.shape[0]):
        plt.plot(axis, np.abs(values[i]), linewidth=0.8, alpha=0.5, label=names[i])
    plt.plot(axis, np.abs(values).mean(axis=0), color="black", linewidth=2.0, label="Mean")

    plt.axvline(0.0, color="black", linestyle="--", linewidth=1.0)
    plt.xlabel("Time relative to R-peak (ms)")
    plt.ylabel("|attribution|")
    plt.title(title)
    plt.grid(True, alpha=0.3)
    plt.legend(loc="upper right", fontsize="small", ncol=2)
    plt.tight_layout()

    finalize_figure(save_path)