ch5: robust MC (no env.is_terminal dependency)

Author: srikanthbaride

ch5_monte_carlo/examples/mc_control_es_gridworld.py

@@ -1,37 +1,53 @@
-# ch5_monte_carlo/examples/mc_control_es_gridworld.py
-import numpy as np
+import numpy as np
 from ch4_dynamic_programming.gridworld import GridWorld4x4
 
-__all__ = ["mc_es_control", "generate_episode_es"]
+__all__ = ["mc_es_control", "generate_episode_es", "ACTIONS"]
 
-ACTIONS = [(0, 1), (0, -1), (1, 0), (-1, 0)]  # R,L,D,U (must match ch4 env)
+# Must match env's action ordering
+ACTIONS = [(0, 1), (0, -1), (1, 0), (-1, 0)]  # R, L, D, U
+
+def _is_terminal(env: GridWorld4x4, s) -> bool:
+    """Robust terminal check even if env.is_terminal is absent."""
+    if hasattr(env, "is_terminal"):
+        return bool(env.is_terminal(s))
+    st = s if isinstance(s, tuple) else env.i2s[int(s)]
+    return st == env.goal
+
+def _step(env: GridWorld4x4, s, a):
+    """Use env.step if present; otherwise use P/R (deterministic)."""
+    if hasattr(env, "step"):
+        return env.step(s, a)
+    s_idx = env.s2i[s] if isinstance(s, tuple) else int(s)
+    probs = env.P[s_idx, a]
+    sp_idx = int(np.argmax(probs))
+    r = float(env.R[s_idx, a, sp_idx])
+    return env.i2s[sp_idx], r
 
 def _greedy_action(q_row: np.ndarray) -> int:
     return int(np.argmax(q_row))
 
 def generate_episode_es(env: GridWorld4x4, Q: np.ndarray, gamma: float, max_steps: int = 10000):
     """
     Exploring starts: start from a random non-terminal state and random action,
-    then follow greedy policy w.r.t. Q thereafter. Returns (states, actions, returns).
+    then follow greedy policy thereafter. Returns (states, actions, returns).
     """
     rng = np.random.default_rng()
-    non_terminal = [s for s in env.S if not env.is_terminal(s)]
+    non_terminal = [s for s in env.S if not _is_terminal(env, s)]
     s = non_terminal[rng.integers(len(non_terminal))]
     a = int(rng.integers(len(env.A)))
 
     states = [s]
     actions = [a]
-    rewards = [0.0]  # so rewards[t+1] aligns with action taken at t
+    rewards = [0.0]  # so rewards[t+1] aligns with action at t
 
     steps = 0
-    while not env.is_terminal(s) and steps < max_steps:
-        sp, r = env.step(s, a)
+    while not _is_terminal(env, s) and steps < max_steps:
+        sp, r = _step(env, s, a)
         rewards.append(float(r))
         s = sp
-        if env.is_terminal(s):
+        if _is_terminal(env, s):
             break
-        s_idx = env.s2i[s]
-        a = _greedy_action(Q[s_idx])
+        a = _greedy_action(Q[env.s2i[s]])
         states.append(s)
         actions.append(a)
         steps += 1
