API Reference: ecg_sdk.core.signal

ecg_sdk.core.signal.ECGSignal dataclass

Standardized container representing an Electrocardiogram (ECG) recording.

Attributes:

Name	Type	Description
data	ndarray	2D numpy array of shape (n_samples, n_leads) representing voltage in units. If initialized with 1D array, automatically reshaped to (n_samples, 1).
sampling_rate	float	Sampling rate in Hz (samples per second), \(f_s\).
lead_names	List[str]	List of strings identifying each lead (e.g. ['MLII', 'V1']).
patient_id	Optional[str]	Optional string identifying the subject/patient.
record_name	Optional[str]	Optional string identifier of the recording.
units	str	Measurement voltage unit, defaults to "mV".
metadata	Dict[str, Any]	Dictionary holding arbitrary recording metadata or clinical annotations.

Source code in ecg_sdk/core/signal.py

@dataclass
class ECGSignal:
    """
    Standardized container representing an Electrocardiogram (ECG) recording.

    Attributes:
        data: 2D numpy array of shape (n_samples, n_leads) representing voltage in `units`.
              If initialized with 1D array, automatically reshaped to (n_samples, 1).
        sampling_rate: Sampling rate in Hz (samples per second), $f_s$.
        lead_names: List of strings identifying each lead (e.g. ['MLII', 'V1']).
        patient_id: Optional string identifying the subject/patient.
        record_name: Optional string identifier of the recording.
        units: Measurement voltage unit, defaults to "mV".
        metadata: Dictionary holding arbitrary recording metadata or clinical annotations.
    """

    data: np.ndarray
    sampling_rate: float
    lead_names: List[str] = field(default_factory=list)
    patient_id: Optional[str] = None
    record_name: Optional[str] = None
    units: str = "mV"
    metadata: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        # Ensure data is float numpy array
        self.data = np.asarray(self.data, dtype=np.float64)

        if self.data.ndim == 1:
            self.data = self.data.reshape(-1, 1)
        elif self.data.ndim != 2:
            raise ValueError(f"ECG data must be 1D or 2D array, got shape {self.data.shape}")

        if self.sampling_rate <= 0:
            raise ValueError(f"Sampling rate must be positive, got {self.sampling_rate} Hz")

        n_leads = self.data.shape[1]
        if not self.lead_names:
            if n_leads == 1:
                self.lead_names = ["Lead_I"]
            elif n_leads == 2:
                self.lead_names = ["Lead_I", "Lead_II"]
            elif n_leads == 12:
                self.lead_names = [
                    "I",
                    "II",
                    "III",
                    "aVR",
                    "aVL",
                    "aVF",
                    "V1",
                    "V2",
                    "V3",
                    "V4",
                    "V5",
                    "V6",
                ]
            else:
                self.lead_names = [f"Lead_{i + 1}" for i in range(n_leads)]
        elif len(self.lead_names) != n_leads:
            raise ValueError(
                f"Length of lead_names ({len(self.lead_names)}) does not match data leads ({n_leads})"
            )

    @property
    def n_samples(self) -> int:
        """Total number of discrete time samples."""
        return self.data.shape[0]

    @property
    def n_leads(self) -> int:
        """Total number of leads / channels."""
        return self.data.shape[1]

    @property
    def duration(self) -> float:
        """Duration of the recording in seconds."""
        return self.n_samples / self.sampling_rate

    @property
    def time_vector(self) -> np.ndarray:
        """1D array containing timestamps in seconds for each sample."""
        return np.arange(self.n_samples, dtype=np.float64) / self.sampling_rate

    def get_lead(self, lead: Union[str, int]) -> np.ndarray:
        """
        Extract a single lead as a 1D numpy array.

        Args:
            lead: Integer lead index or string lead name.

        Returns:
            1D numpy array of shape (n_samples,).
        """
        if isinstance(lead, int):
            if 0 <= lead < self.n_leads:
                return self.data[:, lead].copy()
            raise IndexError(f"Lead index {lead} out of bounds for {self.n_leads} leads.")
        elif isinstance(lead, str):
            # 1. Exact match
            if lead in self.lead_names:
                idx = self.lead_names.index(lead)
                return self.data[:, idx].copy()

            # 2. Case-insensitive and alias matching
            target = lead.strip().lower().replace(" ", "_")
            for i, name in enumerate(self.lead_names):
                norm_name = name.strip().lower().replace(" ", "_")
                if (
                    norm_name == target
                    or norm_name == f"lead_{target}"
                    or target == f"lead_{norm_name}"
                ):
                    return self.data[:, i].copy()
                # Also match Roman numerals II <-> Lead_II <-> MLII
                if target == "ii" and norm_name in ("lead_ii", "mlii", "ii"):
                    return self.data[:, i].copy()
                if target == "i" and norm_name in ("lead_i", "i"):
                    return self.data[:, i].copy()

            raise KeyError(f"Lead name '{lead}' not found in {self.lead_names}")

        else:
            raise TypeError(f"Lead must be int or str, got {type(lead)}")

    def slice(self, start_time: float = 0.0, end_time: Optional[float] = None) -> ECGSignal:
        """
        Extract a time window from the recording.

        Args:
            start_time: Start time in seconds.
            end_time: End time in seconds. If None, slices to the end.

        Returns:
            A new ECGSignal instance containing the sliced window.
        """
        start_idx = max(0, int(round(start_time * self.sampling_rate)))
        if end_time is None:
            end_idx = self.n_samples
        else:
            end_idx = min(self.n_samples, int(round(end_time * self.sampling_rate)))

        if start_idx >= end_idx:
            raise ValueError(
                f"Invalid slice window: start_time ({start_time}s) >= end_time ({end_time}s)"
            )

        sliced_data = self.data[start_idx:end_idx, :].copy()
        new_metadata = dict(self.metadata)
        new_metadata["slice_start_time"] = start_time
        new_metadata["slice_end_time"] = end_time if end_time is not None else self.duration

        return ECGSignal(
            data=sliced_data,
            sampling_rate=self.sampling_rate,
            lead_names=list(self.lead_names),
            patient_id=self.patient_id,
            record_name=self.record_name,
            units=self.units,
            metadata=new_metadata,
        )

    def resample(self, target_fs: float) -> ECGSignal:
        """
        Resample the signal to a new sampling frequency using Fourier method.

        Args:
            target_fs: Target sampling frequency in Hz.

        Returns:
            New ECGSignal instance resampled at target_fs.
        """
        if target_fs <= 0:
            raise ValueError(f"Target sampling rate must be positive, got {target_fs}")
        if target_fs == self.sampling_rate:
            return self.copy()

        num_target_samples = int(round(self.n_samples * target_fs / self.sampling_rate))
        resampled_data = scipy_signal.resample(self.data, num_target_samples, axis=0)

        new_metadata = dict(self.metadata)
        new_metadata["original_fs"] = self.sampling_rate
        new_metadata["resampled_fs"] = target_fs

        # Scale annotation sample indices (R-peaks) proportionally to maintain exact alignment
        if "annotations" in self.metadata and isinstance(self.metadata["annotations"], dict):
            ann = dict(self.metadata["annotations"])
            if "r_peaks" in ann and ann["r_peaks"] is not None:
                scale_factor = target_fs / self.sampling_rate
                ann["r_peaks"] = np.round(np.array(ann["r_peaks"]) * scale_factor).astype(int)
            new_metadata["annotations"] = ann

        return ECGSignal(
            data=resampled_data,
            sampling_rate=float(target_fs),
            lead_names=list(self.lead_names),
            patient_id=self.patient_id,
            record_name=self.record_name,
            units=self.units,
            metadata=new_metadata,
        )

    def normalize(self, method: str = "zscore") -> ECGSignal:
        """
        Normalize each lead independently.

        Args:
            method: 'zscore' (mean=0, std=1), 'minmax' (scaled to [0, 1]), or 'robust' (median & IQR).

        Returns:
            New normalized ECGSignal instance.
        """
        normalized_data = np.zeros_like(self.data)
        for i in range(self.n_leads):
            col = self.data[:, i]
            if method == "zscore":
                mean_val = np.mean(col)
                std_val = np.std(col)
                normalized_data[:, i] = (col - mean_val) / (std_val + 1e-8)
            elif method == "minmax":
                min_val = np.min(col)
                max_val = np.max(col)
                denom = max_val - min_val
                normalized_data[:, i] = (col - min_val) / (denom + 1e-8)
            elif method == "robust":
                median_val = np.median(col)
                q75, q25 = np.percentile(col, [75, 25])
                iqr = q75 - q25
                normalized_data[:, i] = (col - median_val) / (iqr + 1e-8)
            else:
                raise ValueError(
                    f"Unknown normalization method '{method}'. Choose 'zscore', 'minmax', or 'robust'."
                )

        new_metadata = dict(self.metadata)
        new_metadata["normalization"] = method

        return ECGSignal(
            data=normalized_data,
            sampling_rate=self.sampling_rate,
            lead_names=list(self.lead_names),
            patient_id=self.patient_id,
            record_name=self.record_name,
            units="normalized",
            metadata=new_metadata,
        )

    def summary(self) -> Dict[str, Any]:
        """
        Generate statistical summary dictionary of the recording.
        """
        lead_stats = {}
        for i, lead in enumerate(self.lead_names):
            col = self.data[:, i]
            lead_stats[lead] = {
                "mean": float(np.mean(col)),
                "std": float(np.std(col)),
                "min": float(np.min(col)),
                "max": float(np.max(col)),
                "peak_to_peak": float(np.ptp(col)),
            }

        return {
            "record_name": self.record_name,
            "patient_id": self.patient_id,
            "sampling_rate": self.sampling_rate,
            "n_samples": self.n_samples,
            "n_leads": self.n_leads,
            "duration_sec": self.duration,
            "units": self.units,
            "leads": lead_stats,
        }

    def to_dataframe(self) -> pd.DataFrame:
        """
        Export the signal data and time vector to a pandas DataFrame.
        """
        df = pd.DataFrame(self.data, columns=self.lead_names)
        df.insert(0, "time_sec", self.time_vector)
        return df

    def copy(self) -> ECGSignal:
        """Return a deep copy of the ECGSignal instance."""
        return ECGSignal(
            data=self.data.copy(),
            sampling_rate=self.sampling_rate,
            lead_names=list(self.lead_names),
            patient_id=self.patient_id,
            record_name=self.record_name,
            units=self.units,
            metadata=dict(self.metadata),
        )

    def plot_clinical(
        self,
        lead: Optional[Union[str, int]] = 0,
        start_time: float = 0.0,
        duration: Optional[float] = 10.0,
        display_mode: str = "paper",
        speed_mm_s: float = 25.0,
        gain_mm_mv: float = 10.0,
        grid_style: str = "clinical_pink",
        voltage_range: Optional[Tuple[float, float]] = None,
        ylim: Optional[Tuple[float, float]] = None,
        plot_height: Optional[float] = None,
        height_per_lead: Optional[float] = None,
        figsize: Optional[Tuple[float, float]] = None,
        r_peaks: Optional[np.ndarray] = None,
        waves: Optional[Dict[str, np.ndarray]] = None,
        annotations: Optional[List[Dict[str, Any]]] = None,
        time_tick_step: Optional[float] = None,
        voltage_tick_step: Optional[float] = None,
        sparse_ticks: bool = True,
        show_calibration_pulse: bool = True,
        show_header: bool = True,
        title: Optional[str] = None,
        save_path: Optional[str] = None,
        **kwargs: Any,
    ):
        """
        Plot this recording on standard clinical ECG paper.
        """
        from ecg_sdk.visualization.plotter import plot_clinical_ecg

        return plot_clinical_ecg(
            signal=self,
            lead=lead,
            start_time=start_time,
            duration=duration,
            display_mode=display_mode,
            speed_mm_s=speed_mm_s,
            gain_mm_mv=gain_mm_mv,
            grid_style=grid_style,
            voltage_range=voltage_range,
            ylim=ylim,
            plot_height=plot_height,
            height_per_lead=height_per_lead,
            figsize=figsize,
            r_peaks=r_peaks,
            waves=waves,
            annotations=annotations,
            time_tick_step=time_tick_step,
            voltage_tick_step=voltage_tick_step,
            sparse_ticks=sparse_ticks,
            show_calibration_pulse=show_calibration_pulse,
            show_header=show_header,
            title=title,
            save_path=save_path,
            **kwargs,
        )

    def filter(
        self,
        lowcut: float = 0.5,
        highcut: float = 45.0,
        notch_freq: Optional[float] = 50.0,
        notch_q: float = 30.0,
        order: int = 4,
    ) -> ECGSignal:
        """
        Apply zero-phase bandpass and optional powerline notch filtering to all leads.
        """
        from ecg_sdk.preprocessing.filters import filter_ecg

        return filter_ecg(
            self,
            lowcut=lowcut,
            highcut=highcut,
            notch_freq=notch_freq,
            notch_q=notch_q,
            order=order,
        )

    def remove_baseline(
        self,
        method: str = "cascaded_median",
        window1_ms: float = 200.0,
        window2_ms: float = 600.0,
        poly_order: int = 3,
        return_baseline: bool = False,
    ):
        """
        Remove low-frequency baseline drift from all leads.
        """
        from ecg_sdk.preprocessing.baseline import remove_baseline_wander

        return remove_baseline_wander(
            self,
            method=method,
            window1_ms=window1_ms,
            window2_ms=window2_ms,
            poly_order=poly_order,
            return_baseline=return_baseline,
        )

    def assess_quality(
        self,
        acceptance_threshold: float = 0.60,
    ):
        """
        Evaluate multi-metric Signal Quality Index (SQI) across all channels.
        """
        from ecg_sdk.preprocessing.quality import assess_signal_quality

        return assess_signal_quality(
            self,
            acceptance_threshold=acceptance_threshold,
        )

    def __repr__(self) -> str:

        return (
            f"ECGSignal(record='{self.record_name}', leads={self.lead_names}, "
            f"fs={self.sampling_rate}Hz, duration={self.duration:.2f}s, shape={self.data.shape})"
        )

