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

def test_sznajd_model(self):
        g = nx.complete_graph(100)
        model = sm.SznajdModel(g)
        config = mc.Configuration()
        config.add_model_parameter("percentage_infected", 0.2)
        model.set_initial_status(config)
        iterations = model.iteration_bunch(10)
        self.assertEqual(len(iterations), 10)
        iterations = model.iteration_bunch(10, node_status=False)
        self.assertEqual(len(iterations), 10)

        g = nx.complete_graph(100)
        g = g.to_directed()
        model = sm.SznajdModel(g)
        config = mc.Configuration()
        config.add_model_parameter("percentage_infected", 0.2)
        model.set_initial_status(config)
        iterations = model.iteration_bunch(10)
        self.assertEqual(len(iterations), 10)
        iterations = model.iteration_bunch(10, node_status=False)
        self.assertEqual(len(iterations), 10)

def test_cognitive_model(self):
        g = nx.complete_graph(100)
        model = cm.CognitiveOpDynModel(g)
        config = mc.Configuration()
        config.add_model_parameter("I", 0.15)
        config.add_model_parameter("B_range_min", 0)
        config.add_model_parameter("B_range_max", 1)
        config.add_model_parameter("T_range_min", 0)
        config.add_model_parameter("T_range_max", 1)
        config.add_model_parameter("R_fraction_negative", 1.0 / 3)
        config.add_model_parameter("R_fraction_neutral", 1.0 / 3)
        config.add_model_parameter("R_fraction_positive", 1.0 / 3)
        model.set_initial_status(config)

        iterations = model.iteration_bunch(10)
        self.assertEqual(len(iterations), 10)
        iterations = model.iteration_bunch(10, node_status=False)
        self.assertEqual(len(iterations), 10)

def test_kertesz_model_predefined_blocked(self):
        g = nx.complete_graph(100)
        model = ks.KerteszThresholdModel(g)
        config = mc.Configuration()
        config.add_model_parameter('adopter_rate', 0.4)
        predefined_blocked = [0, 1, 2, 3, 4, 5]
        config.add_model_initial_configuration("Blocked", predefined_blocked)
        config.add_model_parameter('percentage_infected', 0.1)

        threshold = 0.2
        for i in g.nodes():
            config.add_node_configuration("threshold", i, threshold)

        model.set_initial_status(config)
        iteration = model.iteration()
        blocked = [x for x, v in future.utils.iteritems(iteration['status']) if v == -1]
        self.assertEqual(blocked, predefined_blocked)

def induced_graph(partition, graph) :
    """Produce the graph where nodes are the communities

    there is a link of weight w between communities if the sum of the weights of the links between their elements is w

    Parameters
    ----------
    partition : dict
       a dictionary where keys are graph nodes and  values the part the node belongs to
    graph : networkx.Graph
        the initial graph

    Returns
    -------
    g : networkx.Graph
       a networkx graph where nodes are the parts

    Examples
    --------
    >>> n = 5
    >>> g = nx.complete_graph(2*n)
    >>> part = dict([])
    >>> for node in g.nodes() :
    >>>     part[node] = node % 2
    >>> ind = induced_graph(part, g)
    >>> goal = nx.Graph()
    >>> goal.add_weighted_edges_from([(0,1,n*n),(0,0,n*(n-1)/2), (1, 1, n*(n-1)/2)])
    >>> nx.is_isomorphic(int, goal)
    True
    """
    ret = nx.Graph()
    ret.add_nodes_from(partition.values())

    for node1, node2, datas in graph.edges_iter(data = True) :
        weight = datas.get("weight", 1)
        com1 = partition[node1]
        com2 = partition[node2]
        w_prec = ret.get_edge_data(com1, com2, {"weight":0}).get("weight", 1)
        ret.add_edge(com1, com2, weight = w_prec + weight)

    return ret

