code: def observe(self, *, terminal, reward, parallel):
      zero = tf_util.constant(value=0, dtype="int")
      one = tf_util.constant(value=1, dtype="int")
      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
      )
      is_terminal = tf.math.greater(x=terminal[-1], y=zero)
  
      # Input assertions
      assertions = list()
      if self.config.create_tf_assertions:
          assertions.extend(
              self.terminal_spec.tf_assert(
                  x=terminal,
                  batch_size=batch_size,
                  message="Agent.observe: invalid {issue} for terminal input.",
              )
          )
          assertions.extend(
              self.reward_spec.tf_assert(
                  x=reward,
                  batch_size=batch_size,
                  message="Agent.observe: invalid {issue} for terminal input.",
              )
          )
          assertions.extend(
              self.parallel_spec.tf_assert(
                  x=parallel,
                  message="Agent.observe: invalid {issue} for parallel input.",
              )
          )
          # Assertion: at most one terminal
          num_terms = tf.math.count_nonzero(
              input=terminal, dtype=tf_util.get_dtype(type="int")
          )
          assertions.append(
              tf.debugging.assert_less_equal(
                  x=num_terms,
                  y=one,
                  message="Agent.observe: input contains more than one terminal.",
              )
          )
          # Assertion: if terminal, last timestep in batch
          assertions.append(
              tf.debugging.assert_equal(
                  x=tf.math.greater(x=num_terms, y=zero),
                  y=is_terminal,
                  message="Agent.observe: terminal is not the last input timestep.",
              )
          )
  
      with tf.control_dependencies(control_inputs=assertions):
          dependencies = list()
  
          # Reward summary
          if self.summaries == "all" or "reward" in self.summaries:
              with self.summarizer.as_default():
                  x = tf.math.reduce_mean(input_tensor=reward)
                  dependencies.append(
                      tf.summary.scalar(
                          name="reward", data=x, step=self.timesteps
                      )
                  )
  
          # Update episode length/reward
          updates = tf.expand_dims(input=batch_size, axis=0)
          value = tf.tensor_scatter_nd_add(
              tensor=self.episode_length,
              indices=expanded_parallel,
              updates=updates,
          )
          dependencies.append(self.episode_length.assign(value=value))
          # sparse_delta = tf.IndexedSlices(values=batch_size, indices=parallel)
          # dependencies.append(self.episode_length.scatter_add(sparse_delta=sparse_delta))
          sum_reward = tf.math.reduce_sum(
              input_tensor=reward, keepdims=True
          )
          value = tf.tensor_scatter_nd_add(
              tensor=self.episode_return,
              indices=expanded_parallel,
              updates=sum_reward,
          )
          dependencies.append(self.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.episode_return.scatter_add(sparse_delta=sparse_delta))
  
          # Core observe (before terminal handling)
          updated = self.core_observe(
              terminal=terminal, reward=reward, parallel=parallel
          )
          dependencies.append(updated)
  
      # Handle terminal (after core observe and episode reward)
      with tf.control_dependencies(control_inputs=dependencies):
  
          def fn_terminal():
              operations = list()
  
              # Reset internals
              def function(spec, initial):
                  return tf_util.constant(
                      value=initial, dtype=spec.type
                  )
  
              initials = self.internals_spec.fmap(
                  function=function,
                  cls=TensorDict,
                  zip_values=self.initial_internals,
              )
              for (
                  name,
                  previous,
                  initial,
              ) in self.previous_internals.zip_items(initials):
                  updates = tf.expand_dims(input=initial, axis=0)
                  value = tf.tensor_scatter_nd_update(
                      tensor=previous,
                      indices=expanded_parallel,
                      updates=updates,
                  )
                  operations.append(previous.assign(value=value))
                  # sparse_delta = tf.IndexedSlices(values=initial, indices=parallel)
                  # operations.append(previous.scatter_update(sparse_delta=sparse_delta))
  
              # Episode length/reward summaries (before episode reward reset / episodes increment)
              dependencies = list()
              if self.summaries == "all" or "reward" in self.summaries:
                  with self.summarizer.as_default():
                      x = tf.gather(
                          params=self.episode_length, indices=parallel
                      )
                      dependencies.append(
                          tf.summary.scalar(
                              name="episode-length",
                              data=x,
                              step=self.episodes,
                          )
                      )
                      x = tf.gather(
                          params=self.episode_return, indices=parallel
                      )
                      dependencies.append(
                          tf.summary.scalar(
                              name="episode-return",
                              data=x,
                              step=self.episodes,
                          )
                      )
  
              # Reset episode length/reward
              with tf.control_dependencies(control_inputs=dependencies):
                  zeros = tf_util.zeros(shape=(1,), dtype="int")
                  value = tf.tensor_scatter_nd_update(
                      tensor=self.episode_length,
                      indices=expanded_parallel,
                      updates=zeros,
                  )
                  operations.append(
                      self.episode_length.assign(value=value)
                  )
                  # sparse_delta = tf.IndexedSlices(values=zero, indices=parallel)
                  # operations.append(self.episode_length.scatter_update(sparse_delta=sparse_delta))
                  zeros = tf_util.zeros(shape=(1,), dtype="float")
                  value = tf.tensor_scatter_nd_update(
                      tensor=self.episode_return,
                      indices=expanded_parallel,
                      updates=zeros,
                  )
                  operations.append(
                      self.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.episode_return.scatter_update(sparse_delta=sparse_delta))
  
              # Increment episodes counter
              operations.append(
                  self.episodes.assign_add(delta=one, read_value=False)
              )
  
              return tf.group(*operations)
  
          handle_terminal = tf.cond(
              pred=is_terminal, true_fn=fn_terminal, false_fn=tf.no_op
          )
  
      with tf.control_dependencies(control_inputs=(handle_terminal,)):
          episodes = tf_util.identity(input=self.episodes)
          updates = tf_util.identity(input=self.updates)
          return updated, episodes, updates
  
imports: from collections import OrderedDict
  import logging
  import os
  import time
  import h5py
  import numpy
  import tensorflow
  from tensorforce import TensorforceError
  from tensorforce.core import ArrayDict, Module, SignatureDict, TensorDict, TensorSpec, TensorsSpec, tf_function, tf_util, VariableDict
  from tensorforce.core.layers import Layer
label_desc: Use image node_id 18 for calling the Model obj's underlying member method code with example usage: obj.observe() and returns: updated,  episodes,  updates, tf, tf_util