duration property

Duration of the recording in seconds.

n_leads property

Total number of leads / channels.

n_samples property

Total number of discrete time samples.

time_vector property

1D array containing timestamps in seconds for each sample.

assess_quality(acceptance_threshold=0.6)

Evaluate multi-metric Signal Quality Index (SQI) across all channels.

Source code in ecg_sdk/core/signal.py

def assess_quality(
    self,
    acceptance_threshold: float = 0.60,
):
    """
    Evaluate multi-metric Signal Quality Index (SQI) across all channels.
    """
    from ecg_sdk.preprocessing.quality import assess_signal_quality

    return assess_signal_quality(
        self,
        acceptance_threshold=acceptance_threshold,
    )

copy()

Return a deep copy of the ECGSignal instance.

Source code in ecg_sdk/core/signal.py

def copy(self) -> ECGSignal:
    """Return a deep copy of the ECGSignal instance."""
    return ECGSignal(
        data=self.data.copy(),
        sampling_rate=self.sampling_rate,
        lead_names=list(self.lead_names),
        patient_id=self.patient_id,
        record_name=self.record_name,
        units=self.units,
        metadata=dict(self.metadata),
    )

filter(lowcut=0.5, highcut=45.0, notch_freq=50.0, notch_q=30.0, order=4)

