Skip to content

EdgeStateMatrix

EdgeStateMatrix

A class for managing an edge state matrix.

An edge state matrix is square, with the entry (i,j) representing the state of the directed edge between nodes i and j. The state of an edge is one of: 0: The existence of the state is undecided. -1: The edge does not exist. 1: The edge exists.

Self-edges are not allowed. The presence of an edge implies the absence of its inverse.

Source code in src/logos/edge_state_matrix.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
class EdgeStateMatrix:
    """
    A class for managing an edge state matrix.

    An edge state matrix is square, with the entry (i,j) representing the state
    of the directed edge between nodes i and j. The state of an edge is one of:
         0: The existence of the state is undecided.
        -1: The edge does not exist.
         1: The edge exists.

    Self-edges are not allowed. The presence of an edge implies the absence of
    its inverse.
    """

    def __init__(self, variables: list[str]) -> None:
        """
        Initialize the edge state matrix to the right dimensions and mark self-edges
        as rejected and all other edges as undecided.

        Parameters:
            variables: The variables to initialize the edge state matrix based on. This
                list must include variable NAMES, not tags.
        """

        n = len(variables)
        self._variables = variables
        self._m = np.zeros((n, n))
        for i in range(n):
            self._m[i, i] = -1

    @property
    def m(self) -> np.ndarray:
        """
        Returns the edge state matrix.
        """
        return self._m

    @property
    def n(self) -> int:
        """
        Returns the number of nodes.
        """
        return self._m.shape[0]

    def clear_and_set_from_graph(self, graph: nx.DiGraph) -> None:
        """
        Clear the edge state matrix and then set it based on the provided graph.
        In particular, mark all edges in the graph as accepted and all others as rejected.

        Parameters:
            graph: The graph to use to set the edge states.
        """

        self._m = np.zeros((self.n, self.n))
        for edge in graph.edges:
            print("Marking edge as accepted: ", edge)
            self._m[self.idx(edge[0]), self.idx(edge[1])] = 1

        self._m[self._m == 0] = -1

    def clear_and_set_from_matrix(self, m: np.ndarray) -> None:
        """
        Clear the edge state matrix and then set it based on the provided matrix.

        Parameters:
            m: The matrix to use to set the edge states.
        """

        self._m = m

    def idx(self, var: str) -> int:
        """
        Retrieve the index of a variable in the edge state matrix.

        Parameters:
            var: The name or tag of the variable.

        Returns:
            The index of the variable in the edge state matrix.
        """
        return self._variables.index(var)

    def get_edge_state(self, src: str, dst: str) -> str:
        """
        Get the state of a specific edge.

        Parameters:
            src: The name or tag of the source variable.
            dst: The name or tag of the destination variable.

        Returns:
            The state of the edge (Accepted, Rejected, or Undecided).
        """
        src_idx = self.idx(src)
        dst_idx = self.idx(dst)
        return self.edge_state_to_str(self._m[src_idx][dst_idx])

    def edge_state_to_str(self, state: int) -> str:
        """
        Translate between edge value and its interpretation.

        Parameters:
            state: The state of the edge represented as an integer.

        Returns:
            The state of the edge (Accepted, Rejected, or Undecided).
        """
        if state == 0:
            return "Undecided"
        elif state == -1:
            return "Rejected"
        elif state == 1:
            return "Accepted"
        else:
            raise ValueError(f"Invalid edge state {state}")

    def mark_edge(self, src: str, dst: str, state: str) -> list[str]:
        """
        Mark an edge as being in a specified state.

        Parameters:
            src: The name or tag of the source variable.
            dst: The name or tag of the destination variable.
            state: The state to mark the edge with (Accepted, Rejected, or Undecided).

        Returns:
            A list of variables that were removed from the partial causal graph as a result
            of this edge being marked as Accepted.

        Throws:
            ValueError: If `state` is not one of "Accepted", "Rejected", or "Undecided".
        """

        src_idx = self.idx(src)
        dst_idx = self.idx(dst)

        if state == "Accepted":
            self._m[src_idx][dst_idx] = 1
            self._m[dst_idx][src_idx] = -1
            return self._reject_other_variants(src, dst)
        elif state == "Rejected":
            self._m[src_idx][dst_idx] = -1
            return []
        elif state == "Undecided":
            self._m[src_idx][dst_idx] = 0
            return []
        else:
            raise ValueError(f"Invalid edge state {state}")

    def _reject_other_variants(self, src: str, dst: str) -> list[str]:
        """
        Mark any edges that touch a variable different from `src` and `dst`, but sharing
        the same base variable as `src` or `dst`, as rejected. Also remove any such variables
        from the partial causal graph.

        Parameters:
            src: The name or tag of the source variable.
            dst: The name or tag of the destination variable.

        Returns:
            A list of variables that were removed from the partial causal graph as a result
            of this edge being marked as Accepted.
        """

        src_base = PreparedVariableName(src).base_var()
        dst_base = PreparedVariableName(dst).base_var()

        l = []
        for var in self._variables:
            var_base = PreparedVariableName(var).base_var()
            if (var_base == src_base and var != src) or (
                var_base == dst_base and var != dst
            ):
                self._m[self.idx(var), :] = -1
                self._m[:, self.idx(var)] = -1
                l.append(var)

        return l 

    @staticmethod
    def enumerate_with_max_edges(n: int, max_edges: int) -> list[np.ndarray]:
        """
        Enumerate all edge state matrices of dimension `n` with at most `max_edges` accepted edges.

        Parameters:
            n: The dimension of the edge state matrices.
            max_edges: The maximum number of edges to allow in the edge state matrices.

        Returns:
            A list of edge state matrices.
        """
        valid_matrices = {0: [np.full(shape=(n, n), fill_value=-1)]}

        # Enumerate all valid matrices with k edges
        for k in range(1, max_edges + 1):
            valid_matrices[k] = []

            # For each valid matrix with k-1 edges...
            for m in valid_matrices[k - 1]:
                # ...add a new edge in every possible way
                for i in range(n):
                    for j in range(i + 1, n):
                        if m[i, j] < 0 and m[j, i] < 0:
                            forward = m.copy()
                            forward[i, j] = 1
                            valid_matrices[k].append(forward)
                            backward = m.copy()
                            backward[j, i] = 1
                            valid_matrices[k].append(backward)

        # Flatten the collection of matrices into a single list
        returned_matrices = []
        for k in range(1, max_edges + 1):
            returned_matrices.extend(valid_matrices[k])

        return returned_matrices