@@ -67,7 +83,7 @@ def mc_es_control(env: GridWorld4x4, episodes: int = 1500, gamma: float | None =
             s_idx = env.s2i[s]
             key = (s_idx, a)
             if key in seen:
-                continue  # first-visit MC
+                continue
             seen.add(key)
             G = returns[t]
             N[s_idx, a] += 1.0
@@ -77,11 +93,3 @@ def mc_es_control(env: GridWorld4x4, episodes: int = 1500, gamma: float | None =
     pi = np.zeros((S, A), dtype=float)
     pi[np.arange(S), np.argmax(Q, axis=1)] = 1.0
     return Q, pi
-
-# Optional: run this file directly for a quick check
-if __name__ == "__main__":
-    env = GridWorld4x4(step_reward=-1.0, goal=(0, 3), gamma=1.0)
-    Q, pi = mc_es_control(env, episodes=2000, seed=0)
-    start = env.s2i[(0, 0)]
-    print("Q(start):", Q[start])
-    print("Greedy action at start:", int(np.argmax(pi[start])))

ch5_monte_carlo/examples/mc_control_onpolicy_gridworld.py

@@ -1,37 +1,47 @@
-# ch5_monte_carlo/examples/mc_control_onpolicy_gridworld.py
-import numpy as np
+import numpy as np
 from ch4_dynamic_programming.gridworld import GridWorld4x4
 
 __all__ = ["mc_control_onpolicy", "ACTIONS", "generate_episode_onpolicy"]
 
-# Must match the environment's action ordering
 ACTIONS = [(0, 1), (0, -1), (1, 0), (-1, 0)]  # R, L, D, U
 
-def _epsilon_greedy(Q_row: np.ndarray, epsilon: float, rng: np.random.Generator) -> int:
-    if rng.random() < epsilon:
-        return int(rng.integers(len(Q_row)))
-    return int(np.argmax(Q_row))
+def _is_terminal(env: GridWorld4x4, s) -> bool:
+    if hasattr(env, "is_terminal"):
+        return bool(env.is_terminal(s))
+    st = s if isinstance(s, tuple) else env.i2s[int(s)]
+    return st == env.goal
+
+def _step(env: GridWorld4x4, s, a):
+    if hasattr(env, "step"):
+        return env.step(s, a)
+    s_idx = env.s2i[s] if isinstance(s, tuple) else int(s)
+    probs = env.P[s_idx, a]
+    sp_idx = int(np.argmax(probs))
+    r = float(env.R[s_idx, a, sp_idx])
+    return env.i2s[sp_idx], r
+
+def _epsilon_greedy(q_row: np.ndarray, epsilon: float, rng: np.random.Generator) -> int:
+    return int(rng.integers(len(q_row))) if rng.random() < epsilon else int(np.argmax(q_row))
 
 def generate_episode_onpolicy(env: GridWorld4x4, Q: np.ndarray, epsilon: float,
-                              rng: np.random.Generator, max_steps: int = 10_000):
-    """Start from a random non-terminal state; follow ε-greedy w.r.t. Q throughout."""
-    non_terminal = [s for s in env.S if not env.is_terminal(s)]
+                              rng: np.random.Generator, max_steps: int = 10000):
+    """Start from a random non-terminal state; follow ε-greedy w.r.t. Q."""
+    non_terminal = [s for s in env.S if not _is_terminal(env, s)]
     s = non_terminal[rng.integers(len(non_terminal))]
-    S, A = len(env.S), len(env.A)
 
     states, actions, rewards = [s], [], [0.0]
     steps = 0
-    while not env.is_terminal(s) and steps < max_steps:
+    while not _is_terminal(env, s) and steps < max_steps:
         a = _epsilon_greedy(Q[env.s2i[s]], epsilon, rng)
         actions.append(a)
-        sp, r = env.step(s, a)
+        sp, r = _step(env, s, a)
         rewards.append(float(r))
         s = sp
         states.append(s)
         steps += 1
 
     # first-visit returns
-    gamma = env.gamma
+    gamma = float(getattr(env, "gamma", 1.0))
     G = 0.0
     returns = np.zeros(len(actions), dtype=float)
     for t in range(len(actions) - 1, -1, -1):
@@ -43,15 +53,15 @@ def mc_control_onpolicy(env: GridWorld4x4, episodes: int = 5000,
                         epsilon: float = 0.1, gamma: float | None = None,
                         seed: int | None = None):
     """
-    On-policy Monte Carlo control using ε-greedy behavior/target policy (no exploring starts).
+    On-policy MC control using ε-greedy behavior/target policy (no ES).
     Returns:
-        Q:  (S,A) table
-        pi: (S,A) deterministic greedy policy derived from Q
+        Q:  (S,A)
+        pi: (S,A) deterministic greedy policy
     """
     rng = np.random.default_rng(seed)
     S, A = len(env.S), len(env.A)
     if gamma is None:
-        gamma = float(env.gamma)
+        gamma = float(getattr(env, "gamma", 1.0))
 
     Q = np.zeros((S, A), dtype=float)
     N = np.zeros((S, A), dtype=float)
@@ -63,21 +73,13 @@ def mc_control_onpolicy(env: GridWorld4x4, episodes: int = 5000,
             s_idx = env.s2i[s]
             key = (s_idx, a)
             if key in seen:
-                continue  # first-visit MC
+                continue
             seen.add(key)
             G = returns[t]
             N[s_idx, a] += 1.0
             alpha = 1.0 / N[s_idx, a]
             Q[s_idx, a] += alpha * (G - Q[s_idx, a])
 
-    # deterministic greedy policy
     pi = np.zeros((S, A), dtype=float)
     pi[np.arange(S), np.argmax(Q, axis=1)] = 1.0
     return Q, pi
-
-if __name__ == "__main__":
-    env = GridWorld4x4(step_reward=-1.0, goal=(0, 3), gamma=1.0)
-    Q, pi = mc_control_onpolicy(env, episodes=3000, epsilon=0.1, seed=0)
-    s0 = env.s2i[(0, 0)]
-    print("Q(start):", Q[s0])
-    print("Greedy action at start:", int(np.argmax(pi[s0])))