Synthesis Validation

For synthesis validation we have only feature based evaluation metrics. Reminder that synthesis validation is the process of evaluating the quality of the generated data, having a reference dataset.

These metrics require a preprocessing of the data, to extract the features that will be used to compare the generated data with the reference dataset. For more information on this process, please refer to the feature extraction tutorials.

Feature-based

Quality

`pymdma.time_series.measures.synthesis_val.ImprovedPrecision`

Improved Precision Metric for accessing fidelity of generative models.

Objective: Fidelity

Parameters:

Name	Type	Description	Default
`k`	`int`	Number of nearest neighbors to consider in the hypersphere estimation. Defaults to 5.	`5`
`metric`	`str`	The metric to use when calculating distance between instances. For the available metrics, see the documentation of `sklearn.metrics.pairwise_distances`.	`"euclidean"`
`n_workers`	`int`	Number of workers for computing pairwise distances. Defaults to 4.	`4`
`**kwargs`		Additional keyword arguments for compatiblilty.	`{}`

References

Kynkaanniemi et al., Improved Precision and Recall Metric for Assessing Generative Models (2019). https://arxiv.org/abs/1904.06991

Code adapted from: improved-precision-and-recall-metric: Improved Precision and Recall Metric for Assessing Generative Models — Official TensorFlow Implementation. https://github.com/kynkaat/improved-precision-and-recall-metric

Hypersphere estimation code was taken from: generative-evaluation-prdc, Reliable Fidelity and Diversity Metrics for Generative Models. https://github.com/clovaai/generative-evaluation-prdc

Examples:

>>> improved_precision = ImprovedPrecision()
>>> real_features = np.random.rand(100, 100)
>>> fake_features = np.random.rand(100, 100)
>>> result: MetricResult = improved_precision.compute(real_features, fake_features)

Source code in src/pymdma/general/measures/prdc.py