m: np.ndarray property

Returns the edge state matrix.

n: int property

Returns the number of nodes.

__init__(variables)

Initialize the edge state matrix to the right dimensions and mark self-edges as rejected and all other edges as undecided.

Parameters:

Name Type Description Default
variables list[str]

The variables to initialize the edge state matrix based on. This list must include variable NAMES, not tags.

 required
Source code in src/logos/edge_state_matrix.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(self, variables: list[str]) -> None:
    """
    Initialize the edge state matrix to the right dimensions and mark self-edges
    as rejected and all other edges as undecided.

    Parameters:
        variables: The variables to initialize the edge state matrix based on. This
            list must include variable NAMES, not tags.
    """

    n = len(variables)
    self._variables = variables
    self._m = np.zeros((n, n))
    for i in range(n):
        self._m[i, i] = -1

clear_and_set_from_graph(graph)

Clear the edge state matrix and then set it based on the provided graph. In particular, mark all edges in the graph as accepted and all others as rejected.

Parameters:

Name Type Description Default
graph DiGraph

The graph to use to set the edge states.

 required
Source code in src/logos/edge_state_matrix.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def clear_and_set_from_graph(self, graph: nx.DiGraph) -> None:
    """
    Clear the edge state matrix and then set it based on the provided graph.
    In particular, mark all edges in the graph as accepted and all others as rejected.

    Parameters:
        graph: The graph to use to set the edge states.
    """

    self._m = np.zeros((self.n, self.n))
    for edge in graph.edges:
        print("Marking edge as accepted: ", edge)
        self._m[self.idx(edge[0]), self.idx(edge[1])] = 1

    self._m[self._m == 0] = -1

clear_and_set_from_matrix(m)

Clear the edge state matrix and then set it based on the provided matrix.

Parameters:

Name Type Description Default
m ndarray

The matrix to use to set the edge states.

 required
Source code in src/logos/edge_state_matrix.py 
70
71
72
73
74
75
76
77
78
def clear_and_set_from_matrix(self, m: np.ndarray) -> None:
    """
    Clear the edge state matrix and then set it based on the provided matrix.

    Parameters:
        m: The matrix to use to set the edge states.
    """

    self._m = m

idx(var)

Retrieve the index of a variable in the edge state matrix.

Parameters:

Name Type Description Default
var str

The name or tag of the variable.

 required

Returns:

Type Description
int

The index of the variable in the edge state matrix.
Source code in src/logos/edge_state_matrix.py 
80
81
82
83
84
85
86
87
88
89
90
def idx(self, var: str) -> int:
    """
    Retrieve the index of a variable in the edge state matrix.

    Parameters:
        var: The name or tag of the variable.

    Returns:
        The index of the variable in the edge state matrix.
    """
    return self._variables.index(var)

get_edge_state(src, dst)

Get the state of a specific edge.

Parameters:

Name Type Description Default
src str

The name or tag of the source variable.

 required
dst str

The name or tag of the destination variable.

 required

Returns:

Type Description
str

The state of the edge (Accepted, Rejected, or Undecided).
Source code in src/logos/edge_state_matrix.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def get_edge_state(self, src: str, dst: str) -> str:
    """
    Get the state of a specific edge.

    Parameters:
        src: The name or tag of the source variable.
        dst: The name or tag of the destination variable.

    Returns:
        The state of the edge (Accepted, Rejected, or Undecided).
    """
    src_idx = self.idx(src)
    dst_idx = self.idx(dst)
    return self.edge_state_to_str(self._m[src_idx][dst_idx])

edge_state_to_str(state)

Translate between edge value and its interpretation.

Parameters:

Name Type Description Default
state int

The state of the edge represented as an integer.

 required

Returns:

Type Description
str

