Skip to content

CS 188 — Bayes Nets & Tracking: pseudocode → Python

Your pseudocode, written out in the actual API of the project. Constants like PAC, X_RANGE, MAX_NOISE are already defined for you at the top of each starter function — I've kept them so each block reads standalone.

Part 1 — Factors & variable elimination

Q1 — constructBayesNet (inference.py)

def constructBayesNet(gameState):
    # These five are given in the starter code
    PAC = "Pacman"
    GHOST0 = "Ghost0"
    GHOST1 = "Ghost1"
    OBS0 = "Observation0"
    OBS1 = "Observation1"

    X_RANGE = gameState.getWalls().width
    Y_RANGE = gameState.getWalls().height
    MAX_NOISE = 7

    variables = [PAC, GHOST0, GHOST1, OBS0, OBS1]
    edges = [(PAC, OBS0), (GHOST0, OBS0), (PAC, OBS1), (GHOST1, OBS1)]

    positions = [(x, y) for x in range(X_RANGE) for y in range(Y_RANGE)]

    maxManhattan = (X_RANGE - 1) + (Y_RANGE - 1)
    obsDomain = list(range(maxManhattan + MAX_NOISE + 1))   # 0 .. max, inclusive

    variableDomainsDict = {
        PAC:    positions,
        GHOST0: positions,
        GHOST1: positions,
        OBS0:   obsDomain,
        OBS1:   obsDomain,
    }

    return constructEmptyBayesNet(variables, edges, variableDomainsDict)

Watch: range(n) is exclusive, so you need maxManhattan + MAX_NOISE + 1 to actually include the largest reading. Off-by-one here is the usual q1 failure.

Q2 — joinFactors (factorOperations.py)

def joinFactors(factors):
    factors = list(factors)     # may arrive as a set or generator — pin it down first

    # ... starter assertion about no variable being unconditioned in two factors stays here ...

    unconditioned = set()
    conditioned = set()
    for factor in factors:
        unconditioned |= set(factor.unconditionedVariables())
        conditioned   |= set(factor.conditionedVariables())
    conditioned -= unconditioned    # a var conditioned here but unconditioned there ends up unconditioned

    domains = factors[0].variableDomainsDict()
    newFactor = Factor(unconditioned, conditioned, domains)

    for assignment in newFactor.getAllPossibleAssignmentDicts():
        product = 1.0
        for factor in factors:
            product *= factor.getProbability(assignment)   # extra keys are ignored
        newFactor.setProbability(assignment, product)

    return newFactor

Why the subtraction: joining P(A|B) with P(B) gives P(A,B) — B was conditioned in the first factor but unconditioned in the second, and the join resolves it to unconditioned.

Q3 — eliminate (factorOperations.py)

def eliminate(factor, eliminationVariable):
    # ... starter assertions (var is unconditioned; factor has >1 unconditioned var) stay here ...

    unconditioned = set(factor.unconditionedVariables()) - {eliminationVariable}
    conditioned   = set(factor.conditionedVariables())
    domains       = factor.variableDomainsDict()    # same dict, unchanged

    newFactor = Factor(unconditioned, conditioned, domains)

    for assignment in newFactor.getAllPossibleAssignmentDicts():
        total = 0.0
        for value in domains[eliminationVariable]:
            fullAssignment = dict(assignment)               # copy, don't mutate
            fullAssignment[eliminationVariable] = value
            total += factor.getProbability(fullAssignment)
        newFactor.setProbability(assignment, total)

    return newFactor

Keeping the original variableDomainsDict matters — the eliminated variable stays in it even though it's no longer in the factor, and later joins rely on that.

Q4 — inferenceByVariableElimination (inference.py)

def inferenceByVariableElimination(bayesNet, queryVariables, evidenceDict, eliminationOrder):
    # joinFactorsByVariable and eliminate are already bound above by the call-tracking wrapper
    currentFactorsList = bayesNet.getAllCPTsWithEvidence(evidenceDict)

    for var in eliminationOrder:
        currentFactorsList, joinedFactor = joinFactorsByVariable(currentFactorsList, var)

        if len(joinedFactor.unconditionedVariables()) == 1:
            continue        # sums to 1 — throw it away, do NOT call eliminate (it asserts)

        currentFactorsList.append(eliminate(joinedFactor, var))

    fullJoint = joinFactors(currentFactorsList)
    return normalize(fullJoint)

joinFactorsByVariable does the "pull out the factors mentioning var, join them, hand back the rest" step in one call, so you don't filter the list by hand.

The single-unconditioned-variable case is exactly why eliminate has that second assertion — eliminating the last unconditioned variable would leave a factor over nothing.

Part 2 — Exact inference

Q5a — DiscreteDistribution (inference.py)

def normalize(self):
    total = self.total()
    if total == 0:
        return                          # empty or all-zero: leave it alone
    for key in list(self.keys()):
        self[key] = self[key] / total   # in place, returns None

