Adding utility functions

pyproject.toml

@@ -47,6 +47,8 @@ opensmile = "^2.5.0"
 audiomentations = "^0.35.0"
 torch-audiomentations = "^0.11.1"
 sentence-transformers = "^2.7.0"
+jiwer = "^3.0.4"
+speechbrain = "^1.0.0"
 
 [tool.poetry.group.dev]
 optional = true
@@ -143,10 +145,10 @@ pattern = "default-unprefixed"
 
 [tool.codespell]
 skip = [
-  "./poetry.lock",
-  "./docs_style/pdoc-theme/syntax-highlighting.css"
+  "poetry.lock",
+  "docs_style/pdoc-theme/syntax-highlighting.css"
 ]
-ignore-words-list = ["senselab"]
+ignore-words-list = ["senselab", "nd", "astroid", "wil"]
 
 [build-system]
 requires = ["poetry-core>=1.0.0", "poetry-dynamic-versioning>=1.0.0,<2.0.0"]

src/senselab/audio/tasks/speech_to_text_evaluation.py

@@ -0,0 +1,88 @@
+"""This module implements some utilities for evaluating a transcription."""
+
+import jiwer
+
+
+def calculate_wer(reference: str, hypothesis: str) -> float:
+    """Calculate the Word Error Rate (WER) between the reference and hypothesis.
+
+    Args:
+        reference (str): The ground truth text.
+        hypothesis (str): The predicted text.
+
+    Returns:
+        float: The WER score.
+
+    Examples:
+        >>> calculate_wer("hello world", "hello duck")
+        0.5
+    """
+    return jiwer.wer(reference, hypothesis)
+
+
+def calculate_mer(reference: str, hypothesis: str) -> float:
+    """Calculate the Match Error Rate (MER) between the reference and hypothesis.
+
+    Args:
+        reference (str): The ground truth text.
+        hypothesis (str): The predicted text.
+
+    Returns:
+        float: The MER score.
+
+    Examples:
+        >>> calculate_mer("hello world", "hello duck")
+        0.5
+    """
+    return jiwer.mer(reference, hypothesis)
+
+
+def calculate_wil(reference: str, hypothesis: str) -> float:
+    """Calculate the Word Information Lost (WIL) between the reference and hypothesis.
+
+    Args:
+        reference (str): The ground truth text.
+        hypothesis (str): The predicted text.
+
+    Returns:
+        float: The WIL score.
+
+    Examples:
+        >>> calculate_wil("hello world", "hello duck")
+        0.75
+    """
+    return jiwer.wil(reference, hypothesis)
+
+
+def calculate_wip(reference: str, hypothesis: str) -> float:
+    """Calculate the Word Information Preserved (WIP) between the reference and hypothesis.
+
+    Args:
+        reference (str): The ground truth text.
+        hypothesis (str): The predicted text.
+
+    Returns:
+        float: The WIP score.
+
+    Examples:
+        >>> calculate_wip("hello world", "hello duck")
+        0.25
+    """
+    return jiwer.wip(reference, hypothesis)
+
+
+def calculate_cer(reference: str, hypothesis: str) -> float:
+    """Calculate the Character Error Rate (CER) between the reference and hypothesis.
+
+    Args:
+        reference (str): The ground truth text.
+        hypothesis (str): The predicted text.
+
+    Returns:
+        float: The CER score.
+
+    Examples:
+        >>> calculate_cer("hello world", "hello duck")
+        0.45454545454545453
+    """
+    return jiwer.cer(reference, hypothesis)

src/senselab/audio/tasks/speech_to_text_evaluation_pydra.py

