Naive Bayes: Add method predict, fix predict_storage

Author: janezd

Orange/classification/naive_bayes.py

@@ -1,7 +1,8 @@
 import numpy as np
+import scipy.sparse as sp
 
 from Orange.classification import Learner, Model
-from Orange.data import Instance, Storage
+from Orange.data import Instance, Storage, Table
 from Orange.statistics import contingency
 from Orange.preprocess import Discretize, RemoveNaNColumns
 
@@ -47,23 +48,81 @@ def __init__(self, log_cont_prob, class_prob, domain):
         self.class_prob = class_prob
 
     def predict_storage(self, data):
+        if type(data) is Table:  # pylint: disable=unidiomatic-typecheck
+            return self.predict(data.X)
+
         if isinstance(data, Instance):
-            data = [data]
+            data = Table(data.domain, [data])
         if len(data.domain.attributes) == 0:
             probs = np.tile(self.class_prob, (len(data), 1))
         else:
             isnan = np.isnan
-            probs = np.exp(
+            probs = np.atleast_2d(np.exp(
                 np.log(self.class_prob) +
                 np.array([np.zeros_like(self.class_prob)
                           if isnan(ins.x).all() else
-                          np.sum(attr_prob[:, int(attr_val)]
-                                 for attr_val, attr_prob in zip(ins, self.log_cont_prob)
-                                 if not isnan(attr_val))
-                          for ins in data]))
+                          np.sum([attr_prob[:, int(attr_val)]
+                                  for attr_val, attr_prob in
+                                  zip(ins, self.log_cont_prob)
+                                  if not isnan(attr_val)])
+                          for ins in data])))
         probs /= probs.sum(axis=1)[:, None]
         values = probs.argmax(axis=1)
         return values, probs
 
+    def predict(self, X):
+        if not self.log_cont_prob:
+            probs = self._priors(X)
+        elif sp.issparse(X):
+            probs = self._sparse_probs(X)
+        else:
+            probs = self._dense_probs(X)
+        probs = np.exp(probs)
+        probs /= probs.sum(axis=1)[:, None]
+        values = probs.argmax(axis=1)
+        return values, probs
+
+    def _priors(self, data):
+        return np.tile(np.log(self.class_prob), (data.shape[0], 1))
+
+    def _dense_probs(self, data):
+        probs = self._priors(data)
+        zeros = np.zeros((1, probs.shape[1]))
+        for col, attr_prob in zip(data.T, self.log_cont_prob):
+            col = col.copy()
+            col[np.isnan(col)] = attr_prob.shape[1] - 1
+            col = col.astype(int)
+            probs0 = np.vstack((attr_prob.T, zeros))
+            probs += probs0[col]
+        return probs
+
+    def _sparse_probs(self, data):
+        probs = self._priors(data)
+
+        n_vals = max(p.shape[1] for p in self.log_cont_prob) + 1
+        log_prob = np.zeros((len(self.log_cont_prob),
+                             n_vals,
+                             self.log_cont_prob[0].shape[0]))
+        for i, p in enumerate(self.log_cont_prob):
+            p0 = p.T[0].copy()
+            probs[:] += p0
+            log_prob[i, :p.shape[1]] = p.T - p0
+
+        dat = data.data.copy()
+        dat[np.isnan(dat)] = n_vals - 1
+        dat = dat.astype(int)
+
+        if sp.isspmatrix_csr(data):
+            for row, start, end in zip(probs, data.indptr, data.indptr[1:]):
+                row += log_prob[data.indices[start:end],
+                                dat[start:end]].sum(axis=0)
+        else:
+            csc = data.tocsc()
+            for start, end, attr_prob in zip(csc.indptr, csc.indptr[1:],
+                                             log_prob):
+                probs[csc.indices[start:end]] += attr_prob[dat[start:end]]
+
+        return probs
+
 
 NaiveBayesLearner.__returns__ = NaiveBayesModel

Orange/tests/test_naive_bayes.py

@@ -2,6 +2,10 @@
 # pylint: disable=missing-docstring
 
 import unittest
+from unittest.mock import Mock
+
+import numpy as np
+import scipy.sparse as sp
 
 from Orange.classification import NaiveBayesLearner
 from Orange.data import Table, Domain, DiscreteVariable, ContinuousVariable
@@ -11,17 +15,23 @@
 class TestNaiveBayesLearner(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
-        data = Table('titanic')
+        cls.data = data = Table('titanic')
         cls.learner = NaiveBayesLearner()
-        cls.model = cls.learner(data)
         cls.table = data[::20]
 
+    def setUp(self):
+        self.model = self.learner(self.data)
+
     def test_NaiveBayes(self):
         results = CrossValidation(self.table, [self.learner], k=10)
         ca = CA(results)
         self.assertGreater(ca, 0.7)
         self.assertLess(ca, 0.9)
 
+        results = CrossValidation(Table("iris"), [self.learner], k=10)
+        ca = CA(results)
+        self.assertGreater(ca, 0.7)
+
     def test_predict_single_instance(self):
         for ins in self.table:
             self.model(ins)
@@ -53,3 +63,35 @@ def test_allnan_cv(self):
         data = Table('voting')
         results = CrossValidation(data, [self.learner])
         self.assertFalse(any(results.failed))
+
+    def test_sparse(self):
+        _, dense_p = self.model.predict(self.data.X)
+
+        _, csc_p = self.model.predict(sp.csc_matrix(self.data.X))
+        np.testing.assert_almost_equal(dense_p, csc_p)
+
+        _, csr_p = self.model.predict(sp.csr_matrix(self.data.X))
+        np.testing.assert_almost_equal(dense_p, csr_p)
+
+    def test_prediction_routing(self):
+        data = self.data
+        predict = self.model.predict = Mock(return_value=(data.Y, None))
+
+        self.model(data)
+        predict.assert_called()
+        predict.reset_mock()
+
+        self.model(data.X)
+        predict.assert_called()
+        predict.reset_mock()
+
+        self.model.predict_storage(data)
+        predict.assert_called()
+        predict.reset_mock()
+
+        self.model.predict_storage(data[0])
+        predict.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()