def induced_graph(partition, graph) :
    """Produce the graph where nodes are the communities

    there is a link of weight w between communities if the sum of the weights of the links between their elements is w

    Parameters
    ----------
    partition : dict
       a dictionary where keys are graph nodes and  values the part the node belongs to
    graph : networkx.Graph
        the initial graph

    Returns
    -------
    g : networkx.Graph
       a networkx graph where nodes are the parts

    Examples
    --------
    >>> n = 5
    >>> g = nx.complete_graph(2*n)
    >>> part = dict([])
    >>> for node in g.nodes() :
    >>>     part[node] = node % 2
    >>> ind = induced_graph(part, g)
    >>> goal = nx.Graph()
    >>> goal.add_weighted_edges_from([(0,1,n*n),(0,0,n*(n-1)/2), (1, 1, n*(n-1)/2)])
    >>> nx.is_isomorphic(int, goal)
    True
    """
    ret = nx.Graph()
    ret.add_nodes_from(partition.values())

    for node1, node2, datas in graph.edges_iter(data = True) :
        weight = datas.get("weight", 1)
        com1 = partition[node1]
        com2 = partition[node2]
        w_prec = ret.get_edge_data(com1, com2, {"weight":0}).get("weight", 1)
        ret.add_edge(com1, com2, weight = w_prec + weight)

    return ret

def induced_graph(partition, graph) :
    """Produce the graph where nodes are the communities

    there is a link of weight w between communities if the sum of the weights of the links between their elements is w

    Parameters
    ----------
    partition : dict
       a dictionary where keys are graph nodes and  values the part the node belongs to
    graph : networkx.Graph
        the initial graph

    Returns
    -------
    g : networkx.Graph
       a networkx graph where nodes are the parts

    Examples
    --------
    >>> n = 5
    >>> g = nx.complete_graph(2*n)
    >>> part = dict([])
    >>> for node in g.nodes() :
    >>>     part[node] = node % 2
    >>> ind = induced_graph(part, g)
    >>> goal = nx.Graph()
    >>> goal.add_weighted_edges_from([(0,1,n*n),(0,0,n*(n-1)/2), (1, 1, n*(n-1)/2)])
    >>> nx.is_isomorphic(int, goal)
    True
    """
    ret = nx.Graph()
    ret.add_nodes_from(partition.values())

    for node1, node2, datas in graph.edges_iter(data = True) :
        weight = datas.get("weight", 1)
        com1 = partition[node1]
        com2 = partition[node2]
        w_prec = ret.get_edge_data(com1, com2, {"weight":0}).get("weight", 1)
        ret.add_edge(com1, com2, weight = w_prec + weight)

    return ret

def Link_Prediction(index, G, ebunch=None):
    #G = nx.complete_graph(5)
    if index == "RA":
        Rank_List = resource_allocation_index(G, ebunch)
    if index == "AA":
        Rank_List = adamic_adar_index(G, ebunch)
    if index == "CN":
        Rank_List = common_neighbor_index(G, ebunch)
    if index == "LP":
        Rank_List = local_path_index(G, ebunch)
    if index == "SD":
        Rank_List = structure_dependent_index(G, ebunch)
    return Rank_List

def test_voter_model(self):
        g = nx.complete_graph(100)
        model = vm.VoterModel(g)
        config = mc.Configuration()
        config.add_model_parameter("percentage_infected", 0.2)
        model.set_initial_status(config)
        iterations = model.iteration_bunch(10)
        self.assertEqual(len(iterations), 10)
        iterations = model.iteration_bunch(10, node_status=False)
        self.assertEqual(len(iterations), 10)

def test_majorityrule_model(self):
        g = nx.complete_graph(100)
        model = mrm.MajorityRuleModel(g)
        config = mc.Configuration()
        config.add_model_parameter("q", 3)
        config.add_model_parameter("percentage_infected", 0.2)
        model.set_initial_status(config)
        iterations = model.iteration_bunch(10)
        self.assertEqual(len(iterations), 10)
        iterations = model.iteration_bunch(10, node_status=False)
        self.assertEqual(len(iterations), 10)

def test_qvoter_model(self):
        g = nx.complete_graph(100)
        model = qvm.QVoterModel(g)
        config = mc.Configuration()
        config.add_model_parameter("q", 5)
        config.add_model_parameter("percentage_infected", 0.6)
        model.set_initial_status(config)
        iterations = model.iteration_bunch(10)
        self.assertEqual(len(iterations), 10)
        iterations = model.iteration_bunch(10, node_status=False)
        self.assertEqual(len(iterations), 10)

