adds framework for testing predict

src/py/crankshaft/test/test_regression_gwr.py

@@ -1,13 +1,13 @@
 import unittest
+import json
 import numpy as np
 
+
+from crankshaft import random_seeds
 from helper import fixture_file
 from crankshaft.regression import GWR
 from crankshaft.analysis_data_provider import AnalysisDataProvider
 
-from crankshaft import random_seeds
-import json
-
 
 class FakeDataProvider(AnalysisDataProvider):
     def __init__(self, mocked_result):
@@ -16,6 +16,9 @@ class FakeDataProvider(AnalysisDataProvider):
     def get_gwr(self, params):
         return self.mocked_result
 
+    def get_gwr_predict(self, params):
+        return self.mocked_result
+
 
 class GWRTest(unittest.TestCase):
     """Testing class for geographically weighted regression (gwr)"""
@@ -37,10 +40,24 @@ class GWRTest(unittest.TestCase):
                       pctpov IS NOT NULL AND
                       pctblack IS NOT NULL
         """
+        import copy
+        # data packed from https://github.com/TaylorOshan/pysal/blob/1d6af33bda46b1d623f70912c56155064463383f/pysal/examples/georgia/GData_utm.csv
         self.data = json.loads(
               open(fixture_file('gwr_packed_data.json')).read())
+
+        # data packed from https://github.com/TaylorOshan/pysal/blob/a44c5541e2e0d10a99ff05edc1b7f81b70f5a82f/pysal/examples/georgia/georgia_BS_NN_listwise.csv
         self.knowns = json.loads(
               open(fixture_file('gwr_packed_knowns.json')).read())
+
+        # data for GWR prediction
+        self.data_predict = copy.deepcopy(self.data)
+        self.ids_of_unknowns = [13083, ]
+        self.idx_ids_of_unknowns = [self.data_predict[0]['rowid'].index(idx)
+                                    for idx in self.ids_of_unknowns]
+
+        for idx in self.idx_ids_of_unknowns:
+            self.data_predict[0]['dep_var'][idx] = None
+
         # params, with ind_vars in same ordering as query above
         self.params = {'subquery': 'select * from table',
                        'dep_var': 'pctbach',
@@ -54,29 +71,23 @@ class GWRTest(unittest.TestCase):
         """
         """
         gwr = GWR(FakeDataProvider(self.data))
-        gwr_resp = gwr.gwr(self.params['subquery'], self.params['dep_var'],
-                           self.params['ind_vars'], bw=self.params['bw'],
+        gwr_resp = gwr.gwr(self.params['subquery'],
+                           self.params['dep_var'],
+                           self.params['ind_vars'],
+                           bw=self.params['bw'],
                            fixed=self.params['fixed'])
 
         # unpack response
-        coeffs, stand_errs, t_vals, t_vals_filtered, predicteds, residuals, r_squareds, bws, rowids = zip(*gwr_resp)
+        coeffs, stand_errs, t_vals, t_vals_filtered, predicteds, \
+            residuals, r_squareds, bws, rowids = zip(*gwr_resp)
 
-        # known_coeffs = self.knowns['coeffs']
-        # data packed from https://github.com/TaylorOshan/pysal/blob/a44c5541e2e0d10a99ff05edc1b7f81b70f5a82f/pysal/examples/georgia/georgia_BS_NN_listwise.csv
+        # prepare for comparision
         coeff_known_pctpov = self.knowns['est_pctpov']
         tval_known_pctblack = self.knowns['t_pctrural']
         pctpov_se = self.knowns['se_pctpov']
         ids = self.knowns['area_key']
         resp_idx = None
 
-        print sorted(coeff_known_pctpov[:10])
-        print sorted(
-                [json.loads(coeffs[i])['pctpov']
-                 for i in xrange(len(coeffs))][:10])
-        with open('gwr_test_data.json', 'w') as f:
-            print("writing to file")
-            f.write(str(zip(rowids, coeffs)))
-
         # test pctpov coefficient estimates
         for idx, val in enumerate(coeff_known_pctpov):
             resp_idx = rowids.index(ids[idx])
@@ -89,3 +100,31 @@ class GWRTest(unittest.TestCase):
             self.assertAlmostEquals(val,
                                     json.loads(t_vals[resp_idx])['pctrural'],
                                     places=4)
+
+    def test_gwr_predict(self):
+        """
+        """
+        gwr = GWR(FakeDataProvider(self.data_predict))
+        gwr_resp = gwr.gwr_predict(self.params['subquery'],
+                                   self.params['dep_var'],
+                                   self.params['ind_vars'],
+                                   bw=self.params['bw'],
+                                   fixed=self.params['fixed'])
+
+        # unpack response
+        coeffs, stand_errs, t_vals, \
+            r_squareds, predicteds, rowid = zip(*gwr_resp)
+        threshold = 0.05
+
+        print("{0}, {1}, {2}, {3}".format('known', 'predicted', 'diff', 'id'))
+        for i, idx in enumerate(self.idx_ids_of_unknowns):
+
+            known_val = self.data[0]['dep_var'][idx]
+            predicted_val = predicteds[i]
+            test_val = abs(known_val - predicted_val) / known_val
+
+            print("{0}, {1}, {2}, {3}".format(self.data[0]['dep_var'][idx],
+                                              predicteds[i],
+                                              test_val,
+                                              rowid[i]))
+            self.assertTrue(test_val < threshold)