Coverage for src/causalspyne/ancestral_shd.py: 94%

1""" 

2""" 

3 

4from causallearn.utils.DAG2PAG import dag2pag 

5from causallearn.graph.Dag import Dag 

6from causallearn.graph.GraphNode import GraphNode 

7import numpy as np 

8 

9# class ancestral_shd(): 

10# def __init__(mat_gt_ancestral, mat_fci): 

11# self.mat_gt_ancestral = mat_gt_ancestral 

12 

13 

14import numpy as np 

15 

16 

17def structural_hamming_distance(true_dag, true_hidden_nodes, prediction): 

18 """ 

19 Compute the standardized structural Hamming distance between two graphs. 

20 

21 Parameters: 

22 true_dag (numpy.ndarray): Target graph adjacency matrix 

23 true_hidden_nodes (list): list of nodes in DAG to hide 

24 prediction (numpy.ndarray): Predicted graph adjacency matrix (pag.graph 

25 from causal-learn) 

26 double_for_anticausal (bool): Count badly oriented edges as two mistakes 

27 

28 Returns: 

29 int: Structural Hamming Distance 

30 """ 

31 n = len(true_dag) - len(true_hidden_nodes) 

32 if (n, n) != prediction.shape: 

33 raise ValueError("Graphs must have the same number of nodes") 

34 

35 nodes = [GraphNode(f"X{i}") for i in range(len(true_dag))] 

36 hidden_nodes = [nodes[int(ind)] for ind in true_hidden_nodes] 

37 

38 cl_dag = Dag(nodes) 

39 for ch, pa in np.argwhere(true_dag): 

40 cl_dag.add_directed_edge(nodes[pa], nodes[ch]) 

41 true_pag = dag2pag(cl_dag, hidden_nodes) 

42 

43 total_shd = np.sum(true_pag.graph != prediction) 

44 

45 return total_shd / (n**2 - n)