<commit_before># Released under The MIT License (MIT)
# http://opensource.org/licenses/MIT
# Copyright (c) 2013 SCoT Development Team

""" Connectivity Analysis """

import numpy as np
from .utils import memoize


def connectivity(measure_names, b, c=None, nfft=512):
    """ calculate connectivity measures.

    Parameters
    ----------
    measure_names : {str, list of str}
        Name(s) of the connectivity measure(s) to calculate. See :class:`Connectivity` for supported measures.
    b : ndarray, shape = [n_channels, n_channels*model_order]
        VAR model coefficients. See :ref:`var-model-coefficients` for details about the arrangement of coefficients.
    c : ndarray, shape = [n_channels, n_channels], optional
        Covariance matrix of the driving noise process. Identity matrix is used if set to None.
    nfft : int, optional
        Number of frequency bins to calculate. Note that these points cover the range between 0 and half the
        sampling rate.

    Returns
    -------
    result : ndarray, shape = [n_channels, n_channels, `nfft`]
        An ndarray of shape [m, m, nfft] is returned if measures is a string. If measures is a list of strings a
        dictionary is returned, where each key is the name of the measure, and the corresponding values are ndarrays
        of shape [m, m, nfft].

    Notes
    -----
    When using this function it is more efficient to get several measures at once than calling the function multiple times.

    Examples
    --------
    >>> c = connectivity(['DTF', 'PDC'], [[0.3, 0.6], [0.0, 0.9]])
    """
    con = Connectivity(b, c, nfft)
    try:
        return getattr(con, measure_names)()
    except TypeError:
        return {m: getattr(con, m)() for m in measure_names}


#noinspection PyPep8Naming
class Connectivity:
    """ Calculation of connectivity measures
    
    This class calculates various spectral connectivity measures from a vector autoregressive (VAR) model.

    Parameters
    ----------
    b : ndarray, shape = [n_channels, n_channels*model_order]
        VAR model coefficients. See :ref:`var-model-coefficients` for details about the arrangement of coefficients.
    c : ndarray, shape = [n_channels, n_channels], optional
        Covariance matrix of the driving noise process. Identity matrix is used if set to None.
    nfft : int, optional
        Number of frequency bins to calculate. Note that these points cover the range between 0 and half the
        sampling rate.

    Methods
    -------
    :func:`A`
       Spectral representation of the VAR coefficients
    :func:`H`
        Transfer function that turns the innovation process into the VAR process
    :func:`S`
        Cross spectral density
    :func:`logS`
        Logarithm of the cross spectral density (S), for convenience.
    :func:`G`
        Inverse cross spectral density
    :func:`logG`
        Logarithm of the inverse cross spectral density
    :func:`PHI`
        Phase angle
    :func:`COH`
        Coherence
    :func:`pCOH`
        Partial coherence
    :func:`PDC`
        Partial directed coherence
    :func:`ffPDC`
        Full frequency partial directed coherence
    :func:`PDCF`
        PDC factor
    :func:`GPDC`
        Generalized partial directed coherence
    :func:`DTF`
        Directed transfer function
    :func:`ffDTF`
        Full frequency directed transfer function
    :func:`dDTF`
        Direct directed transfer function
    :func:`GDTF`
        Generalized directed transfer function

    Notes
    -----
    Connectivity measures are returned by member functions that take no arguments and return a matrix of
    shape [m,m,nfft]. The first dimension is the sink, the second dimension is the source, and the third dimension is
    the frequency.

    A summary of most supported measures can be found in [1]_.

    References
    ----------
    .. [1] <NAME> et al, “Single-trial connectivity estimation for classification of motor imagery data”,
           *J. Neural Eng.* 10, 2013.
    """

    def __init__(self, b, c=None, nfft=512):
        b = np.asarray(b)
        (m, mp) = b.shape
        p = mp // m
        if m * p != mp:
            raise AttributeError('Second dimension of b must be an integer multiple of the first dimension.')

        if c is None:
            self.c = None
        else:
            self.c = np.asarray(c)

        self.b = np.reshape(b, (m, m, p), 'c')
        self.m = m
        self.p = p
        self.nfft = nfft

    @memoize
    def Cinv(self):
        """ Inverse of the noise covariance
        """
        try:
            return np.linalg.inv(self.c)
        except np.linalg.linalg.LinAlgError:
            print('Warning: non invertible noise covariance matrix c!')
            return np.eye(self.c.shape[0])

    @memoize
    def A(self):
        """ Spectral VAR coefficients

        .. math:: \mathbf{A}(f) = \mathbf{I} - \sum_{k=1}^{p} \mathbf{a}^{(k)} \mathrm{e}^{-2\pi f}
        """
        return np.fft.rfft(np.dstack([np.eye(self.m), -self.b]), self.nfft * 2 - 1)

    @memoize
    def H(self):
        """ VAR transfer function

        .. math:: \mathbf{H}(f) = \mathbf{A}(f)^{-1}
        """
        return _inv3(self.A())

    @memoize
    def S(self):
        """ Cross spectral density

        .. math:: \mathbf{S}(f) = \mathbf{H}(f) \mathbf{C} \mathbf{H}'(f)
        """
        if self.c is None:
            raise RuntimeError('Cross spectral density requires noise covariance matrix c.')
        H = self.H()
        #TODO can we do that more efficiently?
        S = np.einsum('ij..., jk... ->ik...', H, self.c)
        S = np.einsum('ij..., kj... ->ik...', S, H.conj())
        return S

    @memoize
    def logS(self):
        """ Logarithmic cross spectral density

        .. math:: \mathrm{logS}(f) = \log | \mathbf{S}(f) |
        """
        return np.log10(np.abs(self.S()))

    @memoize
    def absS(self):
        """ Absolute cross spectral density

        .. math:: \mathrm{absS}(f) = | \mathbf{S}(f) |
        """
        return np.abs(self.S())

    @memoize
    def G(self):
        """ Inverse cross spectral density

        .. math:: \mathbf{G}(f) = \mathbf{A}(f) \mathbf{C}^{-1} \mathbf{A}'(f)
        """
        if self.c is None:
            raise RuntimeError('Inverse cross spectral density requires invertible noise covariance matrix c.')
        A = self.A()
        #TODO can we do that more efficiently?
        G = np.einsum('ji..., jk... ->ik...', A.conj(), self.Cinv())
        G = np.einsum('ij..., jk... ->ik...', G, A)
        return G<commit_msg>Fixed error caused by 1D signals<commit_after># Released under The MIT License (MIT)