def get_couling_derivate_matrix(self, h, twist_number, s):
        if type(twist_number) == InPhase:
            G = networkx.complete_graph(self.n)
            d = np.zeros((self.n, self.n))
            for ir in range(self.n):
                ir_neigh = G.neighbors(ir)
                d[ir, ir_neigh] = 1
                d[ir, :] = d[ir, :] / np.sum(d[ir, :])
            return d
        else:
            raise Exception('Topology not compatible with state')

def induced_graph(partition, graph, weight="weight"):
    """Produce the graph where nodes are the communities

    there is a link of weight w between communities if the sum of the weights
    of the links between their elements is w

    Parameters
    ----------
    partition : dict
       a dictionary where keys are graph nodes and  values the part the node
       belongs to
    graph : networkx.Graph
        the initial graph
    weight : str, optional
        the key in graph to use as weight. Default to 'weight'


    Returns
    -------
    g : networkx.Graph
       a networkx graph where nodes are the parts

    Examples
    --------
    >>> n = 5
    >>> g = nx.complete_graph(2*n)
    >>> part = dict([])
    >>> for node in g.nodes() :
    >>>     part[node] = node % 2
    >>> ind = induced_graph(part, g)
    >>> goal = nx.Graph()
    >>> goal.add_weighted_edges_from([(0,1,n*n),(0,0,n*(n-1)/2), (1, 1, n*(n-1)/2)])  # NOQA
    >>> nx.is_isomorphic(int, goal)
    True
    """
    ret = nx.Graph()
    ret.add_nodes_from(partition.values())

    for node1, node2, datas in graph.edges_iter(data=True):
        edge_weight = datas.get(weight, 1)
        com1 = partition[node1]
        com2 = partition[node2]
        w_prec = ret.get_edge_data(com1, com2, {weight: 0}).get(weight, 1)
        ret.add_edge(com1, com2, attr_dict={weight: w_prec + edge_weight})

    return ret

def induced_graph(partition, graph, weight="weight"):
    """Produce the graph where nodes are the communities

    there is a link of weight w between communities if the sum of the weights
    of the links between their elements is w

    Parameters
    ----------
    partition : dict
       a dictionary where keys are graph nodes and  values the part the node
       belongs to
    graph : networkx.Graph
        the initial graph
    weight : str, optional
        the key in graph to use as weight. Default to 'weight'


    Returns
    -------
    g : networkx.Graph
       a networkx graph where nodes are the parts

    Examples
    --------
    >>> n = 5
    >>> g = nx.complete_graph(2*n)
    >>> part = dict([])
    >>> for node in g.nodes() :
    >>>     part[node] = node % 2
    >>> ind = induced_graph(part, g)
    >>> goal = nx.Graph()
    >>> goal.add_weighted_edges_from([(0,1,n*n),(0,0,n*(n-1)/2), (1, 1, n*(n-1)/2)])  # NOQA
    >>> nx.is_isomorphic(int, goal)
    True
    """
    ret = nx.Graph()
    ret.add_nodes_from(partition.values())

    for node1, node2, datas in graph.edges_iter(data=True):
        edge_weight = datas.get(weight, 1)
        com1 = partition[node1]
        com2 = partition[node2]
        w_prec = ret.get_edge_data(com1, com2, {weight: 0}).get(weight, 1)
        ret.add_edge(com1, com2, attr_dict={weight: w_prec + edge_weight})

    return ret

