FED Encoders

ECG foundation-model encoder adapters for Fréchet ECG Distance (FED).

Each adapter exposes a single encode(signals) method:

signals : np.ndarray, shape (N, L, T)
    N recordings × L leads × T timepoints.

returns : np.ndarray, shape (N, D)
    One embedding vector per recording.

Weights are downloaded from HuggingFace on first use and cached in the default HuggingFace cache directory (~/.cache/huggingface). Set the HF_HOME environment variable to change the cache location.

Supported backbone names

• "ecgfounder" — ECGFounder (PKUDigitalHealth/ECGFounder, RegNet, 10.7 M ECGs)
• "hubert_ecg" — HuBERT-ECG (Edoardo-BS/hubert-ecg-base, Transformer, 9.1 M ECGs)
• "ecgfm" — ECG-FM (bowang-lab/ECG-FM, wav2vec2-style, 1.5 M ECGs)

SUPPORTED_BACKBONES module-attribute

SUPPORTED_BACKBONES = ('ecgfounder', 'hubert_ecg', 'ecgfm')

get_encoder

get_encoder(name: str, device: str = 'cpu', batch_size: int = 32)

Return an initialised encoder adapter for the requested backbone.

Parameters:

Name	Type	Description	Default
name	str	One of "ecgfounder", "hubert_ecg", "ecgfm".	required
device	str	PyTorch device string, e.g. "cpu" or "cuda".	'cpu'
batch_size	int	Number of recordings to process per forward pass.	32

Returns:

Type	Description
An encoder object with an ``encode(signals: np.ndarray) -> np.ndarray`` method.

Raises:

Type	Description
ValueError	If name is not a recognised backbone.
ImportError	If torch / transformers are not installed.

Source code in ecgen_eval/metrics/encoders/__init__.py

def get_encoder(name: str, device: str = "cpu", batch_size: int = 32):
    """
    Return an initialised encoder adapter for the requested backbone.

    Parameters
    ----------
    name : str
        One of ``"ecgfounder"``, ``"hubert_ecg"``, ``"ecgfm"``.
    device : str
        PyTorch device string, e.g. ``"cpu"`` or ``"cuda"``.
    batch_size : int
        Number of recordings to process per forward pass.

    Returns
    -------
    An encoder object with an ``encode(signals: np.ndarray) -> np.ndarray`` method.

    Raises
    ------
    ValueError
        If *name* is not a recognised backbone.
    ImportError
        If ``torch`` / ``transformers`` are not installed.
    """
    if name == "ecgfounder":
        from ecgen_eval.metrics.encoders.ecgfounder import ECGFounderEncoder
        return ECGFounderEncoder(device=device, batch_size=batch_size)
    elif name == "hubert_ecg":
        from ecgen_eval.metrics.encoders.hubert_ecg import HuBERTECGEncoder
        return HuBERTECGEncoder(device=device, batch_size=batch_size)
    elif name == "ecgfm":
        from ecgen_eval.metrics.encoders.ecgfm import ECGFMEncoder
        return ECGFMEncoder(device=device, batch_size=batch_size)
    else:
        raise ValueError(
            f"Unknown backbone '{name}'. Supported: {SUPPORTED_BACKBONES}"
        )

Adapters

ECGFounderEncoder

ECGFounderEncoder(device: str = 'cpu', batch_size: int = 32, preload: bool = False)

Lazy-loading encoder wrapper for ECGFounder.

The HuggingFace checkpoint is downloaded only when encode() is called for the first time (or when the object is first constructed if preload=True).

Parameters:

Name	Type	Description	Default
device	str	PyTorch device ("cpu" or "cuda").	'cpu'
batch_size	int	Recordings per forward pass.	32
preload	bool	If True, download and load the model immediately on construction.	False

Source code in ecgen_eval/metrics/encoders/ecgfounder.py

def __init__(self, device: str = "cpu", batch_size: int = 32, preload: bool = False):
    self.device = device
    self.batch_size = batch_size
    self._model = None  # loaded lazily

    if preload:
        self._load_model()

encode

encode(signals: ndarray) -> ndarray

Encode ECG signals into embedding vectors.

Parameters:

Name	Type	Description	Default
signals	(ndarray, shape(N, L, T))		required

Returns:

Type	Description
(ndarray, shape(N, D))

Source code in ecgen_eval/metrics/encoders/ecgfounder.py

def encode(self, signals: np.ndarray) -> np.ndarray:
    """
    Encode ECG signals into embedding vectors.

    Parameters
    ----------
    signals : np.ndarray, shape (N, L, T)

    Returns
    -------
    np.ndarray, shape (N, D)
    """
    self._load_model()  # no-op if already loaded

    try:
        import torch
    except ImportError as exc:
        raise ImportError(
            "torch is required for ECGFounder. "
            "Install with: pip install ecgen-eval[fed]"
        ) from exc

    N, L, T = signals.shape
    signals = self._pad_leads(signals)  # → (N, 12, T)

    all_embeddings = []
    self._model.eval()

    with torch.no_grad():
        for start in range(0, N, self.batch_size):
            batch = signals[start : start + self.batch_size]
            x = torch.tensor(batch, dtype=torch.float32).to(self.device)
            emb = self._forward(x)  # (B, D)
            all_embeddings.append(emb.cpu().numpy())

    return np.concatenate(all_embeddings, axis=0)

