1
2
3
4
5
6
7
8
9
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
|
import copy
import numpy as np
import networkx as nx
from beliefs.models.DirectedGraph import DirectedGraph
from beliefs.utils.edges_helper import EdgesHelper
from beliefs.utils.math_helper import is_kronecker_delta
class BayesianModel(DirectedGraph):
"""
Bayesian model stores nodes and edges described by conditional probability
distributions.
"""
def __init__(self, edges, nodes_dict=None):
"""
Input:
edges: list of edge tuples of form ('parent', 'child')
nodes: (optional) dict
a dict key, value pair as {label_id: instance_of_node_class_or_subclass}
"""
if nodes_dict is not None:
super().__init__(edges, nodes_dict.keys())
else:
super().__init__(edges)
self.nodes_dict = nodes_dict
@classmethod
def from_node_class(cls, edges, node_class):
"""Automatically create all nodes from the same node class
Input:
edges: list of edge tuples of form ('parent', 'child')
node_class: (optional) the Node class or subclass from which to
create all the nodes from edges.
"""
nodes = cls.create_nodes(edges, node_class)
return cls.__init__(edges=edges, nodes=nodes)
@staticmethod
def create_nodes(edges, node_class):
"""Returns list of Node instances created from edges using
the default node_class"""
edges_helper = EdgesHelper(edges)
nodes = edges_helper.create_nodes_from_edges(node_class=node_class)
label_to_node = dict()
for node in nodes:
label_to_node[node.label_id] = node
return label_to_node
def set_boundary_conditions(self):
"""
1. Root nodes: if x is a node with no parents, set Pi(x) = prior
probability of x.
2. Leaf nodes: if x is a node with no children, set Lambda(x)
to an (unnormalized) unit vector, of length the cardinality of x.
"""
for root in self.get_roots():
self.nodes_dict[root].pi_agg = self.nodes_dict[root].cpd.values
for leaf in self.get_leaves():
self.nodes_dict[leaf].lambda_agg = np.ones([self.nodes_dict[leaf].cardinality])
@property
def all_nodes_are_fully_initialized(self):
"""
Returns True if, for all nodes in the model, all lambda and pi
messages and lambda_agg and pi_agg are not None, else False.
"""
for node in self.nodes_dict.values():
if not node.is_fully_initialized:
return False
return True
def copy(self):
"""
Returns a copy of the model.
"""
copy_edges = list(self.edges()).copy()
copy_nodes = copy.deepcopy(self.nodes_dict)
copy_model = self.__class__(edges=copy_edges, nodes=copy_nodes)
return copy_model
def get_variables_in_definite_state(self):
"""
Returns a set of labels of all nodes in a definite state, i.e. with
label values that are kronecker deltas.
RETURNS
set of strings (labels)
"""
return {label for label, node in self.nodes_dict.items() if is_kronecker_delta(node.belief)}
def get_unobserved_variables_in_definite_state(self, observed=set()):
"""
Returns a set of labels that are inferred to be in definite state, given
list of labels that were directly observed (e.g. YES/NOs, but not MAYBEs).
INPUT
observed: set of strings, directly observed labels
RETURNS
set of strings, labels inferred to be in a definite state
"""
# Assert that beliefs of directly observed vars are kronecker deltas
for label in observed:
assert is_kronecker_delta(self.nodes_dict[label].belief), \
("Observed label has belief {} but should be kronecker delta"
.format(self.nodes_dict[label].belief))
vars_in_definite_state = self.get_variables_in_definite_state()
assert observed <= vars_in_definite_state, \
"Expected set of observed labels to be a subset of labels in definite state."
return vars_in_definite_state - observed
def _get_ancestors_of(self, observed):
"""Return list of ancestors of observed labels, including the observed labels themselves."""
ancestors = observed.copy()
for label in observed:
ancestors.update(nx.ancestors(self, label))
return ancestors
def reachable_observed_variables(self, source, observed=set()):
"""
Returns list of observed labels (labels with direct evidence to be in a definite
state) that are reachable from the source.
INPUT
source: string, label of node for which to evaluate reachable observed labels
observed: set of strings, directly observed labels
RETURNS
reachable_observed_vars: set of strings, observed labels (variables with direct
evidence) that are reachable from the source label.
"""
ancestors_of_observed = self._get_ancestors_of(observed)
visit_list = set()
visit_list.add((source, 'up'))
traversed_list = set()
reachable_observed_vars = set()
while visit_list:
node, direction = visit_list.pop()
if (node, direction) not in traversed_list:
if node in observed:
reachable_observed_vars.add(node)
traversed_list.add((node, direction))
if direction == 'up' and node not in observed:
for parent in self.predecessors(node):
# causal flow
visit_list.add((parent, 'up'))
for child in self.successors(node):
# common cause flow
visit_list.add((child, 'down'))
elif direction == 'down':
if node not in observed:
# evidential flow
for child in self.successors(node):
visit_list.add((child, 'down'))
if node in ancestors_of_observed:
# common effect flow (activated v-structure)
for parent in self.predecessors(node):
visit_list.add((parent, 'up'))
return reachable_observed_vars
|