Apply zero-phase bandpass and optional powerline notch filtering to all leads.

Source code in ecg_sdk/core/signal.py

def filter(
    self,
    lowcut: float = 0.5,
    highcut: float = 45.0,
    notch_freq: Optional[float] = 50.0,
    notch_q: float = 30.0,
    order: int = 4,
) -> ECGSignal:
    """
    Apply zero-phase bandpass and optional powerline notch filtering to all leads.
    """
    from ecg_sdk.preprocessing.filters import filter_ecg

    return filter_ecg(
        self,
        lowcut=lowcut,
        highcut=highcut,
        notch_freq=notch_freq,
        notch_q=notch_q,
        order=order,
    )

get_lead(lead)

Extract a single lead as a 1D numpy array.

Parameters:

Name	Type	Description	Default
lead	Union[str, int]	Integer lead index or string lead name.	required

Returns:

Type	Description
ndarray	1D numpy array of shape (n_samples,).

Source code in ecg_sdk/core/signal.py

def get_lead(self, lead: Union[str, int]) -> np.ndarray:
    """
    Extract a single lead as a 1D numpy array.

    Args:
        lead: Integer lead index or string lead name.

    Returns:
        1D numpy array of shape (n_samples,).
    """
    if isinstance(lead, int):
        if 0 <= lead < self.n_leads:
            return self.data[:, lead].copy()
        raise IndexError(f"Lead index {lead} out of bounds for {self.n_leads} leads.")
    elif isinstance(lead, str):
        # 1. Exact match
        if lead in self.lead_names:
            idx = self.lead_names.index(lead)
            return self.data[:, idx].copy()

        # 2. Case-insensitive and alias matching
        target = lead.strip().lower().replace(" ", "_")
        for i, name in enumerate(self.lead_names):
            norm_name = name.strip().lower().replace(" ", "_")
            if (
                norm_name == target
                or norm_name == f"lead_{target}"
                or target == f"lead_{norm_name}"
            ):
                return self.data[:, i].copy()
            # Also match Roman numerals II <-> Lead_II <-> MLII
            if target == "ii" and norm_name in ("lead_ii", "mlii", "ii"):
                return self.data[:, i].copy()
            if target == "i" and norm_name in ("lead_i", "i"):
                return self.data[:, i].copy()

        raise KeyError(f"Lead name '{lead}' not found in {self.lead_names}")

    else:
        raise TypeError(f"Lead must be int or str, got {type(lead)}")

