Added unit tests to benchmark transformer_nmf against nimfa nmf.
MODIFY
demos/demo_transformer_nmf.py
demos/demo_transformer_nmf.py
@@ -11,7 +11,7 @@ def simple_example():
# Make two simple matrices
n = 4
a = np.arange(n ** 2).reshape((n, n))
- b = 2 * a + 3
+ b = 2. * a + 3.
# Mix them together
mixture = np.dot(b, a)
@@ -24,15 +24,19 @@ def simple_example():
nussl_nmf.max_num_iterations = n_iter
nussl_nmf.distance_measure = nussl.DistanceType.DIVERGENCE
+ # set up nimfa NMF
+ V = mixture
+ nimfa_nmf = nimfa.Nmf(V, max_iter=n_iter, rank=n_bases, update='divergence')
+
+ # calculate errors
+ errors = []
+ num_iterations = 100
+ for i in range(num_iterations):
# run NU NMF
nussl_nmf.run()
-
signals = nussl_nmf.recombine_calculated_matrices()
- # for sig in signals:
- # print sig
- V = mixture
- nimfa_nmf = nimfa.Nmf(V, max_iter=n_iter, rank=n_bases, update='divergence')
+ #run nimfa NMF
nmf_fit = nimfa_nmf()
W = nmf_fit.basis()
H = nmf_fit.coef()
@@ -40,17 +44,31 @@ def simple_example():
nimfa_est = np.dot(W, H)
nussl_est = np.dot(nussl_nmf.templates, nussl_nmf.activation_matrix)
error = np.abs(nussl_est - nimfa_est) / nimfa_est * 100
+ errors.append(error.flatten())
+
+ errors = np.array(errors).flatten()
+ mean_error = np.mean(errors)
+ std_dev_error = np.std(errors)
+ print "mean", mean_error
+ print "std", std_dev_error
+ std_check = True
+ num_over_std_dev = 0
+ for error in errors:
+ if error > (mean_error + (5 * std_dev_error)):
+ num_over_std_dev += 1
+ print num_over_std_dev
+ if (num_over_std_dev/float(num_iterations)) > .05:
+ std_check = False
+ print (num_over_std_dev / float(num_iterations))
+ print std_check
+ # error statistics
+
- print('nimfa estimate:\n%s\n\n' % nimfa_est)
- print('nussl estimate:\n%s\n\n' % nussl_est)
- print('Errors:\n%s' % error)
+ # print('nimfa estimate:\n%s\n\n' % nimfa_est)
+ # print('nussl estimate:\n%s\n\n' % nussl_est)
+ # print('Errors:\n%s' % error)
- sm = nmf_fit.summary()
- # print( np.dot(W, H))
- x = 1
- y = 2
- z = x + y
if __name__ == '__main__':
simple_example()