def sample(self):
    total = self.total()
    r = random.random() * total         # scaling by total means no need to normalize first
    running = 0.0
    for key, value in self.items():
        running += value
        if running > r:
            return key
    return list(self.keys())[-1]        # float-rounding safety net

> vs >= only differs on zero-probability keys — with >=, a key with probability 0 sitting right where the running total lands could get returned. Use >.

Q5b — getObservationProb (InferenceModule)

def getObservationProb(self, noisyDistance, pacmanPosition, ghostPosition, jailPosition):
    if ghostPosition == jailPosition:
        return 1.0 if noisyDistance is None else 0.0
    if noisyDistance is None:
        return 0.0                      # reading is None but ghost isn't jailed: impossible

    trueDistance = manhattanDistance(pacmanPosition, ghostPosition)
    return busters.getObservationProbability(noisyDistance, trueDistance)

Jail check first, as you noted — it's the only case where None is a valid reading.

Q6 — ExactInference.observeUpdate

B'(g) ∝ P(obs | pac, g) · B(g)

def observeUpdate(self, observation, gameState):
    pacmanPosition = gameState.getPacmanPosition()
    jailPosition = self.getJailPosition()

    for pos in self.allPositions:
        self.beliefs[pos] *= self.getObservationProb(
            observation, pacmanPosition, pos, jailPosition)

    self.beliefs.normalize()

Q7 — ExactInference.elapseTime

B'(g') = Σ_g P(g' | g) · B(g)

def elapseTime(self, gameState):
    newBeliefs = DiscreteDistribution()

    for oldPos in self.allPositions:
        if self.beliefs[oldPos] == 0:
            continue        # skips the expensive getPositionDistribution call entirely
        newPosDist = self.getPositionDistribution(gameState, oldPos)
        for newPos, prob in newPosDist.items():
            newBeliefs[newPos] += prob * self.beliefs[oldPos]

    newBeliefs.normalize()
    self.beliefs = newBeliefs

The continue is the whole optimization: once a few observations have come in, most positions have belief 0 and you skip nearly all the work.

Q8 — GreedyBustersAgent.chooseAction (bustersAgents.py)

def chooseAction(self, gameState):
    pacmanPosition = gameState.getPacmanPosition()
    legal = [a for a in gameState.getLegalPacmanActions()]
    livingGhosts = gameState.getLivingGhosts()
    livingGhostPositionDistributions = [
        beliefs for i, beliefs in enumerate(self.ghostBeliefs)
        if livingGhosts[i + 1]
    ]
    # everything above is starter code

    targets = [dist.argMax() for dist in livingGhostPositionDistributions]

    bestAction = None
    bestDistance = float('inf')
    for action in legal:
        successorPosition = Actions.getSuccessor(pacmanPosition, action)
        d = min(self.distancer.getDistance(successorPosition, t) for t in targets)
        if d < bestDistance:
            bestDistance = d
            bestAction = action

    return bestAction

Note the i + 1 — index 0 of livingGhosts is Pacman, so ghost i lives at index i + 1.

Part 3 — Particle filtering

Q9 — ParticleFilter.initializeUniformly / getBeliefDistribution

def initializeUniformly(self, gameState):
    self.particles = []
    for i in range(self.numParticles):
        self.particles.append(self.legalPositions[i % len(self.legalPositions)])


def getBeliefDistribution(self):
    dist = DiscreteDistribution()
    for particle in self.particles:
        dist[particle] += 1     # DiscreteDistribution defaults missing keys to 0
    dist.normalize()
    return dist

Modular, not random — the autograder checks that the initial spread is exactly even.

Q10 — ParticleFilter.observeUpdate

def observeUpdate(self, observation, gameState):
    pacmanPosition = gameState.getPacmanPosition()
    jailPosition = self.getJailPosition()

    weights = DiscreteDistribution()
    for particle in self.particles:
        weights[particle] += self.getObservationProb(
            observation, pacmanPosition, particle, jailPosition)

    if weights.total() == 0:
        self.initializeUniformly(gameState)     # all particles died — restart
    else:
        weights.normalize()
        self.particles = [weights.sample() for _ in range(self.numParticles)]

Accumulating into a DiscreteDistribution keyed by position (rather than a parallel list) means duplicate particles automatically get proportionally more weight — which is what you want.

Q11 — ParticleFilter.elapseTime

def elapseTime(self, gameState):
    newParticles = []
    for particle in self.particles:
        newPosDist = self.getPositionDistribution(gameState, particle)
        newParticles.append(newPosDist.sample())
    self.particles = newParticles

Speedup for the slow run: many particles sit on the same position, so you're recomputing identical distributions hundreds of times. Cache them:

def elapseTime(self, gameState):
    cache = {}
    newParticles = []
    for particle in self.particles:
        if particle not in cache:
            cache[particle] = self.getPositionDistribution(gameState, particle)
        newParticles.append(cache[particle].sample())
    self.particles = newParticles

Same output distribution, usually a large fraction of the runtime gone.