stormvogel.teaching.policy_iteration

Policy iteration for MDPs, using exact sympy-based DTMC evaluation.

Each step induces a DTMC from the current scheduler, evaluates it exactly via stormvogel.teaching.dtmc_evaluation, then improves the scheduler greedily. Results are exact sympy rationals throughout.

Classes

PI

Policy iteration for reachability probabilities on an MDP.

Functions

initial_scheduler(→ stormvogel.result.Scheduler)

Return a suitable starting scheduler for policy iteration.

policy_evaluation() → dict[stormvogel.model.State, ...)

Evaluate scheduler exactly and return per-state values for mdp.

policy_improvement(→ stormvogel.result.Scheduler)

Return the improved scheduler for values.

visualise_pi_iterations(pi[, max_steps])

Run pi to convergence and return a pandas DataFrame of each step.

Module Contents

stormvogel.teaching.policy_iteration.initial_scheduler(mdp: stormvogel.model.Model, target_label: str, minimize: bool) → stormvogel.result.Scheduler

Return a suitable starting scheduler for policy iteration.

Uses a one-step Bellman lookahead from the zero vector (target states fixed to 1, all others 0) to bias the initial scheduler:

  • minimize — picks the action with the lowest one-step expected value, giving a pessimistic start that avoids the target.

  • maximize — picks the action with the highest one-step expected value, giving an optimistic start that heads towards the target.

Parameters:

  • mdp – The MDP.

  • target_label – Label identifying the target states.

  • minimize – If True, build a pessimistic scheduler; otherwise an optimistic one.

Returns:

A Scheduler for mdp.

stormvogel.teaching.policy_iteration.policy_evaluation(mdp: stormvogel.model.Model, one_states: Iterable[stormvogel.model.State], scheduler: stormvogel.result.Scheduler, reward_model: stormvogel.model.RewardModel | None = None, discount: sympy.Expr = sp.Integer(1)) → dict[stormvogel.model.State, sympy.Expr]

Evaluate scheduler exactly and return per-state values for mdp.

Induces a DTMC from scheduler, solves it via sympy, then lifts the values back to MDP states by matching UUIDs.

Parameters:

  • mdp – The MDP.

  • one_states – Target (absorbing) states fixed to value 1.

  • scheduler – The scheduler to evaluate.

  • reward_model – If provided, solve for expected reward instead of reachability.

  • discount – Discount factor (default 1, i.e. undiscounted).

Returns:

Exact sympy values for every state in mdp.

Raises:

ValueError – If the model is not an MDP or POMDP.

stormvogel.teaching.policy_iteration.policy_improvement(mdp: stormvogel.model.Model, values: Mapping[stormvogel.model.State, sympy.Expr], one_states: Iterable[stormvogel.model.State], minimize: bool, current_scheduler: stormvogel.result.Scheduler) → stormvogel.result.Scheduler

Return the improved scheduler for values.

For each state, pick the action that minimises (or maximises) the expected value under the current value function. If the current action is already optimal (ties included), it is kept — only a strict improvement causes a switch. This tie-breaking rule guarantees that PI terminates: every iteration that changes the scheduler strictly improves the objective.

Parameters:

  • mdp – The MDP.

  • values – Current value for each state (from DTMC evaluation).

  • one_states – States fixed to value 1; any action is kept for them.

  • minimize – If True, minimize; otherwise maximize.

  • current_scheduler – The scheduler used to produce values.

Returns:

A new Scheduler.

class stormvogel.teaching.policy_iteration.PI(mdp: stormvogel.model.Model, target_label: str, minimize: bool, scheduler: stormvogel.result.Scheduler | None = None, reward_model: stormvogel.model.RewardModel | None = None, discount: sympy.Expr = sp.Integer(1))

Policy iteration for reachability probabilities on an MDP.

Usage:

pi = PI(mdp, "target", minimize=False)
while not pi.has_converged():
    scheduler, values = pi.step()
Parameters:

  • mdp – The MDP to optimise.

  • target_label – Label identifying the target (one) states.

  • minimize – If True, minimize reachability; otherwise maximize.

  • scheduler – Starting scheduler. If None (default), initial_scheduler() is called with the same minimize flag.

_mdp
_one_states
_minimize
_scheduler
_reward_model = None
_discount = None
_values: dict[stormvogel.model.State, sympy.Expr] | None = None
_converged = False
property current_scheduler: stormvogel.result.Scheduler
property current_values: dict[stormvogel.model.State, sympy.Expr] | None

Values from the last evaluation step, or None before the first step.

has_converged() → bool

True after a step where the scheduler did not change.

step() → tuple[stormvogel.result.Scheduler, dict[stormvogel.model.State, sympy.Expr]]

Run one policy-iteration step: evaluate then improve.

Returns:

The new scheduler and the exact values under the old one.

Raises:

RuntimeError – If the scheduler cannot induce a DTMC.

stormvogel.teaching.policy_iteration.visualise_pi_iterations(pi: PI, max_steps: int = 50)

Run pi to convergence and return a pandas DataFrame of each step.

Each column group step k shows the scheduler evaluated at that step and the exact values it produced — so the action and value in the same column are always consistent.

Rows are states; columns are a two-level MultiIndex (step k, "value") / (step k, "action").

Parameters:

  • pi – A PI instance (not yet converged).

  • max_steps – Safety limit on the number of steps.

Returns:

A pandas.DataFrame indexed by state friendly name.