complementarity.functor.py

Core Functor for facilitating Mode Complementarity computation.

class rings.complementarity.functor.ComplementarityFunctor(feature_metric, graph_metric, comparator, n_jobs: int = 1, use_edge_information: bool = False, normalize_diameters: bool = True, edge_attr: str | None = None, **kwargs)

A functor for computing complementarity between graph structure and node features.

This class computes complementarity by comparing the metric spaces derived from graph structure and node features using a specified comparator. It quantifies how well node features align with the graph structure, with lower values indicating stronger alignment.

For disconnected graphs, the functor processes each connected component separately and aggregates the results using a weighted average, where component sizes serve as weights. This approach ensures that larger components have proportionally more influence on the final complementarity score.

Parameters:

• feature_metric (callable) – The metric function used to compute distances between node features.

• graph_metric (callable) – The metric function used to compute distances in the graph structure.

• comparator (class) – A comparator class that implements a __call__ method to compare metric spaces.

• n_jobs (int, default=1) – Number of jobs to run in parallel. If 1, no parallelism is used.

• use_edge_information (bool, default=False) – Whether to use edge attributes in graph metric computation.

• normalize_diameters (bool, default=True) – Whether to normalize the diameters of metric spaces before comparison.

• **kwargs (dict) – Additional arguments passed to the comparator and metric functions.

Examples

>>> import numpy as np
>>> import torch
>>> from torch_geometric.data import Data
>>> from rings.complementarity.comparator import MatrixNormComparator
>>> from rings.complementarity.metrics import standard_feature_metrics, shortest_path_distance
>>>
>>> # Create a simple graph where node features correspond to positions in the graph
>>> # A path graph: 0 -- 1 -- 2 -- 3 with features encoding their positions
>>> edge_index = torch.tensor([[0, 1, 1, 2, 2, 3], [1, 0, 2, 1, 3, 2]], dtype=torch.long)
>>> # Node features encode positions: node 0 is at position 0, node 1 at position 1, etc.
>>> x = torch.tensor([[0], [1], [2], [3]], dtype=torch.float)
>>> data = Data(x=x, edge_index=edge_index)
>>>
>>> # Create functor with simple metrics
>>> functor = ComplementarityFunctor(
...     feature_metric='euclidean',  # Use Euclidean distance for features
...     graph_metric='shortest_path_distance',  # Use shortest path for graph
...     comparator=MatrixNormComparator,  # Compare using matrix norm
...     n_jobs=1,
...     normalize_diameters=True,  # Normalize distances for fair comparison
... )
>>>
>>> # Compute complementarity for the graph
>>> result = functor([data])
>>> print(f"Complementarity score: {result['complementarity'].item():.4f}")
Complementarity score: 0.0000
>>>
>>> # The score is 0, indicating perfect alignment between features and structure
>>> # (node features perfectly correspond to their positions in the path)
>>>
>>> # Now create a graph with misaligned features
>>> x_misaligned = torch.tensor([[3], [1], [2], [0]], dtype=torch.float)  # Swap 0 and 3
>>> data_misaligned = Data(x=x_misaligned, edge_index=edge_index)
>>>
>>> # Compute complementarity for the misaligned graph
>>> result = functor([data_misaligned])
>>> print(f"Complementarity score: {result['complementarity'].item():.4f}")
Complementarity score: 0.5000
>>>
>>> # The score is higher, indicating weaker alignment between features and structure
>>> # (node features no longer match their positions in the path)

__init__(feature_metric, graph_metric, comparator, n_jobs: int = 1, use_edge_information: bool = False, normalize_diameters: bool = True, edge_attr: str | None = None, **kwargs)

Initialize internal Module state, shared by both nn.Module and ScriptModule.

forward(batch, as_dataframe: bool = True) → Dict[str, Any] | DataFrame | List[Dict[str, Any]]

Compute complementarity for a batch of graphs in PyTorch Geometric format.

Parameters:

• batch (list) – A batch of graphs in PyTorch Geometric format.

• as_dataframe (bool, default=True) – If True, return results as a pandas DataFrame, otherwise as a dictionary with tensor values or a list of dictionaries.

Returns:

Results of complementarity computation for each graph in the batch. If as_dataframe=True, returns a pandas DataFrame. Otherwise, returns either a dictionary with tensor values or a list of dictionaries.

Return type:

Union[Dict[str, Any], pd.DataFrame, List[Dict[str, Any]]]

Examples

>>> from rings.complementarity.comparator import MatrixNormComparator
>>> import torch_geometric.datasets as datasets
>>> dataset = datasets.TUDataset(root='/tmp/ENZYMES', name='ENZYMES')
>>> functor = ComplementarityFunctor(
...     feature_metric=lambda x, y: np.linalg.norm(x - y),
...     graph_metric=lambda x, y: abs(x - y),
...     comparator=MatrixNormComparator,
...     n_jobs=4,
...     normalize_diameters=True
... )
>>> # Get first 5 graphs as a batch
>>> batch = [dataset[i] for i in range(5)]
>>> results = functor.forward(batch)
>>> print(results)

property invalid_data

Return a dictionary with NaN score and other standardized fields.

Returns:

Dictionary with keys: - score: NaN (invalid data) - pvalue: None - pvalue_adjusted: None - significant: None - method: The norm used

Return type:

dict