仿写 networkx 的功能
# -*- coding: cp936 -*-
'''
简单图 Graph：
要求：
关于节点：
功能1.add_node：
通过 add_node 一次性加一个节点
字符串。数字，不论什么能够被哈希的 python 对象都能够当做节点
包含自己定义的类型或者还有一个图（ 构成超图 ）
格式：
>>> G.add_node( 1 )
功能2.add_nodes_from：
通过其它容器增加节点，比方list， dict，set，文件或者还有一个图
格式：
>>> G.add_nodes_from( [2, 3] )
>>> H = Graph()
>>> H.add_path( range( 10 ) )
>>> G.add_nodes_from( H ) # 构成超图
关于边：
功能1.add_edge：
增加一条边
格式：
>>> G.add_edge( 1, 2 ) # 在节点 1 和节点 2 之间加一条边
功能2.add_edges_from：
增加边集
格式：
>>> G.add_edges_from( [ ( 2, 4 ), ( 1, 3 ) ] ) # 通过边集来加边
>>> G.add_edges_from( H.edges() ) # 通过还有一个容器来加边
注：
当加边操作的两个节点不存在时。会自己主动创建出来
关于属性：
每一个图，边。节点都能够拥有 '键值对' 属性
初始时都为空，可是同意被动态增加和改动属性
能够通过 add_node， add_edge 操作。或者直接对节点，边。图的字典进行改动
功能1.节点属性：
>>> G = Graph( name = 'scheme' )
>>> G.graph
{ 'name': 'scheme' }
>>> G.add_node( 1, city = 'Nanjing' )
>>> G.nodes[1]
{ 'city': 'Nanjing' }
>>> G.node
{ 1: {'city': 'Nanjing'} }
>>> G.add_node( 1 ) # 可是再增加同样的节点时，属性不会消失掉
>>> G.node
{1: {'city': 'Nanjing'}}
>>> G.add_node( 1, nowtime = '8:00' ) # 增加同样的节点，带上新的属性时，原来的属性会被更新
>>> G.node
{1: { 'city': 'Nanjing', 'nowtime': '8:00' } }
>>> G.add_node( 1, nowtime = '8:10' )
>>> G.node
{1: { 'city': 'Nanjing', 'nowtime': '8:10' } }
>>> G.nodes[1]['nowtime'] = '10:00' # 可是当节点 1 不存在时。这样写属性须要报错
>>> del G.nodes[1]['nowtime']
>>> G.add_nodes_from( range( 7, 9 ), city = 'shanghai' )
注：
pass
功能2.边的属性：
>>> G.add_edge( 1, 2 weight = 100 )
>>> G.edge
{ 1: { 2: { 'weight': 100 } }, 2: { 1: { 'weight': 100 } } }
>>> G.add_edges_from( [ ( 1, 3 ), ( 2, 3 ) ], weight = 111 )
>>> G.edge
{ 1: { 2: { 'weight': 100 }, 3: { 'weight': 111 } },
2: { 1: { 'weight': 100 }, 3: { 'weight': 111 } },
3: { 1: { 'weight': 111 }, 2: { 'weight': 111 } } }
>>> G.add_edges_from( [ ( 1, 3, { 'weight': 1 } ), ( 3, 4, { 'weight': 2 } ) ] )
>>> G.edge
{ 1: { 2: { 'weight': 100 }, 3: { 'weight': 1 } },
2: { 1: { 'weight': 100 }, 3: { 'weight': 111 } },
3: { 1: { 'weight': 1 }, 2: { 'weight': 111 },
4: { 'weight': 2 } }, 4: { 3: { 'weight': 2 } } }
>>> G.edge[1][2]['weight'] = 111111 # 同意直接操作
或者
>>> G[1][2]['weight'] = 1111
利用 python 的特性提供快捷操作：
比方：
>>> 1 in G
True
>>> [ n for n in G if n < 3 ] # 节点迭代
[1, 2]
>>> len(G) # 节点个数
5
>>> G[1] # 与该点相邻的全部点的属性
{ 2: { 'weight': 100 }, 3: { 'weight': 1 } }
提供 adjacency_iter 来遍历边：
>>> for node, nbrsdict in G.adjacency_iter():
for nbr, attr in nbrsdict.items():
if 'weight' in attr:
( node, nbr, attr['weight'] )
(1, 2, 100)
(1, 3, 1)
(2, 1, 100)
(2, 3, 111)
(3, 1, 1)
(3, 2, 111)
(3, 4, 2)
(4, 3, 2)
>>> [ ( start, end, attr['weight'])
for start, end, attr in G.edges( data = True ) if 'weight' in attr ]
[(1, 2, 100), (1, 3, 1), (2, 3, 111), (3, 4, 2)]
其它一些功能：
pass
'''
class Graph( object ):
def __init__( self, data = None, **attr ):
self.graph = {}
self.node = {}
self.adj = {}
if data is not None:
pass
self.graph.update( attr )
self.edge = self.adj
@property
def name( self ):
return self.graph.get( 'name', '' )
@name.setter
def name( self, newname ):
self.graph['name'] = newname
def __str__( self ):
return self.name
def __iter__( self ):
return iter( self.node )
def __contains__( self, node ):
try:
return node in self.node
except TypeError:
return False
def __len__( self ):
return len( self.node )
def __getitem__( self, node ):
return self.adj[node]
def add_node( self, node, attr_dict = None, **attr ):
if attr_dict is None:
attr_dict = attr
else:
try:
attr_dict.update( attr )
except AttributeError:
raise Exception( "The attr_dict argument must be a dictionary." )
if node not in self.node:
self.adj[node] = {}
self.node[node] = attr_dict
else:
self.node[node].update( attr_dict )
def add_nodes_from( self, nodes, **attr ):
for node in nodes:
try:
newnode = node not in self.node
except TypeError:
node_, node_dict = node
if node_ not in self.node:
self.adj[node_] = {}
newdict = attr.copy()
newdict.update( node_dict )
self.node[node_] = newdict
else:
olddict = self.node[node_]
olddict.update( attr )
olddict.update( node_dict )
continue
if newnode:
self.adj[node] = {}
self.node[node] = attr.copy()
else:
self.node[node].update( attr )
def remove_node( self, start ):
try:
nbrs = list( adj[start].keys() )
del self.node[start]
except KeyError:
raise Exception( "The node %s is not in the graph."%( start ) )
for end in nbrs:
del self.adj[start][end]
del self.adj[start]
def remove_nodes_from( self, nodes ):
for start in nodes:
try:
del self.node[start]
for end in list( self.adj[start].keys() ):
del self.adj[end][start]
del self.adj[start]
except KeyError:
pass
def nodes( self, show_info = False ):
def nodes_iter( show_info = False ):
if show_info:
return iter( self.node.item() )
return iter( self.node )
return list( nodes_iter( show_info = show_info ) )
def number_of_nodes( self ):
return len( self.node )
def order( self ):
return len( self.node )
def has_node( self, node ):
try:
return node in self.node
except TypeError:
return False
def add_edge( self, start, end, attr_dict = None, **attr ):
if attr_dict is None:
attr_dict = attr
else:
try:
attr_dict.update( attr )
except AttributeError:
raise Exception( "The attr_dict argument must be a dictionary." )
if start not in self.node:
self.adj[start] = {}
self.node[start] = {}
if end not in self.node:
self.adj[end] = {}
self.node[end] = {}
data_dict = self.adj[start].get( end, {} )
data_dict.update( attr_dict )
self.adj[start][end] = data_dict
self.adj[end][start] = data_dict
def add_edges_from( self, edges, attr_dict = None, **attr ):
if attr_dict is None:
attr_dict = attr
else:
try:
attr_dict.update( attr )
except AttributeError:
raise Exception( "The attr_dict argument must be a dictionary." )
for edge in edges:
elem_num = len( edge )
if elem_num == 3:
start, end, start_end_dict = edge
elif elem_num == 2:
start, end = edge
else:
raise Exception( "Edge tuple %s must be a 2-tuple or 3-tuple."%( edge ) )
attr_dict.update( start_end_dict )
self.add_edge( start, end, attr_dict )
def remove_edge( self, start, end ):
try:
del self.adj[start][end]
if start != end:
del adj[end][start]
except KeyError:
raise Exception( "The edge %s-%s is not in the graph"%( start,end ) )
def remove_edges_from( self, edges ):
for edge in edges:
start, end = edge[:2]
if start in self.adj and end in self.adj[start]:
del self.adj[start][end]
if start != end:
del self.adj[end][start]
def has_edge( self, start, end ):
try:
return end in self.adj[start]
except KeyError:
return False
def neighbors( self, node ):
try:
return list( self.adj[node] )
except KeyError:
raise Exception( "The node %s is not in the graph."%( node,) )
def neighbors_iter( self, node ):
try:
return iter( self.adj[n] )
except KeyError:
raise Exception( "The node %s is not in the graph."%( node,) )
def edges( self, nodes = None, show_info = False ):
def edges_iter( nodes, show_info = False ):
seen = {}
if nodes is None:
nodes_nbrs = self.adj.items()
else:
nodes_nbrs = ( ( node, self.adj[node] ) for node in self.nodes_container( nodes ) )
if show_info:
for node, nbrs in nodes_nbrs:
for nbr, data in nbrs.items():
if nbr not in seen:
yield ( node, nbr, data )
seen[node] = 1
else:
for node, nbrs in nodes_nbrs:
for nbr, data in nbrs.items():
if nbr not in seen:
yield ( node, nbr )
seen[node] = 1
del seen
return list( edges_iter( nodes, show_info = show_info ) )
def get_edge_data( self, start, end, default = None ):
try:
return self.adj[start][end]
except KeyError:
return default
def adjacency_list( self ):
return list( map( list, iter( self.adj.values() ) ) )
def adjacency_iter( self ):
return iter( self.adj.items() )
def degree( self, nodes_container = None, weight = None ):
'''
功能：
返回一个节点或者一系列节点的度( degree )
參数：
nodes_container： 能够被迭代的容器，可选，默认值为全部节点
weight： 字符串或者为空，可选。默觉得空。
若是为None，则全部的边的权重都设为 1，
degree 则是全部与节点相邻的 weight 权重的边的个数之和
返回值：
字典或者数字
字典： 一系列节点与它们的键值对
数字： 一个指定节点的度
比如：
>>> G = nx.Graph() # or DiGraph, MultiGraph, MultiDiGraph, etc
>>> G.add_path( [ 0, 1, 2, 3 ] )
>>> G.degree(0)
1
>>> G.degree( [ 0, 1 ] )
{ 0: 1, 1: 2 }
>>> list( G.degree( [ 0, 1 ] ).values() )
[ 1, 2 ]
'''
if nodes_container in self:
return next( self.degree_iter( nodes_container, weight ) )[1]
else:
return dict( self.degree_iter( nodes_container, weight ) )
def degree_iter( self, nodes_container = None, weight = None ):
'''
功能：
返回一个( node, degree ) 的迭代对象
參数：
nodes_container： 能够被迭代的容器。可选。默认值为全部节点
weight： 字符串或者为空。可选，默觉得空，
若是为None，则全部的边的权重都设为 1。
degree 则是全部与节点相邻的 weight 权重的边的个数之和
返回值：
返回一个( node, degree ) 的迭代对象
'''
if nodes_container is None:
nodes_nbrs = self.adj.items()
else:
nodes_nbrs = ( ( node, self.adj[node] )
for node in self.nodes_container_iter(
nodes_container ) )
if weight is None:
for node, nbrs in nodes_nbrs:
yield ( node, len( nbrs ) + ( node in nbrs ) )
else:
for node, nbrs in nodes_nbrs:
yield (node, sum( ( nbrs[nbr].get( weight, 1 ) for nbr in nbrs ) ) +
( node in nbrs and nbrs[node].get( weight, 1 ) ) )
def clear( self ):
self.name = ''
self.adj.clear()
self.node.clear()
self.graph.clear()
def copy( self ):
from copy import deepcopy
return deepcopy( self )
def is_multigraph( self ):
return False
def is_directed( self ):
return False
def subgraph( self, nodes ):
from copy import deepcopy
nodes = self.nodes_container_iter( nodes )
H = self.__class__()
H.graph = deepcopy( self.graph )
for node in nodes:
H.node[node] = deepcopy( self.node[node] )
for node in H.node:
H_nbrs = {}
H.adj[node] = H_nbrs
for nbr, attr in self.adj[node].items():
if nbr in H.adj:
H_nbrs[nbr] = attr
H.adj[nbr][node] = attr
return H
def nodes_with_selfloops( self ):
return [ node for node, nbrs in self.adj.items() if node in nbrs ]
def selfloop_edges( self, show_info = False ):
if show_info:
return [ ( node, node, nbrs[node] )
for node, nbrs in self.adj.items() if node in nbrs ]
else:
return [ ( node, node )
for node, nbrs in self.adj.items() if node in nbrs ]
def number_of_selfloops( self ):
return len( self.selfloop_edges() )
def size( self, weight = None ):
s = sum( self.degree( weight = weight ).values() ) / 2
if weight is None:
return int( s )
else:
return float( s )
def number_of_edges( self, start = None, end = None ):
if start is None:
return int( self.size() )
if end in self.adj[start]:
return 1
else:
return 0
def add_path( self, nodes, **attr ):
node_list = list( nodes )
edges = zip( node_list[:-1], node_list[1:] )
self.add_edges_from( edges, **attr )
def add_cycle( self, nodes, **attr ):
node_list = list( nodes )
edges = zip( node_list, node_list[1:] + [node_list[0]] )
self.add_edges_from( edges, **attr )
def nodes_container_iter( self, nodes_container = None ):
'''
功能：
返回存在图中且在 nodes_container 中的节点的迭代对象
參数：
nodes_container： 能够被迭代的容器，可选。默认值为全部节点
返回值：
nodes_iter： iterator
'''
if nodes_container is None:
container = iter( self.adj.keys() )
elif nodes_container in self:
container = iter( [nodes_container] )
else:
def container_iter( node_list, adj ):
try:
for node in node_list:
if node in adj:
yield node
except TypeError as e:
message = e.args[0]
import sys
sys.stdout.write( message )
if 'iter' in message:
raise Exception( "nodes_container is not a node or a sequence of nodes." )
elif 'hashable' in message:
raise Exception( "Node %s in the sequence nbunch is not a valid node."%n )
else:
raise
container = container_iter( nodes_container, self.adj )
return container