1 '''
2 Created on Feb 1, 2011
3 @author: Shreyas Joshi
4 @organization: Georgia Institute of Technology
5 @contact: shreyasj@gatech.edu
6 @summary: This is an implementation of the K nearest neighbor learning algorithm. The implementation is trivial in that the near neighbors are
7 calculated naively- without any smart tricks. Euclidean distance is used to calculate the distance between two points. The implementation
8 also provides some coarse parallelism. If the par_query function is used then the query points are split up equally amongst threads and their
9 near neighbors are calculated in parallel. If the number of threads to use is not specified then no of threads = no of cores as returned by
10 the cpu_count function. This may not be ideal.
11 @status: complete. "mode" untested
12 '''
13
14 import numpy as np
15 import scipy as sc
16 import math
17 import sys
18 import time
19 from multiprocessing import Pool
20 from multiprocessing import cpu_count
21 data = np.zeros (0)
22 import scipy.stats
23
24 -def par_query (allQueries, k, method='mean', noOfThreads=None):
25 '''
26 @summary: Finds the k- nearest nrighbors in parallel. Based on function "query"
27 @param allQueries: is another 2D numpy array. Each row here is one query point. It has no 'y' values. These have to be calculated.
28 @param k: no. of neighbors to consider
29 @param method: method of combining the 'y' values of the nearest neighbors. Default is mean.
30 @param noOfThreads: optional parameter that specifies how many threads to create. Default value: no. of threads = value returned by cpu_count
31 @return: A numpy array with the predicted 'y' values for the query points. The ith element in the array is the 'y' value for the ith query point.
32 '''
33
34
35
36 if (noOfThreads == None):
37 noOfThreads = cpu_count()
38
39
40
41 pool = Pool (processes=noOfThreads)
42
43 resultList = []
44 query_per_thread = allQueries.shape[0] / noOfThreads
45
46
47 for thread_ctr in range (0, noOfThreads - 1):
48 resultList.append(pool.apply_async(query, (allQueries[math.floor(query_per_thread * thread_ctr): (math.floor(query_per_thread * (thread_ctr + 1 ))),:], k,)))
49
50
51
52 resultList.append (pool.apply_async(query , (allQueries[(math.floor(query_per_thread* (noOfThreads - 1))):, :] ,k ,)))
53
54 pool.close()
55 pool.join()
56
57
58
59
60 answer = resultList[0].get()
61
62 for thread_ctr in range (1, noOfThreads):
63 answer = np.hstack((answer, resultList[thread_ctr].get()))
64
65
66
67 return answer
68
69
70
71 -def query(allQueries, k, method='mean'):
72 '''
73 @summary: A serial implementation of k-nearest neighbors.
74 @param allQueries: is another 2D numpy array. Each row here is one query point. It has no 'y' values. These have to be calculated.
75 @param k: no. of neighbors to consider
76 @param method: method of combining the 'y' values of the nearest neighbors. Default is mean.
77 @return: A numpy array with the predicted 'y' values for the query points. The ith element in the array is the 'y' value for the ith query point. If there is more than one mode then only the first mode is returned.
78 '''
79
80 limit = allQueries.shape [0]
81 data_limit = data.shape[0]
82 omitLastCol = data.shape [1] - 1;
83 answer = np.zeros (limit)
84 temp1 = np.zeros ((data.shape[0], (data.shape[1] -1)))
85 temp2= np.zeros (data.shape[0])
86
87 if (allQueries.shape[1] != (data.shape[1] -1) ):
88 print "ERROR: Data and query points are not of the same dimension"
89 raise ValueError
90
91 if (k < 1):
92 print "ERROR: K should be >= 1"
93 raise ValueError
94
95 if (k > data.shape[0]):
96 print "ERROR: K is greater than the total number of data points."
97 raise ValueError
98
99
100 for ctr in range (0, limit):
101
102
103
104
105
106 temp1[0:data_limit, :] = data [0:data_limit ,0:omitLastCol] - allQueries[ctr, :]
107
108
109 temp1 = temp1*temp1;
110 for ctr2 in range (0, data_limit):
111 temp2[ctr2] = math.sqrt(sum (temp1[ctr2,:]))
112
113
114 index = temp2.argsort ()
115
116
117
118
119
120 if (method == 'mean'):
121
122 answer[ctr] = sum (data[ index [0:k], -1]) / k
123
124 if (method == 'median'):
125 answer [ctr] = np.median (data[index [0:k], -1])
126
127 if (method == 'mode'):
128 answer [ctr] = sc.stats.mode (data[index[0:k],-1])[0][0];
129
130
131 return answer
132
133
134
136 '''
137 @summary: This is the funtion to be called to add data. This function can be called multiple times- to add data whenever you like.
138 @note: Any dimensional data can be added the first time. After that- the data must have the same number of columns as the data that was added the first time.
139 @param newData: A 2D numpy array. Each row is a data point and each column is a dimension. The last dimension corresponds to 'y' values.
140 '''
141
142 global data
143 if (data.shape[0] == 0):
144 data = newData
145 else:
146 try:
147 data= np.vstack ((data, newData))
148 except Exception as ex:
149 print "Type of exception: "+ str(type (ex))
150 print "args: " + str(ex.args)
151
152
153
154
156 '''
157 @summary: This function is just for testing. Will not be used as such...
158 '''
159
160
161 a = np.loadtxt ("/nethome/sjoshi42/knn_naive/data/3_D_1000_diskQueryPoints.txt")
162 b= np.loadtxt ("/nethome/sjoshi42/knn_naive/data/3_D_128000_diskDataPoints.txt")
163 addEvidence(b)
164
165 answer = par_query(a, 5 ,'mode')
166
167
168 for i in range (0, answer.shape[0]):
169 print answer[i]
170
171
172
173 print "The answer is: "
174
175 if __name__ == '__main__':
176 main (sys.argv)
177