@@ -0,0 +1,17 @@
+"""This module defines a pydra API for the speech to text evaluation task."""
+
+import pydra
+
+from senselab.audio.tasks.speech_to_text_evaluation import (
+    calculate_cer,
+    calculate_mer,
+    calculate_wer,
+    calculate_wil,
+    calculate_wip,
+)
+
+calculate_wer_pt = pydra.mark.task(calculate_wer)
+calculate_mer_pt = pydra.mark.task(calculate_mer)
+calculate_wil_pt = pydra.mark.task(calculate_wil)
+calculate_wip_pt = pydra.mark.task(calculate_wip)
+calculate_cer_pt = pydra.mark.task(calculate_cer)

src/senselab/utils/tasks/cca_cka.py

@@ -0,0 +1,120 @@
+"""This module is for computing CCA and CKA."""
+
+import torch
+
+
+def compute_cca(features_x: torch.Tensor, features_y: torch.Tensor) -> float:
+    """Compute the mean squared CCA correlation (R^2_{CCA}).
+
+    Args:
+        features_x (torch.Tensor): A num_examples x num_features matrix of features.
+        features_y (torch.Tensor): A num_examples x num_features matrix of features.
+
+    Returns:
+        float: The mean squared CCA correlations between X and Y.
+    """
+    qx, _ = torch.qr(features_x)
+    qy, _ = torch.qr(features_y)
+    result = torch.norm(qx.t() @ qy) ** 2 / min(features_x.shape[1], features_y.shape[1])
+    return result.item() if isinstance(result, torch.Tensor) else float(result)
+
+
+def compute_cka(
+    features_x: torch.Tensor, features_y: torch.Tensor, kernel: str = "linear", threshold: float = 1.0
+) -> float:
+    """Compute CKA between feature matrices.
+
+    Args:
+        features_x (torch.Tensor): A num_examples x num_features matrix of features.
+        features_y (torch.Tensor): A num_examples x num_features matrix of features.
+        kernel (str): Type of kernel to use ('linear' or 'rbf'). Default is 'linear'.
+        threshold (float): Fraction of median Euclidean distance to use as RBF kernel bandwidth
+            (used only if kernel is 'rbf').
+
+    Returns:
+        float: The value of CKA between X and Y.
+    """
+
+    def _gram_linear(x: torch.Tensor) -> torch.Tensor:
+        """Compute Gram (kernel) matrix for a linear kernel.
+
+        Args:
+            x (torch.Tensor): A num_examples x num_features matrix of features.
+
+        Returns:
+            torch.Tensor: A num_examples x num_examples Gram matrix of examples.
+        """
+        return x @ x.t()
+
+    def _gram_rbf(x: torch.Tensor, threshold: float = 1.0) -> torch.Tensor:
+        """Compute Gram (kernel) matrix for an RBF kernel.
+
+        Args:
+            x (torch.Tensor): A num_examples x num_features matrix of features.
+            threshold (float): Fraction of median Euclidean distance to use as RBF kernel bandwidth.
+
+        Returns:
+            torch.Tensor: A num_examples x num_examples Gram matrix of examples.
+        """
+        dot_products = x @ x.t()
+        sq_norms = torch.diag(dot_products)
+        sq_distances = -2 * dot_products + sq_norms[:, None] + sq_norms[None, :]
+        sq_median_distance = torch.median(sq_distances)
+        return torch.exp(-sq_distances / (2 * threshold**2 * sq_median_distance))
+
+    def _center_gram(gram: torch.Tensor) -> torch.Tensor:
+        """Center a symmetric Gram matrix.
+
+        This is equivalent to centering the (possibly infinite-dimensional) features
+        induced by the kernel before computing the Gram matrix.
+
+        Args:
+            gram (torch.Tensor): A num_examples x num_examples symmetric matrix.
+
+        Returns:
+            torch.Tensor: A symmetric matrix with centered columns and rows.
+
+        Raises:
+            ValueError: If the input is not a symmetric matrix.
+        """
+        if not torch.allclose(gram, gram.t()):
+            raise ValueError("Input must be a symmetric matrix.")
+
+        n = gram.size(0)
+        unit = torch.ones(n, n, device=gram.device)
+        eye = torch.eye(n, device=gram.device)
+        unit = unit / n
+        haitch = eye - unit
+        centered_gram = haitch.mm(gram).mm(haitch)
+        return centered_gram
+
+    def _cka(gram_x: torch.Tensor, gram_y: torch.Tensor) -> torch.Tensor:
+        """Compute CKA.
+
+        Args:
+            gram_x (torch.Tensor): A num_examples x num_examples Gram matrix.
+            gram_y (torch.Tensor): A num_examples x num_examples Gram matrix.
+
+        Returns:
+            float: The value of CKA between X and Y.
+        """
+        gram_x = _center_gram(gram_x)
+        gram_y = _center_gram(gram_y)
+
+        scaled_hsic = torch.sum(gram_x * gram_y)
+
+        normalization_x = torch.norm(gram_x)
+        normalization_y = torch.norm(gram_y)
+        return scaled_hsic / (normalization_x * normalization_y)
+
+    if kernel == "linear":
+        gram_x = _gram_linear(features_x)
+        gram_y = _gram_linear(features_y)
+    elif kernel == "rbf":
+        gram_x = _gram_rbf(features_x, threshold)
+        gram_y = _gram_rbf(features_y, threshold)
+    else:
+        raise ValueError("Unsupported kernel type. Use 'linear' or 'rbf'.")
+
+    result = _cka(gram_x, gram_y)
+    return result.item() if isinstance(result, torch.Tensor) else float(result)