normalize(method='zscore')

Normalize each lead independently.

Parameters:

Name	Type	Description	Default
method	str	'zscore' (mean=0, std=1), 'minmax' (scaled to [0, 1]), or 'robust' (median & IQR).	'zscore'

Returns:

Type	Description
ECGSignal	New normalized ECGSignal instance.

Source code in ecg_sdk/core/signal.py

def normalize(self, method: str = "zscore") -> ECGSignal:
    """
    Normalize each lead independently.

    Args:
        method: 'zscore' (mean=0, std=1), 'minmax' (scaled to [0, 1]), or 'robust' (median & IQR).

    Returns:
        New normalized ECGSignal instance.
    """
    normalized_data = np.zeros_like(self.data)
    for i in range(self.n_leads):
        col = self.data[:, i]
        if method == "zscore":
            mean_val = np.mean(col)
            std_val = np.std(col)
            normalized_data[:, i] = (col - mean_val) / (std_val + 1e-8)
        elif method == "minmax":
            min_val = np.min(col)
            max_val = np.max(col)
            denom = max_val - min_val
            normalized_data[:, i] = (col - min_val) / (denom + 1e-8)
        elif method == "robust":
            median_val = np.median(col)
            q75, q25 = np.percentile(col, [75, 25])
            iqr = q75 - q25
            normalized_data[:, i] = (col - median_val) / (iqr + 1e-8)
        else:
            raise ValueError(
                f"Unknown normalization method '{method}'. Choose 'zscore', 'minmax', or 'robust'."
            )

    new_metadata = dict(self.metadata)
    new_metadata["normalization"] = method

    return ECGSignal(
        data=normalized_data,
        sampling_rate=self.sampling_rate,
        lead_names=list(self.lead_names),
        patient_id=self.patient_id,
        record_name=self.record_name,
        units="normalized",
        metadata=new_metadata,
    )

