from __future__ import annotations
import base64
from io import BytesIO
from typing import List, Any, Tuple, Optional, Dict
import networkx as nx
import pandas as pd
import numpy as np
import json
import math
from pathlib import Path
import matplotlib as mpl
import matplotlib.colors as mcolors
import gseapy as gp
from gseapy import dotplot
from matplotlib import pyplot as plt
from networkx.classes.graph import _Node
[docs]
class PPINetwork:
"""
A protein-protein interaction (PPI) network wrapper around NetworkX graphs.
This class provides a comprehensive interface for working with PPI networks,
including network construction from STRING data, computation of global, node-level,
and edge-level metrics, export to various formats, and enrichment analysis (GSEA).
Attributes:
G (nx.Graph): The underlying NetworkX graph object.
node_metrics_df (pd.DataFrame): DataFrame containing node-level metrics.
edge_metrics_df (pd.DataFrame): DataFrame containing edge-level metrics.
graph_metrics (dict): Dictionary of global network metrics.
gsea_results_df (pd.DataFrame): GSEA enrichment results.
ranked_genes_df (pd.DataFrame): Genes ranked by a network metric.
score_attr (str): Name of the edge attribute used as the interaction score.
Example:
>>> loader = StringLoader(protein_query=["TP53", "BRCA1"], species=9606)
>>> data = loader.retrieve_data().standardize_data_format().get_data()
>>> ppin = PPINetwork(data=data)
>>> ppin.compute_all_node_metrics()
>>> print(ppin.node_metrics_df.head())
"""
def __init__(
self,
data: Optional[dict] = None,
G: Optional[nx.Graph] = None,
score_attr: str = "score",
):
"""
Initialize the PPINetwork object.
Either a data dictionary (from StringLoader) or a NetworkX graph must be provided.
Args:
data: Dictionary containing "edges" and "nodes" DataFrames from StringLoader.
Defaults to None.
G: An existing NetworkX graph object. Defaults to None.
score_attr: Name of the edge attribute to use as the interaction score.
Defaults to "score".
Raises:
ValueError: If both data and G are None.
Note:
If G is provided, it takes precedence over data.
"""
self.gsea_results_df = None
self.ranked_genes_df = None
if data is None and G is None:
raise ValueError("Either data or G must be provided.")
self.score_attr = score_attr
if G is not None:
self.G = G.copy()
else:
self.G = self._build_string_networkx_graph(data)
self.node_metrics_df = pd.DataFrame()
self.edge_metrics_df = pd.DataFrame()
self.graph_metrics = {}
self._weights_prepared = False
######################
# Graph construction #
######################
@staticmethod
def _build_string_networkx_graph(data: dict) -> nx.Graph:
"""
Construct a NetworkX graph from STRING data.
Creates an undirected graph with nodes and edges from the provided
STRING data dictionary.
Args:
data: Dictionary with "edges" and "nodes" keys, each containing
a pandas DataFrame (from StringLoader.get_data()).
Returns:
nx.Graph: A NetworkX undirected graph with nodes and edges from the data.
Note:
Node IDs are taken from the stringId column. All columns except stringId
become node attributes. For edges, stringId_A and stringId_B become
the edge endpoints, and all other columns become edge attributes.
"""
edges_df = data["edges"]
nodes_df = data["nodes"]
G = nx.Graph()
for _, row in nodes_df.iterrows():
node_id = row["stringId"]
node_attrs = row.to_dict()
node_attrs.pop("stringId", None)
G.add_node(
node_id,
**node_attrs,
)
for _, row in edges_df.iterrows():
source = row["stringId_A"]
target = row["stringId_B"]
edge_attrs = row.to_dict()
edge_attrs.pop("stringId_A", None)
edge_attrs.pop("stringId_B", None)
G.add_edge(
source,
target,
**edge_attrs,
)
return G
###################
# Utility methods #
###################
[docs]
def prepare_string_weights(self) -> nx.Graph:
"""
Adds two edge attributes:
- confidence: normalized DB score
- distance: inverse confidence, useful for shortest paths
The new STRING APIs give already normalized scores, but for coherence
and back compatibility, the confidence will still be calculated
as score / 1000 if the max score is greater than 1.
"""
scores = []
for _, _, attrs in self.G.edges(data=True):
score = attrs.get(self.score_attr)
if score is not None:
scores.append(float(score))
if not scores:
raise ValueError(
f"No edge attribute called {self.score_attr!r} found."
)
max_score = max(scores)
for _, _, attrs in self.G.edges(data=True):
score = float(attrs.get(self.score_attr, 0.0))
if max_score > 1:
confidence = score / 1000.0
else:
confidence = score
confidence = max(confidence, 1e-12)
attrs["confidence"] = confidence
attrs["distance"] = 1.0 / confidence
self._weights_prepared = True
return self.G
def _ensure_weights(self):
if not self._weights_prepared:
self.prepare_string_weights()
def _set_node_metric(self, metric_name, values):
nx.set_node_attributes(
self.G,
values,
metric_name,
)
return values
def _set_graph_metric(self, metric_name, value):
self.graph_metrics[metric_name] = value
return value
@staticmethod
def _is_finite_number(number):
if number is None or isinstance(number, bool):
return False
try:
number_f = float(number)
except (TypeError, ValueError):
return False
return math.isfinite(number_f)
[docs]
def get_numeric_edge_attrs(self) -> List[str]:
"""
Return edge attributes that contain at least one finite numeric value.
"""
attrs = set()
for _, _, edge_attrs in self.G.edges(data=True):
for key, value in edge_attrs.items():
if self._is_finite_number(value):
attrs.add(str(key))
return sorted(attrs)
[docs]
def get_numeric_node_attrs(self) -> List[str]:
"""
Return node attributes that contain at least one finite numeric value
"""
attrs = set()
for _, node_attrs in self.G.nodes(data=True):
for key, value in node_attrs.items():
if self._is_finite_number(value):
attrs.add(str(key))
return sorted(attrs)
[docs]
def get_nodes_metric_value(
self,
metric: str,
sort: bool = False,
) -> Tuple[List[Any], List[str]]:
"""
Read all nodes metric values.
If sort=True, values are sorted by the same metric in descending order.
"""
if not self.node_metrics_df.empty:
if metric not in self.node_metrics_df.columns:
raise AttributeError(
f"Metric '{metric}' has not been calculated for this network!"
)
df = self.node_metrics_df
if sort:
df = df.sort_values(by=metric, ascending=False)
values = df[metric].tolist()
nodes = df["preferredName"].astype(str).tolist()
return values, nodes
node_items = [
(
data[metric],
str(data["preferredName"]),
)
for _, data in self.G.nodes(data=True)
]
if sort:
node_items = sorted(
node_items,
key=lambda item: item[0],
reverse=True,
)
values = [value for value, _ in node_items]
nodes = [node for _, node in node_items]
return values, nodes
[docs]
def get_edges_metric_value(self, attr: str) -> Tuple[List[Any], List[str]]:
"""
Read all edges metric values sorted by the same metric in descending order.
Also return edge identifiers in the same order.
"""
if not self.edge_metrics_df.empty:
if attr not in self.edge_metrics_df.columns:
raise AttributeError(
f"Metric '{attr}' has not been calculated for this network!"
)
df = self.edge_metrics_df.sort_values(by=attr, ascending=False)
values = df[attr].tolist()
edge_ids = [
f"{source}@{target}"
for source, target in zip(df["preferredName_A"], df["preferredName_B"])
]
return values, edge_ids
sorted_edges = sorted(
[
(data[attr], f"{data['preferredName_A']}@{data['preferredName_B']}")
for _, _, data in self.G.edges(data=True)
if attr in data
],
key=lambda item: item[0],
reverse=True,
)
values = [value for value, _ in sorted_edges]
edge_ids = [edge_id for _, edge_id in sorted_edges]
return values, edge_ids
[docs]
def get_nodes_preferred_names(self) -> List[str]:
"""
Return a list of preferred names for all nodes in the network.
"""
return [str(data["preferredName"]) for _, data in self.G.nodes(data=True)]
################
# Node metrics #
################
[docs]
def compute_degree(self) -> Dict[str, int]:
"""
Computes the degree of each node in the network.
"""
values = dict(self.G.degree())
return self._set_node_metric(
"degree",
values,
)
[docs]
def compute_weighted_degree(self) -> Dict[str, int]:
"""
Computes the weighted degree of each node in the network using the 'confidence' edge attribute.
"""
self._ensure_weights()
values = dict(
self.G.degree(
weight="confidence",
)
)
return self._set_node_metric(
"weighted_degree",
values,
)
[docs]
def compute_degree_centrality(self) -> Dict[str, float]:
values = nx.degree_centrality(self.G)
return self._set_node_metric(
"degree_centrality",
values,
)
[docs]
def compute_betweenness_centrality(self, weighted=True) -> Dict[str, float]:
"""
Computes the betweenness centrality of each node in the network.
"""
if weighted:
self._ensure_weights()
values = nx.betweenness_centrality(
self.G,
weight="distance" if weighted else None,
normalized=True,
)
return self._set_node_metric(
"betweenness_centrality",
values,
)
[docs]
def compute_closeness_centrality(self, weighted=True) -> Dict[str, float]:
"""
Computes the closeness centrality of each node in the network.
"""
if weighted:
self._ensure_weights()
values = nx.closeness_centrality(
self.G,
distance="distance" if weighted else None,
wf_improved=True,
)
return self._set_node_metric(
"closeness_centrality",
values,
)
[docs]
def compute_harmonic_centrality(self, weighted=True) -> Dict[str, float]:
"""
Computes the harmonic centrality of each node in the network.
"""
if weighted:
self._ensure_weights()
values = nx.harmonic_centrality(
self.G,
distance="distance" if weighted else None,
)
return self._set_node_metric(
"harmonic_centrality",
values,
)
[docs]
def compute_clustering_coefficient(self, weighted=True) -> Dict[str, float]:
"""
Computes the clustering coefficient of each node in the network.
"""
if weighted:
self._ensure_weights()
values = nx.clustering(
self.G,
weight="confidence" if weighted else None,
)
return self._set_node_metric(
"clustering_coefficient",
values,
)
[docs]
def compute_eigenvector_centrality(self, weighted=True) -> Dict[str, float]:
"""
Computes the eigenvector centrality of each node in the network.
"""
if weighted:
self._ensure_weights()
try:
values = nx.eigenvector_centrality(
self.G,
weight="confidence" if weighted else None,
max_iter=1000,
)
except nx.PowerIterationFailedConvergence:
values = {
node: np.nan
for node in self.G.nodes
}
return self._set_node_metric(
"eigenvector_centrality",
values,
)
[docs]
def compute_katz_centrality(self, alpha=0.01, beta=1.0, weighted=True) -> Dict[str, float]:
"""
Computes the Katz centrality of each node in the network.
"""
if weighted:
self._ensure_weights()
try:
values = nx.katz_centrality(
self.G,
alpha=alpha,
beta=beta,
weight="confidence" if weighted else None,
max_iter=1000,
)
except nx.PowerIterationFailedConvergence:
values = {
node: np.nan
for node in self.G.nodes
}
return self._set_node_metric(
"katz_centrality",
values,
)
[docs]
def compute_core_number(self) -> Dict[str, float]:
"""
Computes the core number of each node in the network.
"""
values = nx.core_number(self.G)
return self._set_node_metric(
"core_number",
values,
)
[docs]
def compute_average_neighbor_degree(self, weighted=True) -> Dict[str, float]:
"""
Computes the average neighbor degree of each node in the network.
"""
if weighted:
self._ensure_weights()
values = nx.average_neighbor_degree(
self.G,
weight="confidence" if weighted else None,
)
return self._set_node_metric(
"average_neighbor_degree",
values,
)
[docs]
def compute_connected_components(self) -> Tuple[dict[Any, Any], dict[Any, Any]]:
"""
Computes the connected components of the network.
"""
components = list(nx.connected_components(self.G))
component_by_node = {}
component_size_by_node = {}
for component_id, component_nodes in enumerate(components):
component_size = len(component_nodes)
for node in component_nodes:
component_by_node[node] = component_id
component_size_by_node[node] = component_size
self._set_node_metric(
"component_id",
component_by_node,
)
self._set_node_metric(
"component_size",
component_size_by_node,
)
return component_by_node, component_size_by_node
[docs]
def compute_communities(self, weighted=True) -> Dict[str, int]:
"""
Computes communities in the network using the greedy modularity maximization algorithm.
"""
if weighted:
self._ensure_weights()
try:
communities = list(
nx.algorithms.community.greedy_modularity_communities(
self.G,
weight="confidence" if weighted else None,
)
)
community_by_node = {}
for community_id, community_nodes in enumerate(communities):
for node in community_nodes:
community_by_node[node] = community_id
except Exception:
community_by_node = {
node: np.nan
for node in self.G.nodes
}
return self._set_node_metric(
"community_id",
community_by_node,
)
[docs]
def compute_bridging_centrality(self, weighted=True) -> Dict[str, float]:
"""
Computes a simple bridging centrality.
It combines:
- betweenness centrality
- bridging coefficient
"""
if weighted:
self._ensure_weights()
betweenness = nx.betweenness_centrality(
self.G,
weight="distance" if weighted else None,
normalized=True,
)
degree = dict(self.G.degree())
bridging_values = {}
for node in self.G.nodes:
neighbors = list(self.G.neighbors(node))
if not neighbors or degree[node] == 0:
bridging_values[node] = 0.0
continue
neighbor_inverse_degree_sum = sum(
1.0 / degree[neighbor]
for neighbor in neighbors
if degree[neighbor] > 0
)
if neighbor_inverse_degree_sum == 0:
bridging_coefficient = 0.0
else:
bridging_coefficient = (
1.0 / degree[node]
) / neighbor_inverse_degree_sum
bridging_values[node] = (
betweenness[node] * bridging_coefficient
)
return self._set_node_metric(
"bridging_centrality",
bridging_values,
)
[docs]
def compute_node_removal_impact(self) -> pd.DataFrame:
"""
Computes the impact of removing each node on the network.
"""
if self.G.number_of_nodes() == 0:
raise ValueError("The graph is empty.")
original_components = list(nx.connected_components(self.G))
original_largest_component_size = len(
max(
original_components,
key=len,
)
)
rows = []
for node in self.G.nodes:
H = self.G.copy()
H.remove_node(node)
if H.number_of_nodes() == 0:
largest_component_size = 0
number_components = 0
else:
components = list(nx.connected_components(H))
largest_component_size = len(
max(
components,
key=len,
)
)
number_components = len(components)
largest_component_loss = (
original_largest_component_size - largest_component_size
)
rows.append(
{
"node": node,
"largest_component_loss": largest_component_loss,
"number_components_after_removal": number_components,
}
)
removal_df = pd.DataFrame(rows)
for metric_name in [
"largest_component_loss",
"number_components_after_removal",
]:
values = removal_df.set_index("node")[metric_name].to_dict()
self._set_node_metric(
metric_name,
values,
)
return removal_df
################
# Edge metrics #
################
[docs]
def compute_edge_betweenness_centrality(self, weighted=True) -> dict[tuple[_Node, _Node], float] | Any:
"""
Computes the edge betweenness centrality of each edge in the network.
"""
if weighted:
self._ensure_weights()
edge_betweenness = nx.edge_betweenness_centrality(
self.G,
weight="distance" if weighted else None,
normalized=True,
)
for u, v in self.G.edges:
value = edge_betweenness.get(
(u, v),
edge_betweenness.get((v, u)),
)
self.G.edges[u, v]["edge_betweenness_centrality"] = value
return edge_betweenness
##################
# Global metrics #
##################
[docs]
def compute_number_of_nodes(self) -> int:
"""
Returns the number of nodes in the network.
"""
return self._set_graph_metric(
"number_of_nodes",
self.G.number_of_nodes(),
)
[docs]
def compute_number_of_edges(self) -> int:
"""
Returns the number of edges in the network.
"""
return self._set_graph_metric(
"number_of_edges",
self.G.number_of_edges(),
)
[docs]
def compute_density(self) -> float:
"""
Returns the density of the network.
"""
return self._set_graph_metric(
"density",
nx.density(self.G),
)
[docs]
def compute_number_connected_components(self) -> int:
"""
Returns the number of connected components in the network.
"""
return self._set_graph_metric(
"number_connected_components",
nx.number_connected_components(self.G),
)
[docs]
def compute_largest_component_size(self) -> int:
"""
Returns the size of the largest connected component in the network.
"""
components = list(nx.connected_components(self.G))
if not components:
value = 0
else:
value = max(
len(component)
for component in components
)
return self._set_graph_metric(
"largest_component_size",
value,
)
[docs]
def compute_average_clustering(self, weighted=True) -> float:
"""
Returns the average clustering coefficient of the network.
"""
if weighted:
self._ensure_weights()
return self._set_graph_metric(
"average_clustering",
nx.average_clustering(
self.G,
weight="confidence" if weighted else None,
),
)
[docs]
def compute_transitivity(self) -> float:
"""
Returns the transitivity of the network.
"""
return self._set_graph_metric(
"transitivity",
nx.transitivity(self.G),
)
[docs]
def compute_is_connected(self) -> float:
"""
Returns whether the network is connected or not.
"""
value = nx.is_connected(self.G)
return self._set_graph_metric(
"is_connected",
value,
)
[docs]
def compute_diameter(self) -> float:
"""
Returns the diameter of the network.
"""
components = list(nx.connected_components(self.G))
if nx.is_connected(self.G):
value = nx.diameter(self.G)
metric_name = "diameter"
else:
largest_component = max(
components,
key=len,
)
H = self.G.subgraph(largest_component).copy()
value = nx.diameter(H)
metric_name = "diameter_largest_component"
return self._set_graph_metric(
metric_name,
value,
)
[docs]
def compute_average_shortest_path_length(self, weighted=True) -> float:
"""
Returns the average shortest path length of the network.
"""
if weighted:
self._ensure_weights()
components = list(nx.connected_components(self.G))
if nx.is_connected(self.G):
value = nx.average_shortest_path_length(
self.G,
weight="distance" if weighted else None,
)
metric_name = "average_shortest_path_length"
else:
largest_component = max(
components,
key=len,
)
H = self.G.subgraph(largest_component).copy()
value = nx.average_shortest_path_length(
H,
weight="distance" if weighted else None,
)
metric_name = "average_shortest_path_length_largest_component"
return self._set_graph_metric(
metric_name,
value,
)
[docs]
def compute_global_efficiency(self) -> float:
"""
Returns the global efficiency of the network.
"""
value = nx.global_efficiency(self.G)
return self._set_graph_metric(
"global_efficiency",
value,
)
######################
# DataFrame builders #
######################
[docs]
def build_node_metrics_df(self) -> pd.DataFrame:
"""
Builds a DataFrame containing node-level metrics.
"""
rows = []
for node, attrs in self.G.nodes(data=True):
row = {
"node": node,
**attrs,
}
rows.append(row)
self.node_metrics_df = pd.DataFrame(rows)
return self.node_metrics_df
[docs]
def build_edge_metrics_df(self) -> pd.DataFrame:
"""
Builds a DataFrame containing edge-level metrics.
"""
rows = []
for u, v, attrs in self.G.edges(data=True):
row = {
"source": u,
"target": v,
**attrs,
}
rows.append(row)
self.edge_metrics_df = pd.DataFrame(rows)
return self.edge_metrics_df
#######################
# Compute all metrics #
#######################
[docs]
def compute_all_node_metrics(self, weighted=True) -> pd.DataFrame:
"""
Computes all node-level metrics and returns a DataFrame with the results.
"""
self.compute_degree()
self.compute_weighted_degree()
self.compute_degree_centrality()
self.compute_betweenness_centrality(weighted=weighted)
self.compute_closeness_centrality(weighted=weighted)
self.compute_harmonic_centrality(weighted=weighted)
self.compute_clustering_coefficient(weighted=weighted)
self.compute_eigenvector_centrality(weighted=weighted)
self.compute_pagerank(weighted=weighted)
self.compute_katz_centrality(weighted=weighted)
self.compute_core_number()
self.compute_average_neighbor_degree(weighted=weighted)
self.compute_connected_components()
self.compute_communities(weighted=weighted)
self.compute_bridging_centrality(weighted=weighted)
self.compute_community_roles()
self.compute_node_removal_impact()
return self.build_node_metrics_df()
[docs]
def compute_all_edge_metrics(self, weighted=True) -> pd.DataFrame:
"""
Computes all edge-level metrics and returns a DataFrame with the results.
"""
self.compute_edge_betweenness_centrality(weighted=weighted)
return self.build_edge_metrics_df()
[docs]
def compute_all_global_metrics(self, weighted=True) -> dict:
"""
Computes all global metrics and returns a dictionary with the results.
"""
self.compute_number_of_nodes()
self.compute_number_of_edges()
self.compute_density()
self.compute_number_connected_components()
self.compute_largest_component_size()
self.compute_average_clustering(weighted=weighted)
self.compute_transitivity()
self.compute_is_connected()
self.compute_diameter()
self.compute_average_shortest_path_length(weighted=weighted)
self.compute_global_efficiency()
return self.graph_metrics
[docs]
def compute_all_metrics(self, weighted=True) -> tuple[pd.DataFrame, pd.DataFrame, dict, nx.Graph]:
"""
Calls all metric computation methods and returns the results as DataFrames and dictionaries.
"""
node_metrics_df = self.compute_all_node_metrics(weighted=weighted)
edge_metrics_df = self.compute_all_edge_metrics(weighted=weighted)
graph_metrics = self.compute_all_global_metrics(weighted=weighted)
return node_metrics_df, edge_metrics_df, graph_metrics, self.G
################
# Hub proteins #
################
[docs]
def get_hub_proteins(
self,
node_attribute: str,
top_n=10,
ascending=False,
) -> pd.DataFrame:
"""
Returns top hub proteins according to one node attribute.
Example attributes:
- degree
- weighted_degree
- betweenness_centrality
- closeness_centrality
- harmonic_centrality
- eigenvector_centrality
- pagerank
- katz_centrality
- core_number
- bridging_centrality
- within_module_z_score
- participation_coefficient
- largest_component_loss
"""
rows = []
for node, attrs in self.G.nodes(data=True):
if node_attribute not in attrs:
raise ValueError(
f"Node attribute {node_attribute!r} not found. "
"Compute the metric before calling get_hub_proteins()."
)
row = {
"node": node,
**attrs,
"hub_attribute": node_attribute,
"hub_score": attrs.get(node_attribute),
}
rows.append(row)
hub_df = pd.DataFrame(rows)
hub_df["hub_score"] = pd.to_numeric(
hub_df["hub_score"],
errors="coerce",
)
hub_df = hub_df.sort_values(
by="hub_score",
ascending=ascending,
)
hub_df["hub_rank"] = range(
1,
len(hub_df) + 1,
)
hub_df = hub_df.head(top_n).copy()
return hub_df
[docs]
def get_consensus_hub_proteins(
self,
top_n=10,
attributes=None,
) -> pd.DataFrame:
"""
Returns hub proteins using a consensus score.
The score is the average percentile rank across several metrics.
"""
if attributes is None:
attributes = [
"weighted_degree",
"betweenness_centrality",
"eigenvector_centrality",
"pagerank",
"katz_centrality",
"core_number",
"bridging_centrality",
"within_module_z_score",
"participation_coefficient",
"largest_component_loss",
]
# TODO
df = self.build_node_metrics_df()
attributes = [
attr
for attr in attributes
if attr in df.columns
]
if not attributes:
raise ValueError("No valid attributes found for consensus hub score.")
percentile_columns = []
for attr in attributes:
percentile_col = f"{attr}_percentile"
df[percentile_col] = pd.to_numeric(
df[attr],
errors="coerce",
).rank(
pct=True,
ascending=True,
)
percentile_columns.append(percentile_col)
df["consensus_hub_score"] = df[percentile_columns].mean(axis=1)
df = df.sort_values(
by="consensus_hub_score",
ascending=False,
)
df["hub_rank"] = range(
1,
len(df) + 1,
)
hub_df = df.head(top_n).copy()
return hub_df
##########
# Export #
##########
def _make_export_safe(self, value):
"""
Converts Python / NumPy / pandas values into Cytoscape-friendly values.
GraphML supports simple scalar attributes.
Complex objects are converted to JSON strings.
Missing or invalid numeric values are converted to empty strings.
"""
if value is None:
return ""
if pd.isna(value) if not isinstance(value, (list, tuple, dict, set)) else False:
return ""
if isinstance(value, np.integer):
return int(value)
if isinstance(value, np.floating):
value = float(value)
if isinstance(value, float):
if math.isnan(value) or math.isinf(value):
return ""
return value
if isinstance(value, np.ndarray):
return json.dumps(
value.tolist(),
ensure_ascii=False,
)
if isinstance(value, (list, tuple, set)):
return json.dumps(
list(value),
ensure_ascii=False,
)
if isinstance(value, dict):
return json.dumps(
{
str(k): self._make_export_safe(v)
for k, v in value.items()
},
ensure_ascii=False,
)
if isinstance(value, bool):
return bool(value)
if isinstance(value, (int, str)):
return value
return str(value)
def _copy_graph_for_export(self):
"""
Creates a sanitized copy of the graph for GraphML export.
All node, edge, and graph attributes are preserved when possible.
Non-scalar attributes are serialized as strings.
"""
H = self.G.copy()
for node, attrs in H.nodes(data=True):
for key, value in list(attrs.items()):
safe_key = str(key)
safe_value = self._make_export_safe(value)
if safe_key != key:
attrs.pop(key, None)
attrs[safe_key] = safe_value
for u, v, attrs in H.edges(data=True):
for key, value in list(attrs.items()):
safe_key = str(key)
safe_value = self._make_export_safe(value)
if safe_key != key:
attrs.pop(key, None)
attrs[safe_key] = safe_value
for key, value in self.graph_metrics.items():
H.graph[str(key)] = self._make_export_safe(value)
H.graph["score_attr"] = self._make_export_safe(self.score_attr)
H.graph["weights_prepared"] = self._make_export_safe(self._weights_prepared)
return H
[docs]
def export_for_cytoscape(
self,
output_dir: str,
basename="ppi_network",
include_tables=True,
include_graph_metrics=True,
) -> dict[str, str]:
"""
Exports the network in Cytoscape-compatible files.
Main output:
- <basename>.graphml
Optional companion files:
- <basename>_nodes.csv
- <basename>_edges.csv
- <basename>_graph_metrics.json
The GraphML file can be loaded in Cytoscape Desktop with:
File > Import > Network from File
"""
output_dir = Path(output_dir)
output_dir.mkdir(
parents=True,
exist_ok=True,
)
graphml_path = output_dir / f"{basename}.graphml"
H = self._copy_graph_for_export()
nx.write_graphml(
H,
graphml_path,
)
exported_files = {
"graphml": str(graphml_path),
}
if include_tables:
node_df = self.build_node_metrics_df().copy()
edge_df = self.build_edge_metrics_df().copy()
for col in node_df.columns:
node_df[col] = node_df[col].map(self._make_export_safe)
for col in edge_df.columns:
edge_df[col] = edge_df[col].map(self._make_export_safe)
nodes_path = output_dir / f"{basename}_nodes.csv"
edges_path = output_dir / f"{basename}_edges.csv"
node_df.to_csv(
nodes_path,
index=False,
)
edge_df.to_csv(
edges_path,
index=False,
)
exported_files["nodes_csv"] = str(nodes_path)
exported_files["edges_csv"] = str(edges_path)
if include_graph_metrics:
graph_metrics_path = output_dir / f"{basename}_graph_metrics.json"
safe_graph_metrics = {
str(key): self._make_export_safe(value)
for key, value in self.graph_metrics.items()
}
safe_graph_metrics["score_attr"] = self._make_export_safe(self.score_attr)
safe_graph_metrics["weights_prepared"] = self._make_export_safe(
self._weights_prepared
)
with open(graph_metrics_path, "w", encoding="utf-8") as f:
json.dump(
safe_graph_metrics,
f,
indent=2,
ensure_ascii=False,
)
exported_files["graph_metrics_json"] = str(graph_metrics_path)
return exported_files
[docs]
def networkx_to_cytoscape_elements(
self,
node_color_attr="pagerank",
node_size_attr="degree",
edge_width_attr="score",
cmap_name="viridis",
min_node_size=35,
max_node_size=100,
min_edge_width=1,
max_edge_width=10
) -> list[dict[str, Any]]:
"""
Convert a NetworkX graph into Cytoscape.js elements.
Initial node color is based on node_color_attr.
Initial node size is based on node_size_attr.
Initial edge width is based on edge_width_attr.
The generated HTML can later recompute these values interactively.
"""
RESERVED_NODE_KEYS = {"id", "source", "target"}
RESERVED_EDGE_KEYS = {"id", "source", "target"}
def make_json_safe(val):
"""
Convert values into JSON-safe Python objects.
Handles numpy-like values, NaN, infinity and non-serializable objects.
"""
if val is None:
return None
if isinstance(val, (str, bool, int)):
return val
if isinstance(val, float):
if math.isfinite(val):
return val
return None
# handles numpy scalar values. i.e. np.float64, np.int64
if hasattr(val, "item"):
try:
return make_json_safe(val.item())
except Exception:
pass
try:
json.dumps(val)
return val
except TypeError:
return str(val)
def safe_float(num, default=.0):
if num is None:
return default
try:
num_f = float(num)
except (TypeError, ValueError):
return default
if math.isnan(num_f) or math.isinf(num_f):
return default
return num_f
def scale_value(
val,
vmin,
vmax,
out_min,
out_max,
):
if vmax == vmin:
return out_min
scaled = (val - vmin) / (vmax - vmin)
return out_min + scaled * (out_max - out_min)
elements = []
node_color_values = [
safe_float(attrs.get(node_color_attr))
for node, attrs in self.G.nodes(data=True)
]
node_size_values = [
safe_float(attrs.get(node_size_attr))
for node, attrs in self.G.nodes(data=True)
]
edge_width_values = [
safe_float(attrs.get(edge_width_attr))
for _, _, attrs in self.G.edges(data=True)
]
color_vmin = min(node_color_values) if node_color_values else 0.0
color_vmax = max(node_color_values) if node_color_values else 1.0
size_vmin = min(node_size_values) if node_size_values else 0.0
size_vmax = max(node_size_values) if node_size_values else 1.0
edge_vmin = min(edge_width_values) if edge_width_values else 0.0
edge_vmax = max(edge_width_values) if edge_width_values else 1.0
if color_vmin == color_vmax:
color_vmax = color_vmin + 1.0
if size_vmin == size_vmax:
size_vmax = size_vmin + 1.0
if edge_vmin == edge_vmax:
edge_vmax = edge_vmin + 1.0
color_norm = mcolors.Normalize(
vmin=color_vmin,
vmax=color_vmax,
)
try:
cmap = mpl.colormaps[cmap_name]
except KeyError:
cmap = mpl.colormaps["viridis"]
for node, attrs in self.G.nodes(data=True):
color_raw_value = safe_float(attrs.get(node_color_attr))
size_raw_value = safe_float(attrs.get(node_size_attr))
color_scaled_value = color_norm(color_raw_value)
size = scale_value(
val=size_raw_value,
vmin=size_vmin,
vmax=size_vmax,
out_min=min_node_size,
out_max=max_node_size,
)
color = mcolors.to_hex(
cmap(color_scaled_value)
)
label = attrs.get(
"preferredName",
str(node),
)
data = {}
for key, value in attrs.items():
key = str(key)
if key in RESERVED_NODE_KEYS:
data[f"attr_{key}"] = make_json_safe(value)
else:
data[key] = make_json_safe(value)
data.update(
{
"id": str(node),
"label": str(label),
"visual_color": color,
"visual_size": float(size),
"node_color_attr": node_color_attr,
"node_color_value": color_raw_value,
"node_size_attr": node_size_attr,
"node_size_value": size_raw_value,
}
)
elements.append(
{
"group": "nodes",
"data": data,
}
)
for i, (u, v, attrs) in enumerate(self.G.edges(data=True)):
edge_raw_value = safe_float(
attrs.get(edge_width_attr),
default=0.0,
)
edge_width = scale_value(
val=edge_raw_value,
vmin=edge_vmin,
vmax=edge_vmax,
out_min=min_edge_width,
out_max=max_edge_width,
)
data = {}
for key, value in attrs.items():
key = str(key)
if key in RESERVED_EDGE_KEYS:
data[f"attr_{key}"] = make_json_safe(value)
else:
data[key] = make_json_safe(value)
data.update(
{
"id": f"edge_{i}",
"source": str(u),
"target": str(v),
"visual_width": float(edge_width),
"edge_width_attr": edge_width_attr,
"edge_width_value": edge_raw_value,
}
)
elements.append(
{
"group": "edges",
"data": data,
}
)
return elements
########
# GSEA #
########
[docs]
def run_gsea(
self,
node_attribute: str,
gene_sets: str = "Reactome_2022",
gene_symbol_attr: str = "preferredName",
min_size: int = 5,
max_size: int = 1000,
permutation_num: int = 1000,
threads: int = 4,
seed: int = 42,
outdir: Optional[str] = None,
ascending: bool = False,
) -> Tuple[pd.DataFrame, pd.DataFrame]:
"""
Run preranked GSEA using a node metric as ranking score.
Args:
node_attribute: Node metric used to rank genes. Examples:
- "degree"
- "weighted_degree"
- "degree_centrality"
- "betweenness_centrality"
- "closeness_centrality"
- "eigenvector_centrality"
- "pagerank"
- "bridging_centrality"
- "core_number"
gene_sets: Gene set collection used by GSEApy. Defaults to "Reactome_2022". Examples:
- "Reactome_2022"
- "KEGG_2021_Human"
- "GO_Biological_Process_2023"
- Path to a .gmt file
- Dictionary of custom gene sets
gene_symbol_attr: Node attribute containing the gene symbol.
In STRING networks this is "preferredName". Defaults to "preferredName".
min_size: Minimum gene-set size. Defaults to 5.
max_size: Maximum gene-set size. Defaults to 1000.
permutation_num: Number of permutations. Defaults to 1000.
threads: Number of CPU threads. Defaults to 4.
seed: Random seed. Defaults to 42.
outdir: Output directory. If None, GSEApy does not write files to disk. Defaults to None.
ascending: If False, highest scores are placed at the top of the ranked list.
This is usually correct for centrality-based GSEA. Defaults to False.
Returns:
Tuple[pd.DataFrame, pd.DataFrame]: A tuple of (ranked_genes_df, gsea_results_df).
Raises:
ValueError: If node_attribute is not found in node attributes.
ValueError: If gene_symbol_attr is not found in node attributes.
ValueError: If no valid gene-score pairs are available for GSEA.
"""
rows = []
for node, attrs in self.G.nodes(data=True):
if node_attribute not in attrs:
raise ValueError(
f"Node attribute {node_attribute!r} not found. "
"Compute the metric before running GSEA."
)
if gene_symbol_attr not in attrs:
raise ValueError(
f"Gene symbol attribute {gene_symbol_attr!r} not found."
)
gene = attrs.get(gene_symbol_attr)
score = attrs.get(node_attribute)
rows.append({
"node": node,
"gene": gene,
"score": score,
})
ranked_genes_df = pd.DataFrame(rows)
ranked_genes_df["gene"] = (
ranked_genes_df["gene"]
.astype(str)
.str.strip()
.str.upper()
)
ranked_genes_df["score"] = pd.to_numeric(
ranked_genes_df["score"],
errors="coerce",
)
ranked_genes_df = ranked_genes_df.dropna(
subset=["gene", "score"]
)
ranked_genes_df = ranked_genes_df[
(ranked_genes_df["gene"] != "")
& (ranked_genes_df["gene"].str.lower() != "nan")
]
if ranked_genes_df.empty:
raise ValueError(
"No valid gene-score pairs available for GSEA."
)
# if multiple proteins map to the same gene keep the highest score
self.ranked_genes_df = (
ranked_genes_df
.groupby("gene", as_index=False)["score"]
.max()
.sort_values(
by="score",
ascending=ascending
)
.reset_index(drop=True)
)
pre_res = gp.prerank(
rnk=ranked_genes_df[["gene", "score"]],
gene_sets=gene_sets,
min_size=min_size,
max_size=max_size,
permutation_num=permutation_num,
threads=threads,
outdir=outdir,
seed=seed,
verbose=True,
)
gsea_results_df = pre_res.res2d.copy()
if "FDR q-val" in gsea_results_df.columns:
gsea_results_df = gsea_results_df.sort_values(
by="FDR q-val",
ascending=True,
)
self.gsea_results_df = gsea_results_df.reset_index(drop=True)
return ranked_genes_df, gsea_results_df
[docs]
def plot_gsea_dotplot(
self,
gsea_results_df: pd.DataFrame,
column: str = "FDR q-val",
title: str = "GSEA enrichment dotplot",
cutoff: float = 0.25,
top_term: int = 10,
size: float = 6,
figsize: tuple = (6, 5),
cmap: str = "viridis",
show_ring: bool = False,
) -> str:
"""
Plot GSEA results as a dotplot using GSEApy.
Args:
gsea_results_df: Result DataFrame returned by run_gsea().
Usually pre_res.res2d.
column: Column used to color the dots. For preranked GSEA use usually:
- "FDR q-val"
- "NOM p-val"
Defaults to "FDR q-val".
title: Plot title. Defaults to "GSEA enrichment dotplot".
cutoff: Only terms with column value <= cutoff are shown. Defaults to 0.25.
top_term: Number of top enriched terms to show. Defaults to 10.
size: Dot size scaling. Defaults to 6.
figsize: Figure size (width, height). Defaults to (6, 5).
cmap: Matplotlib colormap. Defaults to "viridis".
show_ring: Whether to draw an outer ring around dots. Defaults to False.
Returns:
str: Base64-encoded PNG image as a data URI.
Raises:
ValueError: If gsea_results_df is empty.
ValueError: If the specified column is not found in gsea_results_df.
"""
if self.gsea_results_df is None or self.gsea_results_df.empty:
raise ValueError("gsea_results_df is empty")
if column not in self.gsea_results_df.columns:
raise ValueError(
f"Column {column} not found in gsea_results_df"
)
ax = dotplot(
gsea_results_df,
column=column,
title=title,
cutoff=cutoff,
top_term=top_term,
size=size,
figsize=figsize,
cmap=cmap,
ofname=None,
show_ring=show_ring,
)
buffer = BytesIO()
fig = ax.figure
fig.savefig(
buffer,
format="png",
bbox_inches="tight"
)
buffer.seek(0)
image_base64 = base64.b64encode(
buffer.read()
).decode("utf-8")
buffer.close()
plt.close(fig)
return f"data:image/png;base64,{image_base64}"