"""Chapter 1 lab: a gridworld, some policies, and the return. The world (same as the demo in the chapter): col: 0 1 2 3 4 5 6 row 0 . . . . . . . the long, safe way round row 1 . # # # # # . row 2 S . . . . . G the bridge: short, but... row 3 . P P P P P . ...the pits are right underneath S start (row 2, col 0) G goal: reward +1, episode ends P pit: reward -1, ends # wall: bumping into it (or the edge) leaves you where you are every other move gets `step_reward` (0 by default) A state is a (row, col) tuple. An action is an int: 0 up, 1 right, 2 down, 3 left. With probability `slip` the floor is slippery: your action is replaced by one of the two perpendicular actions (each with probability slip / 2). Pressing "right" on the bridge can therefore drop you into the pit below. Work through the tasks in order; run `python -m pytest -q` after each one. """ import numpy as np # Legend: S start, G goal (+1, episode ends), P pit (-1, episode ends), # wall, . empty LAYOUT = [ ".......", ".#####.", "S.....G", ".PPPPP.", ] # Action a is an index into ACTIONS; each entry is (d_row, d_col). UP, RIGHT, DOWN, LEFT = 0, 1, 2, 3 ACTIONS = [(-1, 0), (0, 1), (1, 0), (0, -1)] ACTION_NAMES = ["up", "right", "down", "left"] # --------------------------------------------------------------------------- # Task 1: the return # --------------------------------------------------------------------------- def discounted_return(rewards, gamma): """The discounted return of a list of rewards, measured from the first step. G_0 = r_0 + gamma * r_1 + gamma^2 * r_2 + ... + gamma^(T-1) * r_(T-1) Args: rewards: list (or 1-D array) of floats r_0, ..., r_(T-1). May be empty. gamma: discount factor in [0, 1]. Returns: G_0 as a Python float (0.0 for an empty list). Example: discounted_return([0, 0, 1], 0.9) == 0.81 Tip: going backwards, G_t = r_t + gamma * G_(t+1), avoids computing powers. """ # TODO: implement (3-4 lines) raise NotImplementedError("Task 1: discounted_return") # --------------------------------------------------------------------------- # Task 2: the environment # --------------------------------------------------------------------------- class GridWorld: """The environment. Its whole interface is reset() and step(a). Attributes you can use (set up for you in __init__): self.grid list of lists of characters, e.g. self.grid[2][6] == "G" self.n_rows, self.n_cols self.slip probability that the action is replaced by a perpendicular one self.step_reward reward for any move that does not land on G or P self.rng a numpy Generator; use it for ALL randomness (reproducibility) self.start the start state, (2, 0) self.state the current state (None before the first reset) self.done True when the episode has ended """ def __init__(self, slip=0.0, step_reward=0.0, seed=None, layout=LAYOUT): self.grid = [list(row) for row in layout] self.n_rows = len(self.grid) self.n_cols = len(self.grid[0]) self.slip = slip self.step_reward = step_reward self.rng = np.random.default_rng(seed) self.start = next( (r, c) for r in range(self.n_rows) for c in range(self.n_cols) if self.grid[r][c] == "S" ) self.state = None self.done = True def cell(self, state): """The character at a (row, col) state.""" r, c = state return self.grid[r][c] def _move(self, state, a): """Where action a takes you from `state` if nothing slips. Moving into a wall '#' or off the grid leaves you where you are.""" r, c = state dr, dc = ACTIONS[a] nr, nc = r + dr, c + dc if not (0 <= nr < self.n_rows and 0 <= nc < self.n_cols): return state if self.grid[nr][nc] == "#": return state return (nr, nc) def reset(self): """Start a new episode. Set self.state to the start state and self.done to False. Returns: the start state (a (row, col) tuple). """ # TODO: implement (3 lines) raise NotImplementedError("Task 2: GridWorld.reset") def step(self, a): """Take action a (0 up, 1 right, 2 down, 3 left) from the current state. 1. Slip: with probability self.slip, replace a by one of the two perpendicular actions, (a + 1) % 4 or (a + 3) % 4, each equally likely. Use self.rng.random() for the coin flips. 2. Move with self._move(self.state, actual_action). 3. Reward: +1.0 if you land on 'G', -1.0 if you land on 'P', otherwise self.step_reward. Landing on G or P ends the episode. 4. Update self.state and self.done. Returns: (next_state, reward, done) Raises: RuntimeError if called after the episode has ended (call reset()). """ if self.done: raise RuntimeError("episode is over; call reset()") # TODO: implement (about 12 lines) raise NotImplementedError("Task 2: GridWorld.step") # --------------------------------------------------------------------------- # Task 3: policies # --------------------------------------------------------------------------- # A policy here is a function policy(state, rng) -> action. It must return an # action for EVERY non-terminal state, not just the ones on its favorite path, # because slipping can put you anywhere. def random_policy(state, rng): """Pick one of the 4 actions uniformly at random, using rng (a numpy Generator). Return a Python int.""" # TODO: implement (1 line) raise NotImplementedError("Task 3: random_policy") def bridge_policy(state, rng=None): """A complete example of a hand-written policy: walk along row 2 (the bridge). It is deterministic, so it ignores rng. Read it, then write safe_policy below. """ r, c = state if r < 2: return DOWN if c in (0, 6) else RIGHT # got knocked up: head back down to the bridge if r == 3: return UP # bottom corners: back up return RIGHT # on the bridge: go right def safe_policy(state, rng=None): """Your hand-written policy: take the long way round along the top row. From the start, go up the left column to row 0, right along row 0 to column 6, then down the right column to the goal (10 moves without slips). Decide what to do in every other non-terminal state too: the bridge cells (row 2, cols 1-5), the bottom corners (3, 0) and (3, 6), and so on. Requirement checked by the tests: with slip = 0 it reaches G from the start in exactly 10 steps and never visits row 2, columns 1 to 5. """ # TODO: implement (about 8 lines of if/return) raise NotImplementedError("Task 3: safe_policy") # --------------------------------------------------------------------------- # Task 4: one episode # --------------------------------------------------------------------------- def run_episode(env, policy, gamma=1.0, max_steps=100, rng=None): """Play one episode with `policy` in `env`. reset the env, then repeat: a = policy(state, rng); state, r, done = env.step(a) until done or max_steps steps have been taken (whichever comes first). Args: env: a GridWorld policy: function (state, rng) -> action gamma: discount factor for the return max_steps: cut the episode off after this many steps rng: numpy Generator passed to the policy (make one if None) Returns: (G, rewards): G = discounted_return(rewards, gamma) (a float) and the list of rewards r_0, ..., r_(T-1) received in the episode. """ if rng is None: rng = np.random.default_rng() # TODO: implement (about 8 lines) raise NotImplementedError("Task 4: run_episode") # --------------------------------------------------------------------------- # Task 5: how good is a policy? # --------------------------------------------------------------------------- def estimate_return(env, policy, gamma=1.0, n_episodes=1000, max_steps=100, seed=0): """Monte Carlo estimate of the expected return from the start state. Run n_episodes episodes with run_episode, using ONE numpy Generator made from `seed` for the policy, and average their returns. Returns: (mean, stderr): the sample mean of the returns, and its standard error std(returns, ddof=1) / sqrt(n_episodes). This mean is an estimate of what Chapter 3 will call V^pi(start). """ # TODO: implement (3-4 lines) raise NotImplementedError("Task 5: estimate_return")