msaf/pymf/cnmf.py
def factorize(self, niter=10, compute_w=True, compute_h=True,
                  compute_err=True, show_progress=False):
        """ Factorize s.t. WH = data

            Parameters
            ----------
            niter : int
                    number of iterations.
            show_progress : bool
                    print some extra information to stdout.
            compute_h : bool
                    iteratively update values for H.
            compute_w : bool
                    iteratively update values for W.
            compute_err : bool
                    compute Frobenius norm |data-WH| after each update and store
                    it to .ferr[k].

            Updated Values
            --------------
            .W : updated values for W.
            .H : updated values for H.
            .ferr : Frobenius norm |data-WH| for each iteration.
        """

        if not hasattr(self,'W'):
               self.init_w()

        if not hasattr(self,'H'):
                self.init_h()

        def separate_positive(m):
            return (np.abs(m) + m)/2.0

        def separate_negative(m):
            return (np.abs(m) - m)/2.0

        if show_progress:
            self._logger.setLevel(logging.INFO)
        else:
            self._logger.setLevel(logging.ERROR)

        XtX = np.dot(self.data[:,:].T, self.data[:,:])
        XtX_pos = separate_positive(XtX)
        XtX_neg = separate_negative(XtX)

        self.ferr = np.zeros(niter)
        # iterate over W and H

        for i in range(niter):
            # update H
            XtX_neg_x_W = np.dot(XtX_neg, self.G)
            XtX_pos_x_W = np.dot(XtX_pos, self.G)

            if compute_h:
                H_x_WT = np.dot(self.H.T, self.G.T)
                ha = XtX_pos_x_W + np.dot(H_x_WT, XtX_neg_x_W)
                hb = XtX_neg_x_W + np.dot(H_x_WT, XtX_pos_x_W) + 10**-9
                self.H = (self.H.T*np.sqrt(ha/hb)).T

            # update W
            if compute_w:
                HT_x_H = np.dot(self.H, self.H.T)
                wa = np.dot(XtX_pos, self.H.T) + np.dot(XtX_neg_x_W, HT_x_H)
                wb = np.dot(XtX_neg, self.H.T) + np.dot(XtX_pos_x_W, HT_x_H) + 10**-9

                self.G *= np.sqrt(wa/wb)
                self.W = np.dot(self.data[:,:], self.G)

            if compute_err:
                self.ferr[i] = self.frobenius_norm()
                self._logger.info('Iteration ' + str(i+1) + '/' + str(niter) +
                ' FN:' + str(self.ferr[i]))
            else:
                self._logger.info('Iteration ' + str(i+1) + '/' + str(niter))

            if i > 1 and compute_err:
                if self.converged(i):
                    self.ferr = self.ferr[:i]
                    break
def factorize(self, niter=10, compute_w=True, compute_h=True,
                  compute_err=True, show_progress=False):
        """ Factorize s.t. WH = data

            Parameters
            ----------
            niter : int
                    number of iterations.
            show_progress : bool
                    print some extra information to stdout.
            compute_h : bool
                    iteratively update values for H.
            compute_w : bool
                    iteratively update values for W.
            compute_err : bool
                    compute Frobenius norm |data-WH| after each update and store
                    it to .ferr[k].

            Updated Values
            --------------
            .W : updated values for W.
            .H : updated values for H.
            .ferr : Frobenius norm |data-WH| for each iteration.
        """

        if not hasattr(self,'W'):
               self.init_w()

        if not hasattr(self,'H'):
                self.init_h()

        def separate_positive(m):
            return (np.abs(m) + m)/2.0

        def separate_negative(m):
            return (np.abs(m) - m)/2.0

        if show_progress:
            self._logger.setLevel(logging.INFO)
        else:
            self._logger.setLevel(logging.ERROR)

        XtX = np.dot(self.data[:,:].T, self.data[:,:])
        XtX_pos = separate_positive(XtX)
        XtX_neg = separate_negative(XtX)

        self.ferr = np.zeros(niter)
        # iterate over W and H

        for i in range(niter):
            # update H
            XtX_neg_x_W = np.dot(XtX_neg, self.G)
            XtX_pos_x_W = np.dot(XtX_pos, self.G)

            if compute_h:
                H_x_WT = np.dot(self.H.T, self.G.T)
                ha = XtX_pos_x_W + np.dot(H_x_WT, XtX_neg_x_W)
                hb = XtX_neg_x_W + np.dot(H_x_WT, XtX_pos_x_W) + 10**-9
                self.H = (self.H.T*np.sqrt(ha/hb)).T

            # update W
            if compute_w:
                HT_x_H = np.dot(self.H, self.H.T)
                wa = np.dot(XtX_pos, self.H.T) + np.dot(XtX_neg_x_W, HT_x_H)
                wb = np.dot(XtX_neg, self.H.T) + np.dot(XtX_pos_x_W, HT_x_H) + 10**-9

                self.G *= np.sqrt(wa/wb)
                self.W = np.dot(self.data[:,:], self.G)

            if compute_err:
                self.ferr[i] = self.frobenius_norm()
                self._logger.info('Iteration ' + str(i+1) + '/' + str(niter) +
                ' FN:' + str(self.ferr[i]))
            else:
                self._logger.info('Iteration ' + str(i+1) + '/' + str(niter))

            if i > 1 and compute_err:
                if self.converged(i):
                    self.ferr = self.ferr[:i]
                    break
