<gh_stars>0
# Copyright (c) Microsoft. All rights reserved.

# Licensed under the MIT license. See LICENSE.md file in the project root
# for full license information.
# ==============================================================================

import numpy as np

class ReplayMemory(object):
    """
    ReplayMemory keeps track of the environment dynamic.
    We store all the transitions (s(t), action, s(t+1), reward, done).
    The replay memory allows us to efficiently sample minibatches from it, and generate the correct state representation
    (w.r.t the number of previous frames needed).
    """
    
    def __init__(self, size, sample_shape):
        self._pos = 0
        self._count = 0
        self._max_size = size
        self._state_shape = sample_shape
        self._states = np.zeros((size,) + sample_shape, dtype=np.float32)
        self._actions = np.zeros(size, dtype=np.uint8)
        self._rewards = np.zeros(size, dtype=np.float32)
        self._post_states = np.zeros((size,) + sample_shape, dtype=np.float32)
        self._terminals = np.zeros(size, dtype=np.float32)

    def __len__(self):
        """ Returns the number of items currently present in the memory
        Returns: Int >= 0
        """
        return self._count

    def append(self, state, action, reward, post_state, done):
        """ Appends the specified transition to the memory.

        Attributes:
            state (Tensor[sample_shape]): The state to append
            action (int): An integer representing the action done
            reward (float): An integer representing the reward received for doing this action
            post_state (Tensor[sample_shape]): The post state to append
            done (bool): A boolean specifying if this state is a terminal (episode has finished)
        """
        assert state.shape == self._state_shape, \
            'Invalid state shape (required: %s, got: %s)' % (self._state_shape, state.shape)

        self._states[self._pos] = state
        self._actions[self._pos] = action
        self._rewards[self._pos] = reward
        self._post_states[self._pos] = post_state
        self._terminals[self._pos] = done

        self._count = max(self._count, self._pos + 1)
        self._pos = (self._pos + 1) % self._max_size

    def sample(self, size):
        """ Generate size random integers mapping indices in the memory.
            The returned indices can be retrieved using #get_state().
            See the method #minibatch() if you want to retrieve samples directly.

        Attributes:
            size (int): The minibatch size

        Returns:
             Indexes of the sampled states ([int])
        """
        indexes = np.random.choice(np.arange(self._count), size=size, replace=False)
        return indexes

    def minibatch(self, size):
        """ Generate a minibatch with the number of samples specified by the size parameter.

        Attributes:
            size (int): Minibatch size

        Returns:
            tuple: Tensor[minibatch_size, input_shape...], [int], [float], [bool]
        """
        indexes = self.sample(size)

        pre_states = self._states[indexes]
        actions = self._actions[indexes]
        rewards = self._rewards[indexes]
        post_states = self._post_states[indexes]
        dones = self._terminals[indexes]

        return pre_states, actions, rewards, post_states, dones

# Copyright (c) Microsoft. All rights reserved.

# Licensed under the MIT license. See LICENSE.md file in the project root
# for full license information.
# ==============================================================================

import numpy as np

class ReplayMemory(object):
    """
    ReplayMemory keeps track of the environment dynamic.
    We store all the transitions (s(t), action, s(t+1), reward, done).
    The replay memory allows us to efficiently sample minibatches from it, and generate the correct state representation
    (w.r.t the number of previous frames needed).
    """
    
    def __init__(self, size, sample_shape):
        self._pos = 0
        self._count = 0
        self._max_size = size
        self._state_shape = sample_shape
        self._states = np.zeros((size,) + sample_shape, dtype=np.float32)
        self._actions = np.zeros(size, dtype=np.uint8)
        self._rewards = np.zeros(size, dtype=np.float32)
        self._post_states = np.zeros((size,) + sample_shape, dtype=np.float32)
        self._terminals = np.zeros(size, dtype=np.float32)

    def __len__(self):
        """ Returns the number of items currently present in the memory
        Returns: Int >= 0
        """
        return self._count

    def append(self, state, action, reward, post_state, done):
        """ Appends the specified transition to the memory.

        Attributes:
            state (Tensor[sample_shape]): The state to append
            action (int): An integer representing the action done
            reward (float): An integer representing the reward received for doing this action
            post_state (Tensor[sample_shape]): The post state to append
            done (bool): A boolean specifying if this state is a terminal (episode has finished)
        """
        assert state.shape == self._state_shape, \
            'Invalid state shape (required: %s, got: %s)' % (self._state_shape, state.shape)

        self._states[self._pos] = state
        self._actions[self._pos] = action
        self._rewards[self._pos] = reward
        self._post_states[self._pos] = post_state
        self._terminals[self._pos] = done

        self._count = max(self._count, self._pos + 1)
        self._pos = (self._pos + 1) % self._max_size

    def sample(self, size):
        """ Generate size random integers mapping indices in the memory.
            The returned indices can be retrieved using #get_state().
            See the method #minibatch() if you want to retrieve samples directly.

        Attributes:
            size (int): The minibatch size

        Returns:
             Indexes of the sampled states ([int])
        """
        indexes = np.random.choice(np.arange(self._count), size=size, replace=False)
        return indexes

    def minibatch(self, size):
        """ Generate a minibatch with the number of samples specified by the size parameter.

        Attributes:
            size (int): Minibatch size

        Returns:
            tuple: Tensor[minibatch_size, input_shape...], [int], [float], [bool]
        """
        indexes = self.sample(size)

        pre_states = self._states[indexes]
        actions = self._actions[indexes]
        rewards = self._rewards[indexes]
        post_states = self._post_states[indexes]
        dones = self._terminals[indexes]

        return pre_states, actions, rewards, post_states, dones
# Copyright (c) Microsoft. All rights reserved. # Licensed under the MIT license. See LICENSE.md file in the project root # for full license information. # ============================================================================== ReplayMemory keeps track of the environment dynamic.     We store all the transitions (s(t), action, s(t+1), reward, done).     The replay memory allows us to efficiently sample minibatches from it, and generate the correct state representation     (w.r.t the number of previous frames needed). Returns the number of items currently present in the memory         Returns: Int >= 0 Appends the specified transition to the memory.          Attributes:             state (Tensor[sample_shape]): The state to append             action (int): An integer representing the action done             reward (float): An integer representing the reward received for doing this action             post_state (Tensor[sample_shape]): The post state to append             done (bool): A boolean specifying if this state is a terminal (episode has finished) Generate size random integers mapping indices in the memory.             The returned indices can be retrieved using #get_state().             See the method #minibatch() if you want to retrieve samples directly.          Attributes:             size (int): The minibatch size          Returns:              Indexes of the sampled states ([int]) Generate a minibatch with the number of samples specified by the size parameter.          Attributes:             size (int): Minibatch size          Returns:             tuple: Tensor[minibatch_size, input_shape...], [int], [float], [bool]