def test_rev_node_attr(self):
"""Are node attr preserved when we convert the graphml back to a network?"""
rev = nx.read_graphml(self.path)
mg = self.mg
# Check that node attributes still exist
for n in mg.nodes():
self.assertIn('id', rev.node[n])
self.assertIn('type', rev.node[n])
self.assertIn('records', rev.node[n])
self.assertIsInstance(rev.node[n]['records'], str)
if rev.node[n]['type'] == 'author':
self.assertIn('email', rev.node[n])
# Check that node attributes haven't been added
for n in rev.nodes():
self.assertEqual(len(mg.node[n]), len(rev.node[n]))
# Check that node attribute values are the same
for n in rev.nodes():
for k in rev.node[n]:
if not isinstance(mg.node[n][k], list):
self.assertEqual(mg.node[n][k], rev.node[n][k])
else:
self.assertEqual(";".join(mg.node[n][k]), rev.node[n][k])
python类read_graphml()的实例源码
def test_command_graph_astar(self):
matches = [
((0, 0), (3, 3)),
((1, 11), (4, 14)),
((8, 12), (11, 15)),
]
regions = utils.serialize_json(matches)
result = self.runner.invoke(
cli.graph,
[regions, '-pm', 'astar', self.image_grid]
)
out = nx.parse_graphml(result.output.strip())
graph = nx.read_graphml(fixtures_path('graph_astar.graphml'))
assert nodeset(out) == nodeset(graph)
assert edgeset(out) == edgeset(graph)
def graph2png(infile, outdir, fname=None):
'''
infile: input .gpickle or .graphml file
outdir: path to directory to store output png files
'''
# if file is .gpickle, otherwise load .graphml
try:
graph = nx.read_gpickle(infile)
except:
graph = nx.read_graphml(infile)
# get numpy array equivalent of adjacency matrix
g = nx.adj_matrix(graph).todense()
fig = plt.figure(figsize=(7, 7))
# plot adjacency matrix
p = plt.imshow(g, interpolation='None', cmap='jet')
if fname is None:
fname = os.path.split(infile)[1].split('.')[0] + '.png'
save_location = outdir + fname
plt.savefig(save_location, format='png')
print(fname + ' done!')
def loadGraphs(filenames, verb=False):
"""
Given a list of files, returns a dictionary of graphs
Required parameters:
filenames:
- List of filenames for graphs
Optional parameters:
verb:
- Toggles verbose output statements
"""
# Initializes empty dictionary
if type(filenames) is not list:
filenames = [filenames]
gstruct = OrderedDict()
for idx, files in enumerate(filenames):
if verb:
print("Loading: " + files)
# Adds graphs to dictionary with key being filename
fname = os.path.basename(files)
try:
gstruct[fname] = nx.read_graphml(files)
except:
gstruct[fname] = nx.read_gpickle(files)
return gstruct
def load_gaphml(cls, graph_path):
"""Loads a graph saved as gaphml
Parameters
----------
graph_path: The path of the graph that should be loaded
Returns
-------
NxGraph: Graph object
Examples
--------
>>> g.load_gaphml("graph.gaphml")
"""
return cls(graph_obj=nx.read_graphml(graph_path))
def test_rev(self):
"""Test if taking turning the graphml back into networkx yields the expected edges and nodes"""
rev = nx.read_graphml(self.path)
self.assertSetEqual({'Alice', 'Bobby', 'file01', 'file02', 'file02', 'file03'}, set(rev.nodes()))
self.assertSetEqual({'file01', 'file02'}, set(rev.edge['Alice']))
self.assertSetEqual({'file01', 'file03'}, set(rev.edge['Bobby']))
self.assertSetEqual({'Alice', 'Bobby'}, set(rev.edge['file01']))
self.assertSetEqual({'Alice'}, set(rev.edge['file02']))
self.assertSetEqual({'Bobby'}, set(rev.edge['file03']))
def test_rev_edge_attr(self):
"""Are edge attr preserved when we convert the graphml back to a network?"""
rev = nx.read_graphml(self.path)
mg = self.mg
# Check that edge attributes still exist
for n1, n2, data in self.mg.edges(data=True):
self.assertIn('sha', rev.edge[n1][n2])
# Check that edge attributes are correct (atomic and vector)
for n1, n2, data in mg.edges(data=True):
for k in data:
if not isinstance(data[k], list):
self.assertEqual(data[k], rev.edge[n1][n2][k])
else:
self.assertEqual(gitnet.helpers.list_to_scd(data[k]), rev.edge[n1][n2][k])
owner_graph_metrics.py 文件源码
项目:community-networks-monitoring-tools
作者: netCommonsEU
项目源码
文件源码
阅读 40
收藏 0
点赞 0
评论 0
def initialize_from_files(self, file_list):
""" initialize from a list of graphml files. Each graphml should
contain an "email" field per node that identifies the owner and
a scan_id, scan_time and network global field to identify the scan
and network """
for file in file_list:
try:
g = nx.read_graphml(file)
except:
raise
self.scanTree[g.graph["network"]]["ETX"].append(
[g.graph["scan_id"], g.graph["scan_time"]])
self.routeData[g.graph["network"]][g.graph["scan_id"]]["Graph"] = g
def test_edges_to_graph(self):
graph = nx.read_graphml(fixtures_path('graph.graphml'))
with open(fixtures_path('graph.json'), 'r') as json_graph:
edges = json.load(json_graph)
out = nx.parse_graphml(utils.edges_to_graph(edges))
assert nodeset(out) == nodeset(graph)
assert edgeset(out) == edgeset(graph)
def test_edges_to_graph_graphml(self):
graph = nx.read_graphml(fixtures_path('graph.graphml'))
with open(fixtures_path('graph.json'), 'r') as json_graph:
edges = json.load(json_graph)
out = nx.parse_graphml(utils.edges_to_graph(edges, fmt='graphml'))
assert nodeset(out) == nodeset(graph)
assert edgeset(out) == edgeset(graph)
def from_graphml(cls, path):
"""
Hydrate an instance from a graphml file.
Args:
path (str)
"""
graph = nx.read_graphml(os.path.abspath(path))
return cls(graph)
def load_graphml(data_dir, files):
graphs = []
for i in range(len(files)):
g = nx.read_graphml(join(data_dir,files[i]))
graphs += [g]
return graphs