HuBERTECGEncoder

HuBERTECGEncoder(device: str = 'cpu', batch_size: int = 32, variant: str = 'base', preload: bool = False)

Lazy-loading encoder wrapper for HuBERT-ECG.

The HuggingFace checkpoint is downloaded only when encode() is called for the first time.

Parameters:

Name	Type	Description	Default
device	str	PyTorch device ("cpu" or "cuda").	'cpu'
batch_size	int	Recordings per forward pass.	32
variant	str	"base" (default, 93 M params) or "large" (188 M params).	'base'
preload	bool	If True, download and load the model immediately on construction.	False

Source code in ecgen_eval/metrics/encoders/hubert_ecg.py

def __init__(
    self,
    device: str = "cpu",
    batch_size: int = 32,
    variant: str = "base",
    preload: bool = False,
):
    self.device = device
    self.batch_size = batch_size
    self.variant = variant
    self._model = None
    self._processor = None

    if preload:
        self._load_model()

encode

encode(signals: ndarray) -> ndarray

Encode ECG signals into embedding vectors.

Parameters:

Name	Type	Description	Default
signals	(ndarray, shape(N, L, T))		required

Returns:

Type	Description
(ndarray, shape(N, D))

Source code in ecgen_eval/metrics/encoders/hubert_ecg.py

def encode(self, signals: np.ndarray) -> np.ndarray:
    """
    Encode ECG signals into embedding vectors.

    Parameters
    ----------
    signals : np.ndarray, shape (N, L, T)

    Returns
    -------
    np.ndarray, shape (N, D)
    """
    self._load_model()

    try:
        import torch
    except ImportError as exc:
        raise ImportError(
            "torch is required for HuBERT-ECG. "
            "Install with: pip install ecgen-eval[fed]"
        ) from exc

    N, L, T = signals.shape
    all_embeddings = []
    self._model.eval()

    with torch.no_grad():
        for start in range(0, N, self.batch_size):
            batch = signals[start : start + self.batch_size]  # (B, L, T)
            B = batch.shape[0]

            # Encode each lead independently → average across leads
            lead_embeddings = []
            for li in range(L):
                lead_signal = batch[:, li, :]  # (B, T)
                emb = self._encode_lead(lead_signal)  # (B, D)
                lead_embeddings.append(emb)

            # Stack (L, B, D) → mean over leads → (B, D)
            import torch as _torch
            stacked = _torch.stack(lead_embeddings, dim=0)  # (L, B, D)
            recording_emb = stacked.mean(dim=0)             # (B, D)
            all_embeddings.append(recording_emb.cpu().numpy())

    return np.concatenate(all_embeddings, axis=0)

ECGFMEncoder

ECGFMEncoder(device: str = 'cpu', batch_size: int = 32, preload: bool = False)

Lazy-loading encoder wrapper for ECG-FM.

The HuggingFace checkpoint is downloaded only when encode() is called for the first time.

Parameters:

Name	Type	Description	Default
device	str	PyTorch device ("cpu" or "cuda").	'cpu'
batch_size	int	Recordings per forward pass.	32
preload	bool	If True, download and load the model immediately on construction.	False

Source code in ecgen_eval/metrics/encoders/ecgfm.py

def __init__(self, device: str = "cpu", batch_size: int = 32, preload: bool = False):
    self.device = device
    self.batch_size = batch_size
    self._model = None

    if preload:
        self._load_model()

encode

encode(signals: ndarray) -> ndarray

Encode ECG signals into embedding vectors.

Parameters:

Name	Type	Description	Default
signals	(ndarray, shape(N, L, T))		required

Returns:

Type	Description
(ndarray, shape(N, D))

Source code in ecgen_eval/metrics/encoders/ecgfm.py

def encode(self, signals: np.ndarray) -> np.ndarray:
    """
    Encode ECG signals into embedding vectors.

    Parameters
    ----------
    signals : np.ndarray, shape (N, L, T)

    Returns
    -------
    np.ndarray, shape (N, D)
    """
    self._load_model()

    try:
        import torch
    except ImportError as exc:
        raise ImportError(
            "torch is required for ECG-FM. "
            "Install with: pip install ecgen-eval[fed]"
        ) from exc

    N, L, T = signals.shape
    signals = self._pad_leads(signals)  # → (N, 12, T)

    all_embeddings = []
    self._model.eval()

    with torch.no_grad():
        for start in range(0, N, self.batch_size):
            batch = signals[start : start + self.batch_size]  # (B, 12, T)
            x = torch.tensor(batch, dtype=torch.float32).to(self.device)
            emb = self._forward(x)  # (B, D)
            all_embeddings.append(emb.cpu().numpy())

    return np.concatenate(all_embeddings, axis=0)