class ImprovedPrecision(FeatureMetric):
    """Improved Precision Metric for accessing fidelity of generative models.

    **Objective**: Fidelity

    Parameters
    ----------
    k : int, optional
        Number of nearest neighbors to consider in the hypersphere estimation. Defaults to 5.
    metric : str, optional, default="euclidean"
        The metric to use when calculating distance between instances.
        For the available metrics, see the documentation of `sklearn.metrics.pairwise_distances`.
    n_workers : int, optional
        Number of workers for computing pairwise distances. Defaults to 4.
    **kwargs
        Additional keyword arguments for compatiblilty.

    References
    ----------
    Kynkaanniemi et al., Improved Precision and Recall Metric for Assessing Generative Models (2019).
    https://arxiv.org/abs/1904.06991

    Code adapted from:
    improved-precision-and-recall-metric: Improved Precision and Recall Metric for Assessing Generative Models — Official TensorFlow Implementation.
    https://github.com/kynkaat/improved-precision-and-recall-metric

    Hypersphere estimation code was taken from:
    generative-evaluation-prdc, Reliable Fidelity and Diversity Metrics for Generative Models.
    https://github.com/clovaai/generative-evaluation-prdc

    Examples
    --------
    >>> improved_precision = ImprovedPrecision()
    >>> real_features = np.random.rand(100, 100)
    >>> fake_features = np.random.rand(100, 100)
    >>> result: MetricResult = improved_precision.compute(real_features, fake_features)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = [EvaluationLevel.INSTANCE, EvaluationLevel.DATASET]
    metric_group = MetricGroup.QUALITY

    higher_is_better: bool = True
    min_value: float = 0.0
    max_value: float = 1.0

    def __init__(
        self,
        k: int = 5,
        metric: str = "euclidean",
        n_workers: int = 4,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.k = k
        self.metric = metric
        self.n_workers = n_workers

    def compute(self, real_features: np.ndarray, fake_features: np.ndarray, **kwargs) -> MetricResult:
        """Compute the Improved Precision metric.

        Parameters
        ----------
        real_features : np.ndarray
            Array of shape (n_samples, n_features) containing the real features.
        fake_features : np.ndarray
            Array of shape (n_samples, n_features) containing the fake features.

        Notes
        -----
        Intermediate computations can be stored in the `context` dictionary of the `kwargs` parameter.
        Usefull when calculating multiple metrics that share the same intermediate computations.

        Returns
        -------
        result: MetricResult
            Dataset-level and instance-level results for the precision metric.
        """
        state = kwargs.get("context", {})
        if "real_nn_distances" not in state:
            state["real_nn_distances"] = compute_nearest_neighbour_distances(
                real_features,
                nearest_k=self.k,
                metric=self.metric,
                n_workers=self.n_workers,
            )

        if "real_fake_distances" not in state:
            state["real_fake_distances"] = compute_pairwise_distance(
                real_features,
                fake_features,
                metric=self.metric,
                n_workers=self.n_workers,
            )

        precision = (
            np.logical_or(
                (state["real_fake_distances"] < np.expand_dims(state["real_nn_distances"], axis=1)),
                np.isclose(state["real_fake_distances"], np.expand_dims(state["real_nn_distances"], axis=1)),
            )
            .any(axis=0)
            .astype(int)
        )

        return MetricResult(
            dataset_level={"dtype": OutputsTypes.NUMERIC, "subtype": "float", "value": precision.mean()},
            instance_level={"dtype": OutputsTypes.ARRAY, "subtype": "int", "value": precision.tolist()},
        )

`pymdma.time_series.measures.synthesis_val.ImprovedRecall`

Improved Recall Metric for accessing diversity of generative models.

Objective: Diversity

Parameters:

Name	Type	Description	Default
`k`	`int`	Number of nearest neighbors to consider in the hypersphere estimation. Defaults to 5.	`5`
`metric`	`str`	The metric to use when calculating distance between instances. For the available metrics, see the documentation of `sklearn.metrics.pairwise_distances`.	`"euclidean"`
`n_workers`	`int`	Number of workers for computing pairwise distances. Defaults to 4.	`4`
`**kwargs`		Additional keyword arguments for compatiblilty.	`{}`

References

Kynkaanniemi et al., Improved Precision and Recall Metric for Assessing Generative Models (2019). https://arxiv.org/abs/1904.06991

Code adapted from: improved-precision-and-recall-metric: Improved Precision and Recall Metric for Assessing Generative Models — Official TensorFlow Implementation. https://github.com/kynkaat/improved-precision-and-recall-metric

Hypersphere estimation code was taken from: generative-evaluation-prdc, Reliable Fidelity and Diversity Metrics for Generative Models. https://github.com/clovaai/generative-evaluation-prdc

Examples:

>>> improved_recall = ImprovedRecall()
>>> real_features = np.random.rand(100, 100)
>>> fake_features = np.random.rand(100, 100)
>>> result: MetricResult = improved_recall.compute(real_features, fake_features)

Source code in src/pymdma/general/measures/prdc.py

class ImprovedRecall(FeatureMetric):
    """Improved Recall Metric for accessing diversity of generative models.

    **Objective**: Diversity

    Parameters
    ----------
    k : int, optional
        Number of nearest neighbors to consider in the hypersphere estimation. Defaults to 5.
    metric : str, optional, default="euclidean"
        The metric to use when calculating distance between instances.
        For the available metrics, see the documentation of `sklearn.metrics.pairwise_distances`.
    n_workers : int, optional
        Number of workers for computing pairwise distances. Defaults to 4.
    **kwargs
        Additional keyword arguments for compatiblilty.

    References
    ----------
    Kynkaanniemi et al., Improved Precision and Recall Metric for Assessing Generative Models (2019).
    https://arxiv.org/abs/1904.06991

    Code adapted from:
    improved-precision-and-recall-metric: Improved Precision and Recall Metric for Assessing Generative Models — Official TensorFlow Implementation.
    https://github.com/kynkaat/improved-precision-and-recall-metric

    Hypersphere estimation code was taken from:
    generative-evaluation-prdc, Reliable Fidelity and Diversity Metrics for Generative Models.
    https://github.com/clovaai/generative-evaluation-prdc

    Examples
    --------
    >>> improved_recall = ImprovedRecall()
    >>> real_features = np.random.rand(100, 100)
    >>> fake_features = np.random.rand(100, 100)
    >>> result: MetricResult = improved_recall.compute(real_features, fake_features)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = [EvaluationLevel.INSTANCE, EvaluationLevel.DATASET]
    metric_group = MetricGroup.QUALITY

    higher_is_better: bool = True
    min_value: float = 0.0
    max_value: float = 1.0

    def __init__(
        self,
        k: int = 5,
        metric: str = "euclidean",
        n_workers: int = 4,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.k = k
        self.metric = metric
        self.n_workers = n_workers

    def compute(self, real_features: np.ndarray, fake_features: np.ndarray, **kwargs) -> MetricResult:
        """Compute the Improved Recall metric.

        Parameters
        ----------
        real_features : np.ndarray
            Array of shape (n_samples, n_features) containing the real features.
        fake_features : np.ndarray
            Array of shape (n_samples, n_features) containing the fake features.

        Notes
        -----
        Intermediate computations can be stored in the `context` dictionary of the `kwargs` parameter.
        Usefull when calculating multiple metrics that share the same intermediate computations.

        Returns
        -------
        result: MetricResult
            Dataset-level and instance-level results for the recall metric.
        """
        state = kwargs.get("context", {})
        if "fake_nn_distances" not in state:
            state["fake_nn_distances"] = compute_nearest_neighbour_distances(
                fake_features,
                nearest_k=self.k,
                metric=self.metric,
                n_workers=self.n_workers,
            )

        if "real_fake_distances" not in state:
            state["real_fake_distances"] = compute_pairwise_distance(
                real_features,
                fake_features,
                metric=self.metric,
                n_workers=self.n_workers,
            )

        recall_mask = np.logical_or(
            state["real_fake_distances"] < np.expand_dims(state["fake_nn_distances"], axis=0),
            np.isclose(state["real_fake_distances"], np.expand_dims(state["fake_nn_distances"], axis=0)),
        )
        recall = recall_mask.any(axis=1).astype(int)

        # matrix with (R, F) shape -> .any() -> matrix with (F,) shape
        # an array that indicates for each F sample how many real samples are within its manifold
        recall_counts = recall_mask.sum(axis=0)

        return MetricResult(
            dataset_level={"dtype": OutputsTypes.NUMERIC, "subtype": "float", "value": recall.mean()},
            instance_level={"dtype": OutputsTypes.ARRAY, "subtype": "int", "value": recall_counts.tolist()},
        )

`pymdma.time_series.measures.synthesis_val.Density`

Density Metric for accessing fidelity of the generated samples. Unlike Improved Precision, it is robust towards outliers in the real/reference data.

Objective: Fidelity

Parameters:

Name	Type	Description	Default
`k`	`int`	Number of nearest neighbors to consider in the hypersphere estimation. Defaults to 5.	`5`
`metric`	`str`	The metric to use when calculating distance between instances. For the available metrics, see the documentation of `sklearn.metrics.pairwise_distances`.	`"euclidean"`
`n_workers`	`int`	Number of workers for computing pairwise distances. Defaults to 4.	`4`
`**kwargs`		Additional keyword arguments for compatibility.	`{}`

References

Naeem et al., Reliable Fidelity and Diversity Metrics for Generative Models (2020). https://arxiv.org/abs/2002.09797

Code was adapted from: generative-evaluation-prdc, Reliable Fidelity and Diversity Metrics for Generative Models. https://github.com/clovaai/generative-evaluation-prdc

Examples:

>>> density = Density()
>>> real_features = np.random.rand(100, 100)
>>> fake_features = np.random.rand(100, 100)
>>> result: MetricResult = density.compute(real_features, fake_features)

Source code in src/pymdma/general/measures/prdc.py

class Density(FeatureMetric):
    """Density Metric for accessing fidelity of the generated samples. Unlike
    Improved Precision, it is robust towards outliers in the real/reference
    data.

    **Objective**: Fidelity

    Parameters
    ----------
    k : int, optional
        Number of nearest neighbors to consider in the hypersphere estimation. Defaults to 5.
    metric : str, optional, default="euclidean"
        The metric to use when calculating distance between instances.
        For the available metrics, see the documentation of `sklearn.metrics.pairwise_distances`.
    n_workers : int, optional
        Number of workers for computing pairwise distances. Defaults to 4.
    **kwargs
        Additional keyword arguments for compatibility.

    References
    ----------
    Naeem et al., Reliable Fidelity and Diversity Metrics for Generative Models (2020).
    https://arxiv.org/abs/2002.09797

    Code was adapted from:
    generative-evaluation-prdc, Reliable Fidelity and Diversity Metrics for Generative Models.
    https://github.com/clovaai/generative-evaluation-prdc

    Examples
    --------
    >>> density = Density()
    >>> real_features = np.random.rand(100, 100)
    >>> fake_features = np.random.rand(100, 100)
    >>> result: MetricResult = density.compute(real_features, fake_features)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = [EvaluationLevel.INSTANCE, EvaluationLevel.DATASET]
    metric_group = MetricGroup.QUALITY

    higher_is_better: bool = True
    min_value: float = 0.0
    max_value: float = 1.0

    def __init__(
        self,
        k: int = 5,
        metric: str = "euclidean",
        n_workers: int = 4,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.k = k
        self.metric = metric
        self.n_workers = n_workers

    def compute(self, real_features: np.ndarray, fake_features: np.ndarray, **kwargs) -> MetricResult:
        """Compute the Density metric.

        Parameters
        ----------
        real_features : np.ndarray
            Array of shape (n_samples, n_features) containing the real features.
        fake_features : np.ndarray
            Array of shape (n_samples, n_features) containing the fake features.

        Notes
        -----
        Intermediate computations can be stored in the `context` dictionary of the `kwargs` parameter.
        Usefull when calculating multiple metrics that share the same intermediate computations.

        Returns
        -------
        result: MetricResult
            Dataset-level and instance-level results for the density metric.
        """
        state = kwargs.get("context", {})
        if "real_nn_distances" not in state:
            state["real_nn_distances"] = compute_nearest_neighbour_distances(
                real_features,
                nearest_k=self.k,
                metric=self.metric,
                n_workers=self.n_workers,
            )

        if "real_fake_distances" not in state:
            state["real_fake_distances"] = compute_pairwise_distance(
                real_features,
                fake_features,
                metric=self.metric,
                n_workers=self.n_workers,
            )

        density = np.logical_or(
            (state["real_fake_distances"] < np.expand_dims(state["real_nn_distances"], axis=1)),
            np.isclose(state["real_fake_distances"], np.expand_dims(state["real_nn_distances"], axis=1)),
        )
        density = (1.0 / float(self.k)) * density.sum(axis=0)

        return MetricResult(
            dataset_level={"dtype": OutputsTypes.NUMERIC, "subtype": "float", "value": density.mean()},
            instance_level={"dtype": OutputsTypes.ARRAY, "subtype": "float", "value": density.tolist()},
        )

`pymdma.time_series.measures.synthesis_val.Coverage`

Coverage Metric for accessing diversity of the generated samples. Unlike Improved Recall, it is robust towards outliers in the real/reference data.

Objective: Diversity

Parameters:

Name	Type	Description	Default
`k`	`int`	Number of nearest neighbors to consider in the hypersphere estimation. Defaults to 5.	`5`
`metric`	`str`	The metric to use when calculating distance between instances. For the available metrics, see the documentation of `sklearn.metrics.pairwise_distances`.	`"euclidean"`
`n_workers`	`int`	Number of workers for computing pairwise distances. Defaults to 4.	`4`
`**kwargs`		Additional keyword arguments for compatibility.	`{}`

References

Naeem et al., Reliable Fidelity and Diversity Metrics for Generative Models (2020). https://arxiv.org/abs/2002.09797

Code was adapted from: generative-evaluation-prdc, Reliable Fidelity and Diversity Metrics for Generative Models. https://github.com/clovaai/generative-evaluation-prdc

Examples:

>>> coverage = Coverage()
>>> real_features = np.random.rand(100, 100)
>>> fake_features = np.random.rand(100, 100)
>>> result: MetricResult = coverage.compute(real_features, fake_features)

Source code in src/pymdma/general/measures/prdc.py

class Coverage(FeatureMetric):
    """Coverage Metric for accessing diversity of the generated samples. Unlike
    Improved Recall, it is robust towards outliers in the real/reference data.

    **Objective**: Diversity

    Parameters
    ----------
    k : int, optional
        Number of nearest neighbors to consider in the hypersphere estimation. Defaults to 5.
    metric : str, optional, default="euclidean"
        The metric to use when calculating distance between instances.
        For the available metrics, see the documentation of `sklearn.metrics.pairwise_distances`.
    n_workers : int, optional
        Number of workers for computing pairwise distances. Defaults to 4.
    **kwargs
        Additional keyword arguments for compatibility.

    References
    ----------
    Naeem et al., Reliable Fidelity and Diversity Metrics for Generative Models (2020).
    https://arxiv.org/abs/2002.09797

    Code was adapted from:
    generative-evaluation-prdc, Reliable Fidelity and Diversity Metrics for Generative Models.
    https://github.com/clovaai/generative-evaluation-prdc

    Examples
    --------
    >>> coverage = Coverage()
    >>> real_features = np.random.rand(100, 100)
    >>> fake_features = np.random.rand(100, 100)
    >>> result: MetricResult = coverage.compute(real_features, fake_features)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = [EvaluationLevel.INSTANCE, EvaluationLevel.DATASET]
    metric_group = MetricGroup.QUALITY

    higher_is_better: bool = True
    min_value: float = 0.0
    max_value: float = 1.0

    def __init__(
        self,
        k: int = 5,
        metric: str = "euclidean",
        n_workers: int = 4,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.k = k
        self.metric = metric
        self.n_workers = n_workers

    def compute(self, real_features: np.ndarray, fake_features: np.ndarray, **kwargs) -> MetricResult:
        """Compute the Coverage metric.

        Parameters
        ----------
        real_features : np.ndarray
            Array of shape (n_samples, n_features) containing the real features.
        fake_features : np.ndarray
            Array of shape (n_samples, n_features) containing the fake features.

        Notes
        -----
        Intermediate computations can be stored in the `context` dictionary of the `kwargs` parameter.
        Usefull when calculating multiple metrics that share the same intermediate computations.

        Returns
        -------
        result: MetricResult
            Dataset-level and instance-level results for the coverage metric.
        """
        state = kwargs.get("context", {})
        if "real_nn_distances" not in state:
            state["real_nn_distances"] = compute_nearest_neighbour_distances(
                real_features,
                nearest_k=self.k,
                metric=self.metric,
                n_workers=self.n_workers,
            )

        if "real_fake_distances" not in state:
            state["real_fake_distances"] = compute_pairwise_distance(
                real_features,
                fake_features,
                metric=self.metric,
                n_workers=self.n_workers,
            )

        coverage = np.logical_or(
            state["real_fake_distances"].min(axis=1) < state["real_nn_distances"],
            np.isclose(state["real_fake_distances"].min(axis=1), state["real_nn_distances"]),
        )

        # matrix with (R, F) shape -> .any() -> matrix with (F,) shape
        # an array that indicates for each F in how many real manifolds it is contained in
        coverage_counts = np.logical_or(
            state["real_fake_distances"] < np.expand_dims(state["real_nn_distances"], axis=1),
            np.isclose(state["real_fake_distances"], np.expand_dims(state["real_nn_distances"], axis=1)),
        ).sum(axis=0)

        return MetricResult(
            dataset_level={"dtype": OutputsTypes.NUMERIC, "subtype": "float", "value": coverage.mean()},
            instance_level={"dtype": OutputsTypes.ARRAY, "subtype": "int", "value": coverage_counts.tolist()},
        )

`pymdma.time_series.measures.synthesis_val.FrechetDistance`

Frechet Distance (FD) metric wrapper from the PIQ implementation of FID. Allows the computation of the dispersion and distance ratios for the metric.

Objective: Fidelity, Diversity

Parameters:

Name	Type	Description	Default
`compute_ratios`	`bool`	If set to True, the dispersion and distance ratios will be computed. Defaults to True.	`True`
`**kwargs`	`dict`	Additional keyword arguments for compatibility (unused).	`{}`

Notes

This implementation is based on the PIQ library. The base extractor model is InceptionV3, but this implementation allows for the use of other embedding models (useful when the synthetic data is not compatible with Inception models).

`pymdma.time_series.measures.synthesis_val.GeometryScore`

Geometry Score (GS) metric wrapper from the PIQ implementation of GS.

Objective: Fidelity, Diversity

Parameters:

Name	Type	Description	Default
`sample_size`	`int`	Number of samples to use for the GS computation. Defaults to 128.	`128`
`num_iters`	`int`	Number of iterations to use for the GS computation. Defaults to 1000.	`1000`
`gamma`	`float`	Gamma parameter for the GS computation. Defaults to None.	`None`
`i_max`	`int`	Maximum number of iterations for the GS computation. Defaults to 10.	`10`
`num_workers`	`int`	Number of workers to use for the GS computation. Defaults to 4.	`4`
`**kwargs`	`dict`	Additional keyword arguments for compatibility (unused).	`{}`

References

Kastryulin et al., PyTorch Image Quality: Metrics for Image Quality Assessment (2022). https://arxiv.org/abs/2208.14818

piq, PyTorch Image Quality: Metrics and Measure for Image Quality Assessment, https://github.com/photosynthesis-team/piq

Examples:

>>> gs = GeometryScore()
>>> x_feats = np.random.rand(100, 100)
>>> y_feats = np.random.rand(100, 100)
>>> result: MetricResult = gs.compute(x_feats, y_feats)

Source code in src/pymdma/general/measures/external/piq.py

class GeometryScore(FeatureMetric):
    """Geometry Score (GS) metric wrapper from the PIQ implementation of GS.

    **Objective**: Fidelity, Diversity

    Parameters
    ----------
    sample_size : int, optional
        Number of samples to use for the GS computation. Defaults to 128.
    num_iters : int, optional
        Number of iterations to use for the GS computation. Defaults to 1000.
    gamma : float, optional
        Gamma parameter for the GS computation. Defaults to None.
    i_max : int, optional
        Maximum number of iterations for the GS computation. Defaults to 10.
    num_workers : int, optional
        Number of workers to use for the GS computation. Defaults to 4.
    **kwargs : dict, optional
        Additional keyword arguments for compatibility (unused).

    References
    ----------
    Kastryulin et al., PyTorch Image Quality: Metrics for Image Quality Assessment (2022).
    https://arxiv.org/abs/2208.14818

    piq, PyTorch Image Quality: Metrics and Measure for Image Quality Assessment,
    https://github.com/photosynthesis-team/piq

    Examples
    --------
    >>> gs = GeometryScore()
    >>> x_feats = np.random.rand(100, 100)
    >>> y_feats = np.random.rand(100, 100)
    >>> result: MetricResult = gs.compute(x_feats, y_feats)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = EvaluationLevel.DATASET
    metric_group = MetricGroup.QUALITY

    higher_is_better: bool = False
    min_value: float = 0.0
    max_value: float = np.inf

    def __init__(
        self,
        sample_size: int = 128,
        num_iters: int = 1000,
        gamma: Optional[float] = None,
        i_max: int = 10,
        num_workers: int = 4,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.sample_size = sample_size
        self.num_iters = num_iters
        self.gamma = gamma
        self.i_max = i_max
        self.num_workers = num_workers

        self._gs = _GS(
            sample_size=self.sample_size,
            num_iters=self.num_iters,
            gamma=self.gamma,
            i_max=self.i_max,
            num_workers=self.num_workers,
        )

    def compute(
        self,
        real_features: Union[Tensor, np.ndarray],
        fake_features: Union[Tensor, np.ndarray],
        **kwargs,
    ) -> MetricResult:
        """Compute the Geometry Score metric.

        Parameters
        ----------
        real_features : Union[Tensor, np.ndarray]
            Array-like of shape (n_samples, n_features) containing features of real samples.
        fake_features : Union[Tensor, np.ndarray]
            Array-like of shape (n_samples, n_features) containing features of fake/generated samples.

        Returns
        -------
        result : MetricResult
            Dataset-level GS score.
        """
        real_features = to_tensor(real_features)
        fake_features = to_tensor(fake_features)

        score = self._gs.compute_metric(real_features, fake_features)

        return MetricResult(
            dataset_level={"dtype": OutputsTypes.NUMERIC, "subtype": "float", "value": score.detach().item()},
        )

`pymdma.time_series.measures.synthesis_val.MultiScaleIntrinsicDistance`

Multi-Scale Intrinsic Distance (MSID) metric wrapper from the PIQ implementation of MSID.

Objective: Fidelity, Diversity

Parameters:

Name	Type	Description	Default
`ts`	`Optional[Tensor]`	Tensor of shape (n_samples, n_features) containing the temperature values. Defaults to None.	`None`
`k_neighbours`	`int`	Number of nearest neighbours to consider. Defaults to 5.	`5`
`m_steps`	`int`	Number of steps for the MSID computation. Defaults to 10.	`10`
`niters`	`int`	Number of iterations for the MSID computation. Defaults to 100.	`100`
`rademacher`	`bool`	Whether to use Rademacher distribution for the MSID computation. Defaults to False. When not active will use standard normal for random vectors in Hutchinson.	`False`
`normalized_laplacian`	`bool`	Whether to normalize the laplacian for the MSID computation. Defaults to True.	`True`
`normalize`	`Literal['empty', 'complete', 'er', 'none']`	Normalization strategy for the laplacian. Defaults to "empty".	`'empty'`
`msid_mode`	`Literal['l2', 'max']`	Mode for the MSID computation. Defaults to "max".	`'max'`
`**kwargs`	`dict`	Additional keyword arguments for compatibility (unused).	`{}`

Notes

The results of this metric are based on random approximations, so they are not deterministic. In some datasets the results can be unstable. This can be mitigated by increasing the number of iterations with the niters parameter.

References

Kastryulin et al., PyTorch Image Quality: Metrics for Image Quality Assessment (2022). https://arxiv.org/abs/2208.14818

piq, PyTorch Image Quality: Metrics and Measure for Image Quality Assessment, https://github.com/photosynthesis-team/piq

Examples:

>>> msid = MultiScaleIntrinsicDistance()
>>> x_feats = np.random.rand(100, 100)
>>> y_feats = np.random.rand(100, 100)
>>> result: MetricResult = msid.compute(x_feats, y_feats)

Source code in src/pymdma/general/measures/external/piq.py

class MultiScaleIntrinsicDistance(FeatureMetric):
    """Multi-Scale Intrinsic Distance (MSID) metric wrapper from the PIQ
    implementation of MSID.

    **Objective**: Fidelity, Diversity

    Parameters
    ----------
    ts : Optional[Tensor], optional
        Tensor of shape (n_samples, n_features) containing the temperature values. Defaults to None.
    k_neighbours : int, optional
        Number of nearest neighbours to consider. Defaults to 5.
    m_steps : int, optional
        Number of steps for the MSID computation. Defaults to 10.
    niters : int, optional
        Number of iterations for the MSID computation. Defaults to 100.
    rademacher : bool, optional
        Whether to use Rademacher distribution for the MSID computation. Defaults to False.
        When not active will use standard normal for random vectors in Hutchinson.
    normalized_laplacian : bool, optional
        Whether to normalize the laplacian for the MSID computation. Defaults to True.
    normalize : Literal["empty", "complete", "er", "none"], optional
        Normalization strategy for the laplacian. Defaults to "empty".
    msid_mode : Literal["l2", "max"], optional
        Mode for the MSID computation. Defaults to "max".
    **kwargs : dict, optional
        Additional keyword arguments for compatibility (unused).

    Notes
    -----
    The results of this metric are based on random approximations, so they are not deterministic.
    In some datasets the results can be unstable. This can be mitigated by increasing the
    number of iterations with the `niters` parameter.


    References
    ----------
    Kastryulin et al., PyTorch Image Quality: Metrics for Image Quality Assessment (2022).
    https://arxiv.org/abs/2208.14818

    piq, PyTorch Image Quality: Metrics and Measure for Image Quality Assessment,
    https://github.com/photosynthesis-team/piq

    Examples
    --------
    >>> msid = MultiScaleIntrinsicDistance()
    >>> x_feats = np.random.rand(100, 100)
    >>> y_feats = np.random.rand(100, 100)
    >>> result: MetricResult = msid.compute(x_feats, y_feats)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = EvaluationLevel.DATASET
    metric_group = MetricGroup.QUALITY

    higher_is_better: bool = False
    min_value: float = 0.0
    max_value: float = np.inf

    extractor_model_name: str = "inception_fid"

    def __init__(
        self,
        ts: Optional[Union[Tensor, np.ndarray]] = None,
        k_neighbours: int = 5,
        m_steps: int = 10,
        niters: int = 100,
        rademacher: bool = False,
        normalized_laplacian: bool = True,
        normalize: Literal["empty", "complete", "er", "none"] = "empty",
        msid_mode: Literal["l2", "max"] = "max",
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.ts = torch.from_numpy(ts) if isinstance(ts, np.ndarray) else ts
        self.k_neighbours = k_neighbours
        self.m_steps = m_steps
        self.niters = niters
        self.rademacher = rademacher
        self.normalized_laplacian = normalized_laplacian
        self.normalize = normalize
        self.msid_mode = msid_mode

        self._msid = _MSID(
            ts=self.ts,
            k=self.k_neighbours,
            m=self.m_steps,
            niters=self.niters,
            rademacher=self.rademacher,
            normalized_laplacian=self.normalized_laplacian,
            normalize=self.normalize,
            msid_mode=self.msid_mode,
        )

    def compute(
        self,
        real_features: Union[Tensor, np.ndarray],
        fake_features: Union[Tensor, np.ndarray],
        **kwargs,
    ) -> MetricResult:
        """Compute the Multi-Scale Intrinsic Distance metric.

        Parameters
        ----------
        real_features : Union[Tensor, np.ndarray]
            Array-like of shape (n_samples, n_features) containing features of real samples.
        fake_features : Union[Tensor, np.ndarray]
            Array-like of shape (n_samples, n_features) containing features of fake/generated samples.

        Returns
        -------
        result : MetricResult
            Dataset-level MSID score.
        """
        real_features = to_tensor(real_features)
        fake_features = to_tensor(fake_features)
        score = self._msid.compute_metric(real_features, fake_features)
        return MetricResult(
            dataset_level={"dtype": OutputsTypes.NUMERIC, "subtype": "float", "value": score.detach().item()},
        )

`pymdma.time_series.measures.synthesis_val.PrecisionRecallDistribution`

Computes PRD data from sample embeddings and the maximum F_beta scores for the given precision/recall values.

The points from both distributions are mixed and then clustered. This leads to a pair of histograms of discrete distributions over the cluster centers on which the PRD algorithm is executed.

For PRD, it is recommended that number of points in eval_data and ref_data are equal since unbalanced distributions bias the clustering towards the larger dataset. The check for this condition can be performed by setting the enforce_balance flag to True (recommended).

Regarding the maximum F_beta scores, the maximum F_beta score over all pairs of precision/recall values is useful to compress a PRD plot into a single value which correlate with recall. Whereas, the max_f_beta_inv score over all pairs of precision/recall values compresses the PRD plot into a single value that correlates with precision.

Objective: Fidelity, Diversity

Parameters:

Name	Type	Description	Default
`num_clusters`	`int`	Number of cluster centers to fit. The default value is 2.	`2`
`num_angles`	`int`	Number of angles for which to compute PRD. Must be in [3, 1e6]. The default value is 1001.	`1001`
`num_runs`	`int`	Number of independent runs over which to average the PRD data. The default value is 10.	`10`
`beta`	`int`	Beta parameter for F_beta score. Must be positive. The default value is 8.	`8`
`epsilon`	`float`	Small constant to avoid numerical instability caused by division by 0 when precision and recall are close to zero. The default value is 1e-10.	`1e-10`
`compute_stats`	`bool`	If True, F_beta scores for all precision/recall pairs will be computed. If False, F_beta scores computation is skipped. Default is True.	`True`
`**kwargs`	`dict`	Additional keyword arguments for compatibility (unused).	`{}`

References

Sajjadi, Mehdi SM, et al. Assessing generative models via precision and recall (2018). https://proceedings.neurips.cc/paper_files/paper/2018/file/f7696a9b362ac5a51c3dc8f098b73923-Paper.pdf

Code adapted from: https://github.com/vanderschaarlab/evaluating-generative-models/blob/main/metrics/prd_score.py

Examples:

>>> prd = PrecisionRecallDistribution()
>>> x_feats = np.random.rand(64, 48)
>>> y_feats = np.random.rand(64, 48)
>>> result: MetricResult = prd.compute(x_feats, y_feats)

Source code in src/pymdma/general/measures/prd.py

class PrecisionRecallDistribution(FeatureMetric):
    """Computes PRD data from sample embeddings and the maximum F_beta scores
    for the given precision/recall values.

    The points from both distributions are mixed and then clustered. This leads
    to a pair of histograms of discrete distributions over the cluster centers
    on which the PRD algorithm is executed.

    For PRD, it is recommended that number of points in eval_data and ref_data are equal since
    unbalanced distributions bias the clustering towards the larger dataset. The
    check for this condition can be performed by setting the enforce_balance flag to True
    (recommended).

    Regarding the maximum F_beta scores, the maximum F_beta score over all pairs of precision/recall
    values is useful to compress a PRD plot into a single value which correlate with recall.
    Whereas, the max_f_beta_inv score over all pairs of precision/recall values compresses
    the PRD plot into a single value that correlates with precision.

    **Objective**: Fidelity, Diversity

    Parameters
    ----------
    num_clusters: int, optional
        Number of cluster centers to fit. The default value is 2.
    num_angles: int, optional
        Number of angles for which to compute PRD. Must be in [3, 1e6]. The default value is 1001.
    num_runs: int, optional
        Number of independent runs over which to average the PRD data. The default value is 10.
    beta: int, optional
        Beta parameter for F_beta score. Must be positive. The default value is 8.
    epsilon: float, optional
        Small constant to avoid numerical instability caused by division
        by 0 when precision and recall are close to zero. The default value is 1e-10.
    compute_stats: bool, optional
        If True, F_beta scores for all precision/recall pairs will be computed.
        If False, F_beta scores computation is skipped. Default is True.

    **kwargs : dict, optional
        Additional keyword arguments for compatibility (unused).

    References
    ---------
    Sajjadi, Mehdi SM, et al. Assessing generative models via precision and recall (2018).
    https://proceedings.neurips.cc/paper_files/paper/2018/file/f7696a9b362ac5a51c3dc8f098b73923-Paper.pdf

    Code adapted from:
    https://github.com/vanderschaarlab/evaluating-generative-models/blob/main/metrics/prd_score.py


    Examples
    --------
    >>> prd = PrecisionRecallDistribution()
    >>> x_feats = np.random.rand(64, 48)
    >>> y_feats = np.random.rand(64, 48)
    >>> result: MetricResult = prd.compute(x_feats, y_feats)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = EvaluationLevel.DATASET
    metric_group = MetricGroup.QUALITY

    def __init__(
        self,
        num_clusters: int = 2,
        num_angles: int = 1001,
        num_runs: int = 10,
        epsilon: float = 1e-10,
        beta: int = 8,
        compute_stats: bool = True,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.num_clusters = num_clusters
        self.num_angles = num_angles
        self.num_runs = num_runs
        self.epsilon = epsilon
        self.compute_stats = compute_stats

        assert beta > 0, "Given parameter beta must be positive."
        self.beta = beta

    def _prd_from_embedding(self, target: np.ndarray, reference: np.ndarray):
        eval_data = np.array(target, dtype=np.float64)
        ref_data = np.array(reference, dtype=np.float64)
        precisions = []
        recalls = []
        for _ in range(self.num_runs):
            eval_dist, ref_dist = cluster_into_bins(eval_data, ref_data, self.num_clusters)
            precision, recall = _compute_prd(eval_dist, ref_dist, self.num_angles, self.epsilon)
            precisions.append(precision)
            recalls.append(recall)
        precision = np.mean(precisions, axis=0)
        recall = np.mean(recalls, axis=0)
        return precision, recall

    def compute(self, real_features: np.ndarray, fake_features: np.ndarray, **kwargs) -> MetricResult:
        """Computes PRD data from sample embeddings. Using the PRD, the maximum
        F_beta and F_beta_inv score are also computed. These scores for the
        given precision/recall values correlate, respectively, with recall and
        precision.

        Returns
        -------
        MetricResult
            Instance-level PRD values.
            Dataset-leve max_f_beta and max_f_beta_inv scores.
        """
        warning = None
        if len(fake_features) != len(real_features):
            warning = (
                "The number of points in eval_data %d is not equal to the number of "
                "points in ref_data %d. To disable this exception, set enforce_balance "
                "to False (not recommended)." % (len(fake_features), len(real_features))
            )

        precision, recall = self._prd_from_embedding(fake_features, real_features)

        stats = None
        if self.compute_stats:
            max_f_beta = np.max(_prd_to_f_beta(precision, recall, self.beta, self.epsilon))
            max_f_beta_inv = np.max(_prd_to_f_beta(recall, precision, 1 / self.beta, self.epsilon))
            stats = {
                "max_f_beta": max_f_beta,
                "max_f_beta_inv": max_f_beta_inv,
            }

        return MetricResult(
            dataset_level={
                "dtype": OutputsTypes.KEY_ARRAY,
                "subtype": "float",
                "value": {
                    "precision_values": precision,
                    "recall_values": recall,
                },
                "plot_params": {
                    "x_label": "Recall",
                    "y_label": "Precision",
                    "kind": "line",
                    "x_key": "recall_values",
                    "y_key": "precision_values",
                },
                "stats": stats,
            },
            errors=[warning] if warning else None,
        )

`pymdma.time_series.measures.synthesis_val.WassersteinDistance`

Calculate the Wasserstein distance between two sets of samples.

Objective: Fidelity, Diversity

Parameters:

Name	Type	Description	Default
`compute_ratios`	`bool`	If True, the diversity and dispersion ratios will be computed. If False, ratio computation is skipped. Default is True.	`True`
`**kwargs`	`dict`	Additional keyword arguments for compatibility.	`{}`

Examples:

>>> wasserstein_distance = WassersteinDistance()
>>> real_features = np.random.rand(64, 48) # (n_samples, num_features)
>>> fake_features = np.random.rand(64, 48) # (n_samples, num_features)
>>> result: MetricResult = wasserstein_distance.compute(real_features, fake_features)

Source code in src/pymdma/time_series/measures/synthesis_val/feature/distance.py

class WassersteinDistance(FeatureMetric):
    """Calculate the Wasserstein distance between two sets of samples.

    **Objective**: Fidelity, Diversity

    Parameters
    ----------
    compute_ratios : bool, optional, default=True
        If True, the diversity and dispersion ratios will be computed.
        If False, ratio computation is skipped. Default is True.
    **kwargs : dict, optional
        Additional keyword arguments for compatibility.

    Examples
    --------
    >>> wasserstein_distance = WassersteinDistance()
    >>> real_features = np.random.rand(64, 48) # (n_samples, num_features)
    >>> fake_features = np.random.rand(64, 48) # (n_samples, num_features)
    >>> result: MetricResult = wasserstein_distance.compute(real_features, fake_features)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = EvaluationLevel.DATASET
    metric_group = MetricGroup.QUALITY

    higher_is_better: bool = False
    min_value: float = 0.0
    max_value: float = np.inf

    def __init__(
        self,
        compute_ratios: bool = True,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.compute_ratios = compute_ratios

    def _compute_ratios(self, real_features: np.ndarray, fake_features: np.ndarray, **kwargs):
        """Calculate the diversity and dispersion ratios.

        Parameters
        ----------
        real_features : array-like of shape (n_samples, n_features)
            2D array with features of the original samples.
        fake_features : array-like of shape (n_samples, n_features)
            2D array with features of the fake samples.

        Returns
        -------
        dispersion_ratio : float
            The dispersion ratio between the real features distributions and the fake features_distribution.
            Ideal value is 1 (dispersion between fake samples is equal to the dispersion between real samples).
        distance_ratio : float
            The distance ratio between t real features distributions and the fake features_distribution.
            Ideal value is 1 (distance between real and fake samples is equal to the distance between real samples).
        """

        state = kwargs.get("context", {})

        if any(key not in state for key in {"x_split_1", "x_split_2", "y_split_1", "y_split_2"}):
            state["x_split_1"], state["x_split_2"] = features_splitting(real_features, seed=0)
            state["y_split_1"], state["y_split_2"] = features_splitting(fake_features, seed=0)

        return {
            "dispersion_ratio": dispersion_ratio(
                wasserstein,
                state["x_split_1"],
                state["x_split_2"],
                state["y_split_1"],
                state["y_split_2"],
            ),
            "distance_ratio": distance_ratio(
                wasserstein,
                state["x_split_1"],
                state["x_split_2"],
                state["y_split_1"],
                state["y_split_2"],
            ),
        }

    def compute(self, real_features: np.ndarray, fake_features: np.ndarray, **kwargs) -> MetricResult:
        """Calculate the Wasserstein distance between two sets of samples. If
        compute_distance is True, the diversity and dispersion ratios are also
        computed using the wasserstein distance.

        Parameters
        ----------
        real_features : array-like of shape (n_samples, n_features)
            2D array with features of the original samples.
        fake_features : array-like of shape (n_samples, n_features)
            2D array with features of the fake samples.
        **kwargs : dict, optional
        Additional keyword arguments for compatibility (unused).


        Returns
        -------
        MetricResult
            Dataset-level results for the Wasserstein distance.
        """

        wasserstein_distance = wasserstein(real_features, fake_features)

        ratios = None
        if self.compute_ratios:
            ratios = self._compute_ratios(real_features, fake_features, **kwargs)

        return MetricResult(
            dataset_level={
                "dtype": OutputsTypes.NUMERIC,
                "subtype": "float",
                "value": wasserstein_distance,
                "stats": ratios,
            },
        )

`pymdma.time_series.measures.synthesis_val.MMD`

Calculate the Maximum Mean Discrepancy (MMD) using a specified kernel function.

Objective: Fidelity, Diversity

Parameters:

Name	Type	Description	Default
`compute_ratios`	`bool`	If True, the diversity and dispersion ratios will be computed. If False, ratio computation is skipped. Default is True.	`True`
`kernel`	`str`	The kernel function to use for calculating MMD. Options include: 'multi_gaussian', 'additive_chi2', 'chi2', 'linear', 'poly', 'polynomial', 'rbf', 'laplacian', 'sigmoid', 'cosine'	`'linear'`
`**kwargs`	`dict`	Additional keyword arguments for compatibility.	`{}`

Notes

When using gaussian kernel, the number of samples in both datasets must be the same

References

Gretton, A. et al. "A Kernel Method for the Two-Sample Problem" (2006) https://arxiv.org/pdf/0805.2368

Gretton, A. et al, Optimal kernel choice for large-scale two-sample tests. (NIPS'12) https://proceedings.neurips.cc/paper_files/paper/2012/file/dbe272bab69f8e13f14b405e038deb64-Paper.pdf

Examples:

>>> mmd = MMD(kernel = 'linear')
>>> real_features = np.random.rand(64, 48) # (n_samples, num_features)
>>> fake_features = np.random.rand(64, 48) # (n_samples, num_features)
>>> result: MetricResult = mmd.compute(real_features, fake_features)

Source code in src/pymdma/time_series/measures/synthesis_val/feature/distance.py

class MMD(FeatureMetric):
    """Calculate the Maximum Mean Discrepancy (MMD) using a specified kernel
    function.

    **Objective**: Fidelity, Diversity

    Parameters
    ----------
    compute_ratios : bool, optional, default=True
        If True, the diversity and dispersion ratios will be computed.
        If False, ratio computation is skipped. Default is True.
    kernel : str, optional, default='linear'
        The kernel function to use for calculating MMD. Options include:
        'multi_gaussian', 'additive_chi2', 'chi2', 'linear', 'poly', 'polynomial', 'rbf', 'laplacian', 'sigmoid', 'cosine'
    **kwargs : dict, optional
        Additional keyword arguments for compatibility.

    Notes
    -----
    When using gaussian kernel, the number of samples in both datasets must be the same

    References
    ----------
    Gretton, A. et al. "A Kernel Method for the Two-Sample Problem" (2006)
    https://arxiv.org/pdf/0805.2368

    Gretton, A. et al, Optimal kernel choice for large-scale two-sample tests. (NIPS'12)
    https://proceedings.neurips.cc/paper_files/paper/2012/file/dbe272bab69f8e13f14b405e038deb64-Paper.pdf

    Examples
    --------
    >>> mmd = MMD(kernel = 'linear')
    >>> real_features = np.random.rand(64, 48) # (n_samples, num_features)
    >>> fake_features = np.random.rand(64, 48) # (n_samples, num_features)
    >>> result: MetricResult = mmd.compute(real_features, fake_features)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = EvaluationLevel.DATASET
    metric_group = MetricGroup.QUALITY

    higher_is_better: bool = False
    min_value: float = 0.0
    max_value: float = np.inf

    def __init__(
        self,
        kernel: Literal[
            "multi_gaussian",
            "gaussian",
            "additive_chi2",
            "chi2",
            "linear",
            "poly",
            "polynomial",
            "rbf",
            "laplacian",
            "sigmoid",
            "cosine",
        ] = "linear",
        compute_ratios: bool = True,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.kernel = kernel
        self.compute_ratios = compute_ratios

        if kernel == "linear":
            self.kernel_fn = fast_mmd_linear
        elif kernel == "multi_gaussian":
            self.kernel_fn = mk_mmd
        else:
            self.kernel_fn = mmd_kernel

    def _compute_ratios(self, real_features: np.ndarray, fake_features: np.ndarray, **kwargs):
        """Calculate the diversity and dispersion ratios.

        Parameters
        ----------
        real_features : array-like of shape (n_samples, n_features)
            2D array with features of the original samples.
        fake_features : array-like of shape (n_samples, n_features)
            2D array with features of the fake samples.
        **kwargs : dict, optional
        Additional keyword arguments for compatibility (unused).


        Returns
        -------
        dispersion_ratio : float
            The dispersion ratio between the real features distributions and the fake features_distribution.
            Ideal value is 1 (dispersion between fake samples is equal to the dispersion between real samples).
        distance_ratio : float
            The distance ratio between t real features distributions and the fake features_distribution.
            Ideal value is 1 (distance between real and fake samples is equal to the distance between real samples).
        """

        state = kwargs.get("context", {})

        if any(key not in state for key in {"x_split_1", "x_split_2", "y_split_1", "y_split_2"}):
            state["x_split_1"], state["x_split_2"] = features_splitting(real_features, seed=0)
            state["y_split_1"], state["y_split_2"] = features_splitting(fake_features, seed=0)

        return {
            "dispersion_ratio": dispersion_ratio(
                self.kernel_fn,
                state["x_split_1"],
                state["x_split_2"],
                state["y_split_1"],
                state["y_split_2"],
            ),
            "distance_ratio": distance_ratio(
                self.kernel_fn,
                state["x_split_1"],
                state["x_split_2"],
                state["y_split_1"],
                state["y_split_2"],
            ),
        }

    def compute(self, real_features: np.ndarray, fake_features: np.ndarray, **kwargs) -> MetricResult:
        """Calculate the Maximum Mean Discrepancy (MMD) using a specified
        kernel function.

        Parameters
        ----------
        real_features : array-like of shape [n_samples_x, n_features]
            2D array containing features from samples of the original distribution.
        fake_features : array-like of shape [n_samples_y, n_features]
            2D array containing features from samples of the fake distribution.
        **kwargs : dict, optional
        Additional keyword arguments for compatibility (unused).

        Returns
        -------
        MetricResult
                Dataset-level results for the MMD distance.
        """
        assert self.kernel != "gaussian" or len(real_features) == len(
            fake_features,
        ), "Gaussian kernel requires the same number of samples in each dataset."

        mmd = self.kernel_fn(real_features, fake_features, kernel=self.kernel)

        ratios = None
        if self.compute_ratios:
            ratios = self._compute_ratios(real_features, fake_features, **kwargs)

        return MetricResult(
            dataset_level={
                "dtype": OutputsTypes.NUMERIC,
                "subtype": "float",
                "value": mmd,
                "stats": ratios,
            },
        )

`pymdma.time_series.measures.synthesis_val.CosineSimilarity`

Calculate the cosine similarity between two sets of feature vectors.

Objective: Fidelity, Diversity

Parameters:

Name	Type	Description	Default
`**kwargs`	`dict`	Additional keyword arguments for compatibility.	`{}`

References

Manning, C. D., Raghavan, P., & Schütze, H., An Introduction to Information Retrieval (2008). https://www.cambridge.org/highereducation/books/introduction-to-information-retrieval/669D108D20F556C5C30957D63B5AB65C#overview

Examples:

>>> cosine_sim = CosineSimilarity()
>>> real_features = np.random.rand(64, 48) # (n_samples, num_features)
>>> fake_features = np.random.rand(64, 48) # (n_samples, num_features)
>>> result: MetricResult = cosine_sim.compute(real_features, fake_features)

Source code in src/pymdma/time_series/measures/synthesis_val/feature/distance.py

class CosineSimilarity(FeatureMetric):
    """Calculate the cosine similarity between two sets of feature vectors.

    **Objective**: Fidelity, Diversity

    Parameters
    ----------
    **kwargs : dict, optional
        Additional keyword arguments for compatibility.

    References
    ----------
    Manning, C. D., Raghavan, P., & Schütze, H., An Introduction to Information Retrieval (2008).
    https://www.cambridge.org/highereducation/books/introduction-to-information-retrieval/669D108D20F556C5C30957D63B5AB65C#overview


    Examples
    --------
    >>> cosine_sim = CosineSimilarity()
    >>> real_features = np.random.rand(64, 48) # (n_samples, num_features)
    >>> fake_features = np.random.rand(64, 48) # (n_samples, num_features)
    >>> result: MetricResult = cosine_sim.compute(real_features, fake_features)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = EvaluationLevel.DATASET
    metric_group = MetricGroup.QUALITY

    higher_is_better: bool = False
    min_value: float = 0.0
    max_value: float = np.inf

    def __init__(
        self,
        **kwargs,
    ):
        super().__init__(**kwargs)

    def compute(self, real_features: np.ndarray, fake_features: np.ndarray, **kwargs) -> MetricResult:
        """Calculate the cosine similarity between two sets of feature vectors.

        Parameters
        ----------
        real_features : array-like of shape [n_samples_x, n_features]
            2D array with features of the original samples.
        fake_features : array-like of shape [n_samples_y, n_features]
            2D array with features of the fake samples.

        Returns
        -------
        MetricResult
            Dataset-level results for the cosine similarity.
        """

        return MetricResult(
            dataset_level={
                "dtype": OutputsTypes.NUMERIC,
                "subtype": "float",
                "value": cos_sim_2d(real_features, fake_features),
            },
        )

Privacy

`pymdma.time_series.measures.synthesis_val.Authenticity`

Authenticity Metric for assessing the authenticity of the generated samples. A synthetic sample is considered authentic if it is signficantly distinct from any real sample.

Objective: Privacy

Parameters:

Name	Type	Description	Default
`metric`	`str`	The metric to use when calculating distance between instances. For the available metrics, see the documentation of `sklearn.metrics.pairwise_distances`.	`"euclidean"`
`**kwargs`		Additional keyword arguments for compatibility.	`{}`

Notes

The authenticity metric is computed by checking if any fake sample is closer to a real sample than the real sample is to any other real sample.

References

Alaa et al., How Faithful Is Your Synthetic Data? Sample-Level Metrics for Evaluating and Auditing Generative Models. (2022) https://doi.org/10.48550/arXiv.2102.08921.

Hypersphere estimation code was adapted from: generative-evaluation-prdc, Reliable Fidelity and Diversity Metrics for Generative Models. https://github.com/clovaai/generative-evaluation-prdc

Examples:

>>> authenticity = Authenticity()
>>> real_features = np.random.rand(100, 100)
>>> fake_features = np.random.rand(100, 100)
>>> result: MetricResult = authenticity.compute(real_features, fake_features)

Source code in src/pymdma/general/measures/prdc.py

class Authenticity(FeatureMetric):
    """Authenticity Metric for assessing the authenticity of the generated
    samples. A synthetic sample is considered authentic if it is signficantly
    distinct from any real sample.

    **Objective**: Privacy

    Parameters
    ----------
    metric : str, optional, default="euclidean"
        The metric to use when calculating distance between instances.
        For the available metrics, see the documentation of `sklearn.metrics.pairwise_distances`.
    **kwargs
        Additional keyword arguments for compatibility.

    Notes
    -----
    The authenticity metric is computed by checking if any fake sample is closer to a real sample than the real sample is to any other real sample.

    References
    ----------
    Alaa et al., How Faithful Is Your Synthetic Data? Sample-Level Metrics for Evaluating and Auditing Generative Models. (2022)
    https://doi.org/10.48550/arXiv.2102.08921.

    Hypersphere estimation code was adapted from:
    generative-evaluation-prdc, Reliable Fidelity and Diversity Metrics for Generative Models.
    https://github.com/clovaai/generative-evaluation-prdc

    Examples
    --------
    >>> authenticity = Authenticity()
    >>> real_features = np.random.rand(100, 100)
    >>> fake_features = np.random.rand(100, 100)
    >>> result: MetricResult = authenticity.compute(real_features, fake_features)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = [EvaluationLevel.INSTANCE, EvaluationLevel.DATASET]
    metric_group = MetricGroup.PRIVACY

    higher_is_better: bool = True
    min_value: float = 0.0
    max_value: float = 1.0

    def __init__(
        self,
        metric: str = "euclidean",
        n_workers: int = 4,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.metric = metric
        self.n_workers = n_workers

    def compute(self, real_features: np.ndarray, fake_features: np.ndarray, **kwargs) -> MetricResult:
        """Compute the Authenticity metric.

        Parameters
        ----------
        real_features : np.ndarray
            Array of shape (n_samples, n_features) containing the real features.
        fake_features : np.ndarray
            Array of shape (n_samples, n_features) containing the fake features.

        Notes
        -----
        Intermediate computations can be stored in the `context` dictionary of the `kwargs` parameter.
        Usefull when calculating multiple metrics that share the same intermediate computations.

        Returns
        -------
        result: MetricResult
            Dataset-level and instance-level results for the authenticity metric
        """
        state = kwargs.get("context", {})

        if "real_fake_distances" not in state:
            state["real_fake_distances"] = compute_pairwise_distance(
                real_features,
                fake_features,
                metric=self.metric,
                n_workers=self.n_workers,
            )

        # compute distance to closest real samples
        state["real_closest_real_distances"] = compute_nearest_neighbour_distances(
            real_features,
            nearest_k=1,
            metric=self.metric,
            n_workers=self.n_workers,
        )

        # check if any fake sample is closer to Ri than Ri is to any other Rj
        authenticity = np.logical_or(
            state["real_fake_distances"] < np.expand_dims(state["real_closest_real_distances"], axis=1),
            np.isclose(state["real_fake_distances"], np.expand_dims(state["real_closest_real_distances"], axis=1)),
        )

        # mask of the values that are considered authentic in the fake dataset
        authenticity_mask = ~authenticity.any(axis=0)

        return MetricResult(
            dataset_level={"dtype": OutputsTypes.NUMERIC, "subtype": "float", "value": authenticity_mask.mean()},
            instance_level={
                "dtype": OutputsTypes.ARRAY,
                "subtype": "int",
                "value": authenticity_mask.astype(int).tolist(),
            },
        )

Data-based

Quality

`pymdma.time_series.measures.synthesis_val.DTW`

Computes the Dynamic Time Warping (DTW) distance between two sets of time-series signals, evaluating the similarity between corresponding channels in the target and reference signals. The DTW distance is computed by comparing each target signal with every reference signal, with lower DTW values indicating greater similarity. For each signal pair, the DTW distance is calculated across all channels, with the mean of the distances being taken across all channels and instances. This process yields both instance-level and dataset-level DTW metrics.

Objective: Fidelity, Diversity

Parameters:

Name	Type	Description	Default
`**kwargs`	`dict`	Additional keyword arguments for compatibility with the Metric framework.	`{}`

References

Salvador, S., & Chan, P., FastDTW: Toward Accurate Dynamic Time (2004). https://dl.acm.org/doi/10.5555/1367985.1367993

Examples:

>>> dtw = DTW()
>>> reference_sigs = np.random.rand(64, 1000, 12) # (N, L, C)
>>> target_imgs = np.random.rand(64, 1000, 12) # (N, L, C)
>>> result: MetricResult = dtw.compute(reference_sigs, target_sigs)

Source code in src/pymdma/time_series/measures/synthesis_val/data/reference.py

class DTW(Metric):
    """Computes the Dynamic Time Warping (DTW) distance between two sets of
    time-series signals, evaluating the similarity between corresponding
    channels in the target and reference signals. The DTW distance is computed
    by comparing each target signal with every reference signal, with lower DTW
    values indicating greater similarity. For each signal pair, the DTW
    distance is calculated across all channels, with the mean of the distances
    being taken across all channels and instances. This process yields both
    instance-level and dataset-level DTW metrics.

    **Objective**: Fidelity, Diversity

    Parameters
    ----------
    **kwargs : dict, optional
        Additional keyword arguments for compatibility with the Metric framework.

    References
    ----------
    Salvador, S., & Chan, P., FastDTW: Toward Accurate Dynamic Time (2004).
    https://dl.acm.org/doi/10.5555/1367985.1367993

    Examples
    --------
    >>> dtw = DTW()
    >>> reference_sigs = np.random.rand(64, 1000, 12) # (N, L, C)
    >>> target_imgs = np.random.rand(64, 1000, 12) # (N, L, C)
    >>> result: MetricResult = dtw.compute(reference_sigs, target_sigs)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = [EvaluationLevel.INSTANCE, EvaluationLevel.DATASET]
    metric_group = MetricGroup.QUALITY

    higher_is_better: bool = False
    min_value: float = 0.0
    max_value: float = np.inf

    def __init__(
        self,
        **kwargs,
    ):
        super().__init__(**kwargs)

    def compute(
        self,
        reference_sigs: List[np.ndarray],
        target_sigs: List[np.ndarray],
        **kwargs,
    ) -> MetricResult:
        """Computes Dinamic Time Wrapping.

        Parameters
        ----------
        reference_sigs: (N, L, C) ndarray
            Signals to use as reference.
            List of arrays representing a signal of shape (L, C).
        target_sigs : (N, L, C) ndarray
            Signals compare with reference.
            List of arrays representing a signal of shape (L, C).

        Returns
        -------
        result : MetricResult
            Instance-level dtw.
            Dataset-level dtw.
        """
        instance_dtw = []
        for targ in target_sigs:
            for ref in reference_sigs:
                dtw_values_by_chan = []

                # Compute the dtw for each channel
                for targ_channel, ref_channel in zip(targ.T, ref.T):
                    channel_dtw, _ = fastdtw(targ_channel, ref_channel)

                    dtw_values_by_chan.append(channel_dtw)

                # Compute mean dtw across channels
                mean_dtw = np.mean(dtw_values_by_chan)

            # Correlations by signal
            instance_dtw.append(mean_dtw)

        # Average correlation across signals
        final_dtw = np.mean(instance_dtw)

        return MetricResult(
            instance_level={"dtype": OutputsTypes.ARRAY, "subtype": "float", "value": instance_dtw},
            dataset_level={"dtype": OutputsTypes.NUMERIC, "subtype": "float", "value": final_dtw},
        )

`pymdma.time_series.measures.synthesis_val.CrossCorrelation`

Computes the Cross-Correlation between two sets of signals.

This function calculates the cross-correlation to analyze the relationship between corresponding channels in the target and reference signals. The computation is performed for each signal in the target set against every signal in the reference set, using a specified overlap mode. The computed cross-correlation for each channel can be summarized using one of two reduction methods: 'mean'and 'max'.

For each signal pair, the function calculates the cross-correlation values for each channel using the specified reduction method. It then computes the mean of these values across all channels to provide an instance-level metric and averages these results across all signal pairs to obtain the dataset-level metric.

Objective: Fidelity, Diversity

Parameters:

Name	Type	Description	Default
`mode`	`(full, same, valid)`	Defines how the cross-correlation is computed. Default is 'full'. - 'full': Computes the convolution at every point of overlap, producing an output of size (N + M - 1). Boundary effects may be present. - 'same': Produces an output of length max(M, N), centered on the signals. - 'valid': Produces an output of length max(M, N) - min(M, N) + 1, considering only complete overlaps.	`'full'`
`reduction`	`(mean, max)`	Determines how the cross-correlation is summarized for each channel. Default is 'max'. - 'mean': The average of the cross-correlation values for the channel. - 'max': The maximum cross-correlation value for the channel.	`'mean'`
`**kwargs`	`dict`	Additional keyword arguments for customization.	`{}`

References

Proakis and Manolakis, Digital Signal Processing: Principles, Algorithms, and Applications (1996). https://dl.acm.org/doi/10.5555/227373

Examples:

>>> cc = CrossCorrelation()
>>> reference_sigs = np.random.rand(64, 1000, 12) # (N, L, C)
>>> target_sigs = np.random.rand(64, 1000, 12) # (N, L, C)
>>> result: MetricResult = cc.compute(reference_sigs, target_sigs)

Source code in src/pymdma/time_series/measures/synthesis_val/data/reference.py

class CrossCorrelation(Metric):
    """Computes the Cross-Correlation between two sets of signals.

    This function calculates the cross-correlation to analyze the relationship between
    corresponding channels in the target and reference signals. The computation is performed
    for each signal in the target set against every signal in the reference set, using a
    specified overlap mode. The computed cross-correlation for each channel can be summarized
    using one of two reduction methods: 'mean'and 'max'.

    For each signal pair, the function calculates the cross-correlation values for each channel
    using the specified reduction method. It then computes the mean of these values across all
    channels to provide an instance-level metric and averages these results across all signal
    pairs to obtain the dataset-level metric.

    **Objective**: Fidelity, Diversity

    Parameters
    ----------
    mode : {'full', 'same', 'valid'}, optional
        Defines how the cross-correlation is computed. Default is 'full'.
        - 'full': Computes the convolution at every point of overlap, producing
          an output of size (N + M - 1). Boundary effects may be present.
        - 'same': Produces an output of length max(M, N), centered on the signals.
        - 'valid': Produces an output of length max(M, N) - min(M, N) + 1,
          considering only complete overlaps.
    reduction : {'mean', 'max'}, optional
        Determines how the cross-correlation is summarized for each channel. Default is 'max'.
        - 'mean': The average of the cross-correlation values for the channel.
        - 'max': The maximum cross-correlation value for the channel.
    **kwargs : dict, optional
        Additional keyword arguments for customization.

    References
    ----------
    Proakis and Manolakis, Digital Signal Processing: Principles, Algorithms, and Applications (1996).
    https://dl.acm.org/doi/10.5555/227373

    Examples
    --------
    >>> cc = CrossCorrelation()
    >>> reference_sigs = np.random.rand(64, 1000, 12) # (N, L, C)
    >>> target_sigs = np.random.rand(64, 1000, 12) # (N, L, C)
    >>> result: MetricResult = cc.compute(reference_sigs, target_sigs)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = [EvaluationLevel.INSTANCE, EvaluationLevel.DATASET]
    metric_group = MetricGroup.QUALITY

    higher_is_better: bool = True
    min_value: float = -np.inf
    max_value: float = np.inf

    def __init__(
        self,
        mode: Literal["full", "same", "valid"] = "full",
        reduction: Literal["mean", "max"] = "max",
        **kwargs,
    ):
        super().__init__(**kwargs)
        assert mode in ["full", "same", "valid"], f"Unsupported mode for Cross Correlation: {mode}"
        self.mode = mode
        assert reduction in ["mean", "max"], f"Unsupported criteria for relative tenengrad: {reduction}"
        self.reduction = reduction

    def compute(
        self,
        reference_sigs: List[np.ndarray],
        target_sigs: List[np.ndarray],
        **kwargs,
    ) -> MetricResult:
        """Computes Dinamic Time Wrapping.

        Parameters
        ----------
        reference_sigs: (N, L1, C) ndarray
            Signals to use as reference.
            List of arrays representing a signal of shape (L1, C).
        target_sigs :(N, L2, C) ndarray
            Signals compare with reference.
            List of arrays representing a signal of shape (L2, C).

        Returns
        -------
        result : MetricResult
            Instance-level maximum cross-correlation.
            Dataset-level maximum cross-correlation.
        """
        instance_cross_corr = []

        for targ in target_sigs:
            for ref in reference_sigs:
                reduct_corr_values_by_chan = []

                # Compute the cross-correlation for each channel with different reductions
                for targ_channel, ref_channel in zip(targ.T, ref.T):
                    cross_corr = np.correlate(targ_channel, ref_channel, mode=self.mode)

                    if self.reduction == "max":
                        max_corr_idx = np.argmax(cross_corr)
                        max_corr_value = cross_corr[max_corr_idx]
                        reduct_corr_values_by_chan.append(max_corr_value)

                    else:
                        mean_corr_value = np.mean(cross_corr)
                        reduct_corr_values_by_chan.append(mean_corr_value)

                # Compute the mean of the correlation across channels
                cross_corr = np.mean(reduct_corr_values_by_chan)

            # Correlations by signal
            instance_cross_corr.append(cross_corr)

        # Average correlation across signals
        final_cross_corr = np.mean(instance_cross_corr)

        return MetricResult(
            instance_level={"dtype": OutputsTypes.ARRAY, "subtype": "float", "value": instance_cross_corr},
            dataset_level={"dtype": OutputsTypes.NUMERIC, "subtype": "float", "value": final_cross_corr},
        )

`pymdma.time_series.measures.synthesis_val.SpectralCoherence`

Compute the mean coherence between real and synthetic signals. Returns the average (median-based) coherence across signals.

Objective: Similarity

Parameters:

Name	Type	Description	Default
`fs`	`int`	The sampling frequency of the signal.	`2048`
`nperseg`	`Optional[int]`	Length of each segment used to compute the power spectral density.	`None`
`valid_freq_range`	`tuple`	The range of valid frequencies for the signal.	`(-inf, inf)`
`**kwargs`	`dict`	Additional keyword arguments for compatibility.	`{}`

References

F. P. Carrle, Y. Hollenbenders, and A. Reichenbach, Generation of synthetic EEG data for training algorithms supporting the diagnosis of major depressive disorder (2023). https://www.frontiersin.org/journals/neuroscience/articles/10.3389/fnins.2023.1219133/full.

J. Vetter, J. H. Macke, and R. Gao, Generating realistic neurophysiological time series with denoising diffusion probabilistic models (2024) https://pmc.ncbi.nlm.nih.gov/articles/PMC11573898/

A. Zancanaro, I. Zoppis, S. Manzoni, and G. Cisotto, vEEGNet: A New Deep Learning Model to Classify and Generate EEG (2023) https://www.scitepress.org/DigitalLibrary/Link.aspx?doi=10.5220/0011990800003476.

Examples:

>>> coeherence = SpectralCoherence()
>>> real_data = np.random.rand(64, 1000, 12) # (N, L, C)
>>> fake_data = np.random.rand(64, 1000, 12) # (N, L, C)
>>> result: MetricResult = coherence.compute(real_data, fake_data)

Source code in src/pymdma/time_series/measures/synthesis_val/data/freq_sim.py

class SpectralCoherence(Metric):
    """Compute the mean coherence between real and synthetic signals. Returns
    the average (median-based) coherence across signals.

    **Objective**: Similarity

    Parameters
    ----------
    fs: int, optional, default=2048
        The sampling frequency of the signal.
    nperseg: int, optional, default=None
        Length of each segment used to compute the power spectral density.
    valid_freq_range: tuple, optional, default=(-np.inf, np.inf)
        The range of valid frequencies for the signal.
    **kwargs : dict, optional
        Additional keyword arguments for compatibility.

    References
    ----------
    F. P. Carrle, Y. Hollenbenders, and A. Reichenbach, Generation of synthetic EEG data for training algorithms supporting the diagnosis of major depressive disorder (2023).
    https://www.frontiersin.org/journals/neuroscience/articles/10.3389/fnins.2023.1219133/full.

    J. Vetter, J. H. Macke, and R. Gao, Generating realistic neurophysiological time series with denoising diffusion probabilistic models (2024)
    https://pmc.ncbi.nlm.nih.gov/articles/PMC11573898/

    A. Zancanaro, I. Zoppis, S. Manzoni, and G. Cisotto, vEEGNet: A New Deep Learning Model to Classify and Generate EEG (2023)
    https://www.scitepress.org/DigitalLibrary/Link.aspx?doi=10.5220/0011990800003476.

    Examples
    --------
    >>> coeherence = SpectralCoherence()
    >>> real_data = np.random.rand(64, 1000, 12) # (N, L, C)
    >>> fake_data = np.random.rand(64, 1000, 12) # (N, L, C)
    >>> result: MetricResult = coherence.compute(real_data, fake_data)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = EvaluationLevel.DATASET
    metric_group = MetricGroup.QUALITY

    higher_is_better: bool = False
    min_value: float = 0.0
    max_value: float = np.inf

    def __init__(
        self,
        fs: int = 2048,
        nperseg: Optional[int] = None,
        valid_freq_range: tuple = (-np.inf, np.inf),
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.fs = fs
        self.valid_freq_range = valid_freq_range
        self.nperseg = nperseg

    def _preprocess(self, data: Union[np.ndarray, List[np.ndarray]]):
        """Preprocess data for computation.

        Parameters
        ----------
        data : Union[np.ndarray, List[np.ndarray]]
            Input data, either a 1D, 2D, or 3D array.

        Returns
        -------
        data : np.ndarray
            Preprocessed data, a 2D array.
        """
        data = np.array(data)
        # 1D signals need to be converted to 2D
        if data.ndim == 1:
            data = np.expand_dims(data, axis=1)
        elif data.ndim > 2:
            # Convert to 2D
            data = data.reshape(data.shape[0], -1)
        return data

    def compute(self, real_data: np.ndarray, syn_data: np.ndarray, **kwargs) -> MetricResult:
        """Compute the metric.

        Parameters
        ----------
        real_features : array-like of shape (N, L) or (N, L, C)
            Array with the original samples.
        fake_features : array-like of shape (N, L) or (N, L, C)
            Array with the fake samples.
        **kwargs : dict, optional
            Additional keyword arguments for compatibility (unused).

        Returns
        -------
        MetricResult
            Dataset-level results for the Spectral Coherence.
        """

        coherence_values = []

        real_data = self._preprocess(real_data)
        syn_data = self._preprocess(syn_data)

        # Pair up signals one-to-one
        for real_sig, synth_sig in zip(real_data, syn_data):
            win = min(len(real_sig) // 8, self.fs // 2) if self.nperseg is None else self.nperseg
            f, Cxy = coherence(real_sig, synth_sig, fs=self.fs, nperseg=win, window="boxcar")
            valid_freqs = (f >= self.valid_freq_range[0]) & (f <= self.valid_freq_range[1])
            coherence_values.append(np.median(Cxy[valid_freqs]))

        return MetricResult(
            dataset_level={
                "dtype": OutputsTypes.NUMERIC,
                "subtype": "float",
                "value": np.mean(coherence_values),
            },
        )

`pymdma.time_series.measures.synthesis_val.SpectralWassersteinDistance`

Compute a Wasserstein distance in the frequency domain based on Power Spectral Density (PSD) differences.

Objective: Similarity

Parameters:

Name	Type	Description	Default
`fs`	`int`	The sampling frequency of the signal.	`2048`
`nperseg`	`Optional[int]`	Length of each segment used to compute the power spectral density.	`None`
`**kwargs`	`dict`	Additional keyword arguments for compatibility.	`{}`

References

F. P. Carrle, Y. Hollenbenders, and A. Reichenbach, Generation of synthetic EEG data for training algorithms supporting the diagnosis of major depressive disorder (2023). https://www.frontiersin.org/journals/neuroscience/articles/10.3389/fnins.2023.1219133/full.

Examples:

>>> spectral_wd = SpectralWassersteinDistance()
>>> real_data = np.random.rand(64, 1000, 12) # (N, L, C)
>>> fake_data = np.random.rand(64, 1000, 12) # (N, L, C)
>>> result: MetricResult = spectral_wd.compute(real_data, fake_data)

Source code in src/pymdma/time_series/measures/synthesis_val/data/freq_sim.py

class SpectralWassersteinDistance(Metric):
    """Compute a Wasserstein distance in the frequency domain based on Power
    Spectral Density (PSD) differences.

    **Objective**: Similarity

    Parameters
    ----------
    fs: int, optional, default=2048
        The sampling frequency of the signal.
    nperseg: int, optional, default=None
        Length of each segment used to compute the power spectral density.
    **kwargs : dict, optional
        Additional keyword arguments for compatibility.

    References
    ----------
    F. P. Carrle, Y. Hollenbenders, and A. Reichenbach, Generation of synthetic EEG data for training algorithms supporting the diagnosis of major depressive disorder (2023).
    https://www.frontiersin.org/journals/neuroscience/articles/10.3389/fnins.2023.1219133/full.

    Examples
    --------
    >>> spectral_wd = SpectralWassersteinDistance()
    >>> real_data = np.random.rand(64, 1000, 12) # (N, L, C)
    >>> fake_data = np.random.rand(64, 1000, 12) # (N, L, C)
    >>> result: MetricResult = spectral_wd.compute(real_data, fake_data)
    """

    reference_type = ReferenceType.DATASET
    evaluation_level = EvaluationLevel.DATASET
    metric_group = MetricGroup.QUALITY

    higher_is_better: bool = False
    min_value: float = 0.0
    max_value: float = np.inf

    def __init__(
        self,
        fs: int = 2048,
        nperseg: Optional[int] = None,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.fs = fs
        self.nperseg = nperseg

    def _compute_power_spectral_density(self, data) -> np.ndarray:
        """Compute the power spectral density of the given data.

        Parameters
        ----------
        data: array-like of shape (N,) or (N, L) or (N, L, C)
            The input data.

        Returns
        -------
        psd: array-like of shape (N, L) or (N, L, C)
            The power spectral density of the input data.
        """
        if isinstance(data, np.ndarray) and data.ndim == 1:
            data = [data]

        psd = []
        for sig in data:
            # Use a segment length (nperseg) to compute PSD:
            win = min(len(sig), 4 * self.fs) if self.nperseg is None else self.nperseg
            psd.append(welch(sig, self.fs, nperseg=win, window="boxcar")[1])
        return np.array(psd)

    def compute(self, real_data: np.ndarray, syn_data: np.ndarray, **kwargs) -> MetricResult:
        """Compute the metric.

        Parameters
        ----------
        real_features : array-like of shape (N, L) or (N, L, C)
            Array with the original samples.
        fake_features : array-like of shape (N, L) or (N, L, C)
            Array with the fake samples.
        **kwargs : dict, optional
            Additional keyword arguments for compatibility (unused).

        Returns
        -------
        MetricResult
            Dataset-level results for the Spectral Coherence.
        """
        real_psd = self._compute_power_spectral_density(real_data)
        syn_psd = self._compute_power_spectral_density(syn_data)

        # Align number of frequency bins
        min_len = min(real_psd.shape[1], syn_psd.shape[1])
        real_psd = real_psd[:, :min_len]
        syn_psd = syn_psd[:, :min_len]

        real_psd = real_psd.reshape(-1, min_len)
        syn_psd = syn_psd.reshape(-1, min_len)

        # Cost matrix in PSD space (Euclidean)
        cost_matrix_psd = ot.dist(real_psd, syn_psd, metric="euclidean")
        # Uniform distributions
        a = np.ones(real_psd.shape[0]) / real_psd.shape[0]
        b = np.ones(syn_psd.shape[0]) / syn_psd.shape[0]

        # Sinkhorn regularized Wasserstein distance
        wd_psd = ot.sinkhorn2(a, b, cost_matrix_psd, reg=0.01)

        return MetricResult(
            dataset_level={
                "dtype": OutputsTypes.NUMERIC,
                "subtype": "float",
                "value": wd_psd,
            },
        )