Source code for graphwar.attack.attacker

import abc
from numbers import Number
from typing import Optional, Union
from torch_geometric.data import Data
from torch_geometric.utils import degree, to_scipy_sparse_matrix

import numpy as np
import scipy.sparse as sp
import torch

from graphwar import set_seed


[docs]class Attacker(torch.nn.Module): """Adversarial attacker for graph data. Note that this is an abstract class. Parameters ---------- data : Data PyG-like data denoting the input graph device : str, optional the device of the attack running on, by default "cpu" seed : Optional[int], optional the random seed for reproducing the attack, by default None name : Optional[str], optional name of the attacker, if None, it would be :obj:`__class__.__name__`, by default None kwargs : additional arguments of :class:`graphwar.attack.Attacker`, Raises ------ TypeError unexpected keyword argument in :obj:`kwargs` Examples -------- For example, the attacker model should be defined as follows: >>> from graphwar.attacker import Attacker >>> attacker = Attacker(data, device='cuda') >>> attacker.reset() # reset states >>> attacker.attack(attack_arguments) # attack >>> attacker.data() # get the attacked graph denoted as PyG-like Data """ _max_perturbations: Union[float, int] = 0 _allow_feature_attack: bool = False _allow_structure_attack: bool = True _allow_singleton: bool = True def __init__(self, data: Data, device: str = "cpu", seed: Optional[int] = None, name: Optional[str] = None, **kwargs): """Initialization of an attacker model. """ super().__init__() if kwargs: raise TypeError( f"Got an unexpected keyword argument '{next(iter(kwargs.keys()))}'." ) assert isinstance(data, Data) assert data.x is not None assert data.edge_index is not None assert data.edge_weight is None self.device = torch.device(device) self.ori_data = data.to(self.device) self.adjacency_matrix: sp.csr_matrix = to_scipy_sparse_matrix(data.edge_index, num_nodes=data.num_nodes).tocsr() self.name = name or self.__class__.__name__ self.seed = seed self._degree = degree( data.edge_index[0], num_nodes=data.num_nodes, dtype=torch.float) self.num_nodes = data.num_nodes self.num_edges = data.num_edges self.num_feats = data.x.size(1) self.nodes_set = set(range(self.num_nodes)) set_seed(seed) self._is_reset = False
[docs] def reset(self): """Reset attacker state. Override this method in subclass to implement specific function.""" self._is_reset = True return self
[docs] @abc.abstractmethod def data(self) -> Data: """Get the attacked graph denoted as PyG-like Data. Raises ------ NotImplementedError The subclass does not implement this interface. """ raise NotImplementedError
[docs] @abc.abstractmethod def attack(self) -> "Attacker": """Abstract method. The subclass must override this method to implement specific attack for itself. Raises ------ NotImplementedError The subclass does not implement this interface. """ raise NotImplementedError
def _check_budget(self, num_budgets: Union[float, int], max_perturbations: Union[float, int]) -> int: """Check and return attack budget.""" max_perturbations = max(max_perturbations, self.max_perturbations) if not isinstance(num_budgets, Number) or num_budgets <= 0: raise ValueError( f"'num_budgets' must be a positive scalar. but got '{num_budgets}'." ) if num_budgets > max_perturbations: raise ValueError( f"'num_budgets' should be less than or equal the maximum allowed perturbations: {max_perturbations}." "if you want to use larger budgets, you could set 'attacker.set_max_perturbations(a_larger_budget)'." ) if num_budgets < 1.: assert self._max_perturbations != np.inf num_budgets = max_perturbations * num_budgets return int(num_budgets)
[docs] def set_max_perturbations(self, max_perturbations: Union[float, int] = np.inf, verbose: bool = True) -> "Attacker": """Set the maximum number of allowed perturbations Parameters ---------- max_perturbations : Union[float, int], optional the maximum number of allowed perturbations, by default np.inf verbose : bool, optional whether to verbose the operation, by default True Example ------- >>> attacker.set_max_perturbations(10) """ assert isinstance(max_perturbations, Number), max_perturbations self._max_perturbations = max_perturbations if verbose: print(f"Set maximum perturbations: {max_perturbations}") return self
@property def max_perturbations(self) -> Union[float, int]: """float or int: Maximum allowable perturbation size.""" return self._max_perturbations @property def feat(self) -> torch.Tensor: """Node features of the original graph.""" return self.ori_data.x @property def label(self) -> torch.Tensor: """Node labels of the original graph.""" return self.ori_data.y @property def edge_index(self) -> torch.Tensor: """Edge index of the original graph.""" return self.ori_data.edge_index @property def edge_weight(self) -> torch.Tensor: """Edge weight of the original graph.""" return self.ori_data.edge_weight def _check_feature_matrix_binary(self): """Check if the feature matrix is binary. Raises ------ RuntimeError if the feature matrix is not binary """ feat = self.feat # FIXME: (Jintang Li) this is quite time-consuming in large matrix # so I only check `10` rows of the matrix randomly. feat = feat[torch.randint(0, feat.size(0), size=(10,))] if not torch.unique(feat).tolist() == [0, 1]: raise RuntimeError( "Node feature matrix is required to be a 0-1 binary matrix.") def extra_repr(self) -> str: return f"device={self.device}, seed={self.seed},"