MODIFY
nussl/transformer_nmf.py
nussl/transformer_nmf.py
@@ -7,6 +7,40 @@ import numpy as np
import constants
class TransformerNMF(object):
+ """
+ This is an implementation of the Non-negative Matrix Factorization algorithm for
+ general matrix transformations. This implementation receives an input matrix and a number,
+ num_templates, which defines the number of bases vectors.
+
+ This class provides two implementations of distance measures, EUCLIDEAN and DIVERGENCE,
+ and also allows the user to define distance measure function.
+
+ References:
+ [1] Lee, Daniel D., and H. Sebastian Seung. "Algorithms for non-negative matrix factorization."
+ Advances in neural information processing systems. 2001.
+
+ :param: input_matrix: The matrix to factor into template and activation matrices
+ :type: input_matrix: 2D numpy array
+
+ :param: num_templates: The rank of the templates matrix
+ :type: num_templates: int
+
+ :param: activation_matrix: An optional seed for the activation matrix
+ :type: activation_matrix: 2D numpy array
+
+ :param: templates: An optional seed for the templates matrix
+ :type: templates: 2D numpy array
+
+ :param: distance_measure: Specifies to use euclidean or divergence distance metrics
+ :type: distance_measure: str
+
+ :param: should_update_template: Whether the template matrix should be updated for another iteration
+ :type: should_update_template: bool
+
+ :param: should_update_activation: Whether the activation matrix should be updated for another iteration
+ :type: should_update_activation: bool
+ """
+
def __init__(self, input_matrix, num_templates,
activation_matrix=None, templates=None, distance_measure=None,
should_update_template=None, should_update_activation=None):
@@ -115,6 +149,11 @@ class TransformerNMF(object):
self.templates = self._update_template_divergence()
def _update_activation_euclidean(self):
+ """
+ Computes a new activation matrix using the Lee and Seung multiplicative update algorithm
+ :return: An updated activation matrix based on euclidean distance
+ """
+
# make a new matrix to store results
activation_copy = np.empty_like(self.activation_matrix)
@@ -133,6 +172,11 @@ class TransformerNMF(object):
return activation_copy
def _update_template_euclidean(self):
+ """
+ Computes a new template matrix using the Lee and Seung multiplicative update algorithm
+ :return: An update template matrix based on euclidean distance
+ """
+
# make a new matrix to store results
template_copy = np.empty_like(self.templates)
@@ -151,6 +195,10 @@ class TransformerNMF(object):
return template_copy
def _update_activation_divergent(self):
+ """
+ Computes a new activation matrix using the Lee and Seung multiplicative update algorithm
+ :return: An update activation matrix based on divergence
+ """
# make a new matrix to store results
activation_copy = np.empty_like(self.activation_matrix)
@@ -168,6 +216,10 @@ class TransformerNMF(object):
return activation_copy
def _update_template_divergence(self):
+ """
+ Computes a new template matrix using the Lee and Seung multiplicative update algorithm
+ :return: An update template matrix based on divergence
+ """
# make a new matrix to store results
template_copy = np.empty_like(self.templates)
@@ -185,6 +237,11 @@ class TransformerNMF(object):
return template_copy
def _euclidean_distance(self):
+ """
+ Calculates the euclidean distance from the original mixture to the
+ dot product of the current template and mixture matrices
+ :return: Euclidean distance
+ """
try:
mixture = np.dot(self.templates, self.activation_matrix)
except:
@@ -198,6 +255,12 @@ class TransformerNMF(object):
return sum((self.input_matrix[index] - val) ** 2 for index, val in np.ndenumerate(mixture))
def _divergence(self):
+ def _euclidean_distance(self):
+ """
+ Calculates the divergence from the original mixture to the
+ dot product of the current template and mixture matrices
+ :return: Euclidean distance
+ """
mixture = np.dot(self.activation_matrix, self.templates)
if mixture.shape != self.input_matrix.shape:

