"""

global checkpoint_dir

global log_reward_path

global log_q_values_path

# Add the environment-name to the checkpoint-dir.

checkpoint_dir = os.path.join(checkpoint_base_dir, env_name)

# Create the checkpoint-dir if it does not already exist.

if not os.path.exists(checkpoint_dir):

os.makedirs(checkpoint_dir)

# File-path for the log-file for episode rewards.

log_reward_path = os.path.join(checkpoint_dir, "log_reward.txt")

# File-path for the log-file for Q-values.

"""

# Get the url for the game-environment.

url = _checkpoint_url[env_name]

# Download and extract the file if it does not already exist.

download.maybe_download_and_extract(url=url,

"""

def __init__(self, file_path):

"""Set the path for the log-file. Nothing is saved or loaded yet."""

# Path for the log-file.

self.file_path = file_path

# Data to be read from the log-file by the _read() function.

self.count_episodes = None

self.count_states = None

self.data = None

def _write(self, count_episodes, count_states, msg):

"""

with open(file=self.file_path, mode='a', buffering=1) as file:

msg_annotated = "{0}\t{1}\t{2}\n".format(count_episodes, count_states, msg)

file.write(msg_annotated)

def _read(self):

"""

# Open and read the log-file.

with open(self.file_path) as f:

reader = csv.reader(f, delimiter="\t")

self.count_episodes, self.count_states, *data = zip(*reader)

# Convert the remaining log-data to a NumPy float-array.

"""Log the rewards obtained for episodes during training."""

def __init__(self):

# These will be set in read() below.

self.episode = None

self.mean = None

# Super-class init.

Log.__init__(self, file_path=log_reward_path)

def write(self, count_episodes, count_states, reward_episode, reward_mean):

"""

msg = "{0:.1f}\t{1:.1f}".format(reward_episode, reward_mean)

self._write(count_episodes=count_episodes, count_states=count_states, msg=msg)

def read(self):

"""

# Read the log-file using the super-class.

self._read()

# Get the episode reward.

self.episode = self.data[0]

# Get the mean reward.

"""Log the Q-Values during training."""

def __init__(self):

# These will be set in read() below.

self.min = None

self.mean = None

self.max = None

self.std = None

# Super-class init.

Log.__init__(self, file_path=log_q_values_path)

def write(self, count_episodes, count_states, q_values):

"""

msg = "{0:.3f}\t{1:.3f}\t{2:.3f}\t{3:.3f}".format(np.min(q_values),

np.mean(q_values),

np.max(q_values),

np.std(q_values))

self._write(count_episodes=count_episodes,

count_states=count_states,

msg=msg)

def read(self):

"""

# Read the log-file using the super-class.

self._read()

# Get the logged statistics for the Q-values.

self.min = self.data[0]

self.mean = self.data[1]

self.max = self.data[2]

"""

sys.stdout.write("\r" + msg)

"""

# Get the separate colour-channels.

r, g, b = image[:, :, 0], image[:, :, 1], image[:, :, 2]

# Convert to gray-scale using the Wikipedia formula.

img_gray = 0.2990 * r + 0.5870 * g + 0.1140 * b

"""Pre-process a raw image from the game-environment."""

# Convert image to gray-scale.

img = _rgb_to_grayscale(image)

# Resize to the desired size using SciPy for convenience.

img = scipy.misc.imresize(img, size=state_img_size, interp='bicubic')

"""

def __init__(self, image, decay=0.75):

"""

# Pre-process the image and save it for later use.

# The input image may be 8-bit integers but internally

# we need to use floating-point to avoid image-noise

# caused by recurrent rounding-errors.

img = _pre_process_image(image=image)

self.last_input = img.astype(np.float)

# Set the last output to zero.

self.last_output = np.zeros_like(img)

self.decay = decay

def process(self, image):

"""Process a raw image-frame from the game-environment."""

# Pre-process the image so it is gray-scale and resized.

img = _pre_process_image(image=image)

# Subtract the previous input. This only leaves the

# pixels that have changed in the two image-frames.

img_dif = img - self.last_input

# Copy the contents of the input-image to the last input.

self.last_input[:] = img[:]

# If the pixel-difference is greater than a threshold then

# set the output pixel-value to the highest value (white),

# otherwise set the output pixel-value to the lowest value (black).

# So that we merely detect motion, and don't care about details.

img_motion = np.where(np.abs(img_dif) > 20, 255.0, 0.0)

# Add some of the previous output. This recurrent formula

# is what gives the trace / tail.

output = img_motion + self.decay * self.last_output

# Ensure the pixel-values are within the allowed bounds.

output = np.clip(output, 0.0, 255.0)

# Set the last output.

self.last_output = output

return output

def get_state(self):

"""

# Stack the last input and output images.

state = np.dstack([self.last_input, self.last_output])

# Convert to 8-bit integer.

# This is done to save space in the replay-memory.

state = state.astype(np.uint8)

"""

def __init__(self, size, num_actions, discount_factor=0.97):

"""

# Array for the previous states of the game-environment.

self.states = np.zeros(shape=[size] + state_shape, dtype=np.uint8)

# Array for the Q-values corresponding to the states.

self.q_values = np.zeros(shape=[size, num_actions], dtype=np.float)

# Array for the Q-values before being updated.

# This is used to compare the Q-values before and after the update.

self.q_values_old = np.zeros(shape=[size, num_actions], dtype=np.float)

# Actions taken for each of the states in the memory.

self.actions = np.zeros(shape=size, dtype=np.int)

# Rewards observed for each of the states in the memory.

self.rewards = np.zeros(shape=size, dtype=np.float)

# Whether the life had ended in each state of the game-environment.

self.end_life = np.zeros(shape=size, dtype=np.bool)

# Whether the episode had ended (aka. game over) in each state.

self.end_episode = np.zeros(shape=size, dtype=np.bool)

# Estimation errors for the Q-values. This is used to balance

# the sampling of batches for training the Neural Network,

# so we get a balanced combination of states with high and low

# estimation errors for their Q-values.

self.estimation_errors = np.zeros(shape=size, dtype=np.float)

# Capacity of the replay-memory as the number of states.

self.size = size

# Discount-factor for calculating Q-values.

self.discount_factor = discount_factor

# Reset the number of used states in the replay-memory.

self.num_used = 0

# Threshold for splitting between low and high estimation errors.

self.error_threshold = 0.1

def is_full(self):

"""Return boolean whether the replay-memory is full."""

return self.num_used == self.size

def used_fraction(self):

"""Return the fraction of the replay-memory that is used."""

return self.num_used / self.size

def reset(self):

"""Reset the replay-memory so it is empty."""

self.num_used = 0

def add(self, state, q_values, action, reward, end_life, end_episode):

"""

if not self.is_full():

# Index into the arrays for convenience.

k = self.num_used

# Increase the number of used elements in the replay-memory.

self.num_used += 1

# Store all the values in the replay-memory.

self.states[k] = state

self.q_values[k] = q_values

self.actions[k] = action

self.end_life[k] = end_life

self.end_episode[k] = end_episode

# Note that the reward is limited. This is done to stabilize

# the training of the Neural Network.

self.rewards[k] = np.clip(reward, -1.0, 1.0)

def update_all_q_values(self):

"""

# Copy old Q-values so we can print their statistics later.

# Note that the contents of the arrays are copied.

self.q_values_old[:] = self.q_values[:]

# Process the replay-memory backwards and update the Q-values.

# This loop could be implemented entirely in NumPy for higher speed,

# but it is probably only a small fraction of the overall time usage,

# and it is much easier to understand when implemented like this.

for k in reversed(range(self.num_used-1)):

# Get the data for the k'th state in the replay-memory.

action = self.actions[k]

reward = self.rewards[k]

end_life = self.end_life[k]

end_episode = self.end_episode[k]

# Calculate the Q-value for the action that was taken in this state.

if end_life or end_episode:

# If the agent lost a life or it was game over / end of episode,

# then the value of taking the given action is just the reward

# that was observed in this single step. This is because the

# Q-value is defined as the discounted value of all future game

# steps in a single life of the agent. When the life has ended,

# there will be no future steps.

action_value = reward

else:

# Otherwise the value of taking the action is the reward that

# we have observed plus the discounted value of future rewards

# from continuing the game. We use the estimated Q-values for

# the following state and take the maximum, because we will

# generally take the action that has the highest Q-value.

action_value = reward + self.discount_factor * np.max(self.q_values[k + 1])

# Error of the Q-value that was estimated using the Neural Network.

self.estimation_errors[k] = abs(action_value - self.q_values[k, action])

# Update the Q-value with the better estimate.

self.q_values[k, action] = action_value

self.print_statistics()

def prepare_sampling_prob(self, batch_size=128):

"""

# Get the errors between the Q-values that were estimated using

# the Neural Network, and the Q-values that were updated with the

# reward that was actually observed when an action was taken.

err = self.estimation_errors[0:self.num_used]

# Create an index of the estimation errors that are low.

idx = err<self.error_threshold

self.idx_err_lo = np.squeeze(np.where(idx))

# Create an index of the estimation errors that are high.

self.idx_err_hi = np.squeeze(np.where(np.logical_not(idx)))

# Probability of sampling Q-values with high estimation errors.

# This is either set to the fraction of the replay-memory that

# has high estimation errors - or it is set to 0.5. So at least

# half of the batch has high estimation errors.

prob_err_hi = len(self.idx_err_hi) / self.num_used

prob_err_hi = max(prob_err_hi, 0.5)

# Number of samples in a batch that have high estimation errors.

self.num_samples_err_hi = int(prob_err_hi * batch_size)

# Number of samples in a batch that have low estimation errors.

self.num_samples_err_lo = batch_size - self.num_samples_err_hi

def random_batch(self):

"""

# Random index of states and Q-values in the replay-memory.

# These have LOW estimation errors for the Q-values.

idx_lo = np.random.choice(self.idx_err_lo,

size=self.num_samples_err_lo,

replace=False)

# Random index of states and Q-values in the replay-memory.

# These have HIGH estimation errors for the Q-values.

idx_hi = np.random.choice(self.idx_err_hi,

size=self.num_samples_err_hi,

replace=False)

# Combine the indices.

idx = np.concatenate((idx_lo, idx_hi))

# Get the batches of states and Q-values.

states_batch = self.states[idx]

q_values_batch = self.q_values[idx]

return states_batch, q_values_batch

def all_batches(self, batch_size=128):

"""

# Start index for the current batch.

begin = 0

# Repeat until all batches have been processed.

while begin < self.num_used:

# End index for the current batch.

end = begin + batch_size

# Ensure the batch does not exceed the used replay-memory.

if end > self.num_used:

end = self.num_used

# Progress counter.

progress = end / self.num_used

# Yield the batch indices and completion-counter.

yield begin, end, progress

# Set the start-index for the next batch to the end of this batch.

begin = end

def estimate_all_q_values(self, model):

"""

print("Re-calculating all Q-values in replay memory ...")

# Process the entire replay-memory in batches.

for begin, end, progress in self.all_batches():

# Print progress.

msg = "\tProgress: {0:.0%}"

msg = msg.format(progress)

print_progress(msg)

# Get the states for the current batch.

states = self.states[begin:end]

# Calculate the Q-values using the Neural Network

# and update the replay-memory.

self.q_values[begin:end] = model.get_q_values(states=states)

# Newline.

print()

def print_statistics(self):

"""Print statistics for the contents of the replay-memory."""

print("Replay-memory statistics:")

# Print statistics for the Q-values before they were updated

# in update_all_q_values().

msg = "\tQ-values Before, Min: {0:5.2f}, Mean: {1:5.2f}, Max: {2:5.2f}"

print(msg.format(np.min(self.q_values_old),

np.mean(self.q_values_old),

np.max(self.q_values_old)))

# Print statistics for the Q-values after they were updated

# in update_all_q_values().

msg = "\tQ-values After, Min: {0:5.2f}, Mean: {1:5.2f}, Max: {2:5.2f}"

print(msg.format(np.min(self.q_values),

np.mean(self.q_values),

np.max(self.q_values)))

# Print statistics for the difference in Q-values before and

# after the update in update_all_q_values().

q_dif = self.q_values - self.q_values_old

msg = "\tQ-values Diff., Min: {0:5.2f}, Mean: {1:5.2f}, Max: {2:5.2f}"

print(msg.format(np.min(q_dif),

np.mean(q_dif),

np.max(q_dif)))

# Print statistics for the number of large estimation errors.

# Don't use the estimation error for the last state in the memory,

# because its Q-values have not been updated.

err = self.estimation_errors[:-1]

err_count = np.count_nonzero(err > self.error_threshold)

msg = "\tNumber of large errors > {0}: {1} / {2} ({3:.1%})"

print(msg.format(self.error_threshold, err_count,

self.num_used, err_count / self.num_used))

# How much of the replay-memory is used by states with end_life.

end_life_pct = np.count_nonzero(self.end_life) / self.num_used

# How much of the replay-memory is used by states with end_episode.

end_episode_pct = np.count_nonzero(self.end_episode) / self.num_used

# How much of the replay-memory is used by states with non-zero reward.

reward_nonzero_pct = np.count_nonzero(self.rewards) / self.num_used

# Print those statistics.

msg = "\tend_life: {0:.1%}, end_episode: {1:.1%}, reward non-zero: {2:.1%}"

"""

def __init__(self, start_value, end_value, num_iterations, repeat=False):

"""

# Store arguments in this object.

self.start_value = start_value

self.end_value = end_value

self.num_iterations = num_iterations

self.repeat = repeat

# Calculate the linear coefficient.

self._coefficient = (end_value - start_value) / num_iterations

def get_value(self, iteration):

"""Get the value of the control signal for the given iteration."""

if self.repeat:

iteration %= self.num_iterations

if iteration < self.num_iterations:

value = iteration * self._coefficient + self.start_value

else:

value = self.end_value

"""