Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_sqrtm():
"""Test matrix square root"""
C = 2*np.eye(3)
Ctrue = np.sqrt(2)*np.eye(3)
assert_array_almost_equal(sqrtm(C), Ctrue)
else:
C = init
k = 0
K = sqrtm(C)
crit = numpy.finfo(numpy.float64).max
# stop when J<10^-9 or max iteration = 50
while (crit > tol) and (k < maxiter):
k = k + 1
J = numpy.zeros((Ne, Ne))
for index, Ci in enumerate(covmats):
tmp = numpy.dot(numpy.dot(K, Ci), K)
J += sample_weight[index] * sqrtm(tmp)
Knew = sqrtm(J)
crit = numpy.linalg.norm(Knew - K, ord='fro')
K = Knew
if k == maxiter:
print('Max iter reach')
C = numpy.dot(K, K)
return C
:returns: the mean covariance matrix
References
----------
[1] Barbaresco, F. "Geometric Radar Processing based on Frechet distance:
Information geometry versus Optimal Transport Theory", Radar Symposium
(IRS), 2011 Proceedings International.
"""
sample_weight = _get_sample_weight(sample_weight, covmats)
Nt, Ne, Ne = covmats.shape
if init is None:
C = numpy.mean(covmats, axis=0)
else:
C = init
k = 0
K = sqrtm(C)
crit = numpy.finfo(numpy.float64).max
# stop when J<10^-9 or max iteration = 50
while (crit > tol) and (k < maxiter):
k = k + 1
J = numpy.zeros((Ne, Ne))
for index, Ci in enumerate(covmats):
tmp = numpy.dot(numpy.dot(K, Ci), K)
J += sample_weight[index] * sqrtm(tmp)
Knew = sqrtm(J)
crit = numpy.linalg.norm(Knew - K, ord='fro')
K = Knew
if k == maxiter:
print('Max iter reach')
def untangent_space(T, Cref):
"""Project a set of Tangent space vectors back to the manifold.
:param T: np.ndarray
the Tangent space , a matrix of Ntrials X (channels * (channels + 1)/2)
:param Cref: np.ndarray
The reference covariance matrix
:returns: np.ndarray
A set of Covariance matrix, Ntrials X Nchannels X Nchannels
"""
Nt, Nd = T.shape
Ne = int((numpy.sqrt(1 + 8 * Nd) - 1) / 2)
C12 = sqrtm(Cref)
idx = numpy.triu_indices_from(Cref)
covmats = numpy.empty((Nt, Ne, Ne))
covmats[:, idx[0], idx[1]] = T
for i in range(Nt):
triuc = numpy.triu(covmats[i], 1) / numpy.sqrt(2)
covmats[i] = (numpy.diag(numpy.diag(covmats[i])) + triuc + triuc.T)
covmats[i] = expm(covmats[i])
covmats[i] = numpy.dot(numpy.dot(C12, covmats[i]), C12)
return covmats
if init is None:
C = numpy.mean(covmats, axis=0)
else:
C = init
k = 0
K = sqrtm(C)
crit = numpy.finfo(numpy.float64).max
# stop when J<10^-9 or max iteration = 50
while (crit > tol) and (k < maxiter):
k = k + 1
J = numpy.zeros((Ne, Ne))
for index, Ci in enumerate(covmats):
tmp = numpy.dot(numpy.dot(K, Ci), K)
J += sample_weight[index] * sqrtm(tmp)
Knew = sqrtm(J)
crit = numpy.linalg.norm(Knew - K, ord='fro')
K = Knew
if k == maxiter:
print('Max iter reach')
C = numpy.dot(K, K)
return C
"""
# init
sample_weight = _get_sample_weight(sample_weight, covmats)
Nt, Ne, Ne = covmats.shape
if init is None:
C = numpy.mean(covmats, axis=0)
else:
C = init
k = 0
nu = 1.0
tau = numpy.finfo(numpy.float64).max
crit = numpy.finfo(numpy.float64).max
# stop when J<10^-9 or max iteration = 50
while (crit > tol) and (k < maxiter) and (nu > tol):
k = k + 1
C12 = sqrtm(C)
Cm12 = invsqrtm(C)
J = numpy.zeros((Ne, Ne))
for index in range(Nt):
tmp = numpy.dot(numpy.dot(Cm12, covmats[index, :, :]), Cm12)
J += sample_weight[index] * logm(tmp)
crit = numpy.linalg.norm(J, ord='fro')
h = nu * crit
C = numpy.dot(numpy.dot(C12, expm(nu * J)), C12)
if h < tau:
nu = 0.95 * nu
tau = h
else:
nu = 0.5 * nu
def distance_wasserstein(A, B):
"""Wasserstein distance between two covariances matrices.
.. math::
d = \left( {tr(A + B - 2(A^{1/2}BA^{1/2})^{1/2})}\\right )^{1/2}
:param A: First covariance matrix
:param B: Second covariance matrix
:returns: Wasserstein distance between A and B
"""
B12 = sqrtm(B)
C = sqrtm(numpy.dot(numpy.dot(B12, A), B12))
return numpy.sqrt(numpy.trace(A + B - 2*C))
def distance_wasserstein(A, B):
"""Wasserstein distance between two covariances matrices.
.. math::
d = \left( {tr(A + B - 2(A^{1/2}BA^{1/2})^{1/2})}\\right )^{1/2}
:param A: First covariance matrix
:param B: Second covariance matrix
:returns: Wasserstein distance between A and B
"""
B12 = sqrtm(B)
C = sqrtm(numpy.dot(numpy.dot(B12, A), B12))
return numpy.sqrt(numpy.trace(A + B - 2*C))
def geodesic_riemann(A, B, alpha=0.5):
"""Return the matrix at the position alpha on the riemannian geodesic between A and B :
.. math::
\mathbf{C} = \mathbf{A}^{1/2} \left( \mathbf{A}^{-1/2} \mathbf{B} \mathbf{A}^{-1/2} \\right)^\\alpha \mathbf{A}^{1/2}
C is equal to A if alpha = 0 and B if alpha = 1
:param A: the first coavriance matrix
:param B: the second coavriance matrix
:param alpha: the position on the geodesic
:returns: the covariance matrix
"""
sA = sqrtm(A)
isA = invsqrtm(A)
C = numpy.dot(numpy.dot(isA, B), isA)
D = powm(C, alpha)
E = numpy.dot(numpy.dot(sA, D), sA)
return E
def transport(Covs, Cref, metric='riemann'):
"""Parallel transport of two set of covariance matrix.
"""
C = mean_covariance(Covs, metric=metric)
iC = invsqrtm(C)
E = sqrtm(numpy.dot(numpy.dot(iC, Cref), iC))
out = numpy.array([numpy.dot(numpy.dot(E, c), E.T) for c in Covs])
return out