# http://opensource.org/licenses/MIT
# Copyright (c) 2013 SCoT Development Team

""" Connectivity Analysis """

import numpy as np
from .utils import memoize


def connectivity(measure_names, b, c=None, nfft=512):
    """ calculate connectivity measures.

    Parameters
    ----------
    measure_names : {str, list of str}
        Name(s) of the connectivity measure(s) to calculate. See :class:`Connectivity` for supported measures.
    b : ndarray, shape = [n_channels, n_channels*model_order]
        VAR model coefficients. See :ref:`var-model-coefficients` for details about the arrangement of coefficients.
    c : ndarray, shape = [n_channels, n_channels], optional
        Covariance matrix of the driving noise process. Identity matrix is used if set to None.
    nfft : int, optional
        Number of frequency bins to calculate. Note that these points cover the range between 0 and half the
        sampling rate.

    Returns
    -------
    result : ndarray, shape = [n_channels, n_channels, `nfft`]
        An ndarray of shape [m, m, nfft] is returned if measures is a string. If measures is a list of strings a
        dictionary is returned, where each key is the name of the measure, and the corresponding values are ndarrays
        of shape [m, m, nfft].

    Notes
    -----
    When using this function it is more efficient to get several measures at once than calling the function multiple times.

    Examples
    --------
    >>> c = connectivity(['DTF', 'PDC'], [[0.3, 0.6], [0.0, 0.9]])
    """
    con = Connectivity(b, c, nfft)
    try:
        return getattr(con, measure_names)()
    except TypeError:
        return {m: getattr(con, m)() for m in measure_names}