src/senselab/utils/tasks/cca_cka_pydra.py

@@ -0,0 +1,8 @@
+"""This module defines a pydra API for the CCA and CKA tasks."""
+
+import pydra
+
+from senselab.utils.tasks.cca_cka import compute_cca, compute_cka
+
+compute_cca_pt = pydra.mark.task(compute_cca)
+compute_cka_pt = pydra.mark.task(compute_cka)

src/senselab/utils/tasks/cosine_similarity.py

@@ -0,0 +1,43 @@
+"""This module provides the implementation of cosine similarity."""
+
+import torch
+
+
+def compute_cosine_similarity(tensor1: torch.Tensor, tensor2: torch.Tensor) -> float:
+    """Compute the cosine similarity between two torch tensors.
+
+    Args:
+        tensor1 (Tensor): The first input tensor.
+        tensor2 (Tensor): The second input tensor.
+
+    Returns:
+        float: The cosine similarity between the two input tensors.
+
+    Raises:
+        ValueError: If the input tensors are not of the same shape.
+
+    Examples:
+        >>> tensor1 = torch.tensor([1.0, 2.0, 3.0])
+        >>> tensor2 = torch.tensor([4.0, 5.0, 6.0])
+        >>> cosine_similarity(tensor1, tensor2)
+        0.9746318461970762
+
+        >>> tensor1 = torch.tensor([1.0, 0.0, -1.0])
+        >>> tensor2 = torch.tensor([-1.0, 0.0, 1.0])
+        >>> cosine_similarity(tensor1, tensor2)
+        -1.0
+
+    Note:
+        This function assumes the input tensors are 1-dimensional and have the same shape.
+    """
+    if tensor1.dim() != 1 or tensor2.dim() != 1:
+        raise ValueError("Input tensors must be 1-dimensional")
+    if tensor1.shape != tensor2.shape:
+        raise ValueError("Input tensors must have the same shape")
+
+    dot_product = torch.dot(tensor1, tensor2)
+    norm_tensor1 = torch.norm(tensor1)
+    norm_tensor2 = torch.norm(tensor2)
+
+    cosine_sim = dot_product / (norm_tensor1 * norm_tensor2)
+    return cosine_sim.item()

src/senselab/utils/tasks/cosine_similarity_pydra.py

