def _preprocess_transition_probabilities()

in graspologic/embed/n2v.py [0:0]

• 55 lines of code
• 10 McCabe index (conditional complexity)

    def _preprocess_transition_probabilities(self, weight_default: float = 1.0) -> None:
        """
        Preprocessing of transition probabilities for guiding the random walks.
        """
        graph = self.graph
        is_directed = self.is_directed

        alias_nodes = {}
        total_nodes = len(graph.nodes())
        bucket = 0
        current_node = 0
        quotient = int(total_nodes / 10)

        logging.info(
            f"Beginning preprocessing of transition probabilities for {total_nodes} vertices"
        )
        for node in graph.nodes():
            current_node += 1
            if current_node > bucket * quotient:
                bucket += 1
                logging.info(f"Completed {current_node} / {total_nodes} vertices")

            unnormalized_probs = [
                graph[node][nbr].get("weight", weight_default)
                for nbr in sorted(graph.neighbors(node))
            ]
            norm_const = sum(unnormalized_probs)
            normalized_probs = [
                float(u_prob) / norm_const for u_prob in unnormalized_probs
            ]
            alias_nodes[node] = _alias_setup(normalized_probs)
        logging.info(
            f"Completed preprocessing of transition probabilities for vertices"
        )

        alias_edges = {}

        total_edges = len(graph.edges())
        bucket = 0
        current_edge = 0
        quotient = int(total_edges / 10)

        logging.info(
            f"Beginning preprocessing of transition probabilities for {total_edges} edges"
        )
        if is_directed:
            for edge in graph.edges():
                current_edge += 1
                if current_edge > bucket * quotient:
                    bucket += 1
                    logging.info(f"Completed {current_edge} / {total_edges} edges")

                alias_edges[edge] = self._get_alias_edge(edge[0], edge[1])
        else:
            for edge in graph.edges():
                current_edge += 1
                if current_edge > bucket * quotient:
                    bucket += 1
                    logging.info(f"Completed {current_edge} / {total_edges} edges")

                alias_edges[edge] = self._get_alias_edge(edge[0], edge[1])
                alias_edges[(edge[1], edge[0])] = self._get_alias_edge(edge[1], edge[0])

        logging.info(f"Completed preprocessing of transition probabilities for edges")

        self.alias_nodes = alias_nodes
        self.alias_edges = alias_edges

        return