master-thesis/RecipeAnalysis/AdjacencyMatrix.py

306 lines
8.1 KiB
Python

#!/usr/bin/env python3
# coding: utf-8
# # Adjacency Matrix
import numpy as np
from scipy.sparse import csr_matrix, lil_matrix, coo_matrix
class adj_matrix(object):
def __init__(self, symmetric_indices=False):
self._sym = symmetric_indices
if not symmetric_indices:
self._x_labels = []
self._y_labels = []
self._x_label_index={}
self._y_label_index={}
else:
self._labels = []
self._label_index={}
self._x = []
self._y = []
self._data = []
self._mat = None
self._csr = None
# for a TF-IDF like approach we need also a counter how frequently ingredients
# and actions appear in documents.
self._current_document_labels = set()
self._label_document_count = {}
self._document_count = 0
# building type dependend functions:
self._build_funcs()
def _get_ix(self, label):
i = self._x_label_index.get(label)
if i is None:
i = len(self._x_labels)
self._x_labels.append(label)
self._x_label_index[label] = i
return i
def _get_iy(self, label):
i = self._y_label_index.get(label)
if i is None:
i = len(self._y_labels)
self._y_labels.append(label)
self._y_label_index[label] = i
return i
def _get_i(self, label):
i = self._label_index.get(label)
if i is None:
i = len(self._labels)
self._labels.append(label)
self._label_index[label] = i
return i
def _end_document(self):
self._document_count += 1
# adding all seen labels to our counter:
for label in self._current_document_labels:
self._label_document_count[label] += 1
else:
self._label_document_count[label] = 1
self._current_document_labels = set()
def apply_threshold(self, min_count=5):
csr = self.get_csr()
new_x = []
new_y = []
new_data = []
for i in range(len(self._data)):
if csr[self._x[i],self._y[i]] >= min_count:
new_x.append(self._x[i])
new_y.append(self._y[i])
new_data.append(self._data[i])
self._x = new_x
self._y = new_y
self._data = new_data
def next_document(self):
self._end_document()
def add_entry(self, x, y, data):
if self._sym:
ix = self._get_i(x)
iy = self._get_i(y)
else:
ix = self._get_ix(x)
iy = self._get_iy(y)
self._x.append(ix)
self._y.append(iy)
self._data.append(data)
self._current_document_labels.add(x)
self._current_document_labels.add(y)
def compile(self):
self._csr = self.get_csr()
if self._sym:
self._np_labels = np.array(self._labels)
else:
self._np_x_labels = np.array(self._x_labels)
self._np_y_labels = np.array(self._y_labels)
def compile_to_mat(self):
if self._sym:
sx = len(self._labels)
sy = len(self._labels)
else:
sx = len(self._x_labels)
sy = len(self._y_labels)
self._mat = coo_matrix((self._data, (self._x, self._y)), shape=(sx,sy))
return self._mat
def get_csr(self):
return self.compile_to_mat().tocsr()
def get_labels(self):
if self._sym:
return self._labels
return self._x_labels, self._y_labels
def _build_funcs(self):
def get_sym_adjacent(key):
assert self._csr is not None
c = self._csr
index = self._label_index[key]
i1 = c[index,:].nonzero()[1]
i2 = c[:,index].nonzero()[0]
i = np.concatenate((i1,i2))
names = self._np_labels[i]
counts = np.concatenate((c[index, i1].toarray().flatten(), c[i2, index].toarray().flatten()))
s = np.argsort(-counts)
return names[s], counts[s]
def get_forward_adjacent(key):
assert self._csr is not None
c = self._csr
index = self._x_label_index[key]
i = c[index,:].nonzero()[1]
names = self._np_y_labels[i]
counts = c[index, i].toarray().flatten()
s = np.argsort(-counts)
return names[s], counts[s]
def get_backward_adjacent(key):
assert self._csr is not None
c = self._csr
index = self._y_label_index[key]
i = c[:,index].nonzero()[0]
names = self._np_x_labels[i]
counts = c[i, index].toarray().flatten()
s = np.argsort(-counts)
return names[s], counts[s]
# sum functions:
def sym_sum(key):
return np.sum(self.get_adjacent(key)[1])
def fw_sum(key):
return np.sum(self.get_forward_adjacent(key)[1])
def bw_sum(key):
return np.sum(self.get_backward_adjacent(key)[1])
# normalization stuff:
def fw_normalization_factor(key, quotient_func):
assert self._csr is not None
c = self._csr
ia = self._x_label_index[key]
occurances = c[ia,:].nonzero()[1]
return 1. / quotient_func(c[ia,occurances].toarray())
def bw_normalization_factor(key, quotient_func):
assert self._csr is not None
c = self._csr
ib = m._y_label_index[key]
occurances = c[:,ib].nonzero()[0]
return 1. / quotient_func(c[occurances,ib].toarray())
def sym_normalization_factor(key, quotient_func):
assert self._csr is not None
c = self._csr
ii = m._label_index[key]
fw_occurances = c[ii,:].nonzero()[1]
bw_occurances = c[:,ii].nonzero()[0]
return 1. / quotient_func(np.concatenate(
[c[ii,fw_occurances].toarray().flatten(),
c[bw_occurances,ii].toarray().flatten()]
))
def sym_p_a_given_b(key_a, key_b, quot_func = np.max):
assert self._csr is not None
c = self._csr
ia = m._label_index[key_a]
ib = m._label_index[key_b]
v = c[ia,ib] + c[ib,ia]
return v * self.sym_normalization_factor(key_b, quot_func)
def fw_p_a_given_b(key_a, key_b, quot_func = np.max):
assert self._csr is not None
c = self._csr
ia = m._x_label_index[key_a]
ib = m._y_label_index[key_b]
v = c[ia,ib]
return v * self.bw_normalization_factor(key_b, quot_func)
def bw_p_a_given_b(key_a, key_b, quot_func = np.max):
assert self._csr is not None
c = self._csr
ia = m._y_label_index[key_a]
ib = m._x_label_index[key_b]
v = c[ib,ia]
return v * self.fw_normalization_factor(key_b, quot_func)
if self._sym:
self.get_adjacent = get_sym_adjacent
self.get_sum = sym_sum
self.get_sym_normalization_factor = sym_normalization_factor
self.p_a_given_b = sym_p_a_given_b
else:
self.get_forward_adjacent = get_forward_adjacent
self.get_backward_adjacent = get_backward_adjacent
self.get_fw_sum = fw_sum
self.get_bw_sum = bw_sum
self.get_fw_normalization_factor = fw_normalization_factor
self.get_bw_normalization_factor = bw_normalization_factor
self.fw_p_a_given_b = fw_p_a_given_b
self.bw_p_a_given_b = bw_p_a_given_b