plot_clinical(lead=0, start_time=0.0, duration=10.0, display_mode='paper', speed_mm_s=25.0, gain_mm_mv=10.0, grid_style='clinical_pink', voltage_range=None, ylim=None, plot_height=None, height_per_lead=None, figsize=None, r_peaks=None, waves=None, annotations=None, time_tick_step=None, voltage_tick_step=None, sparse_ticks=True, show_calibration_pulse=True, show_header=True, title=None, save_path=None, **kwargs)

Plot this recording on standard clinical ECG paper.

Source code in ecg_sdk/core/signal.py

def plot_clinical(
    self,
    lead: Optional[Union[str, int]] = 0,
    start_time: float = 0.0,
    duration: Optional[float] = 10.0,
    display_mode: str = "paper",
    speed_mm_s: float = 25.0,
    gain_mm_mv: float = 10.0,
    grid_style: str = "clinical_pink",
    voltage_range: Optional[Tuple[float, float]] = None,
    ylim: Optional[Tuple[float, float]] = None,
    plot_height: Optional[float] = None,
    height_per_lead: Optional[float] = None,
    figsize: Optional[Tuple[float, float]] = None,
    r_peaks: Optional[np.ndarray] = None,
    waves: Optional[Dict[str, np.ndarray]] = None,
    annotations: Optional[List[Dict[str, Any]]] = None,
    time_tick_step: Optional[float] = None,
    voltage_tick_step: Optional[float] = None,
    sparse_ticks: bool = True,
    show_calibration_pulse: bool = True,
    show_header: bool = True,
    title: Optional[str] = None,
    save_path: Optional[str] = None,
    **kwargs: Any,
):
    """
    Plot this recording on standard clinical ECG paper.
    """
    from ecg_sdk.visualization.plotter import plot_clinical_ecg

    return plot_clinical_ecg(
        signal=self,
        lead=lead,
        start_time=start_time,
        duration=duration,
        display_mode=display_mode,
        speed_mm_s=speed_mm_s,
        gain_mm_mv=gain_mm_mv,
        grid_style=grid_style,
        voltage_range=voltage_range,
        ylim=ylim,
        plot_height=plot_height,
        height_per_lead=height_per_lead,
        figsize=figsize,
        r_peaks=r_peaks,
        waves=waves,
        annotations=annotations,
        time_tick_step=time_tick_step,
        voltage_tick_step=voltage_tick_step,
        sparse_ticks=sparse_ticks,
        show_calibration_pulse=show_calibration_pulse,
        show_header=show_header,
        title=title,
        save_path=save_path,
        **kwargs,
    )