def gnm_random_graph(n, m, seed=None, directed=True):
    """Return the random graph G_{n,m}.

    Produces a graph picked randomly out of the set of all graphs
    with n nodes and m edges.

    Parameters
    ----------
    n : int
        The number of nodes.
    m : int
        The number of edges.
    seed : int, optional
        Seed for random number generator (default=None).
    directed : bool, optional (default=False)
        If True return a directed graph
    """
    if directed:
        G=nx.DiGraph()
        g = nx.DiGraph()
    else:
        G=nx.Graph()
        g = nx.Graph()

    G.add_nodes_from(range(n))
    G.name="gnm_random_graph(%s,%s)"%(n,m)

    if seed is not None:
        random.seed(seed)

    if n==1:
        return G

    max_edges=n*(n-1)

    if not directed:
        max_edges/=2.0
    if m>=max_edges:
        return nx.complete_graph(n,create_using=G)

    nlist=G.nodes()
    edge_count=0
    while edge_count < m:
        # generate random edge,u,v
        u = random.choice(nlist)
        v = random.choice(nlist)
        if u>=v or G.has_edge(u,v):
            continue
        else:
            G.add_edge(u,v)
            edge_count = edge_count+1

    permutation = np.random.permutation(n)
    #print permutation
    new_edges = []
    for e in G.edges():
        u,v = e 
        new_edges.append((permutation[u],permutation[v]))
    g.add_edges_from(new_edges)
    print("is_directed_acyclic_graph: %s" % nx.is_directed_acyclic_graph(g))
    return g

def induced_graph(partition, graph, weight="weight"):
    """Produce the graph where nodes are the communities

    there is a link of weight w between communities if the sum of the weights
    of the links between their elements is w

    Parameters
    ----------
    partition : dict
       a dictionary where keys are graph nodes and  values the part the node
       belongs to
    graph : networkx.Graph
        the initial graph
    weight : str, optional
        the key in graph to use as weight. Default to 'weight'


    Returns
    -------
    g : networkx.Graph
       a networkx graph where nodes are the parts

    Examples
    --------
    >>> n = 5
    >>> g = nx.complete_graph(2*n)
    >>> part = dict([])
    >>> for node in g.nodes() :
    >>>     part[node] = node % 2
    >>> ind = induced_graph(part, g)
    >>> goal = nx.Graph()
    >>> goal.add_weighted_edges_from([(0,1,n*n),(0,0,n*(n-1)/2), (1, 1, n*(n-1)/2)])  # NOQA
    >>> nx.is_isomorphic(int, goal)
    True
    """
    ret = nx.Graph()
    ret.add_nodes_from(partition.values())

    for node1, node2, datas in graph.edges_iter(data=True):
        edge_weight = datas.get(weight, 1)
        com1 = partition[node1]
        com2 = partition[node2]
        w_prec = ret.get_edge_data(com1, com2, {weight: 0}).get(weight, 1)
        ret.add_edge(com1, com2, attr_dict={weight: w_prec + edge_weight})

    return ret

def test_config(self):
        g = nx.erdos_renyi_graph(99, 0.1)
        model = th.ThresholdModel(g)
        config = mc.Configuration()
        config.add_model_parameter('percentage_infected', 0.1)
        config.add_model_initial_configuration("Infected", [1, 2, 3])
        config.add_node_set_configuration("partial", {1: 1, 2: 2})
        try:
            model.set_initial_status(config)
        except:
            pass

        config.add_edge_set_configuration("partial", {e: 1 for e in list(g.edges)[:10]})
        try:
            model.set_initial_status(config)
        except:
            pass
        config.add_node_set_configuration("partial", {n: 1 for n in g.nodes})
        config.add_edge_set_configuration("partial", {e: 1 for e in g.edges})
        model.set_initial_status(config)

        g = nx.complete_graph(100)
        model = mrm.MajorityRuleModel(g)
        config = mc.Configuration()
        config.add_model_parameter("percentage_infected", 0.2)
        try:
            model.set_initial_status(config)
        except:
            pass

        g = nx.erdos_renyi_graph(1000, 0.1)
        model = ids.IndependentCascadesModel(g)
        config = mc.Configuration()
        config.add_model_parameter('percentage_infected', 0.1)
        try:
            model.set_initial_status(config)
        except:
            pass

        g = nx.erdos_renyi_graph(1000, 0.1)
        model = th.ThresholdModel(g)
        config = mc.Configuration()
        config.add_model_parameter('percentage_infected', 0.1)
        try:
            model.set_initial_status(config)
        except:
            pass