def factorize ( self , niter = 10 , compute_w = True , compute_h = True , compute_err = True , show_progress = False ) : if not hasattr ( self , 'W' ) : self . init_w ( ) if not hasattr ( self , 'H' ) : self . init_h ( ) def separate_positive ( m ) : return ( np . abs ( m ) + m ) / 2.0 def separate_negative ( m ) : return ( np . abs ( m ) - m ) / 2.0 if show_progress : self . _logger . setLevel ( logging . INFO ) else : self . _logger . setLevel ( logging . ERROR ) XtX = np . dot ( self . data [ : , : ] . T , self . data [ : , : ] ) XtX_pos = separate_positive ( XtX ) XtX_neg = separate_negative ( XtX ) self . ferr = np . zeros ( niter ) # iterate over W and H for i in range ( niter ) : # update H XtX_neg_x_W = np . dot ( XtX_neg , self . G ) XtX_pos_x_W = np . dot ( XtX_pos , self . G ) if compute_h : H_x_WT = np . dot ( self . H . T , self . G . T ) ha = XtX_pos_x_W + np . dot ( H_x_WT , XtX_neg_x_W ) hb = XtX_neg_x_W + np . dot ( H_x_WT , XtX_pos_x_W ) + 10 ** - 9 self . H = ( self . H . T * np . sqrt ( ha / hb ) ) . T # update W if compute_w : HT_x_H = np . dot ( self . H , self . H . T ) wa = np . dot ( XtX_pos , self . H . T ) + np . dot ( XtX_neg_x_W , HT_x_H ) wb = np . dot ( XtX_neg , self . H . T ) + np . dot ( XtX_pos_x_W , HT_x_H ) + 10 ** - 9 self . G *= np . sqrt ( wa / wb ) self . W = np . dot ( self . data [ : , : ] , self . G ) if compute_err : self . ferr [ i ] = self . frobenius_norm ( ) self . _logger . info ( 'Iteration ' + str ( i + 1 ) + '/' + str ( niter ) + ' FN:' + str ( self . ferr [ i ] ) ) else : self . _logger . info ( 'Iteration ' + str ( i + 1 ) + '/' + str ( niter ) ) if i > 1 and compute_err : if self . converged ( i ) : self . ferr = self . ferr [ : i ] break
Factorize s.t. WH = data

            Parameters
            ----------
            niter : int
                    number of iterations.
            show_progress : bool
                    print some extra information to stdout.
            compute_h : bool
                    iteratively update values for H.
            compute_w : bool
                    iteratively update values for W.
            compute_err : bool
                    compute Frobenius norm |data-WH| after each update and store
                    it to .ferr[k].

            Updated Values
            --------------
            .W : updated values for W.
            .H : updated values for H.
            .ferr : Frobenius norm |data-WH| for each iteration.
Factorize s . t . WH = data