remove_baseline(method='cascaded_median', window1_ms=200.0, window2_ms=600.0, poly_order=3, return_baseline=False)

Remove low-frequency baseline drift from all leads.

Source code in ecg_sdk/core/signal.py

def remove_baseline(
    self,
    method: str = "cascaded_median",
    window1_ms: float = 200.0,
    window2_ms: float = 600.0,
    poly_order: int = 3,
    return_baseline: bool = False,
):
    """
    Remove low-frequency baseline drift from all leads.
    """
    from ecg_sdk.preprocessing.baseline import remove_baseline_wander

    return remove_baseline_wander(
        self,
        method=method,
        window1_ms=window1_ms,
        window2_ms=window2_ms,
        poly_order=poly_order,
        return_baseline=return_baseline,
    )

resample(target_fs)

Resample the signal to a new sampling frequency using Fourier method.

Parameters:

Name	Type	Description	Default
target_fs	float	Target sampling frequency in Hz.	required

Returns:

Type	Description
ECGSignal	New ECGSignal instance resampled at target_fs.

Source code in ecg_sdk/core/signal.py

def resample(self, target_fs: float) -> ECGSignal:
    """
    Resample the signal to a new sampling frequency using Fourier method.

    Args:
        target_fs: Target sampling frequency in Hz.

    Returns:
        New ECGSignal instance resampled at target_fs.
    """
    if target_fs <= 0:
        raise ValueError(f"Target sampling rate must be positive, got {target_fs}")
    if target_fs == self.sampling_rate:
        return self.copy()

    num_target_samples = int(round(self.n_samples * target_fs / self.sampling_rate))
    resampled_data = scipy_signal.resample(self.data, num_target_samples, axis=0)

    new_metadata = dict(self.metadata)
    new_metadata["original_fs"] = self.sampling_rate
    new_metadata["resampled_fs"] = target_fs

    # Scale annotation sample indices (R-peaks) proportionally to maintain exact alignment
    if "annotations" in self.metadata and isinstance(self.metadata["annotations"], dict):
        ann = dict(self.metadata["annotations"])
        if "r_peaks" in ann and ann["r_peaks"] is not None:
            scale_factor = target_fs / self.sampling_rate
            ann["r_peaks"] = np.round(np.array(ann["r_peaks"]) * scale_factor).astype(int)
        new_metadata["annotations"] = ann

    return ECGSignal(
        data=resampled_data,
        sampling_rate=float(target_fs),
        lead_names=list(self.lead_names),
        patient_id=self.patient_id,
        record_name=self.record_name,
        units=self.units,
        metadata=new_metadata,
    )

