diff --git a/.gitignore b/.gitignore
index 3519e695..93cc5fc3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,3 +4,5 @@ Temporary_Plots
Unused_Code
.vscode
exp
+.DS_Store
+*.mp4
\ No newline at end of file
diff --git a/AM_Gyms/AM_Env_wrapper.py b/AM_Gyms/AM_Env_wrapper.py
deleted file mode 100644
index ae094955..00000000
--- a/AM_Gyms/AM_Env_wrapper.py
+++ /dev/null
@@ -1,228 +0,0 @@
-# Wrapper to turn Open AI Gym-environments into active measure environments
-import numpy as np
-import math as m
-import matplotlib.pyplot as plt
-
-
-class AM_ENV:
- """Wrapper class for openAI-environments for AM algorithms.
-
- Most imporantly, changes step-function to not return an observation,
- and adds a seperate observe-function.
- """
-
- def __init__(
- self,
- env,
- StateSize,
- ActionSize,
- MeasureCost,
- s_init,
- log_choices=False,
- max_steps=10_000,
- max_reward=1,
- ):
- self.env = env
- self.StateSize = StateSize
- self.ActionSize = (
- ActionSize # Is there any way to get these two from the env. itself?
- )
- self.MeasureCost = MeasureCost
- self.s_init = s_init
- self.obs = 0
- self.max_steps = max_steps
- self.steps_taken = 0
- self.reward_factor = (
- 1.0 / max_reward
- ) # makes sure rewards are always 'normalised'
-
- self.log_choices = log_choices
- if self.log_choices:
- self.choiceTable = np.zeros((self.StateSize, self.ActionSize))
- self.densityTable = np.zeros(
- (self.StateSize)
- ) # This is also just the sum over actions of choice table, but whatever...
- self.accuracyTable = np.zeros((self.StateSize))
-
- #######################################################
- ### Environment Wrapper code: ###
- #######################################################
- def get_vars(self):
- "Returns StateSize, ActionSize, MeasureCost and s_init (-1 if random)"
- return (self.StateSize, self.ActionSize, self.MeasureCost, self.s_init)
-
- def step(self, action, s=None):
- "Perform action on environment, without returning an observation"
- (obs, reward, done, info) = self.env.step(action)
- self.obs = obs
- reward = reward * self.reward_factor
-
- if done:
- self.obs = 0
-
- # Log action (if turned on):
- if (s == None) & self.log_choices:
- done
- # print("Warning: Logger is turned on, but not all required arguments are given. No logging will be performed.")
- elif self.log_choices:
- self.log_action(action, obs, s)
-
- self.steps_taken += 1
- if self.steps_taken >= self.max_steps:
- done = True
-
- return (reward, done)
-
- def measure(self): # For full version should include m as argument
- "Returns current state of environment"
- return (self.obs, self.MeasureCost)
-
- def reset(self):
- self.env.reset()
- self.steps_taken = 0
-
- def getname(self):
- return self.env.getname()
-
- def horizon(self):
- return None
-
- #######################################################
- ### Logging Code: ###
- #######################################################
-
- # Used for debugging Agents running on Frozen Lake environment.
-
- def log_action(self, action, obs, s):
-
- self.choiceTable[obs, action] += 1
- self.densityTable[obs] += 1
- if obs in s:
- self.accuracyTable[obs] = (
- self.accuracyTable[obs] * (self.densityTable[obs] - 1) + s[obs]
- ) / self.densityTable[obs]
- else:
- self.accuracyTable[obs] = (
- self.accuracyTable[obs]
- * (self.densityTable[obs] - 1)
- / self.densityTable[obs]
- )
-
-
-class AM_Visualiser: # Assuming a grid!
- """Class for visualising results on Frozen Lake environments.
- Used for testing only, would not recommend using!"""
-
- # Winter: Blue = low, green = high
-
- def __init__(self, env_wrapper, agent):
- self.StateSize = agent.StateSize
- self.gridSize = m.ceil(m.sqrt(self.StateSize))
-
- self.QTable = agent.QTable
-
- self.density = env_wrapper.densityTable
- self.accuracy = env_wrapper.accuracyTable
- self.choice = env_wrapper.choiceTable
- return
-
- def __action_to_symbol__(self, action):
- if action == 0:
- return "<"
- elif action == 1:
- return "."
- elif action == 2:
- return ">"
- elif action == 3:
- return "^"
-
- def plot_choice_certainty(self):
-
- # Gather data:
- choice, certainty = np.zeros(self.StateSize, dtype=np.int8), np.zeros(
- self.StateSize
- )
- for i in range(self.StateSize):
- choice[i] = np.argmax(self.QTable[i])
- certainty[i] = self.QTable[i, choice[i]] - np.max(
- np.delete(self.QTable[i], choice[i])
- )
-
- certainty = certainty / np.max(certainty) # normalise
- choice = np.vectorize(self.__action_to_symbol__)(choice)
-
- choice = choice.reshape(self.gridSize, self.gridSize)
- certainty = certainty.reshape(self.gridSize, self.gridSize)
-
- # Create Plot:
- plt.axis([-0.5, self.gridSize - 0.5, -0.5, self.gridSize - 0.5])
- plt.imshow(np.flipud(certainty), cmap="winter")
- for x in range(self.gridSize):
- for y in range(self.gridSize):
- plt.text(y, x, np.flipud(choice)[x, y])
-
- plt.savefig("Test_choice_certainty")
- plt.clf()
-
- def plot_choice_density(self):
- choice = np.zeros(self.StateSize, dtype=np.int8)
- for i in range(self.StateSize):
- choice[i] = np.argmax(self.QTable[i])
-
- choice = np.vectorize(self.__action_to_symbol__)(choice)
-
- choice = choice.reshape(self.gridSize, self.gridSize)
- print(self.density)
- density = self.density / np.max([np.max(self.density), 1])
- density = density.reshape(self.gridSize, self.gridSize)
-
- # Create Plot:
- plt.axis([-0.5, self.gridSize - 0.5, -0.5, self.gridSize - 0.5])
- plt.imshow(np.flipud(density), cmap="winter")
- for x in range(self.gridSize):
- for y in range(self.gridSize):
- plt.text(y, x, np.flipud(choice)[x, y])
-
- plt.savefig("Test_choice_density")
- plt.clf()
-
- def plot_choice_maxQ(self):
- choice, maxQ = np.zeros(self.StateSize, dtype=np.int8), np.zeros(self.StateSize)
- for i in range(self.StateSize):
- choice[i] = np.argmax(self.QTable[i])
-
- choice = np.vectorize(self.__action_to_symbol__)(choice)
- maxQ = np.amax(self.QTable, 1)
-
- choice = choice.reshape(self.gridSize, self.gridSize)
- maxQ = maxQ.reshape(self.gridSize, self.gridSize)
-
- # Create Plot:
- plt.axis([-0.5, self.gridSize - 0.5, -0.5, self.gridSize - 0.5])
- plt.imshow(np.flipud(maxQ), cmap="winter")
- for x in range(self.gridSize):
- for y in range(self.gridSize):
- plt.text(y, x, np.flipud(choice)[x, y])
-
- plt.savefig("Test_choice_maxQ")
- plt.clf()
-
- def plot_choice_state_accuracy(self):
- choice, acc = np.zeros(self.StateSize, dtype=np.int8), np.zeros(self.StateSize)
- choice = np.argmax(self.QTable, 1)
- acc = self.accuracy
-
- choice = np.vectorize(self.__action_to_symbol__)(choice)
-
- choice = choice.reshape(self.gridSize, self.gridSize)
- acc = acc.reshape(self.gridSize, self.gridSize)
-
- # Create Plot:
- plt.axis([-0.5, self.gridSize - 0.5, -0.5, self.gridSize - 0.5])
- plt.imshow(np.flipud(acc), cmap="winter")
- for x in range(self.gridSize):
- for y in range(self.gridSize):
- plt.text(y, x, np.flipud(choice)[x, y])
-
- plt.savefig("Test_choice_accuracy")
- plt.clf()
diff --git a/AM_Gyms/ActiveMeasurementWrapper.py b/AM_Gyms/ActiveMeasurementWrapper.py
new file mode 100644
index 00000000..d96c2f3d
--- /dev/null
+++ b/AM_Gyms/ActiveMeasurementWrapper.py
@@ -0,0 +1,84 @@
+import numpy as np
+import gymnasium as gym
+from gymnasium.spaces import Space
+from typing import Callable
+
+
+# Slightly desaturate an RGB image by blending it with its grayscale version.
+def desaturate_rgb(rgb, alpha=0.5):
+ gray = np.dot(rgb[..., :3], [0.2989, 0.5870, 0.1140])
+ gray_rgb = np.stack((gray, gray, gray), axis=-1)
+ desaturated = (1 - alpha) * rgb + alpha * gray_rgb
+ return np.clip(desaturated, 0, 1 if rgb.dtype.kind == "f" else 255)
+
+
+class ActiveMeasurementWrapper(gym.Wrapper):
+
+ def __init__(
+ self,
+ env: gym.Env,
+ observation_function: Callable[
+ [Space, Space], Space
+ ] = lambda observation, measurement: (observation if measurement else None),
+ measurement_cost: Callable[[Space], int] | int = 0.05,
+ initial_state=-1,
+ ):
+ """Custom Active Measurement Wrapper
+
+ Classic AM:
+ - Provide no observation_function
+ - Let measurement_cost be an integer
+ then it returns the whole observation if measured for cost of of measurement cost
+
+ For more customization:
+ - Let observation_function: observation -> measurement_action -> new observation
+ be a custom observation function dependent on the custom measurement action
+ - Let measurement_cost be dependent on the measurement function
+ """
+ super().__init__(env)
+ self.observation_function = observation_function
+ if type(measurement_cost) is float:
+ self.measurement_cost = lambda measurement_action: (
+ measurement_cost if measurement_action else 0
+ )
+ else:
+ self.measurement_cost = measurement_cost
+ self.initial_state = initial_state
+ self.last_step_measured = False
+
+ def reset(self, seed=None, options=None):
+ self.env.reset(seed=seed, options=options)
+ self.last_step_measured = False
+ # do not return observation here
+ return None, None
+
+ def step(self, action):
+ control_action, measurement_action = action
+ self.last_step_measured = measurement_action
+ observation, reward, terminated, truncated, info = self.env.step(control_action)
+ return (
+ self.observation_function(observation, measurement_action),
+ reward - self.measurement_cost(measurement_action),
+ terminated,
+ truncated,
+ info,
+ )
+
+ def render(self):
+ if self.env.render_mode == "rgb_array":
+ img = self.env.render()
+ if not self.last_step_measured:
+ return desaturate_rgb(img, 0.65)
+ else:
+ return img
+ elif self.env.render_mode in ["ansi", "text"]:
+ return f"Measure action {self.last_step_measured}:\n" + self.env.render()
+ self.env.render()
+
+ def get_vars(self):
+ return (
+ self.env.observation_space,
+ self.env.action_space,
+ self.measurement_cost,
+ self.initial_state,
+ )
diff --git a/AM_Gyms/Blackjack.py b/AM_Gyms/Blackjack.py
index f9002271..2c2b3bec 100644
--- a/AM_Gyms/Blackjack.py
+++ b/AM_Gyms/Blackjack.py
@@ -1,180 +1,21 @@
-# Open-ai standard gym, but with non-factorised state space.
+import gymnasium as gym
+from gymnasium import spaces
-import os
-import re
-from typing import Optional
-import numpy as np
-
-import gym
-from gym import spaces
-from gym.error import DependencyNotInstalled
-
-
-def cmp(a, b):
- return float(a > b) - float(a < b)
-
-
-# 1 = Ace, 2-10 = Number cards, Jack/Queen/King = 10
-deck = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 10, 10, 10]
-
-
-def draw_card(np_random):
- return int(np_random.choice(deck))
-
-
-def draw_hand(np_random):
- return [draw_card(np_random), draw_card(np_random)]
-
-
-def usable_ace(hand): # Does this hand have a usable ace?
- return 1 in hand and sum(hand) + 10 <= 21
-
-
-def sum_hand(hand): # Return current hand total
- if usable_ace(hand):
- return sum(hand) + 10
- return sum(hand)
-
-
-def is_bust(hand): # Is this hand a bust?
- return sum_hand(hand) > 21
-
-
-def score(hand): # What is the score of this hand (0 if bust)
- return 0 if is_bust(hand) else sum_hand(hand)
-
-
-def is_natural(hand): # Is this hand a natural blackjack?
- return sorted(hand) == [1, 10]
-
-
-class BlackjackEnv(gym.Env):
- """
- Blackjack is a card game where the goal is to beat the dealer by obtaining cards
- that sum to closer to 21 (without going over 21) than the dealers cards.
- ### Description
- Card Values:
- - Face cards (Jack, Queen, King) have a point value of 10.
- - Aces can either count as 11 (called a 'usable ace') or 1.
- - Numerical cards (2-9) have a value equal to their number.
- This game is played with an infinite deck (or with replacement).
- The game starts with the dealer having one face up and one face down card,
- while the player has two face up cards.
- The player can request additional cards (hit, action=1) until they decide to stop (stick, action=0)
- or exceed 21 (bust, immediate loss).
- After the player sticks, the dealer reveals their facedown card, and draws
- until their sum is 17 or greater. If the dealer goes bust, the player wins.
- If neither the player nor the dealer busts, the outcome (win, lose, draw) is
- decided by whose sum is closer to 21.
- ### Action Space
- There are two actions: stick (0), and hit (1).
- ### Observation Space
- The observation consists of a 3-tuple containing: the player's current sum,
- the value of the dealer's one showing card (1-10 where 1 is ace),
- and whether the player holds a usable ace (0 or 1).
- This environment corresponds to the version of the blackjack problem
- described in Example 5.1 in Reinforcement Learning: An Introduction
- by Sutton and Barto (http://incompleteideas.net/book/the-book-2nd.html).
- ### Rewards
- - win game: +1
- - lose game: -1
- - draw game: 0
- - win game with natural blackjack:
- +1.5 (if natural is True)
- +1 (if natural is False)
- ### Arguments
- ```
- gym.make('Blackjack-v1', natural=False, sab=False)
- ```
- `natural=False`: Whether to give an additional reward for
- starting with a natural blackjack, i.e. starting with an ace and ten (sum is 21).
- `sab=False`: Whether to follow the exact rules outlined in the book by
- Sutton and Barto. If `sab` is `True`, the keyword argument `natural` will be ignored.
- If the player achieves a natural blackjack and the dealer does not, the player
- will win (i.e. get a reward of +1). The reverse rule does not apply.
- If both the player and the dealer get a natural, it will be a draw (i.e. reward 0).
- ### Version History
- * v0: Initial versions release (1.0.0)
- """
-
- metadata = {
- "render_modes": ["human", "rgb_array", "single_rgb_array"],
- "render_fps": 4,
- }
-
- def __init__(self, render_mode: Optional[str] = None, natural=False, sab=False):
- self.action_space = spaces.Discrete(2)
- self.observation_space = spaces.Tuple(
- (spaces.Discrete(32), spaces.Discrete(11), spaces.Discrete(2))
+# extension upon gymnasium blackjack that converts observations to integers
+class BlackjackEnv(gym.ObservationWrapper):
+ def __init__(self, **kwargs):
+ self.env = gym.make(
+ "Blackjack-v1",
+ **kwargs,
)
+ super().__init__(self.env)
+ # 11 possible dealer hand * 32 possible player hand * 2 for usable ace = 704
+ self.observation_space = spaces.Discrete(704)
+ self.action_space = spaces.Discrete(2)
- # Flag to payout 1.5 on a "natural" blackjack win, like casino rules
- # Ref: http://www.bicyclecards.com/how-to-play/blackjack/
- self.natural = natural
-
- # Flag for full agreement with the (Sutton and Barto, 2018) definition. Overrides self.natural
- self.sab = sab
-
- def step(self, action):
- assert self.action_space.contains(action)
- if action: # hit: add a card to players hand and return
- self.player.append(draw_card(self.np_random))
- if is_bust(self.player):
- terminated = True
- reward = -1.0
- else:
- terminated = False
- reward = 0.0
- else: # stick: play out the dealers hand, and score
- terminated = True
- while sum_hand(self.dealer) < 17:
- self.dealer.append(draw_card(self.np_random))
- reward = cmp(score(self.player), score(self.dealer))
- if self.sab and is_natural(self.player) and not is_natural(self.dealer):
- # Player automatically wins. Rules consistent with S&B
- reward = 1.0
- elif (
- not self.sab
- and self.natural
- and is_natural(self.player)
- and reward == 1.0
- ):
- # Natural gives extra points, but doesn't autowin. Legacy implementation
- reward = 1.5
- return self._get_obs(), reward, terminated, False
-
- def _get_obs(self):
- return sum_hand(self.player) * 4 + self.dealer[0] * 2 + usable_ace(self.player)
-
- return (sum_hand(self.player), self.dealer[0], usable_ace(self.player))
-
- def reset(
- self,
- seed: Optional[int] = None,
- return_info: bool = False,
- options: Optional[dict] = None,
- ):
- super().reset(seed=seed)
- self.dealer = draw_hand(self.np_random)
- self.player = draw_hand(self.np_random)
-
- dealer_card_value = self.dealer[0]
-
- suits = ["C", "D", "H", "S"]
- self.dealer_top_card_suit = self.np_random.choice(suits)
-
- if dealer_card_value == 1:
- self.dealer_top_card_value_str = "A"
- elif dealer_card_value == 10:
- self.dealer_top_card_value_str = self.np_random.choice(["J", "Q", "K"])
- else:
- self.dealer_top_card_value_str = str(dealer_card_value)
-
- if not return_info:
- return self._get_obs()
- else:
- return self._get_obs(), {}
-
- def getname(self):
- return "Blackjack"
+ def observation(self, obs):
+ player_hand, dealer_hand, usable_ace = obs
+ # we have 10 bits, respectively: 4 bits dealer hand - 5 bits player hand - 1 bit usable acce
+ # we put dealer hand at beginning because that does not use all bits
+ return player_hand * 2 + dealer_hand * 64 + usable_ace
diff --git a/AM_Gyms/ModelLearner.py b/AM_Gyms/ModelLearner.py
index 51b1ca04..5a7a97e1 100644
--- a/AM_Gyms/ModelLearner.py
+++ b/AM_Gyms/ModelLearner.py
@@ -1,6 +1,5 @@
import numpy as np
-
-from AM_Gyms.AM_Env_wrapper import AM_ENV
+from gymnasium import Env
def build_dictionary(statesize, actionsize, array: np.ndarray = None):
@@ -17,7 +16,7 @@ def build_dictionary(statesize, actionsize, array: np.ndarray = None):
class ModelLearner:
"""Class for learning ACNO-MDP"""
- def __init__(self, env: AM_ENV, df=0.90):
+ def __init__(self, env: Env, df=0.90):
# Set up AM-environment
self.env = env
diff --git a/AM_Gyms/PartialMeasurementLakeTest.py b/AM_Gyms/PartialMeasurementLakeTest.py
new file mode 100644
index 00000000..ee09172d
--- /dev/null
+++ b/AM_Gyms/PartialMeasurementLakeTest.py
@@ -0,0 +1,43 @@
+from frozen_lake_v2 import FrozenLakeEnv_v2
+from ActiveMeasurementWrapper import ActiveMeasurementWrapper
+from TextEpisodeRecorder import TextEpisodeRecorder
+
+""""
+Test file for testing partial measurements. In this instance for Lake to only observe x or y
+"""
+
+LEFT = 0
+DOWN = 1
+RIGHT = 2
+UP = 3
+
+
+def obs_function(observation, measurement):
+ x = observation % 4
+ y = observation // 4
+ observe_x, observe_y = measurement
+ return (x if observe_x else None, y if observe_y else None)
+
+
+def measurement_cost(measurement):
+ observe_x, observe_y = measurement
+ cost = 0
+ if observe_x:
+ cost += 0.05
+ if observe_y:
+ cost += 0.05
+ return cost
+
+
+env = FrozenLakeEnv_v2(render_mode="ansi", map_name="4x4")
+env = ActiveMeasurementWrapper(
+ env, observation_function=obs_function, measurement_cost=measurement_cost
+)
+env = TextEpisodeRecorder(env, folder="./episodes")
+for i in range(50):
+ termination, truncation = False, False
+ _ = env.reset(seed=123)
+ while not (termination or truncation):
+ obs, rew, termination, truncation, info = env.step(
+ (env.action_space.sample(), (True, True))
+ )
diff --git a/AM_Gyms/TextEpisodeRecorder.py b/AM_Gyms/TextEpisodeRecorder.py
new file mode 100644
index 00000000..f8986bf9
--- /dev/null
+++ b/AM_Gyms/TextEpisodeRecorder.py
@@ -0,0 +1,70 @@
+import os
+import gymnasium as gym
+from gymnasium.utils.save_video import capped_cubic_video_schedule
+
+
+class TextEpisodeRecorder(gym.Wrapper):
+
+ def __init__(
+ self,
+ env,
+ folder: str = "./",
+ episode_trigger=capped_cubic_video_schedule,
+ name_prefix: str = "training",
+ ):
+ super().__init__(env)
+
+ if env.render_mode not in {"text", "ansi"}:
+ raise ValueError(
+ f"Render mode is {env.render_mode}, which is incompatible with TextEpisodeRecorder, should be text or ansi.",
+ )
+
+ self.folder = folder
+ self.episode_trigger = episode_trigger
+ self.name_prefix = name_prefix
+
+ self.recording = False
+ self.episode = -1
+ self.file = None
+
+ def reset(self, seed=None, options=None):
+ obs, info = self.env.reset(seed=seed, options=options)
+ self.episode += 1
+
+ if self.recording:
+ self.stop_recording()
+ if self.episode_trigger(self.episode):
+ self.start_recording()
+
+ return obs, info
+
+ def step(self, action):
+ obs, rew, terminated, truncated, info = self.env.step(action)
+ if self.recording:
+ self.record_frame()
+ return obs, rew, terminated, truncated, info
+
+ def record_frame(self):
+ text = self.env.render()
+ self.file.write(text)
+
+ def stop_recording(self):
+ self.file.close()
+ self.recording = False
+
+ def start_recording(self):
+ full_file_name = (
+ self.folder + "/" + self.name_prefix + str(self.episode) + ".txt"
+ )
+ # Ensure the folder exists
+ os.makedirs(self.folder, exist_ok=True)
+
+ # initialize file, overwrite if it exists
+ self.file = open(full_file_name, "w")
+ self.file.close()
+
+ # now open it in append mode so we can append at every action
+ self.file = open(full_file_name, "a")
+
+ self.recording = True
+ self.record_frame()
diff --git a/AM_Gyms/fire_escape.py b/AM_Gyms/fire_escape.py
new file mode 100644
index 00000000..ee02dba6
--- /dev/null
+++ b/AM_Gyms/fire_escape.py
@@ -0,0 +1,137 @@
+import gymnasium as gym
+from gymnasium import spaces
+from gymnasium.utils import seeding
+import numpy as np
+
+LEFT = 0
+DOWN = 1
+RIGHT = 2
+UP = 3
+MEASURE = 4
+
+
+class FireEscape(gym.Env):
+ """ """
+
+ def __init__(
+ self,
+ size=5,
+ fires=3,
+ measure_cost=0.1,
+ render_mode=None,
+ ):
+ self.size = size
+ self.fires = fires
+ self.measure_cost = measure_cost
+ self.render_mode = render_mode
+
+ self.player = (0, 0)
+ self.generate_random_fires()
+
+ # 4 move actions, one measure action to detect fire
+ self.action_space = spaces.Discrete(5)
+ # observation space is state and whether there is smoke
+ # or, state + locations of fire
+ self.observation_space = spaces.OneOf(
+ (
+ spaces.Tuple(
+ (spaces.Discrete(self.size * self.size), spaces.Discrete(2))
+ ),
+ spaces.Tuple(
+ (spaces.Discrete(self.size * self.size), spaces.MultiBinary(4))
+ ),
+ ),
+ )
+
+ def int_to_space(self, n: int):
+ return (n // self.size, n % self.size)
+
+ def space_to_int(self, space):
+ x, y = space
+ return x * self.size + y
+
+ def generate_random_fires(self):
+ self.fire_locations = np.full((self.size, self.size), False)
+ for i in range(self.fires):
+ # random fire that is not in initial or final position
+ # we disregard the possibility that two fires occur in the same place
+ # todo: check if the fires do not block all paths
+ n = self.np_random.integers(1, self.size * self.size - 2)
+ x, y = self.int_to_space(n)
+ self.fire_locations[x][y] = True
+
+ def seed(self, seed=None):
+ super().reset(seed=seed)
+ self.np_random, seed = seeding.np_random(seed)
+ return [seed]
+
+ def step(self, action):
+ x, y = self.player
+ if action == LEFT:
+ if x > 0:
+ x -= 1
+ elif action == DOWN:
+ if y > 0:
+ y -= 1
+ elif action == RIGHT:
+ if x < self.size - 1:
+ x += 1
+ elif action == UP:
+ if y < self.size - 1:
+ y += 1
+ elif action == MEASURE:
+ left = down = right = up = False
+ if (x > 0) and self.fire_locations[x - 1][y]:
+ left = True
+ if (y > 0) and self.fire_locations[x][y - 1]:
+ down = True
+ if (x < self.size - 1) and self.fire_locations[x + 1][y]:
+ right = True
+ if (y < self.size - 1) and self.fire_locations[x][y + 1]:
+ up = True
+ return (self.player, (left, down, right, up)), -self.measure_cost, False, {}
+
+ self.player = (x, y)
+
+ if self.render_mode is not None:
+ self.render()
+
+ # detect smoke (i.e. fire in adjacent cell)
+ smoke = False
+ if (
+ ((x > 0) and self.fire_locations[x - 1][y])
+ or ((x < self.size - 1) and self.fire_locations[x + 1][y])
+ or (y > 0 and self.fire_locations[x][y - 1])
+ or (y < self.size - 1 and self.fire_locations[x][y + 1])
+ ):
+ smoke = True
+
+ if self.fire_locations[x][y]:
+ # player is in fire, episode is over, reward = 0
+ return (self.player, smoke), 0, True, {}
+ if self.player == (self.size - 1, self.size - 1):
+ # player is at the end and has won, reward = 1
+ return (self.player, smoke), 1, True, {}
+
+ # game not over, regular observation
+ return (self.player, smoke), 0, False, {}
+
+ def render(self):
+ if self.render_mode == "ansi":
+ for y in range(self.size - 1, -1, -1):
+ for x in range(self.size):
+ if (x, y) == self.player:
+ print("+", end="")
+ elif self.fire_locations[x][y]:
+ print("x", end="")
+ else:
+ print("_", end="")
+ print("")
+
+ def reset(self, seed=None):
+ super().reset(seed=seed)
+
+ self.player = (0, 0)
+ self.generate_random_fires()
+
+ return self.to_s(self.components)
diff --git a/AM_Gyms/frozen_lake.py b/AM_Gyms/frozen_lake.py
deleted file mode 100644
index 400ef744..00000000
--- a/AM_Gyms/frozen_lake.py
+++ /dev/null
@@ -1,395 +0,0 @@
-from contextlib import closing
-from io import StringIO
-from os import path
-from typing import List, Optional
-
-import numpy as np
-
-from gym import Env, spaces, utils
-from gym.envs.toy_text.utils import categorical_sample
-from gym.error import DependencyNotInstalled
-
-LEFT = 0
-DOWN = 1
-RIGHT = 2
-UP = 3
-
-MAPS = {
- "4x4": ["SFFF", "FHFH", "FFFH", "HFFG"],
- "8x8": [
- "SFFFFFFF",
- "FFFFFFFF",
- "FFFHFFFF",
- "FFFFFHFF",
- "FFFHFFFF",
- "FHHFFFHF",
- "FHFFHFHF",
- "FFFHFFFG",
- ],
-}
-
-
-# DFS to check that it's a valid path.
-def is_valid(board: List[List[str]], max_size: int) -> bool:
- frontier, discovered = [], set()
- frontier.append((0, 0))
- while frontier:
- r, c = frontier.pop()
- if not (r, c) in discovered:
- discovered.add((r, c))
- directions = [(1, 0), (0, 1), (-1, 0), (0, -1)]
- for x, y in directions:
- r_new = r + x
- c_new = c + y
- if r_new < 0 or r_new >= max_size or c_new < 0 or c_new >= max_size:
- continue
- if board[r_new][c_new] == "G":
- return True
- if board[r_new][c_new] != "H":
- frontier.append((r_new, c_new))
- return False
-
-
-def generate_random_map(size: int = 8, p: float = 0.8) -> List[str]:
- """Generates a random valid map (one that has a path from start to goal)
- Args:
- size: size of each side of the grid
- p: probability that a tile is frozen
- Returns:
- A random valid map
- """
- valid = False
- board = [] # initialize to make pyright happy
-
- while not valid:
- p = min(1, p)
- board = np.random.choice(["F", "H"], (size, size), p=[p, 1 - p])
- board[0][0] = "S"
- board[-1][-1] = "G"
- valid = is_valid(board, size)
- return ["".join(x) for x in board]
-
-
-class FrozenLakeEnv(Env):
- """
- Frozen lake involves crossing a frozen lake from Start(S) to Goal(G) without falling into any Holes(H)
- by walking over the Frozen(F) lake.
- The agent may not always move in the intended direction due to the slippery nature of the frozen lake.
- ### Action Space
- The agent takes a 1-element vector for actions.
- The action space is `(dir)`, where `dir` decides direction to move in which can be:
- - 0: LEFT
- - 1: DOWN
- - 2: RIGHT
- - 3: UP
- ### Observation Space
- The observation is a value representing the agent's current position as
- current_row * nrows + current_col (where both the row and col start at 0).
- For example, the goal position in the 4x4 map can be calculated as follows: 3 * 4 + 3 = 15.
- The number of possible observations is dependent on the size of the map.
- For example, the 4x4 map has 16 possible observations.
- ### Rewards
- Reward schedule:
- - Reach goal(G): +1
- - Reach hole(H): 0
- - Reach frozen(F): 0
- ### Arguments
- ```
- gym.make('FrozenLake-v1', desc=None, map_name="4x4", is_slippery=True)
- ```
- `desc`: Used to specify custom map for frozen lake. For example,
- desc=["SFFF", "FHFH", "FFFH", "HFFG"].
- A random generated map can be specified by calling the function `generate_random_map`. For example,
- ```
- from gym.envs.toy_text.frozen_lake import generate_random_map
- gym.make('FrozenLake-v1', desc=generate_random_map(size=8))
- ```
- `map_name`: ID to use any of the preloaded maps.
- "4x4":[
- "SFFF",
- "FHFH",
- "FFFH",
- "HFFG"
- ]
- "8x8": [
- "SFFFFFFF",
- "FFFFFFFF",
- "FFFHFFFF",
- "FFFFFHFF",
- "FFFHFFFF",
- "FHHFFFHF",
- "FHFFHFHF",
- "FFFHFFFG",
- ]
- `is_slippery`: True/False. If True will move in intended direction with
- probability of 1/3 else will move in either perpendicular direction with
- equal probability of 1/3 in both directions.
- For example, if action is left and is_slippery is True, then:
- - P(move left)=1/3
- - P(move up)=1/3
- - P(move down)=1/3
- ### Version History
- * v1: Bug fixes to rewards
- * v0: Initial versions release (1.0.0)
- """
-
- metadata = {
- "render_modes": ["human", "ansi", "rgb_array"],
- "render_fps": 4,
- }
-
- def __init__(
- self,
- render_mode: Optional[str] = None,
- desc=None,
- map_name="4x4",
- is_slippery=True,
- ):
- if desc is None and map_name is None:
- desc = generate_random_map()
- elif desc is None:
- desc = MAPS[map_name]
- self.desc = desc = np.asarray(desc, dtype="c")
- self.nrow, self.ncol = nrow, ncol = desc.shape
- self.reward_range = (0, 1)
- self.is_slippery = is_slippery
-
- nA = 4
- nS = nrow * ncol
-
- self.initial_state_distrib = np.array(desc == b"S").astype("float64").ravel()
- self.initial_state_distrib /= self.initial_state_distrib.sum()
-
- self.P = {s: {a: [] for a in range(nA)} for s in range(nS)}
-
- def to_s(row, col):
- return row * ncol + col
-
- def inc(row, col, a):
- if a == LEFT:
- col = max(col - 1, 0)
- elif a == DOWN:
- row = min(row + 1, nrow - 1)
- elif a == RIGHT:
- col = min(col + 1, ncol - 1)
- elif a == UP:
- row = max(row - 1, 0)
- return (row, col)
-
- def update_probability_matrix(row, col, action):
- newrow, newcol = inc(row, col, action)
- newstate = to_s(newrow, newcol)
- newletter = desc[newrow, newcol]
- terminated = bytes(newletter) in b"GH"
- reward = float(newletter == b"G")
- return newstate, reward, terminated
-
- for row in range(nrow):
- for col in range(ncol):
- s = to_s(row, col)
- for a in range(4):
- li = self.P[s][a]
- letter = desc[row, col]
- if letter in b"GH":
- li.append((1.0, s, 0, True))
- else:
- if is_slippery:
- for b in [(a - 1) % 4, a, (a + 1) % 4]:
- li.append(
- (1.0 / 3.0, *update_probability_matrix(row, col, b))
- )
- else:
- li.append((1.0, *update_probability_matrix(row, col, a)))
-
- self.observation_space = spaces.Discrete(nS)
- self.action_space = spaces.Discrete(nA)
-
- self.render_mode = render_mode
-
- # pygame utils
- self.window_size = (min(64 * ncol, 512), min(64 * nrow, 512))
- self.cell_size = (
- self.window_size[0] // self.ncol,
- self.window_size[1] // self.nrow,
- )
- self.window_surface = None
- self.clock = None
- self.hole_img = None
- self.cracked_hole_img = None
- self.ice_img = None
- self.elf_images = None
- self.goal_img = None
- self.start_img = None
-
- def step(self, a):
- transitions = self.P[self.s][a]
- i = categorical_sample([t[0] for t in transitions], self.np_random)
- p, s, r, t = transitions[i]
- self.s = s
- self.lastaction = a
-
- if self.render_mode == "human":
- self.render()
- return (int(s), r, t, (False, {"prob": p}))
-
- def reset(
- self,
- *,
- seed: Optional[int] = None,
- options: Optional[dict] = None,
- ):
- super().reset(seed=seed)
- self.s = categorical_sample(self.initial_state_distrib, self.np_random)
- self.lastaction = None
-
- if self.render_mode == "human":
- self.render()
- return int(self.s), {"prob": 1}
-
- def render(self):
- if self.render_mode == "ansi":
- return self._render_text()
- else: # self.render_mode in {"human", "rgb_array"}:
- return self._render_gui(self.render_mode)
-
- def _render_gui(self, mode):
- try:
- import pygame
- except ImportError:
- raise DependencyNotInstalled(
- "pygame is not installed, run `pip install gym[toy_text]`"
- )
-
- if self.window_surface is None:
- pygame.init()
-
- if mode == "human":
- pygame.display.init()
- pygame.display.set_caption("Frozen Lake")
- self.window_surface = pygame.display.set_mode(self.window_size)
- elif mode == "rgb_array":
- self.window_surface = pygame.Surface(self.window_size)
-
- assert (
- self.window_surface is not None
- ), "Something went wrong with pygame. This should never happen."
-
- if self.clock is None:
- self.clock = pygame.time.Clock()
- if self.hole_img is None:
- file_name = path.join(path.dirname(__file__), "img/hole.png")
- self.hole_img = pygame.transform.scale(
- pygame.image.load(file_name), self.cell_size
- )
- if self.cracked_hole_img is None:
- file_name = path.join(path.dirname(__file__), "img/cracked_hole.png")
- self.cracked_hole_img = pygame.transform.scale(
- pygame.image.load(file_name), self.cell_size
- )
- if self.ice_img is None:
- file_name = path.join(path.dirname(__file__), "img/ice.png")
- self.ice_img = pygame.transform.scale(
- pygame.image.load(file_name), self.cell_size
- )
- if self.goal_img is None:
- file_name = path.join(path.dirname(__file__), "img/goal.png")
- self.goal_img = pygame.transform.scale(
- pygame.image.load(file_name), self.cell_size
- )
- if self.start_img is None:
- file_name = path.join(path.dirname(__file__), "img/stool.png")
- self.start_img = pygame.transform.scale(
- pygame.image.load(file_name), self.cell_size
- )
- if self.elf_images is None:
- elfs = [
- path.join(path.dirname(__file__), "img/elf_left.png"),
- path.join(path.dirname(__file__), "img/elf_down.png"),
- path.join(path.dirname(__file__), "img/elf_right.png"),
- path.join(path.dirname(__file__), "img/elf_up.png"),
- ]
- self.elf_images = [
- pygame.transform.scale(pygame.image.load(f_name), self.cell_size)
- for f_name in elfs
- ]
-
- desc = self.desc.tolist()
- assert isinstance(desc, list), f"desc should be a list or an array, got {desc}"
- for y in range(self.nrow):
- for x in range(self.ncol):
- pos = (x * self.cell_size[0], y * self.cell_size[1])
- rect = (*pos, *self.cell_size)
-
- self.window_surface.blit(self.ice_img, pos)
- if desc[y][x] == b"H":
- self.window_surface.blit(self.hole_img, pos)
- elif desc[y][x] == b"G":
- self.window_surface.blit(self.goal_img, pos)
- elif desc[y][x] == b"S":
- self.window_surface.blit(self.start_img, pos)
-
- pygame.draw.rect(self.window_surface, (180, 200, 230), rect, 1)
-
- # paint the elf
- bot_row, bot_col = self.s // self.ncol, self.s % self.ncol
- cell_rect = (bot_col * self.cell_size[0], bot_row * self.cell_size[1])
- last_action = self.lastaction if self.lastaction is not None else 1
- elf_img = self.elf_images[last_action]
-
- if desc[bot_row][bot_col] == b"H":
- self.window_surface.blit(self.cracked_hole_img, cell_rect)
- else:
- self.window_surface.blit(elf_img, cell_rect)
-
- if mode == "human":
- pygame.event.pump()
- pygame.display.update()
- self.clock.tick(self.metadata["render_fps"])
- elif mode == "rgb_array":
- return np.transpose(
- np.array(pygame.surfarray.pixels3d(self.window_surface)), axes=(1, 0, 2)
- )
-
- @staticmethod
- def _center_small_rect(big_rect, small_dims):
- offset_w = (big_rect[2] - small_dims[0]) / 2
- offset_h = (big_rect[3] - small_dims[1]) / 2
- return (
- big_rect[0] + offset_w,
- big_rect[1] + offset_h,
- )
-
- def _render_text(self):
- desc = self.desc.tolist()
- outfile = StringIO()
-
- row, col = self.s // self.ncol, self.s % self.ncol
- desc = [[c.decode("utf-8") for c in line] for line in desc]
- desc[row][col] = utils.colorize(desc[row][col], "red", highlight=True)
- if self.lastaction is not None:
- outfile.write(f" ({['Left', 'Down', 'Right', 'Up'][self.lastaction]})\n")
- else:
- outfile.write("\n")
- outfile.write("\n".join("".join(line) for line in desc) + "\n")
-
- with closing(outfile):
- return outfile.getvalue()
-
- def close(self):
- if self.window_surface is not None:
- import pygame
-
- pygame.display.quit()
- pygame.quit()
-
- def getname(self):
- if self.is_slippery:
- variant_name = "slippery"
- else:
- variant_name = "det"
-
- return "Frozen_{}_{}".format(self.nrow, variant_name)
-
-
-# Elf and stool from https://franuka.itch.io/rpg-snow-tileset
-# All other assets by Mel Tillery http://www.cyaneus.com/
diff --git a/AM_Gyms/frozen_lake_v2.py b/AM_Gyms/frozen_lake_v2.py
index a5aa8c6b..dafc2ddc 100644
--- a/AM_Gyms/frozen_lake_v2.py
+++ b/AM_Gyms/frozen_lake_v2.py
@@ -5,9 +5,12 @@
import numpy as np
-from gym import Env, spaces, utils
-from gym.envs.toy_text.utils import categorical_sample
-from gym.error import DependencyNotInstalled
+import gymnasium as gym
+from gymnasium import Env, spaces, utils
+from gymnasium.envs.toy_text.utils import categorical_sample
+from gymnasium.error import DependencyNotInstalled
+from gymnasium.utils import seeding
+
LEFT = 0
DOWN = 1
@@ -29,6 +32,7 @@
}
+# DFS to check that it's a valid path.
def is_valid(board: List[List[str]], max_size: int) -> bool:
frontier, discovered = [], set()
frontier.append((0, 0))
@@ -49,20 +53,27 @@ def is_valid(board: List[List[str]], max_size: int) -> bool:
return False
-def generate_random_map(size: int = 8, p: float = 0.8) -> List[str]:
+def generate_random_map(
+ size: int = 8, p: float = 0.8, seed: Optional[int] = None
+) -> List[str]:
"""Generates a random valid map (one that has a path from start to goal)
+
Args:
size: size of each side of the grid
p: probability that a tile is frozen
+ seed: optional seed to ensure the generation of reproducible maps
+
Returns:
A random valid map
"""
valid = False
board = [] # initialize to make pyright happy
+ np_random, _ = seeding.np_random(seed)
+
while not valid:
p = min(1, p)
- board = np.random.choice(["F", "H"], (size, size), p=[p, 1 - p])
+ board = np_random.choice(["F", "H"], (size, size), p=[p, 1 - p])
board[0][0] = "S"
board[-1][-1] = "G"
valid = is_valid(board, size)
@@ -72,19 +83,17 @@ def generate_random_map(size: int = 8, p: float = 0.8) -> List[str]:
class FrozenLakeEnv_v2(Env):
"""
This is a variant on the Frozen Lake environment from OpenAI.
- A complete description on the original evironment can be found at https://www.gymlibrary.ml/environments/toy_text/frozen_lake/
+ A complete description on the original evironment can be found at https://gymnasium.farama.org/environments/toy_text/frozen_lake/
In this variant, behavious of 'slippery' environments is slighly altered:
Instead of the 3 possibilities in the original, a step in some direction now has
a 1/2 chance of going to that spot, and a 1/2 chance to taking 2 steps in that direction.
In case of the latter, if the space that gets 'skipped' is a hole the run terminates as though
the current state is a hole.
Also, if going 2 spaces would result in going outside the playingfield, the chance of going forward one space becomes 1.
-
- (Also, some options and rendering functions in the original have been removed from this version.)
"""
metadata = {
- "render_modes": ["human", "ansi", "rgb_array", "single_rgb_array"],
+ "render_modes": ["human", "ansi", "rgb_array"],
"render_fps": 4,
}
@@ -95,12 +104,13 @@ def __init__(
map_name="4x4",
is_slippery=True,
):
- if desc == None:
+ if desc is None and map_name is None:
+ desc = generate_random_map()
+ elif desc is None:
desc = MAPS[map_name]
self.desc = desc = np.asarray(desc, dtype="c")
self.nrow, self.ncol = nrow, ncol = desc.shape
self.reward_range = (0, 1)
- self.is_slippery = is_slippery
nA = 4
nS = nrow * ncol
@@ -125,12 +135,12 @@ def inc(row, col, a):
return (row, col)
def update_probability_matrix(row, col, action):
- newrow, newcol = inc(row, col, action)
- newstate = to_s(newrow, newcol)
- newletter = desc[newrow, newcol]
- terminated = bytes(newletter) in b"GH"
- reward = float(newletter == b"G")
- return newstate, reward, terminated
+ new_row, new_col = inc(row, col, action)
+ new_state = to_s(new_row, new_col)
+ new_letter = desc[new_row, new_col]
+ terminated = bytes(new_letter) in b"GH"
+ reward = float(new_letter == b"G")
+ return new_state, reward, terminated
for row in range(nrow):
for col in range(ncol):
@@ -168,6 +178,8 @@ def update_probability_matrix(row, col, action):
self.observation_space = spaces.Discrete(nS)
self.action_space = spaces.Discrete(nA)
+ self.render_mode = render_mode
+
# pygame utils
self.window_size = (min(64 * ncol, 512), min(64 * nrow, 512))
self.cell_size = (
@@ -189,31 +201,171 @@ def step(self, a):
p, s, r, t = transitions[i]
self.s = s
self.lastaction = a
- return (int(s), r, t, (False, {"prob": p}))
+
+ if self.render_mode == "human":
+ self.render()
+ # truncation=False as the time limit is handled by the `TimeLimit` wrapper added during `make`
+ return int(s), r, t, False, {"prob": p}
def reset(
self,
*,
seed: Optional[int] = None,
- return_info: bool = False,
options: Optional[dict] = None,
):
super().reset(seed=seed)
self.s = categorical_sample(self.initial_state_distrib, self.np_random)
self.lastaction = None
- if not return_info:
- return int(self.s)
+ if self.render_mode == "human":
+ self.render()
+ return int(self.s), {"prob": 1}
+
+ def render(self):
+ if self.render_mode is None:
+ assert self.spec is not None
+ gym.logger.warn(
+ "You are calling render method without specifying any render mode. "
+ "You can specify the render_mode at initialization, "
+ f'e.g. gym.make("{self.spec.id}", render_mode="rgb_array")'
+ )
+ return
+
+ if self.render_mode == "ansi":
+ return self._render_text()
+ else: # self.render_mode in {"human", "rgb_array"}:
+ return self._render_gui(self.render_mode)
+
+ def _render_gui(self, mode):
+ try:
+ import pygame
+ except ImportError as e:
+ raise DependencyNotInstalled(
+ 'pygame is not installed, run `pip install "gymnasium[toy-text]"`'
+ ) from e
+
+ if self.window_surface is None:
+ pygame.init()
+
+ if mode == "human":
+ pygame.display.init()
+ pygame.display.set_caption("Frozen Lake")
+ self.window_surface = pygame.display.set_mode(self.window_size)
+ elif mode == "rgb_array":
+ self.window_surface = pygame.Surface(self.window_size)
+
+ assert (
+ self.window_surface is not None
+ ), "Something went wrong with pygame. This should never happen."
+
+ if self.clock is None:
+ self.clock = pygame.time.Clock()
+ if self.hole_img is None:
+ file_name = path.join(path.dirname(__file__), "img/hole.png")
+ self.hole_img = pygame.transform.scale(
+ pygame.image.load(file_name), self.cell_size
+ )
+ if self.cracked_hole_img is None:
+ file_name = path.join(path.dirname(__file__), "img/cracked_hole.png")
+ self.cracked_hole_img = pygame.transform.scale(
+ pygame.image.load(file_name), self.cell_size
+ )
+ if self.ice_img is None:
+ file_name = path.join(path.dirname(__file__), "img/ice.png")
+ self.ice_img = pygame.transform.scale(
+ pygame.image.load(file_name), self.cell_size
+ )
+ if self.goal_img is None:
+ file_name = path.join(path.dirname(__file__), "img/goal.png")
+ self.goal_img = pygame.transform.scale(
+ pygame.image.load(file_name), self.cell_size
+ )
+ if self.start_img is None:
+ file_name = path.join(path.dirname(__file__), "img/stool.png")
+ self.start_img = pygame.transform.scale(
+ pygame.image.load(file_name), self.cell_size
+ )
+ if self.elf_images is None:
+ elfs = [
+ path.join(path.dirname(__file__), "img/elf_left.png"),
+ path.join(path.dirname(__file__), "img/elf_down.png"),
+ path.join(path.dirname(__file__), "img/elf_right.png"),
+ path.join(path.dirname(__file__), "img/elf_up.png"),
+ ]
+ self.elf_images = [
+ pygame.transform.scale(pygame.image.load(f_name), self.cell_size)
+ for f_name in elfs
+ ]
+
+ desc = self.desc.tolist()
+ assert isinstance(desc, list), f"desc should be a list or an array, got {desc}"
+ for y in range(self.nrow):
+ for x in range(self.ncol):
+ pos = (x * self.cell_size[0], y * self.cell_size[1])
+ rect = (*pos, *self.cell_size)
+
+ self.window_surface.blit(self.ice_img, pos)
+ if desc[y][x] == b"H":
+ self.window_surface.blit(self.hole_img, pos)
+ elif desc[y][x] == b"G":
+ self.window_surface.blit(self.goal_img, pos)
+ elif desc[y][x] == b"S":
+ self.window_surface.blit(self.start_img, pos)
+
+ pygame.draw.rect(self.window_surface, (180, 200, 230), rect, 1)
+
+ # paint the elf
+ bot_row, bot_col = self.s // self.ncol, self.s % self.ncol
+ cell_rect = (bot_col * self.cell_size[0], bot_row * self.cell_size[1])
+ last_action = self.lastaction if self.lastaction is not None else 1
+ elf_img = self.elf_images[last_action]
+
+ if desc[bot_row][bot_col] == b"H":
+ self.window_surface.blit(self.cracked_hole_img, cell_rect)
else:
- return int(self.s), {"prob": 1}
+ self.window_surface.blit(elf_img, cell_rect)
- def getname(self):
- if self.is_slippery:
- variant_name = "semi-slippery"
+ if mode == "human":
+ pygame.event.pump()
+ pygame.display.update()
+ self.clock.tick(self.metadata["render_fps"])
+ elif mode == "rgb_array":
+ return np.transpose(
+ np.array(pygame.surfarray.pixels3d(self.window_surface)),
+ axes=(1, 0, 2),
+ )
+
+ @staticmethod
+ def _center_small_rect(big_rect, small_dims):
+ offset_w = (big_rect[2] - small_dims[0]) / 2
+ offset_h = (big_rect[3] - small_dims[1]) / 2
+ return (
+ big_rect[0] + offset_w,
+ big_rect[1] + offset_h,
+ )
+
+ def _render_text(self):
+ desc = self.desc.tolist()
+ outfile = StringIO()
+
+ row, col = self.s // self.ncol, self.s % self.ncol
+ desc = [[c.decode("utf-8") for c in line] for line in desc]
+ desc[row][col] = utils.colorize(desc[row][col], "red", highlight=True)
+ if self.lastaction is not None:
+ outfile.write(f" ({['Left', 'Down', 'Right', 'Up'][self.lastaction]})\n")
else:
- variant_name = "det"
+ outfile.write("\n")
+ outfile.write("\n".join("".join(line) for line in desc) + "\n")
+
+ with closing(outfile):
+ return outfile.getvalue()
+
+ def close(self):
+ if self.window_surface is not None:
+ import pygame
- return "Lake_{}_{}".format(self.nrow, variant_name)
+ pygame.display.quit()
+ pygame.quit()
# Elf and stool from https://franuka.itch.io/rpg-snow-tileset
diff --git a/AM_Gyms/img/cracked_hole.png b/AM_Gyms/img/cracked_hole.png
new file mode 100644
index 00000000..55930420
Binary files /dev/null and b/AM_Gyms/img/cracked_hole.png differ
diff --git a/AM_Gyms/img/elf_down.png b/AM_Gyms/img/elf_down.png
new file mode 100644
index 00000000..afa3daf1
Binary files /dev/null and b/AM_Gyms/img/elf_down.png differ
diff --git a/AM_Gyms/img/elf_left.png b/AM_Gyms/img/elf_left.png
new file mode 100644
index 00000000..bc9e22ea
Binary files /dev/null and b/AM_Gyms/img/elf_left.png differ
diff --git a/AM_Gyms/img/elf_right.png b/AM_Gyms/img/elf_right.png
new file mode 100644
index 00000000..83640315
Binary files /dev/null and b/AM_Gyms/img/elf_right.png differ
diff --git a/AM_Gyms/img/elf_up.png b/AM_Gyms/img/elf_up.png
new file mode 100644
index 00000000..933f1f00
Binary files /dev/null and b/AM_Gyms/img/elf_up.png differ
diff --git a/AM_Gyms/img/goal.png b/AM_Gyms/img/goal.png
new file mode 100644
index 00000000..709e4b48
Binary files /dev/null and b/AM_Gyms/img/goal.png differ
diff --git a/AM_Gyms/img/hole.png b/AM_Gyms/img/hole.png
new file mode 100644
index 00000000..c7afd1db
Binary files /dev/null and b/AM_Gyms/img/hole.png differ
diff --git a/AM_Gyms/img/ice.png b/AM_Gyms/img/ice.png
new file mode 100644
index 00000000..95dbc74a
Binary files /dev/null and b/AM_Gyms/img/ice.png differ
diff --git a/AM_Gyms/img/stool.png b/AM_Gyms/img/stool.png
new file mode 100644
index 00000000..f304d797
Binary files /dev/null and b/AM_Gyms/img/stool.png differ
diff --git a/AM_Gyms/k_out_of_n.py b/AM_Gyms/k_out_of_n.py
new file mode 100644
index 00000000..cfe0727d
--- /dev/null
+++ b/AM_Gyms/k_out_of_n.py
@@ -0,0 +1,132 @@
+import gymnasium as gym
+from gymnasium import spaces
+from gymnasium.utils import seeding
+
+
+class KOutOfN(gym.Env):
+ """k out of n problem
+ we start with n components all functioning, denoted as 0
+ every step there is some chance, depending on the other components, for components deteriorate, their state to increase by one
+ Action consists of n bits to define whether to do nothing (0) or repair (1)
+
+ Rewards are as follows:
+ if k out of n components are working (not smax), then reward 1
+ repairing a component costs 0.25, a broken component costs 0.5
+ """
+
+ def __init__(
+ self,
+ n=5,
+ k=3,
+ smax=4,
+ repair_cost=0.25,
+ break_cost=0.5,
+ max_steps=100,
+ render_mode=None,
+ ):
+ self.n = n
+ self.k = k
+ self.smax = smax
+ self.repair_cost = repair_cost
+ self.break_cost = break_cost
+ self.max_steps = max_steps
+ self.render_mode = render_mode
+
+ # start with all components repaired
+ self.components = [0] * self.n
+ self.current_step = 0
+ # action is per component whether to repair or do nothing
+ self.action_space = spaces.MultiBinary(self.n)
+ # observation space is per component its value
+ # but, observations must be integers
+ self.observation_space = spaces.Discrete(self.smax**self.n)
+
+ self.last_action = None
+
+ # list of components to state integer
+ def to_s(self, components: list[int]):
+ s = 0
+ for i in range(self.n):
+ s += components[i] * self.smax**i
+ return s
+
+ # state integer to list of components
+ def to_components(self, s: int):
+ components = [0] * self.n
+ for i in range(self.n - 1, -1, -1):
+ components[i] = s // self.smax**i
+ s %= self.smax**i
+ return components
+
+ # action integer to list of actions per component
+ def to_action(self, a: int):
+ action = [0] * self.n
+ for i in range(self.n - 1, -1, -1):
+ action[i] = a // 2**i
+ a %= 2**i
+ return action
+
+ def step(self, action):
+ action = self.to_action(action)
+ done = False
+ self.current_step += 1
+ if self.current_step == 100:
+ done = True
+
+ self.last_action = action
+
+ # process action, calculate next state
+ next_components = [0] * self.n
+ for i in range(self.n):
+ if action[i] == 1:
+ next_components[i] = 0
+ elif self.components[i] == self.smax - 1:
+ # broken component stays broken
+ next_components[i] = self.smax - 1
+ else:
+ broken_neighbors = (
+ self.components[(i - 1) % self.n] == self.smax - 1
+ ) + (self.components[(i - 1) % self.n] == self.smax + 1)
+ p_degrade = 0
+ if broken_neighbors == 0:
+ p_degrade = 0.2
+ elif broken_neighbors == 1:
+ p_degrade = 0.5
+ elif broken_neighbors == 2:
+ p_degrade = 0.9
+
+ if self.np_random.random() <= p_degrade:
+ next_components[i] = self.components[i] + 1
+ else:
+ next_components[i] = self.components[i]
+ self.components = next_components
+
+ # calculating reward
+ reward = 0
+ functioning_components = 0
+ for i in range(self.n):
+ if action[i] == 1:
+ reward -= self.repair_cost
+
+ if self.components[i] == self.smax - 1:
+ reward -= self.break_cost
+ else:
+ functioning_components += 1
+ if functioning_components >= self.k:
+ # positive reward for at least k functioning components
+ reward += 1
+
+ return self.to_s(self.components), reward, done, False, {}
+
+ def reset(self, seed=None, options=None):
+ super().reset(seed=seed)
+
+ # start with all components repaired
+ self.components = [0] * self.n
+ self.current_step = 0
+
+ return self.to_s(self.components), None
+
+ def render(self):
+ # assume ansi, so this returns a string
+ return f"Action {str(self.last_action)}: {str(self.components)}\n"
diff --git a/BAM_QMDP.py b/BAM_QMDP.py
index 906266fe..e3381fd4 100644
--- a/BAM_QMDP.py
+++ b/BAM_QMDP.py
@@ -6,17 +6,13 @@
A full and formal description can be found in the accompanying paper.
"""
-import warnings
-import sys
-
-warnings.filterwarnings("error")
from csv import QUOTE_ALL
from functools import total_ordering
import numpy as np
import math as m
import time
-from AM_Gyms.AM_Env_wrapper import AM_ENV
+import gymnasium as gym
class BAM_QMDP:
@@ -27,11 +23,22 @@ class BAM_QMDP:
#######################################################
def __init__(
- self, env: AM_ENV, epsilon=0.0, nmbr_particles=100, offline_training_steps=0
+ self,
+ env: gym.Env,
+ MeasureCost,
+ InitialState=-1,
+ epsilon=0.0,
+ nmbr_particles=100,
+ offline_training_steps=0,
):
# Environment arguments:
self.env = env
- self.StateSize, self.ActionSize, self.MeasureCost, self.s_init = env.get_vars()
+ self.MeasureCost = MeasureCost
+ self.s_init = InitialState
+
+ # we assume spaces are discrete, otherwise algorithms do not work
+ self.StateSize = env.observation_space.n
+ self.ActionSize = env.action_space.n
self.StateSize = self.StateSize + 1 # Adding a Done-state
self.doneState = self.StateSize - 1
@@ -52,10 +59,8 @@ def __init__(
) # Minimum Measurement Regret for which a measurement is taken (currently equal to measurement cost)
self.optimisticPenalty = 1 # Maximum return estimate (Rewards in all environments are normalised such that this is always 1)
- if offline_training_steps > 0:
- self.otsteps = round(0.2 * self.StateSize) # Offline_training_steps
- else:
- self.otsteps = 0
+ self.otsteps = offline_training_steps
+
self.offline_epsilon = 0.25
self.epsilon_measure = 0.0
self.max_steps_without_measuring = self.StateSize
@@ -79,7 +84,7 @@ def init_run_variables(self):
# Value Estimation Tables
self.QTable = (
- np.ones((self.StateSize, self.ActionSize), dtype=np.longfloat)
+ np.ones((self.StateSize, self.ActionSize), dtype=np.longdouble)
* self.optimisticPenalty
) # Q-table as used by other functions, includes initial bias
@@ -87,10 +92,9 @@ def init_run_variables(self):
(self.StateSize, self.ActionSize)
) # Record average immidiate reward for (s,a) (called \hat{R} in report)
self.Qmax = np.zeros(
- (self.StateSize), dtype=np.longfloat
+ (self.StateSize), dtype=np.longdouble
) # Q-value of optimal action as given by Q (used for readability)
self.QCounter = np.zeros((self.StateSize, self.ActionSize))
- self.Qorder = range(self.StateSize)
self.QTable[self.doneState] = 0
@@ -111,12 +115,11 @@ def init_run_variables(self):
self.ChangedStates = {}
self.T = np.zeros(
- (self.StateSize, self.ActionSize, self.StateSize), dtype=np.longfloat
+ (self.StateSize, self.ActionSize, self.StateSize), dtype=np.longdouble
) # States to be checked in global Q update
# Other vars:
self.totalReward = 0 # reward over all episodes
self.totalSteps = 0 # steps over all episodes
- self.init_episode_variables()
def init_episode_variables(self):
"Initialises all episode-specific variables"
@@ -134,7 +137,7 @@ def init_episode_variables(self):
### RUN FUNCTIONS: ###
#######################################################
- def run_episode(self):
+ def run_episode(self, episode, total_episodes):
"Performes one episode of BAM-QMPD algorithm."
# Initialise all variables:
self.init_episode_variables()
@@ -177,16 +180,16 @@ def run_episode(self):
or self.steps_taken > self.max_steps_without_measuring
)
- # 4: Take Action:
+ # 4: Get Action:
if np.random.rand() < self.epsilon:
action = m.floor(np.random.randint(self.ActionSize))
measure = True
- (reward, self.is_done) = self.env.step(action)
- cost = 0
# 5: Measure
if measure:
- s_next, cost = self.env.measure()
+ s_next, reward, self.is_done, truncated, info = self.env.step(
+ (action, measure)
+ )
next_action_known = False
self.measurements_taken += 1
@@ -200,20 +203,19 @@ def run_episode(self):
# 7: Update P:
self.update_T(s, b_next, action, self.is_done)
+ else:
+ _, reward, self.is_done, truncated, info = self.env.step(
+ (action, measure)
+ )
- # 8: Update Q
-
- self.update_Q_lastStep_only(s, action, reward, isDone=self.is_done)
-
- # if self.otsteps > 0:
- # for i in range(self.otsteps):
- # self.train_offline()
+ # 8: Update Q
+ self.update_Q_lastStep_only(s, action, reward, isDone=self.is_done)
# 9: Update variables for next step:
history.append((s, action))
s = b_next
- self.episodeReward += reward - cost
+ self.episodeReward += reward
self.steps_taken += 1
self.totalSteps += 1
@@ -222,53 +224,24 @@ def run_episode(self):
for b, a in reversed(history):
self.update_Q_lastStep_only(b, a, isReal=False)
- if self.otsteps > 0:
- for i in range(self.otsteps):
- self.train_after_episode()
+ for i in range(self.otsteps):
+ self.train_offline()
self.totalReward += self.episodeReward
- returnVars = (self.episodeReward, self.steps_taken, self.measurements_taken)
- return returnVars
-
- def run(
- self, nmbr_episodes, get_full_results=False, print_info=False, logmessages=True
- ):
- "Performs the specified number of episodes of BAM-QMDP."
- self.init_run_variables()
- epreward, epsteps, epms = (
- np.zeros((nmbr_episodes)),
- np.zeros((nmbr_episodes)),
- np.zeros((nmbr_episodes)),
- )
- for i in range(nmbr_episodes):
- log_nmbr = 100
- if i > 0 and i % log_nmbr == 0 and logmessages:
- print(
- "{} / {} runs complete (current avg reward = {}, nmbr steps = {}, nmbr measures = {})".format(
- i,
- nmbr_episodes,
- np.average(epreward[(i - log_nmbr) : i]),
- np.average(epsteps[(i - log_nmbr) : i]),
- np.average(epms[(i - log_nmbr) : i]),
- )
- )
-
- epreward[i], epsteps[i], epms[i] = self.run_episode()
+ return (self.episodeReward, self.steps_taken, self.measurements_taken)
- if False:
- print(
- """
+ def print_info(self):
+ "Print info for debugging, can be run from Run.py"
+ print(
+ """
Run complete:
Alpha table: {}
QTable: {}
Rewards Table: {}
- """.format(
- self.alpha, self.QTable, self.QTableRewards
- )
+ """.format(
+ self.alpha, self.QTable, self.QTableRewards
)
- if get_full_results:
- return (self.totalReward, epreward, epsteps, epms)
- return self.totalReward
+ )
#######################################################
### HELPER FUNCTIONS: ###
@@ -471,37 +444,15 @@ def update_Q_lastStep_only(self, S1, action, reward=0, isDone=False, isReal=True
self.QTable[s1, action] = 1
def train_offline(self):
- "Performs Dyna-style oflline training of Q-values using current transition function"
- for i in range(self.otsteps):
- # Choose random state and action
- s = np.random.randint(self.StateSize - 1)
- S_dict = {s: 1}
+ "Performs Dyna-style oflline training of Q-values using current transition function once"
+ # Choose random state and action
+ s = np.random.randint(self.StateSize - 1)
+ S_dict = {s: 1}
- if np.random.rand() < self.offline_epsilon:
- a = np.random.randint(self.ActionSize)
- else:
- a = self.get_action(S_dict)
-
- if np.sum(self.alpha[s, a]) > 5:
- # b_next = self.guess_next_state(S_dict,a)
- self.update_Q_lastStep_only(S_dict, a, isReal=False)
-
- def train_after_episode(self):
- self.update_Qorder()
- S = np.array(range(self.StateSize))[self.Qorder]
- idxs = np.random.choice(self.StateSize, size=self.otsteps, replace=False)
- mask = np.zeros(self.StateSize, dtype="bool")
- mask[idxs] = True
- S = S[mask]
-
- for s in S:
- if np.random.rand() < self.offline_epsilon:
- a = np.random.randint(self.ActionSize)
- else:
- a = self.get_action({s: 1})
- if np.sum(self.alpha[s, a]) > 5:
- # b_next = self.guess_next_state(S_dict,a)
- self.update_Q_lastStep_only({s: 1}, a, isReal=False)
+ if np.random.rand() < self.offline_epsilon:
+ a = np.random.randint(self.ActionSize)
+ else:
+ a = self.get_action(S_dict)
- def update_Qorder(self):
- self.Qorder = np.argsort(self.Qmax[self.Qorder])
+ if np.sum(self.alpha[s, a]) > 5:
+ self.update_Q_lastStep_only(S_dict, a, isReal=False)
diff --git a/Baselines/AMRL_Agent.py b/Baselines/AMRL_Agent.py
index ed6e8099..4bc87e1b 100644
--- a/Baselines/AMRL_Agent.py
+++ b/Baselines/AMRL_Agent.py
@@ -1,15 +1,30 @@
### Implementation of AMRL-Algorithm as described in https://arxiv.org/abs/2005.12697
import numpy as np
+import gymnasium as gym
class AMRL_Agent:
"""Creates a AMRL-Agent, as described in https://arxiv.org/abs/2005.12697"""
- def __init__(self, env, epsilon=0.1, m_bias=0.1, turn_greedy=True, greedy_perc=0.9):
+ def __init__(
+ self,
+ env: gym.Env,
+ MeasureCost,
+ InitialState=-1,
+ epsilon=0.1,
+ m_bias=0.1,
+ turn_greedy=True,
+ greedy_perc=0.9,
+ ):
# load all environment-specific variables
self.env = env
- self.StateSize, self.ActionSize, self.measureCost, self.s_init = env.get_vars()
+ self.measureCost = MeasureCost
+ self.s_init = InitialState
+
+ # we assume spaces are discrete, otherwise algorithms do not work
+ self.StateSize = env.observation_space.n
+ self.ActionSize = env.action_space.n
# load all algo-specific vars (if provided)
self.epsilon, self.m_bias = epsilon, m_bias
@@ -21,9 +36,9 @@ def __init__(self, env, epsilon=0.1, m_bias=0.1, turn_greedy=True, greedy_perc=0
self.df = 0.95
# Create all episode and run-specific variables
- self.reset_Run_Variables()
+ self.init_run_variables()
- def reset_Run_Variables(self):
+ def init_run_variables(self):
# Variables for one run
self.QTable = np.zeros((self.StateSize, self.ActionSize, self.MeasureSize))
self.QTable[:, :, 1] = self.m_bias
@@ -36,14 +51,13 @@ def reset_Run_Variables(self):
self.TriesTable = np.zeros((self.StateSize, self.ActionSize, self.StateSize))
self.totalReward = 0
# Variables for one epoch
- self.reset_Epoch_Vars()
+ self.init_episode_variables()
- def reset_Epoch_Vars(self):
+ def init_episode_variables(self):
self.currentReward = 0
self.steps_taken = 0
self.measurements_taken = 0
self.env.reset()
- self.totalReward += self.currentReward
# TODO: add variable to keep track of 'actual reward' without costs, and see how this gets effected
def update_TransTable(self, s1, s2, action):
@@ -112,8 +126,13 @@ def find_nonOptimal_actionPair(self, s):
np.random.randint(0, self.MeasureSize),
)
- def train_epoch(self):
+ def run_episode(self, episode, total_episodes):
"""Training algorithm of AMRL as given in paper"""
+ if self.turn_greedy and episode / total_episodes > self.greedy_perc:
+ self.be_greedy = True
+
+ self.init_episode_variables()
+
s_current = self.s_init
done = False
while not done:
@@ -131,42 +150,23 @@ def train_epoch(self):
# Update reward, Q-table and s_next
if measure:
- (reward, done) = self.env.step(action)
- (obs, cost) = self.env.measure()
+ obs, reward, done, truncated, info = self.env.step((action, measure))
self.update_TransTable(s_current, obs, action)
self.measurements_taken += 1
s_next = obs
else:
- (reward, done) = self.env.step(action)
+ _, reward, done, truncated, info = self.env.step((action, measure))
s_next = self.guess_current_State(s_current, action)
self.update_QTable(s_current, action, measure, s_next, reward, done)
s_current = s_next
- self.currentReward += (
- reward - self.measureCost * measure
- ) # this could be cleaner...
+ self.currentReward += reward
self.steps_taken += 1
if not done:
print("max nmbr of steps exceded!")
- # Reset after epoch, return reward and #steps
self.totalReward += self.currentReward
- (rew, steps, ms) = self.currentReward, self.steps_taken, self.measurements_taken
- self.reset_Epoch_Vars()
- return (rew, steps, ms)
-
- def run(self, nmbr_epochs, get_intermediate_results=False):
- self.reset_Run_Variables()
- rewards, steps, ms = (
- np.zeros((nmbr_epochs)),
- np.zeros((nmbr_epochs)),
- np.zeros((nmbr_epochs)),
- )
- for i in range(nmbr_epochs):
- rewards[i], steps[i], ms[i] = self.train_epoch()
- if self.turn_greedy and i / nmbr_epochs > self.greedy_perc:
- self.be_greedy = True
- # print((self.TransTable, self.QTriesTable, self.QTable)) # Debug stuff
- if get_intermediate_results:
- return (self.totalReward, rewards, steps, ms)
- return self.totalReward
+ return self.currentReward, self.steps_taken, self.measurements_taken
+
+ def print_info(self):
+ print((self.TransTable, self.QTriesTable, self.QTable)) # Debug stuff
diff --git a/Baselines/DynaQ.py b/Baselines/DynaQ.py
index dd9f70e2..fb8f5b5a 100644
--- a/Baselines/DynaQ.py
+++ b/Baselines/DynaQ.py
@@ -4,14 +4,14 @@
"""
from AM_Gyms.ModelLearner import ModelLearner
-from AM_Gyms.AM_Env_wrapper import AM_ENV
+from gymnasium import Env
import numpy as np
class QBasic:
"""Class for standard Q-learning of AM-environments"""
- def __init__(self, ENV: AM_ENV):
+ def __init__(self, ENV: Env):
self.env = ENV
self.StateSize, self.ActionSize, self.MeasureCost, self.s_init = (
self.env.get_vars()
@@ -117,7 +117,7 @@ def update_Q(self, s, action, reward, obs):
class QDyna(QBasic):
- def __init__(self, ENV: AM_ENV):
+ def __init__(self, ENV: Env):
super().__init__(ENV)
self.R_counter = np.zeros((self.StateSize, self.ActionSize))
self.trainingSteps = 10
diff --git a/GetAgent.py b/GetAgent.py
new file mode 100644
index 00000000..ff858df6
--- /dev/null
+++ b/GetAgent.py
@@ -0,0 +1,61 @@
+import gymnasium as gym
+
+from Baselines.AMRL_Agent import AMRL_Agent as AMRL
+from BAM_QMDP import BAM_QMDP
+
+# from Baselines.ACNO_generalised.Observe_then_plan_agent import ACNO_Agent_OTP
+
+# from Baselines.DRQN import DRQN_Agent requires torch to be downloaded, so kept turned off
+from Baselines.DynaQ import QBasic, QOptimistic, QDyna
+
+# from Baselines.ACNO_generalised.ACNO_ENV import ACNO_ENV
+
+
+def get_agent(ENV: gym.Env, algo_name, MeasureCost, InitialState):
+ if algo_name == "AMRL":
+ agent = AMRL(
+ ENV,
+ MeasureCost=MeasureCost,
+ InitialState=InitialState,
+ turn_greedy=True,
+ )
+ # AMRL-Q, alter so it is completely greedy in last steps.
+ elif algo_name == "AMRL_greedy":
+ agent = AMRL(
+ ENV,
+ MeasureCost=MeasureCost,
+ InitialState=InitialState,
+ turn_greedy=False,
+ )
+ # BAM_QMDP, named Dyna-ATMQ in paper. Variant with no offline training
+ elif algo_name == "BAM_QMDP":
+ agent = BAM_QMDP(
+ ENV,
+ offline_training_steps=0,
+ MeasureCost=MeasureCost,
+ InitialState=InitialState,
+ )
+ # BAM_QMDP, named Dyna-ATMQ in paper. Variant with 25 offline training steps per real step
+ elif algo_name == "BAM_QMDP+":
+ agent = BAM_QMDP(
+ ENV,
+ offline_training_steps=25,
+ MeasureCost=MeasureCost,
+ InitialState=InitialState,
+ )
+ # Observe-then-plan agent from ACNO-paper. As used in paper, slight alterations made from original
+ elif algo_name == "ACNO_OTP":
+ ENV_ACNO = ACNO_ENV(ENV)
+ agent = ACNO_Agent_OTP(ENV_ACNO)
+ # A number of generic RL-agents. We did not include these in the paper.
+ # elif algo_name == "DRQN":
+ # agent = DRQN_Agent(ENV)
+ elif algo_name == "QBasic":
+ agent = QBasic(ENV)
+ elif algo_name == "QOptimistic":
+ agent = QOptimistic(ENV)
+ elif algo_name == "QDyna":
+ agent = QDyna(ENV)
+ else:
+ print("Agent {} not recognised, please try again!".format(algo_name))
+ return agent
diff --git a/GetEnv.py b/GetEnv.py
new file mode 100644
index 00000000..c5e1b33a
--- /dev/null
+++ b/GetEnv.py
@@ -0,0 +1,124 @@
+import numpy as np
+import gymnasium as gym
+
+# Environments
+from AM_Gyms.NchainEnv import NChainEnv
+from AM_Gyms.Loss_Env import Measure_Loss_Env
+from AM_Gyms.frozen_lake_v2 import FrozenLakeEnv_v2
+from AM_Gyms.Sepsis.SepsisEnv import SepsisEnv
+from AM_Gyms.Blackjack import BlackjackEnv
+from AM_Gyms.k_out_of_n import KOutOfN
+from gymnasium.envs.toy_text.frozen_lake import generate_random_map
+from AM_Gyms.ActiveMeasurementWrapper import ActiveMeasurementWrapper
+
+
+def get_env(env_name, env_gen, env_variant, env_size, remake_env_opt, seed=None):
+ "Returns ActiveMeasurement env as specified in global (user-specified) vars"
+
+ np.random.seed(seed)
+
+ # if no measure_cost is provided
+ default_measure_cost = 0.05
+
+ remake_env = False
+
+ # Basically, just a big messy pile of if/else statements (Not using match for pre 3.10 python users)
+
+ # Loss-environment, called Measure Regret environment in paper.
+ if env_name == "Loss":
+ env = Measure_Loss_Env()
+ InitialState = 0
+ default_measure_cost = 0.1
+ # Frozen lake environment (includes all variants)
+ elif env_name == "Lake":
+ InitialState = 0
+ default_measure_cost = 0.05
+ if env_size == 0:
+ print("Using standard size map (4x4)")
+ env_size = 4
+
+ if env_gen == "random":
+ map_name = None
+ desc = generate_random_map(size=env_size)
+ elif env_gen == "standard":
+ if env_size != 4 and env_size != 8:
+ print("Standard map type can only be used for sizes 4 and 8")
+ else:
+ map_name = "{}x{}".format(env_size, env_size)
+ desc = None
+ else:
+ print("Using random map")
+ map_name = None
+ desc = generate_random_map(size=env_size)
+
+ if map_name is None and remake_env_opt:
+ remake_env = True
+
+ if env_variant == "det":
+ env = gym.make(
+ "FrozenLake-v1",
+ desc=desc,
+ map_name=map_name,
+ is_slippery=False,
+ render_mode="rgb_array",
+ )
+ elif env_variant == "slippery":
+ env = gym.make(
+ "FrozenLake-v1",
+ desc=desc,
+ map_name=map_name,
+ is_slippery=True,
+ render_mode="rgb_array",
+ )
+ elif env_variant == "semi-slippery":
+ env = FrozenLakeEnv_v2(
+ desc=desc, map_name=map_name, render_mode="rgb_array"
+ )
+ else: # default = deterministic
+ print("Environment var not recognised! (using deterministic variant)")
+ env = gym.make(
+ "FrozenLake-v1",
+ desc=desc,
+ map_name=map_name,
+ is_slippery=False,
+ render_mode="rgb_array",
+ )
+ # Taxi environment, as used in AMRL-Q paper. Not used in paper
+ elif env_name == "Taxi":
+ env = gym.make("Taxi-v3", render_mode="rgb_array")
+ InitialState = -1
+ default_measure_cost = 0.01 / 20
+ elif env_name == "CliffWalking":
+ env = gym.make("CliffWalking-v0", render_mode="rgb_array", is_slippery=True)
+ InitialState = 36
+ default_measure_cost = 0.01 / 20
+ # Chain environment, as used in AMRL-Q paper. Not used in paper
+ elif env_name == "Chain":
+ env = NChainEnv(env_size)
+ InitialState = 0
+ default_measure_cost = 0.01 / 20
+ # Sepsis environment, as used in ACNO-paper. Not used in paper
+ elif env_name == "Sepsis":
+ env = SepsisEnv()
+ InitialState = -1
+ default_measure_cost = 0.05
+ # Standard OpenAI Gym blackjack environment. Not used in paper
+ elif env_name == "Blackjack":
+ env = BlackjackEnv(render_mode="rgb_array")
+ InitialState = -1
+ default_measure_cost = 0.05
+ elif env_name == "KOutOfN":
+ smax = 4
+ n = 4
+ if env_size != 0:
+ n = env_size
+ env = KOutOfN(n=n, smax=smax, render_mode="ansi")
+ default_measure_cost = 0.05
+ InitialState = 0
+ else:
+ print("Environment {} not recognised, please try again!".format(env_name))
+ return
+
+ env = ActiveMeasurementWrapper(env)
+
+ return env, InitialState, default_measure_cost, remake_env
diff --git a/Run.py b/Run.py
index 9ba2936c..cb60fd05 100644
--- a/Run.py
+++ b/Run.py
@@ -1,62 +1,18 @@
"""
File for running & gathering data on Active-Measuring algorithms.
For a brief description of how to use it, see the Readme-file in this repo.
-
"""
-######################################################
-### Imports ###
-######################################################
-
-
-# File structure stuff
-import sys
-import os
-
-sys.path.append(os.path.join(sys.path[0], "Baselines"))
-sys.path.append(os.path.join(sys.path[0], "Baselines", "ACNO_generalised"))
-
-# External modules
import numpy as np
-import gym
-import matplotlib.pyplot as plt
import time as t
import datetime
import json
import argparse
-from typing import List, Optional
-import os
-
-# Agents
-import Baselines.AMRL_Agent as amrl
-from BAM_QMDP import BAM_QMDP
+from gymnasium.wrappers import RecordVideo
+from AM_Gyms.TextEpisodeRecorder import TextEpisodeRecorder
-# from Baselines.ACNO_generalised.Observe_then_plan_agent import ACNO_Agent_OTP
-
-# from Baselines.DRQN import DRQN_Agent requires torch to be downloaded, so kept turned off
-from Baselines.DynaQ import QBasic, QOptimistic, QDyna
-
-# Environments
-from AM_Gyms.NchainEnv import NChainEnv
-from AM_Gyms.Loss_Env import Measure_Loss_Env
-from AM_Gyms.frozen_lake_v2 import FrozenLakeEnv_v2
-from AM_Gyms.Sepsis.SepsisEnv import SepsisEnv
-from AM_Gyms.Blackjack import BlackjackEnv
-from AM_Gyms.frozen_lake import FrozenLakeEnv, generate_random_map, is_valid
-
-# Environment wrappers
-from AM_Gyms.AM_Env_wrapper import AM_ENV as wrapper
-from AM_Gyms.AM_Env_wrapper import AM_Visualiser as visualiser
-
-# from Baselines.ACNO_generalised.ACNO_ENV import ACNO_ENV
-
-
-# JSON encoder
-class NumpyEncoder(json.JSONEncoder):
- def default(self, obj):
- if isinstance(obj, np.ndarray):
- return obj.tolist()
- return json.JSONEncoder.default(self, obj)
+from GetEnv import get_env
+from GetAgent import get_agent
######################################################
@@ -84,13 +40,13 @@ def default(self, obj):
default=-1.0,
help="Cost of measuring (default: use as specified by environment)",
)
-parser.add_argument("-nmbr_eps", default=500, help="nmbr of episodes per run")
-parser.add_argument("-nmbr_runs", default=1, help="nmbr of runs to perform")
+parser.add_argument("-nmbr_eps", default=500, help="Number of episodes per run")
+parser.add_argument("-nmbr_runs", default=1, help="Number of runs to perform")
parser.add_argument(
"-f", default=None, help="File name (default: generated automatically)"
)
parser.add_argument(
- "-rep", default="./Data/", help="Repository to store data (default: ./Data"
+ "-rep", default="./Data/", help="Repository to store data (default: ./Data)"
)
parser.add_argument(
"-env_remake",
@@ -98,6 +54,26 @@ def default(self, obj):
help="Option to make a new (random) environment each run or not",
)
parser.add_argument("-save", default=True, help="Option to save or not save data.")
+parser.add_argument(
+ "-record_videos",
+ default=False,
+ help="Save videos of every episode that is a power of 3 until 1000 and then every 1000 episodes.",
+)
+parser.add_argument(
+ "-record_text",
+ default=False,
+ help="Save text output of every episode that is a power of 3 until 1000 and then every 1000 episodes.",
+)
+parser.add_argument(
+ "-video_directory",
+ default="videos",
+ help="Directory to store the videos of the episodes in (default: ./videos).",
+)
+parser.add_argument(
+ "-video_prefix",
+ default="training",
+ help="Prefix for video filenames.",
+)
# Unpacking for use in this file:
args = parser.parse_args()
@@ -106,207 +82,80 @@ def default(self, obj):
env_variant = args.env_var
env_size = int(args.env_size)
env_gen = str(args.env_gen)
-MeasureCost = float(args.m_cost)
+measure_cost = float(args.m_cost)
nmbr_eps = int(args.nmbr_eps)
nmbr_runs = int(args.nmbr_runs)
file_name = args.f
rep_name = args.rep
remake_env_opt = True
-
-if args.env_remake in ["False", "false"]:
+if args.env_remake in ["False", "false", "0"]:
remake_env_opt = False
-
-if args.save == "False" or args.save == "false":
+doSave = True
+if args.save in ["False", "false", "0"]:
doSave = False
-else:
- doSave = True
-
-# Create name for Data file
-envFullName = env_name
-if env_size != 0:
- envFullName += "_" + env_gen + str(env_size)
+record_videos = False
+if args.record_videos in ["True", "true", "1"]:
+ record_videos = True
+record_text = False
+if args.record_text in ["True", "true", "1"]:
+ record_text = True
+video_directory = args.video_directory
+video_prefix = args.video_prefix
-if env_variant != "None":
- envFullName += "_" + env_variant
######################################################
-### Intitialise Environment ###
+### Getting environment and agent ###
######################################################
-# Lake Envs
-s_init = 0
-MeasureCost_Lake_default = 0.05
-MeasureCost_Taxi_default = 0.01 / 20
-MeasureCost_Chain_default = 0.05
-remake_env = False
-env_folder_name = os.path.join(os.getcwd(), "AM_Gyms", "Learned_Models")
-
-
-def get_env(seed=None):
- "Returns AM_Env as specified in global (user-specified) vars"
- global MeasureCost
- global remake_env
- global env_size
- global env_full_name
-
- # Required for making robust env through generic-gym class
- has_terminal_state = True
- terminal_prob = 0.0
-
- np.random.seed(seed)
-
- # Basically, just a big messy pile of if/else statements (Not using match for pre 3.10 python users)
-
- # Loss-environment, called Measure Regret environment in paper.
- if env_name == "Loss":
- env = Measure_Loss_Env()
- StateSize, ActionSize, s_init = 4, 2, 0
- if MeasureCost == -1:
- MeasureCost = 0.1
-
- # Frozen lake environment (includes all variants)
- elif env_name == "Lake":
- ActionSize, s_init = 4, 0
- if MeasureCost == -1:
- MeasureCost = MeasureCost_Lake_default
- if env_size == 0:
- print("Using standard size map (4x4)")
- env_size = 4
- StateSize = 4**2
- else:
- StateSize = env_size**2
-
- if env_gen == "random":
- map_name = None
- desc = generate_random_map(size=env_size)
- elif env_gen == "standard":
- if env_size != 4 and env_size != 8:
- print("Standard map type can only be used for sizes 4 and 8")
- else:
- map_name = "{}x{}".format(env_size, env_size)
- desc = None
- else:
- print("Using random map")
- map_name = None
- desc = generate_random_map(size=env_size)
-
- if map_name is None and remake_env_opt:
- remake_env = True
-
- if env_variant == "det":
- env = FrozenLakeEnv(desc=desc, map_name=map_name, is_slippery=False)
- elif env_variant == "slippery":
- env = FrozenLakeEnv(desc=desc, map_name=map_name, is_slippery=True)
- elif env_variant == "semi-slippery":
- env = FrozenLakeEnv_v2(desc=desc, map_name=map_name)
- elif env_variant == None:
- env = FrozenLakeEnv(desc=desc, map_name=map_name, is_slippery=False)
- else: # default = deterministic
- print("Environment var not recognised! (using deterministic variant)")
- env = FrozenLakeEnv(desc=desc, map_name=map_name, is_slippery=False)
-
- # Taxi environment, as used in AMRL-Q paper. Not used in paper
- elif env_name == "Taxi":
- env = gym.make("Taxi-v3")
- StateSize, ActionSize, s_init = 500, 6, -1
- if MeasureCost == -1:
- MeasureCost = MeasureCost_Taxi_default
-
- # Chain environment, as used in AMRL-Q paper. Not used in paper
- elif env_name == "Chain":
- if env_size == "10":
- StateSize = 10
- elif env_size == "20":
- StateSize = 20
- elif env_size == "30":
- StateSize = 30
- elif env_size == "50":
- StateSize = 50
- elif env_size == other: # default
- print("env_map not recognised!")
- StateSize = 20
-
- env = NChainEnv(StateSize)
- ActionSize, s_init = 2, 0
- if MeasureCost == -1:
- MeasureCost = MeasureCost_Chain_default
-
- # Sepsis environment, as used in ACNO-paper. Not used in paper
- elif env_name == "Sepsis":
- env = SepsisEnv()
- StateSize, ActionSize, s_init = 720, 8, -1
- if MeasureCost == -1:
- MeasureCost = 0.05
-
- # Standard OpenAI Gym blackjack environment. Not used in paper
- elif env_name == "Blackjack":
- env = BlackjackEnv()
- StateSize, ActionSize, s_init = 704, 2, -1
- if MeasureCost == -1:
- MeasureCost = 0.05
-
- else:
- print("Environment {} not recognised, please try again!".format(env_name))
- return
-
- ENV = wrapper(env, StateSize, ActionSize, MeasureCost, s_init)
- args.m_cost = MeasureCost
-
- return ENV
+env, InitialState, default_measure_cost, remake_env = get_env(
+ env_name, env_gen, env_variant, env_size, remake_env_opt, seed=0
+)
+if measure_cost == -1:
+ measure_cost = default_measure_cost
+
+if record_videos:
+ env = RecordVideo(
+ env,
+ video_folder=video_directory,
+ name_prefix=video_prefix,
+ # create custom episode trigger:
+ # episode_trigger=lambda x: x % 2 == 0,
+ # specify video length for videos to span multiple episodes:
+ # video_length=5000,
+ )
+if record_text:
+ env = TextEpisodeRecorder(env, folder=video_directory, name_prefix=video_prefix)
+
+agent = get_agent(env, algo_name, measure_cost, InitialState)
######################################################
-### Defining Agents ###
+### Exporting Results ###
######################################################
-# Both final names and previous/working names are implemented here
-def get_agent(seed=None):
-
- ENV = get_env(seed)
- if algo_name == "AMRL":
- agent = amrl.AMRL_Agent(ENV, turn_greedy=True)
- # AMRL-Q, alter so it is completely greedy in last steps.
- elif algo_name == "AMRL_greedy":
- agent = amrl.AMRL_Agent(ENV, turn_greedy=False)
- # BAM_QMDP, named Dyna-ATMQ in paper. Variant with no offline training
- elif algo_name == "BAM_QMDP":
- agent = BAM_QMDP(ENV, offline_training_steps=0)
- # BAM_QMDP, named Dyna-ATMQ in paper. Variant with 25 offline training steps per real step
- elif algo_name == "BAM_QMDP+":
- agent = BAM_QMDP(ENV, offline_training_steps=5)
- # Observe-then-plan agent from ACNO-paper. As used in paper, slight alterations made from original
- elif algo_name == "ACNO_OTP":
- ENV_ACNO = ACNO_ENV(ENV)
- agent = ACNO_Agent_OTP(ENV_ACNO)
- # A number of generic RL-agents. We did not include these in the paper.
- # elif algo_name == "DRQN":
- # agent = DRQN_Agent(ENV)
- elif algo_name == "QBasic":
- agent = QBasic(ENV)
- elif algo_name == "QOptimistic":
- agent = QOptimistic(ENV)
- elif algo_name == "QDyna":
- agent = QDyna(ENV)
- else:
- print("Agent {} not recognised, please try again!".format(algo_name))
- return agent
+# JSON encoder
+class NumpyEncoder(json.JSONEncoder):
+ def default(self, obj):
+ if isinstance(obj, np.ndarray):
+ return obj.tolist()
+ return json.JSONEncoder.default(self, obj)
-######################################################
-### Exporting Results ###
-######################################################
+# Create name for Data file
+envFullName = env_name
+if env_size != 0:
+ envFullName += "_" + env_gen + str(env_size)
+
+if env_variant != "None":
+ envFullName += "_" + env_variant
# Automatically creates filename is not specified by user
if file_name == None:
file_name = "AMData_{}_{}_{}.json".format(
- algo_name, envFullName, str(int(float(args.m_cost) * 100)).zfill(3)
+ algo_name, envFullName, str(int(float(measure_cost) * 100)).zfill(3)
)
-# Set measurecost if not set by environment.
-if args.m_cost == -1:
- args.m_cost == MeasureCost
-
def PR_to_data(pr_time):
"Prints timecode as used in datafiles"
@@ -359,30 +208,56 @@ def export_data(rewards, steps, measures, t_start):
)
)
-agent = get_agent(0)
-for i in range(nmbr_runs):
+for run in range(nmbr_runs):
t_this_start = t.perf_counter()
- (r_tot, rewards[i], steps[i], measures[i]) = agent.run(nmbr_eps, True)
- rewards_avg[i], steps_avg[i], measures_avg[i] = (
- np.average(rewards[i]),
- np.average(steps[i]),
- np.average(measures[i]),
+ # (r_tot, rewards[i], steps[i], measures[i]) = agent.run(nmbr_eps, True)
+
+ agent.init_run_variables()
+ # execute episodes for this run
+ for episode in range(nmbr_eps):
+ log_nmbr = 100
+ if episode > 0 and episode % log_nmbr == 0:
+ print(
+ "{} / {} episodes complete (current avg reward = {}, nmbr steps = {}, nmbr measures = {})".format(
+ episode,
+ nmbr_eps,
+ np.average(rewards[run][(episode - log_nmbr) : episode]),
+ np.average(steps[run][(episode - log_nmbr) : episode]),
+ np.average(measures[run][(episode - log_nmbr) : episode]),
+ )
+ )
+ # debugging:
+ # agent.print_info()
+ rewards[run][episode], steps[run][episode], measures[run][episode] = (
+ agent.run_episode(episode, nmbr_eps)
+ )
+
+ rewards_avg[run], steps_avg[run], measures_avg[run] = (
+ np.average(rewards[run]),
+ np.average(steps[run]),
+ np.average(measures[run]),
)
t_this_end = t.perf_counter()
if doSave:
- export_data(rewards[: i + 1], steps[: i + 1], measures[: i + 1], t_start)
+ export_data(rewards[: run + 1], steps[: run + 1], measures[: run + 1], t_start)
print(
"Run {0} done with average reward {2}! (in {1} s, with {3} steps and {4} measurements avg.)\n".format(
- i + 1,
+ run + 1,
t_this_end - t_this_start,
- rewards_avg[i],
- steps_avg[i],
- measures_avg[i],
+ rewards_avg[run],
+ steps_avg[run],
+ measures_avg[run],
)
)
- if remake_env and i < nmbr_runs - 1:
- agent = get_agent(i + 1)
+ if remake_env and run < nmbr_runs - 1:
+ env, InitialState, default_measure_cost, remake_env = get_env(
+ env_name, env_gen, env_variant, env_size, remake_env_opt, seed=run + 1
+ )
+ if measure_cost == -1:
+ measure_cost = default_measure_cost
+ agent = get_agent(env, algo_name, measure_cost, InitialState)
+
print(
"Agent Done! ({0} runs in {1} s, with average reward {2}, steps {3}, measures {4})\n\n".format(
nmbr_runs,
diff --git a/requirements.txt b/requirements.txt
index 102c29e9..6400bfac 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,5 +1,5 @@
numpy
-gym
+gymnasium
scipy
matplotlib
future