mwc/lab_reinforcement_learning
Template
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Updates

Browse Source
This commit is contained in:
Chris Proctor
2026-06-26 20:59:25 -04:00
parent a9385f8296
commit 482f4f6cfa
5 changed files with 191 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

162
games/frogger/__init__.py
View File

@@ -1,19 +1,37 @@
"""Frogger: guide a frog across busy traffic lanes to reach the far side.
The frog starts at the bottom row. Cars move across the lanes between the
start and goal. Each lane has cars moving at different speeds in alternating
directions. The frog earns +10 for each row advanced, +50 for reaching the
top row, and -10 for being hit by a car or falling off the edge. Episodes end
when the frog reaches the top, gets hit, or energy runs out.
The frog starts at the bottom row and must cross to the top. Traffic lanes
alternate direction; safe lanes (no cars) appear every few rows as rest zones.
The frog scores 100 each time it reaches the top and respawns at the bottom.
It earns +1 reward for each upward step and +100 for crossing, but the game
ends immediately if a car hits it.
"""
from random import randint
from retro.game import Game
BOARD_WIDTH = 20
BOARD_HEIGHT = 12
NUM_LANES = BOARD_HEIGHT - 2
START_ENERGY = 200
FRAMERATE = 8
GAME_DURATION = 60 * FRAMERATE # ~60 seconds
# One entry per traffic row (rows 1–10). None marks a safe lane with no cars.
# Each traffic entry is (direction, speed, num_cars, car_length).
# direction: 1 = right, -1 = left
# speed: steps between moves (higher = slower)
# num_cars: how many cars in the lane
# car_length: how many cells each car occupies
LANES = [
( 1, 2, 2, 2), # row 1 fast
(-1, 4, 3, 2), # row 2 slow, crowded
( 1, 1, 1, 3), # row 3 very fast
None, # row 4 safe zone
(-1, 3, 2, 2), # row 5 medium
( 1, 2, 2, 3), # row 6 fast, long cars
(-1, 5, 3, 2), # row 7 slow, crowded
None, # row 8 safe zone
( 1, 1, 2, 2), # row 9 very fast
(-1, 3, 2, 3), # row 10 medium, long cars
]
class Frog:
@@ -28,133 +46,141 @@ class Frog:
RIGHT = (1, 0)
def __init__(self):
self._direction = self.UP
self._pending_direction = None
self._frames_remaining = GAME_DURATION
def handle_keystroke(self, keystroke, game):
if keystroke.name == "KEY_UP":
self._direction = self.UP
self._pending_direction = self.UP
elif keystroke.name == "KEY_DOWN":
self._direction = self.DOWN
self._pending_direction = self.DOWN
elif keystroke.name == "KEY_LEFT":
self._direction = self.LEFT
self._pending_direction = self.LEFT
elif keystroke.name == "KEY_RIGHT":
self._direction = self.RIGHT
self._pending_direction = self.RIGHT
def play_turn(self, game):
self._frames_remaining -= 1
game.state['time_left'] = self._frames_remaining // FRAMERATE
if self._frames_remaining <= 0:
game.end()
return
if self._pending_direction is None:
return
dx, dy = self._pending_direction
self._pending_direction = None
bw, bh = game.board_size
x, y = self.position
dx, dy = self._direction
nx, ny = x + dx, y + dy
if not (0 <= nx < bw):
game.state['reward'] -= 10
game.state['energy'] -= 50
self._reset(game)
if not (0 <= nx < bw) or ny >= bh:
game.state['reward'] -= 5
return
if not (0 <= ny < bh):
if ny < 0:
game.state['score'] += 50
game.state['reward'] += 50
else:
game.state['reward'] -= 5
self._reset(game)
if ny < 0:
game.state['score'] += 100
game.state['reward'] += 100
self._respawn(game)
return
prev_y = y
self.position = (nx, ny)
game.state['energy'] -= 1
game.state['reward'] -= 0.01
if ny < prev_y:
advancement = prev_y - ny
game.state['score'] += advancement * 10
game.state['reward'] += advancement * 5
if ny < y:
game.state['reward'] += 1
for agent in game.agents:
if hasattr(agent, '_is_car') and agent.position == self.position:
game.state['reward'] -= 10
game.state['energy'] -= 50
self._reset(game)
game.end()
return
bw, bh = game.board_size
fx, fy = self.position
game.state['frog_x'] = fx / bw
game.state['frog_y'] = fy / bh
if game.state['energy'] <= 0:
game.end()
def _reset(self, game):
def _respawn(self, game):
bw, bh = game.board_size
self.position = (bw // 2, bh - 1)
self._direction = self.UP
if game.state['energy'] <= 0:
game.end()
self._pending_direction = None
class Car:
class CarSegment:
"""One cell of a multi-cell car. Moved by the lead Car each turn."""
_is_car = True
character = 'X'
color = "red_on_black"
def __init__(self, lane, speed, direction, start_x, board_width):
self.name = f"Car {lane}_{start_x}"
def __init__(self, name, position):
self.name = name
self.position = position
class Car:
"""Lead cell of a multi-cell car. Moves itself and all trailing segments."""
_is_car = True
character = 'X'
color = "red_on_black"
def __init__(self, name, lane, speed, direction, start_x, board_width, segments):
self.name = name
self._lane = lane
self._speed = speed
self._direction = direction
self._board_width = board_width
self._step = 0
self.position = (start_x, lane)
self.segments = segments
def play_turn(self, game):
self._step += 1
if self._step < self._speed:
return
self._step = 0
bw = self._board_width
x, y = self.position
x = (x + self._direction) % self._board_width
self.position = (x, y)
self.position = ((x + self._direction) % bw, y)
for seg in self.segments:
sx, sy = seg.position
seg.position = ((sx + self._direction) % bw, sy)
frog = game.get_agent_by_name("Frog")
if frog.position == self.position:
frog._reset(game)
car_cells = {self.position} | {s.position for s in self.segments}
if frog.position in car_cells:
game.state['reward'] -= 10
game.state['energy'] -= 50
game.end()
def create_game():
bw, bh = BOARD_WIDTH, BOARD_HEIGHT
frog = Frog()
frog.position = (bw // 2, bh - 1)
agents = [frog]
car_id = 0
for lane_idx, row in enumerate(range(1, bh - 1)):
direction = 1 if lane_idx % 2 == 0 else -1
speed = 2 + (lane_idx % 3)
num_cars = 2 + (lane_idx % 3)
spec = LANES[lane_idx] if lane_idx < len(LANES) else None
if spec is None:
continue
direction, speed, num_cars, car_length = spec
spacing = bw // num_cars
for i in range(num_cars):
start_x = (i * spacing + lane_idx * 3) % bw
agents.append(Car(row, speed, direction, start_x, bw))
car_id += 1
segments = []
for j in range(1, car_length):
seg_x = (start_x - direction * j) % bw
seg = CarSegment(f"Car_{row}_{i}_seg{j}", (seg_x, row))
segments.append(seg)
agents.append(seg)
agents.append(Car(f"Car_{row}_{i}", row, speed, direction, start_x, bw, segments))
game = Game(
return Game(
agents,
{
'score': 0,
'reward': 0.0,
'energy': START_ENERGY,
'frog_x': (bw // 2) / bw,
'frog_y': (bh - 1) / bh,
'time_left': GAME_DURATION // FRAMERATE,
},
board_size=(bw, bh),
framerate=8,
show_state=['score', 'energy'],
framerate=FRAMERATE,
show_state=['score', 'time_left'],
)
return game
if __name__ == '__main__':

73
games/snake/observation.py
View File

@@ -2,9 +2,82 @@ import numpy as np
from retro.views.headless import HeadlessView
from retro_gamer.observation import egocentric_board, encode_board, encode_state
HEAD_CHARS = frozenset({">", "<", "^", "v"})
CHARACTER_SET = ["@", "*", ">", "<", "^", "v"]
CHARACTER_SET_NORMALIZED = ["@", "*", "H"]
RADIUS = 8
DIRECTIONS = [(1, 0), (0, -1), (-1, 0), (0, 1)] # RIGHT, UP, LEFT, DOWN
RADIUS_WIDE = 8 # 17×17 egocentric window
RADIUS_NARROW = 3 # 7×7 egocentric window
def direction_observation(game):
"""Normalized 3-char board plus apple_dx, apple_dy, and a one-hot direction.
Returns board in channel-first (C, H, W) order for the CNN, followed by
apple_dx, apple_dy, and 4 one-hot bits encoding the snake's heading.
The board collapses all four head characters to 'H', keeping 3 channels
instead of 6 while the one-hot direction restores that information cheaply.
"""
view = HeadlessView()
view.on_game_start(game)
view.render(game)
normalized = [
["H" if c in HEAD_CHARS else c for c in row]
for row in view.board_characters
]
board_vec = encode_board(normalized, CHARACTER_SET_NORMALIZED).transpose(2, 0, 1).flatten()
extras = encode_state(game.state, ["apple_dx", "apple_dy"])
head = game.get_agent_by_name("Snake head")
direction_onehot = np.array(
[1.0 if head.direction == d else 0.0 for d in DIRECTIONS],
dtype=np.float32,
)
return np.concatenate([board_vec, extras, direction_onehot])
def normalized_observation(game):
"""Full board with all head chars collapsed to 'H', plus apple_dx and apple_dy.
Returns board in channel-first (C, H, W) order for the CNN, followed by extras.
Character set: ['@' apple, '*' body, 'H' head] — 3 channels instead of 6.
"""
view = HeadlessView()
view.on_game_start(game)
view.render(game)
normalized = [
["H" if c in HEAD_CHARS else c for c in row]
for row in view.board_characters
]
board_vec = encode_board(normalized, CHARACTER_SET_NORMALIZED).transpose(2, 0, 1).flatten()
extras = encode_state(game.state, ["apple_dx", "apple_dy"])
return np.concatenate([board_vec, extras])
def egocentric_cnn_observation(game):
"""17×17 egocentric window in channel-first (CHW) format for CNN, plus apple_dx/dy."""
view = HeadlessView()
view.on_game_start(game)
view.render(game)
head = game.get_agent_by_name("Snake head")
cropped = egocentric_board(view.board_characters, head.position, RADIUS_WIDE)
board_vec = encode_board(cropped, CHARACTER_SET).transpose(2, 0, 1).flatten()
extras = encode_state(game.state, ["apple_dx", "apple_dy"])
return np.concatenate([board_vec, extras])
def narrow_egocentric_observation(game):
"""7×7 egocentric window (flat, for MLP), plus apple_dx and apple_dy."""
view = HeadlessView()
view.on_game_start(game)
view.render(game)
head = game.get_agent_by_name("Snake head")
cropped = egocentric_board(view.board_characters, head.position, RADIUS_NARROW)
board_vec = encode_board(cropped, CHARACTER_SET).flatten()
extras = encode_state(game.state, ["apple_dx", "apple_dy"])
return np.concatenate([board_vec, extras])
def egocentric_observation(game):
"""17×17 window centered on the snake's head, plus apple_dx and apple_dy."""

1
games/snake/pyproject.toml
View File

@@ -8,3 +8,4 @@ dependencies = ["retro-games>=2.5.0"]
[tool.retro-gamer]
actions = ["KEY_RIGHT", "KEY_UP", "KEY_LEFT", "KEY_DOWN"]
reward = "reward"
character_set = ["@", "*", "<", ">", "^", "v"]

1
games/snake_v1/pyproject.toml
View File

@@ -8,3 +8,4 @@ dependencies = ["retro-games>=2.5.0"]
[tool.retro-gamer]
actions = ["KEY_RIGHT", "KEY_UP", "KEY_LEFT", "KEY_DOWN"]
reward = "reward"
character_set = ["@", "*", "<", ">", "^", "v"]