MODIFY
tests/test_transformer_nmf.py
tests/test_transformer_nmf.py
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-import nussl
import unittest
+import numpy as np
+import nussl
+
+import nimfa
+
+class DuetUnitTests(unittest.TestCase):
+
+ def test_small_mixture_matrix(self):
+ """
+ A simple example of NMF using euclidean and divergence in nussl with a 4 by 4 matrix
+ """
+ # Make two simple matrices
+ n = 4
+ a = np.arange(n ** 2).reshape((n, n))
+ b = 2. * a + 3.
+
+ # Mix them together
+ mixture = np.dot(b, a)
+
+ # Set up NMFs and run on euclidean while calculating error
+ n_bases = 2
+ iterations = 100
+ # Set error parameters
+ num_std_deviations = 5
+ percent_over_std_dev = .05
+
+ #Run on euclidean
+ type = "euclidean"
+ errors = self.calculate_nmfs_error(mixture, n_bases, type, iterations)
+ error_check_euclidean = self.analyze_errors(errors, iterations, num_std_deviations, percent_over_std_dev)
+
+ # Run on divergence
+ type = "divergence"
+ errors = self.calculate_nmfs_error(mixture, n_bases, type, iterations)
+ error_check_divergence = self.analyze_errors(errors, iterations, num_std_deviations, percent_over_std_dev)
+
+ assert error_check_euclidean == True
+ assert error_check_divergence == True
+
+ def test_big_mixture_matrix(self):
+ """
+ A simple example of NMF using euclidean and divergence in nussl with a 10 by 10 matrix
+ """
+ # Make two simple matrices
+ n = 10
+ a = np.arange(n ** 2).reshape((n, n))
+ b = 2. * a + 3.
+
+ # Mix them together
+ mixture = np.dot(b, a)
+
+ # Set up NMFs and run on euclidean while calculating error
+ n_bases = 2
+ iterations = 100
+ # Set error parameters
+ num_std_deviations = 10
+ percent_over_std_dev = .80
+
+ # Run on euclidean
+ type = "euclidean"
+ errors = self.calculate_nmfs_error(mixture, n_bases, type, iterations)
+ error_check_euclidean = self.analyze_errors(errors, iterations, num_std_deviations, percent_over_std_dev)
+
+ # Run on divergence
+ type = "divergence"
+ errors = self.calculate_nmfs_error(mixture, n_bases, type, iterations)
+ error_check_divergence = self.analyze_errors(errors, iterations, num_std_deviations, percent_over_std_dev)
+
+ assert error_check_euclidean == True
+ assert error_check_divergence == True
+
+ def test_big_random_matrix(self):
+ V = np.random.rand(10, 10)
+ # Set up NMFs and run on euclidean while calculating error
+ n_bases = 2
+ iterations = 100
+ # Set error parameters
+ num_std_deviations = 10
+ percent_over_std_dev = .50
+
+ # Run on euclidean
+ type = "euclidean"
+ errors = self.calculate_nmfs_error(V, n_bases, type, iterations)
+ error_check_euclidean = self.analyze_errors(errors, iterations, num_std_deviations, percent_over_std_dev)
+
+ # Run on divergence
+ type = "divergence"
+ errors = self.calculate_nmfs_error(V, n_bases, type, iterations)
+ error_check_divergence = self.analyze_errors(errors, iterations, num_std_deviations, percent_over_std_dev)
+
+ assert error_check_euclidean == True
+ assert error_check_divergence == True
+
+ @staticmethod
+ def calculate_nmfs_error(mixture, n_bases, type, iterations):
+ nussl_nmf = nussl.TransformerNMF(mixture, n_bases)
+ nussl_nmf.should_use_epsilon = False
+ nussl_nmf.max_num_iterations = iterations
+ if type == "euclidean":
+ nussl_nmf.distance_measure = nussl.DistanceType.EUCLIDEAN
+ elif type == "divergence":
+ nussl_nmf.distance_measure = nussl.DistanceType.DIVERGENCE
+ else:
+ return "Use either euclidean or divergence!"
+ # set up nimfa NMF
+ V = mixture
+ nimfa_nmf = nimfa.Nmf(V, max_iter=iterations, rank=n_bases, update=type)
+
+ # calculate errors
+ errors = []
+ for i in range(iterations):
+ # run NU NMF
+ nussl_nmf.run()
+ signals = nussl_nmf.recombine_calculated_matrices()
+
+ #run nimfa NMF
+ nmf_fit = nimfa_nmf()
+ W = nmf_fit.basis()
+ H = nmf_fit.coef()
+
+ nimfa_est = np.dot(W, H)
+ nussl_est = np.dot(nussl_nmf.templates, nussl_nmf.activation_matrix)
+ error = np.abs(nussl_est - nimfa_est) / nimfa_est * 100
+ errors.append(error.flatten())
+ return errors
+
+ @staticmethod
+ def analyze_errors(errors, iterations, num_std_deviations, percent_over_std_dev):
+ # analyze mean and std deviation of errors
+ errors = np.array(errors).flatten()
+ mean_error = np.mean(errors)
+ std_dev_error = np.std(errors)
+ std_check = True
+ num_over_std_dev = 0
+
+ for error in errors:
+ if error > (mean_error + (num_std_deviations * std_dev_error)):
+ num_over_std_dev += 1
+
+ if (num_over_std_dev / float(iterations)) > percent_over_std_dev:
+ std_check = False
+
+ return std_check
+
+if __name__ == '__main__':
+ simple_example()

Added unit tests to benchmark transformer_nmf against nimfa nmf.