"""Probability models (Chapter 12-13)"""
import copy
import random
from collections import defaultdict
from functools import reduce
import numpy as np
from utils4e import product, probability, extend
# ______________________________________________________________________________
# Chapter 12 Qualifying Uncertainty
# 12.1 Acting Under Uncertainty
def DTAgentProgram(belief_state):
"""A decision-theoretic agent. [Figure 12.1]"""
def program(percept):
belief_state.observe(program.action, percept)
program.action = max(belief_state.actions(), key=belief_state.expected_outcome_utility)
return program.action
program.action = None
return program
# ______________________________________________________________________________
# 12.2 Basic Probability Notation
class ProbDist:
"""A discrete probability distribution. You name the random variable
in the constructor, then assign and query probability of values.
>>> P = ProbDist('Flip'); P['H'], P['T'] = 0.25, 0.75; P['H']
0.25
>>> P = ProbDist('X', {'lo': 125, 'med': 375, 'hi': 500})
>>> P['lo'], P['med'], P['hi']
(0.125, 0.375, 0.5)
"""
def __init__(self, varname='?', freqs=None):
"""If freqs is given, it is a dictionary of values - frequency pairs,
then ProbDist is normalized."""
self.prob = {}
self.varname = varname
self.values = []
if freqs:
for (v, p) in freqs.items():
self[v] = p
self.normalize()
def __getitem__(self, val):
"""Given a value, return P(value)."""
try:
return self.prob[val]
except KeyError:
return 0
def __setitem__(self, val, p):
"""Set P(val) = p."""
if val not in self.values:
self.values.append(val)
self.prob[val] = p
def normalize(self):
"""Make sure the probabilities of all values sum to 1.
Returns the normalized distribution.
Raises a ZeroDivisionError if the sum of the values is 0."""
total = sum(self.prob.values())
if not np.isclose(total, 1.0):
for val in self.prob:
self.prob[val] /= total
return self
def show_approx(self, numfmt='{:.3g}'):
"""Show the probabilities rounded and sorted by key, for the
sake of portable doctests."""
return ', '.join([('{}: ' + numfmt).format(v, p)
for (v, p) in sorted(self.prob.items())])
def __repr__(self):
return "P({})".format(self.varname)
# ______________________________________________________________________________
# 12.3 Inference Using Full Joint Distributions
class JointProbDist(ProbDist):
"""A discrete probability distribute over a set of variables.
>>> P = JointProbDist(['X', 'Y']); P[1, 1] = 0.25
>>> P[1, 1]
0.25
>>> P[dict(X=0, Y=1)] = 0.5
>>> P[dict(X=0, Y=1)]
0.5"""
def __init__(self, variables):
self.prob = {}
self.variables = variables
self.vals = defaultdict(list)
def __getitem__(self, values):
"""Given a tuple or dict of values, return P(values)."""
values = event_values(values, self.variables)
return ProbDist.__getitem__(self, values)
def __setitem__(self, values, p):
"""Set P(values) = p. Values can be a tuple or a dict; it must
have a value for each of the variables in the joint. Also keep track
of the values we have seen so far for each variable."""
values = event_values(values, self.variables)
self.prob[values] = p
for var, val in zip(self.variables, values):
if val not in self.vals[var]:
self.vals[var].append(val)
def values(self, var):
"""Return the set of possible values for a variable."""
return self.vals[var]
def __repr__(self):
return "P({})".format(self.variables)
def event_values(event, variables):
"""Return a tuple of the values of variables in event.
>>> event_values ({'A': 10, 'B': 9, 'C': 8}, ['C', 'A'])
(8, 10)
>>> event_values ((1, 2), ['C', 'A'])
(1, 2)
"""
if isinstance(event, tuple) and len(event) == len(variables):
return event
else:
return tuple([event[var] for var in variables])
def enumerate_joint_ask(X, e, P):
"""Return a probability distribution over the values of the variable X,
given the {var:val} observations e, in the JointProbDist P. [Section 12.3]
>>> P = JointProbDist(['X', 'Y'])
>>> P[0,0] = 0.25; P[0,1] = 0.5; P[1,1] = P[2,1] = 0.125
>>> enumerate_joint_ask('X', dict(Y=1), P).show_approx()
'0: 0.667, 1: 0.167, 2: 0.167'
"""
assert X not in e, "Query variable must be distinct from evidence"
Q = ProbDist(X) # probability distribution for X, initially empty
Y = [v for v in P.variables if v != X and v not in e] # hidden variables.
for xi in P.values(X):
Q[xi] = enumerate_joint(Y, extend(e, X, xi), P)
return Q.normalize()
def enumerate_joint(variables, e, P):
"""Return the sum of those entries in P consistent with e,
provided variables is P's remaining variables (the ones not in e)."""
if not variables:
return P[e]
Y, rest = variables[0], variables[1:]
return sum([enumerate_joint(rest, extend(e, Y, y), P)
for y in P.values(Y)])
# ______________________________________________________________________________
# 12.4 Independence
def is_independent(variables, P):
"""
Return whether a list of variables are independent given their distribution P
P is an instance of JoinProbDist
>>> P = JointProbDist(['X', 'Y'])
>>> P[0,0] = 0.25; P[0,1] = 0.5; P[1,1] = P[1,0] = 0.125
>>> is_independent(['X', 'Y'], P)
False
"""
for var in variables:
event_vars = variables[:]
event_vars.remove(var)
event = {}
distribution = enumerate_joint_ask(var, event, P)
events = gen_possible_events(event_vars, P)
for e in events:
conditional_distr = enumerate_joint_ask(var, e, P)
if conditional_distr.prob != distribution.prob:
return False
return True
def gen_possible_events(vars, P):
"""Generate all possible events of a collection of vars according to distribution of P"""
events = []
def backtrack(vars, P, temp):
if not vars:
events.append(temp)
return
var = vars[0]
for val in P.values(var):
temp[var] = val
backtrack([v for v in vars if v != var], P, copy.copy(temp))
backtrack(vars, P, {})
return events
# ______________________________________________________________________________
# Chapter 13 Probabilistic Reasoning
# 13.1 Representing Knowledge in an Uncertain Domain
class BayesNet:
"""Bayesian network containing only boolean-variable nodes."""
def __init__(self, node_specs=None):
"""
Nodes must be ordered with parents before children.
:param node_specs: an nested iterable object, each element contains (variable name, parents name, cpt)
for each node
"""
self.nodes = []
self.variables = []
node_specs = node_specs or []
for node_spec in node_specs:
self.add(node_spec)
def add(self, node_spec):
"""
Add a node to the net. Its parents must already be in the
net, and its variable must not.
Initialize Bayes nodes by detecting the length of input node specs
"""
if len(node_spec) >= 5:
node = ContinuousBayesNode(*node_spec)
else:
node = BayesNode(*node_spec)
assert node.variable not in self.variables
assert all((parent in self.variables) for parent in node.parents)
self.nodes.append(node)
self.variables.append(node.variable)
for parent in node.parents:
self.variable_node(parent).children.append(node)
def variable_node(self, var):
"""
Return the node for the variable named var.
>>> burglary.variable_node('Burglary').variable
'Burglary'
"""
for n in self.nodes:
if n.variable == var:
return n
raise Exception("No such variable: {}".format(var))
def variable_values(self, var):
"""Return the domain of var."""
return [True, False]
def __repr__(self):
return 'BayesNet({0!r})'.format(self.nodes)
class BayesNode:
"""
A conditional probability distribution for a boolean variable,
P(X | parents). Part of a BayesNet.
"""
def __init__(self, X, parents, cpt):
"""
:param X: variable name,
:param parents: a sequence of variable names or a space-separated string. Representing the names of parent nodes
:param cpt: the conditional probability table, takes one of these forms:
* A number, the unconditional probability P(X=true). You can
use this form when there are no parents.
* A dict {v: p, ...}, the conditional probability distribution
P(X=true | parent=v) = p. When there's just one parent.
* A dict {(v1, v2, ...): p, ...}, the distribution P(X=true |
parent1=v1, parent2=v2, ...) = p. Each key must have as many
values as there are parents. You can use this form always;
the first two are just conveniences.
In all cases the probability of X being false is left implicit,
since it follows from P(X=true).
>>> X = BayesNode('X', '', 0.2)
>>> Y = BayesNode('Y', 'P', {T: 0.2, F: 0.7})
>>> Z = BayesNode('Z', 'P Q',
... {(T, T): 0.2, (T, F): 0.3, (F, T): 0.5, (F, F): 0.7})
"""
if isinstance(parents, str):
parents = parents.split()
# We store the table always in the third form above.
if isinstance(cpt, (float, int)): # no parents, 0-tuple
cpt = {(): cpt}
elif isinstance(cpt, dict):
# one parent, 1-tuple
if cpt and isinstance(list(cpt.keys())[0], bool):
cpt = {(v,): p for v, p in cpt.items()}
assert isinstance(cpt, dict)
for vs, p in cpt.items():
assert isinstance(vs, tuple) and len(vs) == len(parents)
assert all(isinstance(v, bool) for v in vs)
assert 0