Source code for tf_G.graph.graph

import numpy as np
import tensorflow as tf

from tf_G.utils.callbacks.update_edge_notifier import UpdateEdgeNotifier
from tf_G.utils.tensorflow_object import TensorFlowObject, TF_type


def __str__(self) -> str:
  return str(self.run_tf(self.L_tf))


[docs]class Graph(TensorFlowObject, UpdateEdgeNotifier): """ Graph class implemented in the top of TensorFlow. The class codifies the graph using an square matrix of 2-D shape and provides functionality operating with this matrix. Attributes: sess (:obj:`tf.Session`): This attribute represents the session that runs the TensorFlow operations. name (str): This attribute represents the name of the object in TensorFlow's op Graph. writer (:obj:`tf.summary.FileWriter`): This attribute represents a TensorFlow's Writer, that is used to obtain stats. The default value is `None`. _listeners (:obj:`set`): The set of objects that will be notified when an edge modifies it weight. n (int): Represents the cardinality of the vertex set as Python `int`. n_tf (:obj:`tf.Tensor`): Represents the cardinality of the vertex set as 0-D Tensor. m (int): Represents the cardinality of the edge set as Python `int`. A_tf (:obj:`tf.Tensor`): Represents the Adjacency matrix of the graph as 2-D Tensor with shape [n,n]. out_degrees_tf (:obj:`tf.Tensor`): Represents the out-degrees of the vertices of the graph as 2-D Tensor with shape [n, 1] in_degrees_tf (:obj:`tf.Tensor`): Represents the in-degrees of the vertices of the graph as 2-D Tensor with shape [1, n] """
[docs] def __init__(self, sess: tf.Session, name: str, writer: tf.summary.FileWriter = None, edges_np: np.ndarray = None, n: int = None, is_sparse: bool = False) -> None: """ Class Constructor of the Graph This method is called to construct a Graph object. This block of code initializes all the variables necessaries for this class to properly works. This class can be initialized using an edge list, that fill the graph at this moment, or can be construct it from the cardinality of vertices set given by `n` parameter. Args: sess (:obj:`tf.Session`): This attribute represents the session that runs the TensorFlow operations. name (str): This attribute represents the name of the object in TensorFlow's op Graph. writer (:obj:`tf.summary.FileWriter`, optional): This attribute represents a TensorFlow's Writer, that is used to obtain stats. The default value is `None`. edges_np (:obj:`np.ndarray`, optional): The edge set of the graph codifies as `edges_np[:,0]` represents the sources and `edges_np[:,1]` the destinations of the edges. The default value is `None`. n (int, optional): Represents the cardinality of the vertex set. The default value is `None`. is_sparse (bool, optional): Use sparse Tensors if it's set to `True`. The default value is False` Not implemented yet. Show the Todo for more information. Todo: * Implement variables as sparse when it's possible. Waiting to TensorFlow for it. """ TensorFlowObject.__init__(self, sess, name, writer, is_sparse) UpdateEdgeNotifier.__init__(self) if edges_np is not None: if n is not None: self.n = max(n, int(edges_np.max(axis=0).max() + 1)) else: self.n = int(edges_np.max(axis=0).max() + 1) self.m = int(edges_np.shape[0]) A_init = tf.scatter_nd(edges_np.tolist(), self.m * [1.0], [self.n, self.n]) elif n is not None: self.n = n self.m = 0 A_init = tf.zeros([self.n, self.n]) else: raise ValueError('Graph constructor must be have edges or n') self.n_tf = tf.Variable(float(self.n), tf.float32, name=self.name + "_n") self.A_tf = tf.Variable(A_init, tf.float64, name=self.name + "_A") self.out_degrees_tf = tf.Variable( tf.reduce_sum(self.A_tf, 1, keep_dims=True), name=self.name + "_d_out") self.in_degrees_tf = tf.Variable( tf.reduce_sum(self.A_tf, 0, keep_dims=True), name=self.name + "_d_in") self.run_tf(tf.variables_initializer([self.A_tf, self.n_tf])) self.run_tf(tf.variables_initializer([ self.out_degrees_tf, self.in_degrees_tf]))
def __str__(self) -> str: """ Transforms the graph to a string. This method is used to print the graph on the command line. It codifies the laplacian matrix of the graph as string. Returns: (str): representing the laplacian matrix to visualize it. """ return str(self.run_tf(self.L_tf)) @property def L_tf(self) -> tf.Tensor: """ This method returns the Laplacian of the graph. The method generates a 2-D Array containing the laplacian matrix of the graph Returns: (:obj:`tf.Tensor`): A 2-D Tensor with [n,n] shape where n is the cardinality of the vertex set """ return tf.diag(self.out_degrees_tf_vector) - self.A_tf @property def is_not_sink_tf(self) -> tf.Tensor: """ This method returns if a vertex is a sink vertex as vector. The method generates a 1-D Tensor containing the boolean values that indicates if the vertex at position `i` is a sink vertex. Returns: (:obj:`tf.Tensor`): A 1-D Tensor with the same length as cardinality of the vertex set. """ return tf.not_equal(self.out_degrees_tf_vector, 0)
[docs] def is_not_sink_tf_vertex(self, vertex: int) -> TF_type: """ This method returns if a vertex is a sink vertex as vector. The method generates a 1-D Tensor containing the boolean values that indicates if the vertex at position `i` is a sink vertex. Args: vertex (int): The index of the vertex that wants to know if is sink. Returns: (:obj:`tf.Tensor`): A 0-D Tensor that represents if a vertex is a sink vertex """ return tf.not_equal( tf.reshape([self.out_degrees_tf_vertex(vertex)], [1]), 0)
@property def out_degrees_np(self) -> np.ndarray: """ This method returns the degree of all vertex as vector. The method generates a 1-D Array containing the out-degree of the vertex `i` at position `i` Returns: (:obj:`np.ndarray`): A 1-D Array with the same length as cardinality of the vertex set. """ return self.run_tf(self.out_degrees_tf)
[docs] def out_degrees_tf_vertex(self, vertex: int) -> tf.Tensor: """ This method returns the degree of all vertex as vector. The method generates a 0-D Array containing the out-degree of the vertex i. Args: vertex (int): The index of the vertex that wants the degree. Returns: (:obj:`np.ndarray`): A 1-D Array with the same length as cardinality of the vertex set. """ return tf.gather(self.out_degrees_tf, [vertex])
@property def edge_list_tf(self) -> tf.Tensor: """ Method that returns the edge set of the graph as list. This method return all the edges of the graph codified as 2-D matrix in which the first dimension represents each edge and second dimension the source and destination vertices of each edge. Returns: (:obj:`tf.Tensor`): A 2-D Tensor with the he same length as cardinality of the edge set in the first dimension and 2 in the second. """ return tf.cast(tf.where(tf.not_equal(self.A_tf, 0)), tf.int64) @property def edge_list_np(self) -> np.ndarray: """ Method that returns the edge set of the graph as list. This method return all the edges of the graph codified as 2-D matrix in which the first dimension represents each edge and second dimension the source and destination vertices of each edge. Returns: (:obj:`np.ndarray`): A 2-D Array with the he same length as cardinality of the edge set in the first dimension and 2 in the second. """ return self.run_tf(self.edge_list_tf) @property def L_pseudo_inverse_tf(self) -> tf.Tensor: """ Method that returns the pseudo inverse of the Laplacian matrix. This method calculates the pseudo inverse matrix of the Laplacian of the Graph. It generates a matrix of the same shape as the Laplacian matrix, i.e. [n, n] where n is the cardinality of the vertex set. Returns: (:obj:`tf.Tensor`): A 2-D square Tensor with the he same length as cardinality of the vertex set representing the laplacian pseudo inverse. """ return tf.py_func(np.linalg.pinv, [self.L_tf], tf.float32)
[docs] def A_tf_vertex(self, vertex: int) -> tf.Tensor: """ Method that returns the adjacency of an individual vertex. This method extracts the corresponding row referred to the `vertex` passed as parameter. It constructs a vector that contains the weight of the edge between `vertex` (obtained as parameter) and the vertex at position `i` in the vector. Args: vertex (int): The index of the vertex that wants the degree. Returns: (:obj:`tf.Tensor`): A 1-D Tensor with the same length as the cardinality of the vertex set. """ return tf.gather(self.A_tf, [vertex])
@property def in_degrees_np(self) -> np.ndarray: """ This method returns the in-degree of all vertex as vector. The method generates a 1-D Array containing the in-degree of the vertex `i` at position `i` Returns: (:obj:`np.ndarray`): A 1-D Array with the same length as cardinality of the vertex set. """ return self.run_tf(self.in_degrees_tf) @property def in_degrees_tf_vector(self): """ The in-degrees of the vertices of the graph Method that returns the in-degrees of the vertices of the graph as 1-D Tensor with shape [n] Returns: (:obj:`tf.Tensor`): A 1-D Tensor with the same length as the cardinality of the vertex set. """ return tf.reshape(self.in_degrees_tf, [self.n]) @property def out_degrees_tf_vector(self): """ The out-degrees of the vertices of the graph Method that returns the out-degrees of the vertices of the graph as 1-D Tensor with shape [n] Returns: (:obj:`tf.Tensor`): A 1-D Tensor with the same length as the cardinality of the vertex set. """ return tf.reshape(self.out_degrees_tf, [self.n])
[docs] def append(self, src: int, dst: int) -> None: """ Append an edge to the graph. This method process an input edge adding it to the graph updating all the variables necessaries to maintain the graph in correct state. Args: src (int): The id of the source vertex of the edge. dst (int): The id of the destination vertex of the edge. Returns: This method returns nothing. """ if src and dst is None: raise ValueError( "tf_G and dst must not be None ") self.run_tf([tf.scatter_nd_add(self.A_tf, [[src, dst]], [1.0]), tf.scatter_nd_add(self.out_degrees_tf, [[src, 0]], [1.0]), tf.scatter_nd_add(self.in_degrees_tf, [[0, dst]], [1.0])]) self.m += 1 self._notify(np.array([src, dst]), 1)
[docs] def remove(self, src: int, dst: int) -> None: """ Remove an edge to the graph. This method process an input edge deleting it to the graph updating all the variables necessaries to maintain the graph in correct state. Args: src (int): The id of the source vertex of the edge. dst (int): The id of the destination vertex of the edge. Returns: This method returns nothing. """ if src and dst is None: raise ValueError( "tf_G and dst must not be None ") self.run_tf([tf.scatter_nd_add(self.A_tf, [[src, dst]], [-1.0]), tf.scatter_nd_add(self.out_degrees_tf, [[src, 0]], [-1.0]), tf.scatter_nd_add(self.in_degrees_tf, [[0, dst]], [-1.0])]) self.m -= 1 self._notify(np.array([src, dst]), -1)