ActorGraphTransformer.py

class ActorGraphTransformer:
    """
    A class for transforming a DataFrame into an actor graph and calculating various graph metrics.

    Attributes:
        df_d3 (pandas.DataFrame): The input DataFrame containing the graph data.
        graph (networkx.Graph): The graph representation of the DataFrame.
        actor_graph_metrics_df (pandas.DataFrame): DataFrame containing the calculated graph metrics for each actor.
        actor_graph_metrics_dict (dict): Dictionary containing the graph metrics for each actor.
        edge_frequency_dict (dict): Dictionary containing the frequency of edges in the graph.

    Methods:
        __create_graph(): Create a graph from the DataFrame.
        __create_graph_df(): Calculate various graph metrics for each actor and return a DataFrame.
        __create_actor_metrics_dict(): Create a dictionary of graph metrics for each actor.
        __get_edge_frequency_dict(): Calculate the frequency of edges in the graph.
        get_actor_graph_metrics_df(): Get the DataFrame containing graph metrics for each actor.
        get_actor_graph_metrics_dict(): Get the dictionary of graph metrics for each actor.
        get_edge_frequency_dict(): Get the dictionary containing the frequency of edges in the graph.
    """

    def __init__(self, df_d3):
        """
        Initialize the ActorGraphTransformer class.

        Args:
            df_d3 (pandas.DataFrame): The input DataFrame containing the graph data.
        """
        self.df_d3 = df_d3
        self.graph = self.__create_graph()
        self.actor_graph_metrics_df = self.__create_graph_df()
        self.actor_graph_metrics_dict = self.__create_actor_metrics_dict()
        self.edge_frequency_dict = self.__get_edge_frequency_dict()

    def __create_graph(self):
        """
        Create a graph from the DataFrame.

        Returns:
            networkx.Graph: The graph representation of the DataFrame.

        Raises:
            ValueError: If an error occurs while creating the graph.
        """
        try:
            graph = nx.from_pandas_edgelist(self.df_d3, 'source', 'target', 'weight')
            return graph
        except Exception as e:
            raise ValueError("Error creating the graph: " + str(e))

    def __create_graph_df(self):
        """
        Calculate various graph metrics for each actor and return a DataFrame.

        Returns:
            pandas.DataFrame: DataFrame containing the calculated graph metrics for each actor.

        Raises:
            ValueError: If an error occurs while creating the actor attributes DataFrame.
        """
        try:
            actor_attributes = pd.DataFrame(index=self.graph.nodes)

            actor_attributes['DegreeCentrality'] = pd.Series(nx.degree_centrality(self.graph))
            actor_attributes['BetweennessCentrality'] = pd.Series(nx.betweenness_centrality(self.graph))
            actor_attributes['EigenvectorCentrality'] = pd.Series(nx.eigenvector_centrality(self.graph))
            actor_attributes['ClusteringCoefficient'] = pd.Series(nx.clustering(self.graph))

            return round(actor_attributes, 3).reset_index().rename(columns={'index': 'Actor'})
        except Exception as e:
            raise ValueError("Error creating actor attributes: " + str(e))

    def __create_actor_metrics_dict(self):
        """
        Create a dictionary of graph metrics for each actor.

        Returns:
            dict: Dictionary containing the graph metrics for each actor.

        Raises:
            ValueError: If an error occurs while creating the actor metrics dictionary.
        """
        try:
            actor_dict = {}
            for actor, attributes in self.actor_graph_metrics_df.iterrows():
                actor_dict[str(actor).replace(" ", "_")] = attributes.to_dict()
            return actor_dict
        except Exception as e:
            raise ValueError("Error creating actor metrics dictionary: " + str(e))

    def __get_edge_frequency_dict(self):
        """
        Calculate the frequency of edges in the graph.

        Returns:
            dict: Dictionary containing the frequency of edges in the graph.

        Raises:
            ValueError: If an error occurs while getting the edge frequency dictionary.
        """
        try:
            source_counts = self.df_d3['source'].value_counts().to_dict()
            target_counts = self.df_d3['target'].value_counts().to_dict()
            result = {key: source_counts.get(key, 0) + target_counts.get(key, 0) for key in source_counts}
            return result
        except Exception as e:
            raise ValueError("Error getting edge frequency dictionary: " + str(e))

    def get_actor_graph_metrics_df(self):
        """
        Get the DataFrame containing graph metrics for each actor.

        Returns:
            pandas.DataFrame: DataFrame containing graph metrics for each actor.
        """
        return self.actor_graph_metrics_df

    def get_actor_graph_metrics_dict(self):
        """
        Get the dictionary of graph metrics for each actor.

        Returns:
            dict: Dictionary containing graph metrics for each actor.
        """
        return self.actor_graph_metrics_dict

    def get_edge_frequency_dict(self):
        """
        Get the dictionary containing the frequency of edges in the graph.

        Returns:
            dict: Dictionary containing the frequency of edges in the graph.
        """
        return self.edge_frequency_dict