def save_graph(self, graphname, fmt='gpickle'):
"""
Saves the graph to disk
**Positional Arguments:**
graphname:
- Filename for the graph
**Optional Arguments:**
fmt:
- Output graph format
"""
self.g.graph['ecount'] = nx.number_of_edges(self.g)
if fmt == 'gpickle':
nx.write_gpickle(self.g, graphname)
elif fmt == 'graphml':
nx.write_graphml(self.g, graphname)
else:
raise ValueError('graphml is the only format currently supported')
pass
python类write_gpickle()的实例源码
def save_graph(self, save_path, file_type="pickle"):
""" Saves the graph into a file
Parameters
----------
file_type
save_path: string,
The path where the file shpuld be saved
Examples
--------
>>> g.save_graph("Facebook.", "graphml")
"""
if file_type == "pickle":
nx.write_gpickle(self._graph, save_path)
elif file_type == "sgraph":
self.save_as_sgraph(save_path)
elif file_type == "graphml":
nx.write_graphml(self._graph, save_path)
else:
msg = "The file type %s is unknown" % file_type
raise TypeError(msg)
def cache_write(self, cache_folder):
os.makedirs(cache_folder)
pickle.dump(self.contentProvider,
open(cache_folder + '/contentProvider.cache', 'wb'),
protocol=2)
pickle.dump(self.contentNodes,
open(cache_folder + '/contentNodes.cache', 'wb'),
protocol=2)
pickle.dump(self.accessNodes,
open(cache_folder + '/accessNodes.cache', 'wb'),
protocol=2)
nx.write_gpickle(self.netGraph, cache_folder + '/asGraph.cache')
pickle.dump(self.as2ip,
open(cache_folder + '/as2ip.cache', 'wb'),
protocol=2)
return None
def tearDown(self):
nx.write_gpickle = self.real_write_gpickle
dbt.utils.dependency_projects = self.real_dependency_projects
dbt.clients.system.find_matching = self.real_find_matching
dbt.clients.system.load_file_contents = self.real_load_file_contents
def write_graph(self, outfile):
nx.write_gpickle(self.graph, outfile)
def draw_graph(mst, mst_a, mol_names, dir_names, method):
import networkx as nx
G = nx.from_scipy_sparse_matrix(mst)
if method == 'mcs':
corr = 1
else:
corr = 0
for i, j in zip(mst.nonzero()[0], mst.nonzero()[1]):
G.edge[i][j]['label'] = '%.1f' % (mst_a[i][j] - corr)
G.edge[i][j]['len'] = '3.0'
for n in G.nodes():
G.node[n]['shape'] = 'box'
G.node[n]['label'] = ('<'
'<table border="0" cellspacing="-20" cellborder="0">'
'<tr><td><img src="%s"/></td></tr>'
'<tr><td bgcolor="#F0F0F0">%s</td></tr>'
'</table>>' % (os.path.join(dir_names[n],
mol_names[n] + os.extsep + 'svg'),
mol_names[n]) )
print('Writing networkx graph pickle file %s...' % GPICKLE_FILE)
nx.write_gpickle(G, GPICKLE_FILE)
print('Writing DOT file %s...' % DOT_FILE)
nx.write_dot(G, DOT_FILE)
def saveRealGraphSeries(G, file_prefix='graphs/day_'):
for idx in range(len(G)):
f_name = file_prefix + str(idx) + "_graph.gpickle"
# cPickle.dump(G[idx], open(f_name, 'wb'))
nx.write_gpickle(G[idx], f_name)
def saveDynamicSBmGraph(file_perfix, dynamic_graphs):
length = len(dynamic_graphs)
graph_files = ['%s_%d_graph.gpickle' % (file_perfix, i) for i in xrange(length)]
info_files = ['%s_%d_node.pkl' % (file_perfix, i) for i in xrange(length)]
for i in xrange(length):
# save graph
nx.write_gpickle(dynamic_graphs[i][0], graph_files[i])
# save additional node info
with open(info_files[i], 'wb') as fp:
node_infos = {}
node_infos['community'] = dynamic_graphs[i][1]
node_infos['perturbation'] = dynamic_graphs[i][2]
pickle.dump(node_infos, fp)
def setUp(self):
dbt.flags.STRICT_MODE = True
def mock_write_gpickle(graph, outfile):
self.graph_result = graph
self.real_write_gpickle = nx.write_gpickle
nx.write_gpickle = mock_write_gpickle
self.graph_result = None
self.profiles = {
'test': {
'outputs': {
'test': {
'type': 'postgres',
'threads': 4,
'host': 'database',
'port': 5432,
'user': 'root',
'pass': 'password',
'dbname': 'dbt',
'schema': 'dbt_test'
}
},
'target': 'test'
}
}
self.real_dependency_projects = dbt.utils.dependency_projects
dbt.utils.dependency_projects = MagicMock(return_value=[])
self.mock_models = []
self.mock_content = {}
def mock_find_matching(root_path, relative_paths_to_search,
file_pattern):
if 'sql' not in file_pattern:
return []
to_return = []
if 'models' in relative_paths_to_search:
to_return = to_return + self.mock_models
return to_return
self.real_find_matching = dbt.clients.system.find_matching
dbt.clients.system.find_matching = MagicMock(
side_effect=mock_find_matching)
def mock_load_file_contents(path):
return self.mock_content[path]
self.real_load_file_contents = dbt.clients.system.load_file_contents
dbt.clients.system.load_file_contents = MagicMock(
side_effect=mock_load_file_contents)