The state of the edge (Accepted, Rejected, or Undecided).
Source code in src/logos/edge_state_matrix.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def edge_state_to_str(self, state: int) -> str:
    """
    Translate between edge value and its interpretation.

    Parameters:
        state: The state of the edge represented as an integer.

    Returns:
        The state of the edge (Accepted, Rejected, or Undecided).
    """
    if state == 0:
        return "Undecided"
    elif state == -1:
        return "Rejected"
    elif state == 1:
        return "Accepted"
    else:
        raise ValueError(f"Invalid edge state {state}")

mark_edge(src, dst, state)

Mark an edge as being in a specified state.

Parameters:

Name Type Description Default
src str

The name or tag of the source variable.

 required
dst str

The name or tag of the destination variable.

 required
state str

The state to mark the edge with (Accepted, Rejected, or Undecided).

 required

Returns:

Type Description
list[str]

A list of variables that were removed from the partial causal graph as a result
list[str]

of this edge being marked as Accepted.
Throws

ValueError: If state is not one of "Accepted", "Rejected", or "Undecided".

Source code in src/logos/edge_state_matrix.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def mark_edge(self, src: str, dst: str, state: str) -> list[str]:
    """
    Mark an edge as being in a specified state.

    Parameters:
        src: The name or tag of the source variable.
        dst: The name or tag of the destination variable.
        state: The state to mark the edge with (Accepted, Rejected, or Undecided).

    Returns:
        A list of variables that were removed from the partial causal graph as a result
        of this edge being marked as Accepted.

    Throws:
        ValueError: If `state` is not one of "Accepted", "Rejected", or "Undecided".
    """

    src_idx = self.idx(src)
    dst_idx = self.idx(dst)

    if state == "Accepted":
        self._m[src_idx][dst_idx] = 1
        self._m[dst_idx][src_idx] = -1
        return self._reject_other_variants(src, dst)
    elif state == "Rejected":
        self._m[src_idx][dst_idx] = -1
        return []
    elif state == "Undecided":
        self._m[src_idx][dst_idx] = 0
        return []
    else:
        raise ValueError(f"Invalid edge state {state}")

_reject_other_variants(src, dst)

Mark any edges that touch a variable different from src and dst, but sharing the same base variable as src or dst, as rejected. Also remove any such variables from the partial causal graph.

Parameters:

Name Type Description Default
src str

The name or tag of the source variable.

 required
dst str

The name or tag of the destination variable.

 required

Returns:

Type Description
list[str]

A list of variables that were removed from the partial causal graph as a result
list[str]

of this edge being marked as Accepted.
Source code in src/logos/edge_state_matrix.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def _reject_other_variants(self, src: str, dst: str) -> list[str]:
    """
    Mark any edges that touch a variable different from `src` and `dst`, but sharing
    the same base variable as `src` or `dst`, as rejected. Also remove any such variables
    from the partial causal graph.

    Parameters:
        src: The name or tag of the source variable.
        dst: The name or tag of the destination variable.

    Returns:
        A list of variables that were removed from the partial causal graph as a result
        of this edge being marked as Accepted.
    """

    src_base = PreparedVariableName(src).base_var()
    dst_base = PreparedVariableName(dst).base_var()

    l = []
    for var in self._variables:
        var_base = PreparedVariableName(var).base_var()
        if (var_base == src_base and var != src) or (
            var_base == dst_base and var != dst
        ):
            self._m[self.idx(var), :] = -1
            self._m[:, self.idx(var)] = -1
            l.append(var)

    return l

enumerate_with_max_edges(n, max_edges) staticmethod

Enumerate all edge state matrices of dimension n with at most max_edges accepted edges.

Parameters:

Name Type Description Default
n int

The dimension of the edge state matrices.

 required
max_edges int

The maximum number of edges to allow in the edge state matrices.

 required

Returns:

Type Description
list[ndarray]

A list of edge state matrices.
Source code in src/logos/edge_state_matrix.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
@staticmethod
def enumerate_with_max_edges(n: int, max_edges: int) -> list[np.ndarray]:
    """
    Enumerate all edge state matrices of dimension `n` with at most `max_edges` accepted edges.

    Parameters:
        n: The dimension of the edge state matrices.
        max_edges: The maximum number of edges to allow in the edge state matrices.

    Returns:
        A list of edge state matrices.
    """
    valid_matrices = {0: [np.full(shape=(n, n), fill_value=-1)]}

    # Enumerate all valid matrices with k edges
    for k in range(1, max_edges + 1):
        valid_matrices[k] = []

        # For each valid matrix with k-1 edges...
        for m in valid_matrices[k - 1]:
            # ...add a new edge in every possible way
            for i in range(n):
                for j in range(i + 1, n):
                    if m[i, j] < 0 and m[j, i] < 0:
                        forward = m.copy()
                        forward[i, j] = 1
                        valid_matrices[k].append(forward)
                        backward = m.copy()
                        backward[j, i] = 1
                        valid_matrices[k].append(backward)

    # Flatten the collection of matrices into a single list
    returned_matrices = []
    for k in range(1, max_edges + 1):
        returned_matrices.extend(valid_matrices[k])

    return returned_matrices