@@ -0,0 +1,7 @@
+"""This module defines a pydra API for computing cosine similarity."""
+
+import pydra
+
+from senselab.utils.tasks.cosine_similarity import compute_cosine_similarity
+
+cosine_similarity_pt = pydra.mark.task(compute_cosine_similarity)

src/senselab/utils/tasks/cross_correlation.py

@@ -0,0 +1,51 @@
+"""This module contains functions for computing the normalized cross-correlation between two signals."""
+
+import numpy as np
+import torch
+from scipy.signal import correlate
+
+
+def compute_normalized_cross_correlation(signal1: torch.Tensor, signal2: torch.Tensor) -> torch.Tensor:
+    """Calculate the normalized cross-correlation between two signals.
+
+    Args:
+        signal1 (torch.Tensor): The first input signal as a PyTorch tensor.
+        signal2 (torch.Tensor): The second input signal as a PyTorch tensor.
+
+    Returns:
+        torch.Tensor: The normalized cross-correlation value between the two input signals.
+
+    Examples:
+        >>> signal1 = torch.tensor([1.0, 2.0, 3.0, 4.0, 5.0])
+        >>> signal2 = torch.tensor([2.0, 3.0, 4.0])
+        >>> normalized_cross_correlation(signal1, signal2)
+        Tensor([0.30151134, 0.51298918, 0.77459667, 0.9486833 , 0.90453403, 0.70710678, 0.43643578])
+
+    Note:
+        This function assumes the input signals are one-dimensional
+        and contain sufficient elements for meaningful cross-correlation.
+    """
+    # Ensure the inputs are 1D tensors
+    if signal1.ndim != 1 or signal2.ndim != 1:
+        raise ValueError("Input signals must be one-dimensional")
+
+    # Convert PyTorch tensors to NumPy arrays
+    signal1 = signal1.numpy()
+    signal2 = signal2.numpy()
+
+    # Calculate the energy of each signal
+    energy_signal1 = np.sum(signal1**2)
+    energy_signal2 = np.sum(signal2**2)
+
+    # Check for zero energy to avoid division by zero
+    if energy_signal1 == 0 or energy_signal2 == 0:
+        raise ZeroDivisionError("One of the input signals has zero energy, causing division by zero in normalization")
+
+    # Compute the cross-correlation
+    cross_correlation = correlate(signal1, signal2)
+
+    # Calculate the normalized cross-correlation
+    normalized_cross_correlation = cross_correlation / np.sqrt(energy_signal1 * energy_signal2)
+
+    print(normalized_cross_correlation)
+    return torch.Tensor(normalized_cross_correlation)

src/senselab/utils/tasks/cross_correlation_pydra.py

@@ -0,0 +1,7 @@
+"""This module defines a pydra API for computing cross correlation between two signals."""
+
+import pydra
+
+from senselab.utils.tasks.cross_correlation import compute_normalized_cross_correlation
+
+compute_normalized_cross_correlation_pt = pydra.mark.task(compute_normalized_cross_correlation)

src/senselab/utils/tasks/eer.py

@@ -0,0 +1,19 @@
+"""This module implements some utilities for computing the Equal Error Rate (EER)."""
+
+from typing import Tuple
+
+import torch
+from speechbrain.utils.metric_stats import EER
+
+
+def compute_eer(predictions: torch.Tensor, targets: torch.Tensor) -> Tuple[float, float]:
+    """Compute the Equal Error Rate (EER).
+
+    Args:
+        predictions (torch.Tensor): A 1D tensor of predictions.
+        targets (torch.Tensor): A 1D tensor of targets.
+
+    Returns:
+        Tuple[float, float]: The EER and the threshold for the EER.
+    """
+    return EER(predictions, targets)

src/senselab/utils/tasks/eer_pydra.py

@@ -0,0 +1,7 @@
+"""This module defines a pydra API for computing EER."""
+
+import pydra
+
+from senselab.utils.tasks.eer import compute_eer
+
+compute_eer_pt = pydra.mark.task(compute_eer)