slice(start_time=0.0, end_time=None)

Extract a time window from the recording.

Parameters:

Name	Type	Description	Default
start_time	float	Start time in seconds.	0.0
end_time	Optional[float]	End time in seconds. If None, slices to the end.	None

Returns:

Type	Description
ECGSignal	A new ECGSignal instance containing the sliced window.

Source code in ecg_sdk/core/signal.py

def slice(self, start_time: float = 0.0, end_time: Optional[float] = None) -> ECGSignal:
    """
    Extract a time window from the recording.

    Args:
        start_time: Start time in seconds.
        end_time: End time in seconds. If None, slices to the end.

    Returns:
        A new ECGSignal instance containing the sliced window.
    """
    start_idx = max(0, int(round(start_time * self.sampling_rate)))
    if end_time is None:
        end_idx = self.n_samples
    else:
        end_idx = min(self.n_samples, int(round(end_time * self.sampling_rate)))

    if start_idx >= end_idx:
        raise ValueError(
            f"Invalid slice window: start_time ({start_time}s) >= end_time ({end_time}s)"
        )

    sliced_data = self.data[start_idx:end_idx, :].copy()
    new_metadata = dict(self.metadata)
    new_metadata["slice_start_time"] = start_time
    new_metadata["slice_end_time"] = end_time if end_time is not None else self.duration

    return ECGSignal(
        data=sliced_data,
        sampling_rate=self.sampling_rate,
        lead_names=list(self.lead_names),
        patient_id=self.patient_id,
        record_name=self.record_name,
        units=self.units,
        metadata=new_metadata,
    )

summary()

Generate statistical summary dictionary of the recording.

Source code in ecg_sdk/core/signal.py

def summary(self) -> Dict[str, Any]:
    """
    Generate statistical summary dictionary of the recording.
    """
    lead_stats = {}
    for i, lead in enumerate(self.lead_names):
        col = self.data[:, i]
        lead_stats[lead] = {
            "mean": float(np.mean(col)),
            "std": float(np.std(col)),
            "min": float(np.min(col)),
            "max": float(np.max(col)),
            "peak_to_peak": float(np.ptp(col)),
        }

    return {
        "record_name": self.record_name,
        "patient_id": self.patient_id,
        "sampling_rate": self.sampling_rate,
        "n_samples": self.n_samples,
        "n_leads": self.n_leads,
        "duration_sec": self.duration,
        "units": self.units,
        "leads": lead_stats,
    }

to_dataframe()

Export the signal data and time vector to a pandas DataFrame.

Source code in ecg_sdk/core/signal.py

def to_dataframe(self) -> pd.DataFrame:
    """
    Export the signal data and time vector to a pandas DataFrame.
    """
    df = pd.DataFrame(self.data, columns=self.lead_names)
    df.insert(0, "time_sec", self.time_vector)
    return df