Georgia Institute of Technology| CS 6601| Probability Complete Solutions| Latest

"""Testing pbnt. Run this before anything else to get pbnt to work!"""
import sys

if('pbnt/combined' not in sys.path):
sys.path.append('pbnt/combined')
from exampleinference import inferenceExample

inferenceExample()
# Should output:
# ('The marginal probability of sprinkler=false:', 0.80102921)
#('The marginal probability of wetgrass=false | cloudy=False, rain=True:',
0.055)

'''
WRITE YOUR CODE BELOW. DO NOT CHANGE ANY FUNCTION HEADERS FROM THE NOTEBOOK.
'''

from Node import BayesNode
from Graph import BayesNet
from numpy import zeros, float32
from numpy import random
from Distribution import DiscreteDistribution, ConditionalDiscreteDistribution
from Inference import JunctionTreeEngine, EnumerationEngine
from numpy import zeros, float32


def make_power_plant_net():
"""Create a Bayes Net representation of the above power plant problem.
Use the following as the name attribute: "alarm","faulty alarm",
"gauge","faulty gauge", "temperature". (for the tests to work.)
"""
nodes = []

A_node = BayesNode(0, 2, name="alarm")
F_A_node = BayesNode(1, 2, name="faulty alarm")
G_node = BayesNode(2, 2, name="gauge")
F_G_node = BayesNode(3, 2, name="faulty gauge")
T_node = BayesNode(4, 2, name="temperature")

# Add children and parents

A_node.add_parent(F_A_node)
A_node.add_parent(G_node)
G_node.add_parent(T_node)
G_node.add_parent(F_G_node)
F_G_node.add_parent(T_node)

T_node.add_child(F_G_node)
T_node.add_child(G_node)
F_G_node.add_child(G_node)
G_node.add_child(A_node)
F_A_node.add_child(A_node)

nodes = [A_node, F_A_node, G_node, F_G_node, T_node]

return BayesNet(nodes)

def set_probability(bayes_net):
"""Set probability distribution for each node in the power plant system."""
A_node = bayes_net.get_node_by_name("alarm")
F_A_node = bayes_net.get_node_by_name("faulty alarm")
G_node = bayes_net.get_node_by_name("gauge")
F_G_node = bayes_net.get_node_by_name("faulty gauge")
T_node = bayes_net.get_node_by_name("temperature")

, nodes = [A_node, F_A_node, G_node, F_G_node, T_node]

# probability of temperature being normal (false) or hot (True)

T_distribution = DiscreteDistribution(T_node)
T_index = T_distribution.generate_index([],[])
T_distribution[T_index] = [0.8,0.2] # 1=True=Hot 20% of the time
T_node.set_dist(T_distribution)

# Faulty gauge

# Set values for P(F_G | T) aka probabllity of faulty gauge given temp is
hot (or normal)

F_G_dist_CPH = zeros([T_node.size(), F_G_node.size()], dtype=float32)
F_G_dist_CPH[0,:] = [0.95, 0.05] # When T is False(Normal)
F_G_dist_CPH[1,:] = [0.20, 0.80] # when T is true(Hot)
F_G_distribution = ConditionalDiscreteDistribution(nodes=[T_node, F_G_node],
table=F_G_dist_CPH)
F_G_node.set_dist(F_G_distribution)

# Faulty Alarm 15% of the time

F_A_distribution = DiscreteDistribution(F_A_node)
F_A_index = F_A_distribution.generate_index([],[])
F_A_distribution[F_A_index] = [0.85,0.15] # 0=False, 1=True
F_A_node.set_dist(F_A_distribution)

# Gauge

G_dist_CPH = zeros([T_node.size(), F_G_node.size(), G_node.size()],
dtype=float32)
G_dist_CPH[0,0,:] = [0.95, 0.05] # temp normal, gauge not faulty
G_dist_CPH[0,1,:] = [0.20, 0.80] # temp normal, gauge faulty
G_dist_CPH[1,0,:] = [0.05, 0.95] # temp high, gauge not faulty
G_dist_CPH[1,1,:] = [0.80, 0.20] # temp high, gauge faulty
G_distribution = ConditionalDiscreteDistribution(nodes=[T_node, F_G_node,
G_node], table=G_dist_CPH)
G_node.set_dist(G_distribution)

# Alarm

A_dist_CPH = zeros([F_A_node.size(), G_node.size(), A_node.size()],
dtype=float32)
A_dist_CPH[0, 0, :] = [0.90, 0.10]
A_dist_CPH[0, 1, :] = [0.10, 0.90]
A_dist_CPH[1, 0, :] = [0.55, 0.45]
A_dist_CPH[1, 1, :] = [0.45, 0.55]
A_distribution = ConditionalDiscreteDistribution(nodes=[F_A_node, G_node,
A_node], table=A_dist_CPH)
A_node.set_dist(A_distribution)

return bayes_net

def get_alarm_prob(bayes_net, alarm_rings):
"""Calculate the marginal
probability of the alarm
ringing (T/F) in the
power plant system."""

A_node = bayes_net.get_node_by_name("alarm")
engine = EnumerationEngine(bayes_net)
Q = engine.marginal(A_node)[0]
index = Q.generate_index([alarm_rings],range(Q.nDims))