text: """
   Copyright (c) 2022 Intel Corporation
   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at
        http://www.apache.org/licenses/LICENSE-2.0
   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
  """
  
  
  from __future__ import absolute_import
  from collections import deque, namedtuple
  import warnings
  import random
  
  import numpy as np
  
  # [reference] https://github.com/matthiasplappert/keras-rl/blob/master/rl/memory.py
  
  # This is to be understood as a transition: Given `state0`, performing `action`
  # yields `reward` and results in `state1`, which might be `terminal`.
  Experience = namedtuple('Experience', 'state0, action, reward, state1, terminal1')
  
  
  def sample_batch_indexes(low, high, size):
      if high - low >= size:
          # We have enough data. Draw without replacement, that is each index is unique in the
          # batch. We cannot use `np.random.choice` here because it is horribly inefficient as
          # the memory grows. See https://github.com/numpy/numpy/issues/2764 for a discussion.
          # `random.sample` does the same thing (drawing without replacement) and is way faster.
          # try:
          #     r = xrange(low, high)
          # except NameError:
          r = range(low, high)
          batch_idxs = random.sample(r, size)
      else:
          # Not enough data. Help ourselves with sampling from the range, but the same index
          # can occur multiple times. This is not good and should be avoided by picking a
          # large enough warm-up phase.
          warnings.warn(
              'Not enough entries to sample without replacement.'
              'Consider increasing your warm-up phase to avoid oversampling!')
          batch_idxs = np.random.random_integers(low, high - 1, size=size)
      assert len(batch_idxs) == size
      return batch_idxs
  
  
  class RingBuffer:
      def __init__(self, maxlen):
          self.maxlen = maxlen
          self.length = 0
          self.data = []
  
      def __len__(self):
          return self.length
  
      def __getitem__(self, idx):
          if idx < 0 or idx >= self.length:
              raise KeyError()
          return self.data[idx]
  
      def __delitem__(self, subscript):
          del self.data[subscript]
          self.length = len(self.data)
  
      def append(self, v):
          if self.length < self.maxlen:
              # We have space, simply append to data.
              self.data.append(v)
              self.length+=1
  
          elif self.length == self.maxlen:
              # No space, remove the first item then append
              self.data.pop(0)
              self.data.append(v)
          else:
              # This should never happen.
              raise RuntimeError()
  
  
  def zeroed_observation(observation):
      if hasattr(observation, 'shape'):
          return np.zeros(observation.shape)
      if hasattr(observation, '__iter__'):
          out = []
          for x in observation:
              out.append(zeroed_observation(x))
          return out
      return 0.
  
  
  class Memory:
      def __init__(self, window_length, ignore_episode_boundaries=False):
          self.window_length = window_length
          self.ignore_episode_boundaries = ignore_episode_boundaries
  
          self.recent_observations = deque(maxlen=window_length)
          self.recent_terminals = deque(maxlen=window_length)
  
      def sample(self, batch_size, batch_idxs=None):
          raise NotImplementedError()
  
      def append(self, observation, action, reward, terminal, training=True):
          self.recent_observations.append(observation)
          self.recent_terminals.append(terminal)
  
      def get_recent_state(self, current_observation):
          # This code is slightly complicated by the fact that subsequent observations might be
          # from different episodes. We ensure that an experience never spans multiple episodes.
          # This is probably not that important in practice but it seems cleaner.
          state = [current_observation]
          idx = len(self.recent_observations) - 1
          for offset in range(0, self.window_length - 1):
              current_idx = idx - offset
              current_terminal = self.recent_terminals[current_idx - 1] if current_idx - 1 >= 0 else False
              if current_idx < 0 or (not self.ignore_episode_boundaries and current_terminal):
                  # The previously handled observation was terminal, don't add the current one.
                  # Otherwise we would leak into a different episode.
                  break
              state.insert(0, self.recent_observations[current_idx])
          while len(state) < self.window_length:
              state.insert(0, zeroed_observation(state[0]))
          return state
  
      def get_config(self):
          config = {
              'window_length': self.window_length,
              'ignore_episode_boundaries': self.ignore_episode_boundaries,
          }
          return config
  
  
  class SequentialMemory(Memory):
      def __init__(self, limit, **kwargs):
          super().__init__(**kwargs)
  
          self.limit = limit
  
          # Do not use deque to implement the memory. This data structure may seem convenient but
          # it is way too slow on random access. Instead, we use our own ring buffer implementation.
          self.actions = RingBuffer(limit)
          self.rewards = RingBuffer(limit)
          self.terminals = RingBuffer(limit)
          self.observations = RingBuffer(limit)
  
      def sample(self, batch_size, batch_idxs=None):
          if batch_idxs is None:
              # Draw random indexes such that we have at least a single entry before each
              # index.
              batch_idxs = sample_batch_indexes(0, self.nb_entries - 1, size=batch_size)
          batch_idxs = np.array(batch_idxs) + 1
          assert np.min(batch_idxs) >= 1
          assert np.max(batch_idxs) < self.nb_entries
          assert len(batch_idxs) == batch_size
  
          # Create experiences
          experiences = []
          for idx in batch_idxs:
              terminal0 = self.terminals[idx - 2] if idx >= 2 else False
              while terminal0:
                  # Skip this transition because the environment was reset here. Select a new, random
                  # transition and use this instead. This may cause the batch to contain the same
                  # transition twice.
                  idx = sample_batch_indexes(1, self.nb_entries, size=1)[0]
                  terminal0 = self.terminals[idx - 2] if idx >= 2 else False
              assert 1 <= idx < self.nb_entries
  
              # This code is slightly complicated by the fact that subsequent observations might be
              # from different episodes. We ensure that an experience never spans multiple episodes.
              # This is probably not that important in practice but it seems cleaner.
              state0 = [self.observations[idx - 1]]
              for offset in range(0, self.window_length - 1):
                  current_idx = idx - 2 - offset
                  current_terminal = self.terminals[current_idx - 1] if current_idx - 1 > 0 else False
                  if current_idx < 0 or (not self.ignore_episode_boundaries and current_terminal):
                      # The previously handled observation was terminal, don't add the current one.
                      # Otherwise we would leak into a different episode.
                      break
                  state0.insert(0, self.observations[current_idx])
              while len(state0) < self.window_length:
                  state0.insert(0, zeroed_observation(state0[0]))
              action = self.actions[idx - 1]
              reward = self.rewards[idx - 1]
              terminal1 = self.terminals[idx - 1]
  
              # Okay, now we need to create the follow-up state. This is state0 shifted on timestep
              # to the right. Again, we need to be careful to not include an observation from the next
              # episode if the last state is terminal.
              state1 = [np.copy(x) for x in state0[1:]]
              state1.append(self.observations[idx])
  
              assert len(state0) == self.window_length
              assert len(state1) == len(state0)
              experiences.append(Experience(state0=state0, action=action, reward=reward,
                                            state1=state1, terminal1=terminal1))
          assert len(experiences) == batch_size
          return experiences
  
      def sample_and_split(self, batch_size, batch_idxs=None):
          experiences = self.sample(batch_size, batch_idxs)
  
          state0_batch = []
          reward_batch = []
          action_batch = []
          terminal1_batch = []
          state1_batch = []
          for e in experiences:
              state0_batch.append(e.state0)
              state1_batch.append(e.state1)
              reward_batch.append(e.reward)
              action_batch.append(e.action)
              terminal1_batch.append(0. if e.terminal1 else 1.)
  
          # Prepare and validate parameters.
          state0_batch = np.array(state0_batch, 'double').reshape(batch_size, -1)
          state1_batch = np.array(state1_batch, 'double').reshape(batch_size, -1)
          terminal1_batch = np.array(terminal1_batch, 'double').reshape(batch_size, -1)
          reward_batch = np.array(reward_batch, 'double').reshape(batch_size, -1)
          action_batch = np.array(action_batch, 'double').reshape(batch_size, -1)
  
          return state0_batch, action_batch, reward_batch, state1_batch, terminal1_batch
  
      def append(self, observation, action, reward, terminal, training=True):
          super().append(observation, action, reward, terminal, training=training)
  
          # This needs to be understood as follows: in `observation`, take `action`, obtain `reward`
          # and weather the next state is `terminal` or not.
          if training:
              self.observations.append(observation)
              self.actions.append(action)
              self.rewards.append(reward)
              self.terminals.append(terminal)
  
      def discard(self, subscript):
          if not isinstance(subscript, slice):
              raise ValueError("discard function only supports input of type slice")
  
          del self.observations[subscript]
          del self.actions[subscript]
          del self.rewards[subscript]
          del self.terminals[subscript]
  
      @property
      def nb_entries(self):
          return len(self.observations)
  
      def get_config(self):
          config = super().get_config()
          config['limit'] = self.limit
          return config
  