code: def core_observe(self, *, terminal, reward, parallel):
      zero = tf_util.constant(value=0, dtype="int")
      one = tf_util.constant(value=1, dtype="int")
      buffer_index = tf.gather(
          params=self.buffer_index, indices=parallel
      )
      batch_size = tf_util.cast(
          x=tf.shape(input=terminal)[0], dtype="int"
      )
      expanded_parallel = tf.expand_dims(
          input=tf.expand_dims(input=parallel, axis=0), axis=1
      )
      if self.circular_buffer:
          buffer_start = tf.gather(
              params=self.buffer_start, indices=parallel
          )
  
      # Assertion: size of terminal equals number of buffered timesteps
      assertions = list()
      # if self.config.create_tf_assertions:
      #     if self.circular_buffer:
      #         maybe_one = tf.minimum(x=buffer_index, y=self.reward_horizon.value())
      #         assertions.append(tf.debugging.assert_equal(
      #             x=batch_size, y=(buffer_index - buffer_start - maybe_one),
      #             message="Agent.observe: number of observe-timesteps has to be equal to number "
      #                     "of buffered act-timesteps."
      #         ))
      #     else:
      #         assertions.append(tf.debugging.assert_equal(
      #             x=batch_size, y=buffer_index,
      #             message="Agent.observe: number of observe-timesteps has to be equal to number "
      #                     "of buffered act-timesteps."
      #         ))
  
      if self.config.buffer_observe == "episode":
          # Observe inputs are always buffered in agent until episode is terminated
          # --> Call core_experience directly, no need for terminal/reward buffers
  
          def fn_nonterminal():
              # Should not be called
              return tf.debugging.assert_equal(x=batch_size, y=zero)
  
          def fn_terminal():
              # Gather values from buffers, and episode experience
              function = lambda x: x[parallel, :buffer_index]
              states = self.states_buffer.fmap(
                  function=function, cls=TensorDict
              )
              internals = self.internals_buffer.fmap(
                  function=function, cls=TensorDict
              )
              auxiliaries = self.auxiliaries_buffer.fmap(
                  function=function, cls=TensorDict
              )
              actions = self.actions_buffer.fmap(
                  function=function, cls=TensorDict
              )
              return self.core_experience(
                  states=states,
                  internals=internals,
                  auxiliaries=auxiliaries,
                  actions=actions,
                  terminal=terminal,
                  reward=reward,
              )
  
      elif (
          self.reward_horizon == "episode"
          or self.parallel_interactions > 1
      ):
          # Observe inputs need to be buffered until episode is terminated
          # --> Call core_experience if terminal, otherwise buffer terminal/reward
          batch_parallel = tf.fill(dims=(batch_size,), value=parallel)
  
          def fn_nonterminal():
              # Update terminal/reward buffers
              assignments = list()
              indices = tf.range(
                  start=(buffer_index - batch_size), limit=buffer_index
              )
              indices = tf.stack(
                  values=(batch_parallel, indices), axis=1
              )
              value = tf.tensor_scatter_nd_update(
                  tensor=self.terminal_buffer,
                  indices=indices,
                  updates=terminal,
              )
              assignments.append(
                  self.terminal_buffer.assign(value=value)
              )
              value = tf.tensor_scatter_nd_update(
                  tensor=self.reward_buffer,
                  indices=indices,
                  updates=reward,
              )
              assignments.append(self.reward_buffer.assign(value=value))
              return tf.group(assignments)
  
          def fn_terminal():
              # Gather values from buffers, and episode experience
              function = lambda x: x[parallel, :buffer_index]
              states = self.states_buffer.fmap(
                  function=function, cls=TensorDict
              )
              internals = self.internals_buffer.fmap(
                  function=function, cls=TensorDict
              )
              auxiliaries = self.auxiliaries_buffer.fmap(
                  function=function, cls=TensorDict
              )
              actions = self.actions_buffer.fmap(
                  function=function, cls=TensorDict
              )
              episode_terminal = self.terminal_buffer[
                  parallel, : buffer_index - batch_size
              ]
              episode_reward = self.reward_buffer[
                  parallel, : buffer_index - batch_size
              ]
              episode_terminal = tf.concat(
                  values=(episode_terminal, terminal), axis=0
              )
              episode_reward = tf.concat(
                  values=(episode_reward, reward), axis=0
              )
              return self.core_experience(
                  states=states,
                  internals=internals,
                  auxiliaries=auxiliaries,
                  actions=actions,
                  terminal=episode_terminal,
                  reward=episode_reward,
              )
  
      else:
          # Observe inputs are buffered temporarily and return is computed as soon as possible
          # --> Call core_experience if terminal, otherwise ???
          capacity = tf_util.constant(
              value=self.buffer_capacity, dtype="int"
          )
          reward_horizon = self.reward_horizon.value()
          reward_discount = self.reward_discount.value()
          batch_parallel = tf.fill(dims=(batch_size,), value=parallel)
  
          def fn_nonterminal():
              # Update terminal/reward buffers
              assignments = list()
              indices = tf.range(
                  start=(buffer_index - batch_size), limit=buffer_index
              )
              indices = tf.math.mod(x=indices, y=capacity)
              indices = tf.stack(
                  values=(batch_parallel, indices), axis=1
              )
              value = tf.tensor_scatter_nd_update(
                  tensor=self.terminal_buffer,
                  indices=indices,
                  updates=terminal,
              )
              assignments.append(
                  self.terminal_buffer.assign(value=value)
              )
              value = tf.tensor_scatter_nd_update(
                  tensor=self.reward_buffer,
                  indices=indices,
                  updates=reward,
              )
              assignments.append(self.reward_buffer.assign(value=value))
              with tf.control_dependencies(control_inputs=assignments):
                  # Number of completed timesteps to process
                  num_complete = (
                      buffer_index - buffer_start - reward_horizon
                  )
  
                  def true_fn():
                      return self._nonterminal_experience(
                          parallel=parallel,
                          buffer_start=buffer_start,
                          buffer_index=buffer_index,
                          reward_horizon=reward_horizon,
                          num_complete=num_complete,
                          reward_discount=reward_discount,
                      )
  
                  return tf.cond(
                      pred=(num_complete > zero),
                      true_fn=true_fn,
                      false_fn=tf.no_op,
                  )
  
          def fn_terminal():
              # Gather values from buffers
              indices = tf.range(start=buffer_start, limit=buffer_index)
              indices = tf.math.mod(x=indices, y=capacity)
              function = lambda x: tf.gather(
                  params=x[parallel], indices=indices
              )
              states = self.states_buffer.fmap(
                  function=function, cls=TensorDict
              )
              internals = self.internals_buffer.fmap(
                  function=function, cls=TensorDict
              )
              auxiliaries = self.auxiliaries_buffer.fmap(
                  function=function, cls=TensorDict
              )
              actions = self.actions_buffer.fmap(
                  function=function, cls=TensorDict
              )
              indices = tf.range(
                  buffer_start, buffer_index - batch_size
              )
              indices = tf.math.mod(x=indices, y=capacity)
              episode_terminal = tf.gather(
                  params=self.terminal_buffer[parallel], indices=indices
              )
              episode_reward = tf.gather(
                  params=self.reward_buffer[parallel], indices=indices
              )
              episode_terminal = tf.concat(
                  values=(episode_terminal, terminal), axis=0
              )
              episode_reward = tf.concat(
                  values=(episode_reward, reward), axis=0
              )
  
              # Episode experience
              experienced = self.core_experience(
                  states=states,
                  internals=internals,
                  auxiliaries=auxiliaries,
                  actions=actions,
                  terminal=episode_terminal,
                  reward=episode_reward,
              )
  
              # Increment buffer start index
              with tf.control_dependencies(control_inputs=(indices,)):
                  zeros = tf_util.zeros(shape=(1,), dtype="int")
                  value = tf.tensor_scatter_nd_update(
                      tensor=self.buffer_start,
                      indices=expanded_parallel,
                      updates=zeros,
                  )
                  assignment = self.buffer_start.assign(value=value)
                  # sparse_delta = tf.IndexedSlices(values=zero, indices=parallel)
                  # assignment = self.buffer_start.scatter_update(sparse_delta=sparse_delta)
  
              return tf.group((experienced, assignment))
  
      def fn_terminal_continuation():
          # Appropriate terminal function above
          operations = [fn_terminal()]
  
          # Reset buffer index
          with tf.control_dependencies(control_inputs=operations):
              updates = tf_util.zeros(shape=(1,), dtype="int")
              indices = tf.expand_dims(
                  input=tf.expand_dims(input=parallel, axis=0), axis=1
              )
              value = tf.tensor_scatter_nd_update(
                  tensor=self.buffer_index,
                  indices=indices,
                  updates=updates,
              )
              operations.append(self.buffer_index.assign(value=value))
              # sparse_delta = tf.IndexedSlices(values=zero, indices=parallel)
              # operations.append(self.buffer_index.scatter_update(sparse_delta=sparse_delta))
  
          # Preprocessed episode reward summaries (before preprocessed episode reward reset)
          if self.reward_processing is not None:
              dependencies = list()
              if (
                  self.summaries == "all"
                  or "reward" in self.summaries
                  or self.tracking == "all"
                  or "reward" in self.tracking
              ):
                  if (
                      self.summaries == "all"
                      or "reward" in self.summaries
                  ):
                      summarizer = self.summarizer.as_default()
                      summarizer.__enter__()
                  else:
                      summarizer = None
                  x = tf.gather(
                      params=self.preprocessed_episode_return,
                      indices=parallel,
                  )
                  if summarizer is not None:
                      dependencies.append(
                          tf.summary.scalar(
                              name="preprocessed-episode-return",
                              data=x,
                              step=self.episodes,
                          )
                      )
                  dependencies.extend(
                      self.track(
                          label="reward",
                          name="preprocessed-episode-return",
                          data=x,
                      )
                  )
                  if summarizer is not None:
                      summarizer.__exit__(None, None, None)
  
              # Reset preprocessed episode reward
              with tf.control_dependencies(control_inputs=dependencies):
                  zeros = tf_util.zeros(shape=(1,), dtype="float")
                  value = tf.tensor_scatter_nd_update(
                      tensor=self.preprocessed_episode_return,
                      indices=expanded_parallel,
                      updates=zeros,
                  )
                  operations.append(
                      self.preprocessed_episode_return.assign(
                          value=value
                      )
                  )
                  # zero_float = tf_util.constant(value=0.0, dtype='float')
                  # sparse_delta = tf.IndexedSlices(values=zero_float, indices=parallel)
                  # operations.append(
                  #     self.preprocessed_episode_return.scatter_update(sparse_delta=sparse_delta)
                  # )
  
          # Reset preprocessors
          for preprocessor in self.state_preprocessing.values():
              operations.append(preprocessor.reset())
          if self.reward_processing is not None:
              operations.append(self.reward_processing.reset())
  
          return tf.group(*operations)
  
      # Reward preprocessing
      dependencies = assertions
      if self.reward_processing is not None:
          with tf.control_dependencies(control_inputs=dependencies):
              dependencies = list()
              true = tf_util.constant(value=True, dtype="bool")
              reward = self.reward_processing.apply(
                  x=reward, deterministic=true, independent=False
              )
  
              # Preprocessed reward summary
              if (
                  self.summaries == "all"
                  or "reward" in self.summaries
                  or self.tracking == "all"
                  or "reward" in self.tracking
              ):
                  if (
                      self.summaries == "all"
                      or "reward" in self.summaries
                  ):
                      summarizer = self.summarizer.as_default()
                      summarizer.__enter__()
                  else:
                      summarizer = None
                  x = tf.math.reduce_mean(input_tensor=reward, axis=0)
                  if summarizer is not None:
                      dependencies.append(
                          tf.summary.scalar(
                              name="preprocessed-reward",
                              data=x,
                              step=self.timesteps,
                          )
                      )
                  dependencies.extend(
                      self.track(
                          label="reward",
                          name="preprocessed-reward",
                          data=x,
                      )
                  )
                  if summarizer is not None:
                      summarizer.__exit__(None, None, None)
  
              # Update preprocessed episode reward
              sum_reward = tf.math.reduce_sum(
                  input_tensor=reward, keepdims=True
              )
              value = tf.tensor_scatter_nd_add(
                  tensor=self.preprocessed_episode_return,
                  indices=expanded_parallel,
                  updates=sum_reward,
              )
              dependencies.append(
                  self.preprocessed_episode_return.assign(value=value)
              )
              # sum_reward = tf.math.reduce_sum(input_tensor=reward)
              # sparse_delta = tf.IndexedSlices(values=sum_reward, indices=parallel)
              # dependencies.append(
              #     self.preprocessed_episode_return.scatter_add(sparse_delta=sparse_delta)
              # )
  
      # Handle terminal vs non-terminal (after preprocessed episode reward)
      with tf.control_dependencies(control_inputs=dependencies):
          is_terminal = (
              tf.concat(values=([zero], terminal), axis=0)[-1] > zero
          )
          experienced = tf.cond(
              pred=is_terminal,
              true_fn=fn_terminal_continuation,
              false_fn=fn_nonterminal,
          )
  
      # Handle periodic update
      with tf.control_dependencies(control_inputs=(experienced,)):
          if self.update_frequency is None:
              updated = tf_util.constant(value=False, dtype="bool")
  
          else:
              frequency = self.update_frequency.value()
              start = self.update_start.value()
  
              if self.update_unit == "timesteps":
                  # Timestep-based batch
                  past_horizon = tf.math.maximum(
                      x=self.policy.past_horizon(on_policy=False),
                      y=self.baseline.past_horizon(on_policy=False),
                  )
                  unit = self.timesteps
                  start = tf.math.maximum(
                      x=start, y=(frequency + past_horizon)
                  )
                  if self.reward_horizon == "episode":
                      min_start = tf.where(
                          condition=(self.episodes > zero),
                          x=start,
                          y=(unit + one),
                      )
                      start = tf.math.maximum(x=start, y=min_start)
                  else:
                      two = tf_util.constant(value=2, dtype="int")
                      start = tf.where(
                          condition=(self.episodes > zero),
                          x=zero,
                          y=(start + two * self.reward_horizon.value()),
                      )
                  if self.config.buffer_observe == "episode":
                      min_start = tf.where(
                          condition=(self.episodes > zero),
                          x=start,
                          y=(unit + one),
                      )
                      start = tf.math.maximum(x=start, y=min_start)
                  else:
                      buffer_observe = tf_util.constant(
                          value=self.config.buffer_observe, dtype="int"
                      )
                      start = tf.math.maximum(x=start, y=buffer_observe)
  
              elif self.update_unit == "episodes":
                  # Episode-based batch
                  start = tf.math.maximum(x=start, y=frequency)
                  # (Episode counter is only incremented at the end of observe)
                  unit = self.episodes + tf.where(
                      condition=is_terminal, x=one, y=zero
                  )
  
              unit = unit - start
              is_frequency = tf.math.greater_equal(
                  x=unit, y=(self.last_update + frequency)
              )
  
              def perform_update():
                  assignment = self.last_update.assign(
                      value=unit, read_value=False
                  )
                  with tf.control_dependencies(
                      control_inputs=(assignment,)
                  ):
                      return self.core_update()
  
              def no_update():
                  return tf_util.constant(value=False, dtype="bool")
  
              updated = tf.cond(
                  pred=is_frequency,
                  true_fn=perform_update,
                  false_fn=no_update,
              )
  
      with tf.control_dependencies(control_inputs=(updated,)):
          return tf_util.identity(input=updated)
  
imports: import logging
  import tensorflow
  from tensorforce import TensorforceError
  from tensorforce.core import ModuleDict, memory_modules, optimizer_modules, parameter_modules, SignatureDict, TensorDict, TensorSpec, TensorsSpec, tf_function, tf_util, VariableDict
  from tensorforce.core.models import Model
  from tensorforce.core.networks import Preprocessor
  from tensorforce.core.objectives import objective_modules
  from tensorforce.core.policies import policy_modules, StochasticPolicy
label_desc: Use image node_id 13 for calling the TensorforceModel obj's underlying member method code with example usage: obj.core_observe() and returns: tf, tf_util, tf, self, tf, self, tf, tf_util, tf, self, self