|
| 1 | + |
| 2 | +import logging |
| 3 | +from multiprocessing import Process, Queue, current_process |
| 4 | + |
| 5 | + |
| 6 | +logger = logging.getLogger(__name__) |
| 7 | + |
| 8 | + |
| 9 | +class ItemItemMatrix(object): |
| 10 | + """Object representation of the item-item matrix |
| 11 | + """ |
| 12 | + |
| 13 | + def __init__(self, data, combinfunc, symmetric=False, diagonal=None, num_processes=1): |
| 14 | + """Takes a list of data and generates a 2D-matrix using the supplied |
| 15 | + combination function to calculate the values. |
| 16 | +
|
| 17 | + PARAMETERS |
| 18 | + data - the list of items |
| 19 | + combinfunc - the function that is used to calculate teh value in a |
| 20 | + cell. It has to cope with two arguments. |
| 21 | + symmetric - Whether it will be a symmetric matrix along the diagonal. |
| 22 | + For example, if the list contains integers, and the |
| 23 | + combination function is abs(x-y), then the matrix will |
| 24 | + be symmetric. |
| 25 | + Default: False |
| 26 | + diagonal - The value to be put into the diagonal. For some |
| 27 | + functions, the diagonal will stay constant. An example |
| 28 | + could be the function "x-y". Then each diagonal cell |
| 29 | + will be "0". If this value is set to None, then the |
| 30 | + diagonal will be calculated. Default: None |
| 31 | + num_processes |
| 32 | + - If you want to use multiprocessing to split up the work |
| 33 | + and run combinfunc() in parallel, specify num_processes |
| 34 | + > 1 and this number of workers will be spun up, the work |
| 35 | + split up amongst them evenly. Default: 1 |
| 36 | + """ |
| 37 | + self.data = data |
| 38 | + self.combinfunc = combinfunc |
| 39 | + self.symmetric = symmetric |
| 40 | + self.diagonal = diagonal |
| 41 | + self.num_processes = num_processes |
| 42 | + self.use_multiprocessing = num_processes > 1 |
| 43 | + if self.use_multiprocessing: |
| 44 | + self.task_queue = Queue() |
| 45 | + self.done_queue = Queue() |
| 46 | + self.genmatrix() |
| 47 | + |
| 48 | + def worker(self): |
| 49 | + """Multiprocessing task function run by worker processes |
| 50 | + """ |
| 51 | + tasks_completed = 0 |
| 52 | + for task in iter(self.task_queue.get, 'STOP'): |
| 53 | + col_index, item, item2 = task |
| 54 | + result = (col_index, self.combinfunc(item, item2)) |
| 55 | + self.done_queue.put(result) |
| 56 | + tasks_completed += 1 |
| 57 | + logger.info("Worker %s performed %s tasks", |
| 58 | + current_process().name, |
| 59 | + tasks_completed) |
| 60 | + |
| 61 | + def genmatrix(self): |
| 62 | + self.matrix = [] |
| 63 | + logger.info("Generating matrix for %s items - O(n^2)", len(self.data)) |
| 64 | + if self.use_multiprocessing: |
| 65 | + logger.info("Using multiprocessing on %s processes!", self.num_processes) |
| 66 | + |
| 67 | + if self.use_multiprocessing: |
| 68 | + logger.info("Spinning up %s workers", self.num_processes) |
| 69 | + processes = [Process(target=self.worker) for i in xrange(self.num_processes)] |
| 70 | + [process.start() for process in processes] |
| 71 | + |
| 72 | + for row_index, item in enumerate(self.data): |
| 73 | + logger.debug("Generating row %s/%s (%0.2f%%)", |
| 74 | + row_index, |
| 75 | + len(self.data), |
| 76 | + 100.0 * row_index / len(self.data)) |
| 77 | + row = {} |
| 78 | + if self.use_multiprocessing: |
| 79 | + num_tasks_queued = num_tasks_completed = 0 |
| 80 | + for col_index, item2 in enumerate(self.data): |
| 81 | + if self.diagonal is not None and col_index == row_index: |
| 82 | + # This is a cell on the diagonal |
| 83 | + row[col_index] = self.diagonal |
| 84 | + elif self.symmetric and col_index < row_index: |
| 85 | + # The matrix is symmetric and we are "in the lower left |
| 86 | + # triangle" - fill this in after (in case of multiprocessing) |
| 87 | + pass |
| 88 | + # Otherwise, this cell is not on the diagonal and we do indeed |
| 89 | + # need to call combinfunc() |
| 90 | + elif self.use_multiprocessing: |
| 91 | + # Add that thing to the task queue! |
| 92 | + self.task_queue.put((col_index, item, item2)) |
| 93 | + num_tasks_queued += 1 |
| 94 | + # Start grabbing the results as we go, so as not to stuff all of |
| 95 | + # the worker args into memory at once (as Queue.get() is a |
| 96 | + # blocking operation) |
| 97 | + if num_tasks_queued > self.num_processes: |
| 98 | + col_index, result = self.done_queue.get() |
| 99 | + row[col_index] = result |
| 100 | + num_tasks_completed += 1 |
| 101 | + else: |
| 102 | + # Otherwise do it here, in line |
| 103 | + row[col_index] = self.combinfunc(item, item2) |
| 104 | + |
| 105 | + if self.symmetric: |
| 106 | + # One more iteration to get symmetric lower left triangle |
| 107 | + for col_index, item2 in enumerate(self.data): |
| 108 | + if col_index >= row_index: |
| 109 | + break |
| 110 | + # post-process symmetric "lower left triangle" |
| 111 | + row[col_index] = self.matrix[col_index][row_index] |
| 112 | + |
| 113 | + if self.use_multiprocessing: |
| 114 | + # Grab the remaining worker task results |
| 115 | + while num_tasks_completed < num_tasks_queued: |
| 116 | + col_index, result = self.done_queue.get() |
| 117 | + row[col_index] = result |
| 118 | + num_tasks_completed += 1 |
| 119 | + |
| 120 | + row_indexed = [row[index] for index in xrange(len(self.data))] |
| 121 | + self.matrix.append(row_indexed) |
| 122 | + |
| 123 | + if self.use_multiprocessing: |
| 124 | + logger.info("Stopping/joining %s workers", self.num_processes) |
| 125 | + [self.task_queue.put('STOP') for i in xrange(self.num_processes)] |
| 126 | + [process.join() for process in processes] |
| 127 | + |
| 128 | + logger.info("Matrix generated") |
| 129 | + |
| 130 | + def printmatrix(self): |
| 131 | + """ |
| 132 | + Prints out a 2-dimensional list of data cleanly. |
| 133 | + This is useful for debugging. |
| 134 | +
|
| 135 | + PARAMETERS |
| 136 | + data - the 2D-list to display |
| 137 | + """ |
| 138 | + # determine maximum length |
| 139 | + maxlen = 0 |
| 140 | + colcount = len(self.data[0]) |
| 141 | + for col in self.data: |
| 142 | + for cell in col: |
| 143 | + maxlen = max(len(str(cell)), maxlen) |
| 144 | + # print data |
| 145 | + format = " %%%is |" % maxlen |
| 146 | + format = "|" + format * colcount |
| 147 | + for row in self.data: |
| 148 | + print format % tuple(row) |
0 commit comments