Python networkx 模块，write_gpickle() 实例源码

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

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)