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.

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