#noinspection PyPep8Naming
class Connectivity:
    """ Calculation of connectivity measures
    
    This class calculates various spectral connectivity measures from a vector autoregressive (VAR) model.

    Parameters
    ----------
    b : ndarray, shape = [n_channels, n_channels*model_order]
        VAR model coefficients. See :ref:`var-model-coefficients` for details about the arrangement of coefficients.
    c : ndarray, shape = [n_channels, n_channels], optional
        Covariance matrix of the driving noise process. Identity matrix is used if set to None.
    nfft : int, optional
        Number of frequency bins to calculate. Note that these points cover the range between 0 and half the
        sampling rate.

    Methods
    -------
    :func:`A`
       Spectral representation of the VAR coefficients
    :func:`H`
        Transfer function that turns the innovation process into the VAR process
    :func:`S`
        Cross spectral density
    :func:`logS`
        Logarithm of the cross spectral density (S), for convenience.
    :func:`G`
        Inverse cross spectral density
    :func:`logG`
        Logarithm of the inverse cross spectral density
    :func:`PHI`
        Phase angle
    :func:`COH`
        Coherence
    :func:`pCOH`
        Partial coherence
    :func:`PDC`
        Partial directed coherence
    :func:`ffPDC`
        Full frequency partial directed coherence
    :func:`PDCF`
        PDC factor
    :func:`GPDC`
        Generalized partial directed coherence
    :func:`DTF`
        Directed transfer function
    :func:`ffDTF`
        Full frequency directed transfer function
    :func:`dDTF`
        Direct directed transfer function
    :func:`GDTF`
        Generalized directed transfer function

    Notes
    -----
    Connectivity measures are returned by member functions that take no arguments and return a matrix of
    shape [m,m,nfft]. The first dimension is the sink, the second dimension is the source, and the third dimension is
    the frequency.

    A summary of most supported measures can be found in [1]_.

    References
    ----------
    .. [1] <NAME> et al, “Single-trial connectivity estimation for classification of motor imagery data”,
           *J. Neural Eng.* 10, 2013.
    """

    def __init__(self, b, c=None, nfft=512):
        b = np.asarray(b)
        (m, mp) = b.shape
        p = mp // m
        if m * p != mp:
            raise AttributeError('Second dimension of b must be an integer multiple of the first dimension.')

        if c is None:
            self.c = None
        else:
            self.c = np.atleast_2d(c)

        self.b = np.reshape(b, (m, m, p), 'c')
        self.m = m
        self.p = p
        self.nfft = nfft

    @memoize
    def Cinv(self):
        """ Inverse of the noise covariance
        """
        try:
            return np.linalg.inv(self.c)
        except np.linalg.linalg.LinAlgError:
            print('Warning: non invertible noise covariance matrix c!')
            return np.eye(self.c.shape[0])

    @memoize
    def A(self):
        """ Spectral VAR coefficients

        .. math:: \mathbf{A}(f) = \mathbf{I} - \sum_{k=1}^{p} \mathbf{a}^{(k)} \mathrm{e}^{-2\pi f}
        """
        return np.fft.rfft(np.dstack([np.eye(self.m), -self.b]), self.nfft * 2 - 1)

    @memoize
    def H(self):
        """ VAR transfer function

        .. math:: \mathbf{H}(f) = \mathbf{A}(f)^{-1}
        """
        return _inv3(self.A())

    @memoize
    def S(self):
        """ Cross spectral density

        .. math:: \mathbf{S}(f) = \mathbf{H}(f) \mathbf{C} \mathbf{H}'(f)
        """
        if self.c is None:
            raise RuntimeError('Cross spectral density requires noise covariance matrix c.')
        H = self.H()
        #TODO can we do that more efficiently?
        S = np.einsum('ia..., ab..., jb... ->ij...', H, self.c, H.conj())
        return S

    @memoize
    def logS(self):
        """ Logarithmic cross spectral density

        .. math:: \mathrm{logS}(f) = \log | \mathbf{S}(f) |
        """
        return np.log10(np.abs(self.S()))

    @memoize
    def absS(self):
        """ Absolute cross spectral density

        .. math:: \mathrm{absS}(f) = | \mathbf{S}(f) |
        """
        return np.abs(self.S())

    @memoize
    def G(self):
        """ Inverse cross spectral density

        .. math:: \mathbf{G}(f) = \mathbf{A}(f) \mathbf{C}^{-1} \mathbf{A}'(f)
        """
        if self.c is None:
            raise RuntimeError('Inverse cross spectral density requires invertible noise covariance matrix c.')
        A = self.A()
        #TODO can we do that more efficiently?
        G = np.einsum('ji..., jk... ->ik...', A.conj(), self.Cinv())
        G = np.einsum('ij..., jk... ->ik...', G, A)
        return G
{'fraction_non_alphanumeric': 0.12060033521255524, 'fraction_numerical': 0.0060947737315252175, 'mean_word_length': 2.207182995357928, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '471181', 'n_tokens_mistral': 3987, 'n_tokens_neox': 3800, 'n_words': 1360}