class_docstr: Recorder wrapper (specification key: `recorder`).
  
  Args:
      fn_act (callable[states -> actions]): Act-function mapping states to actions which is
          supposed to be recorded.
class_docstr_tok: ["Recorder","wrapper","(","specification","key",":","`","recorder","`",")",".","Args",":","fn_act","(","callable","[","states","-",">","actions","]",")",":","Act-function","mapping","states","to","actions","which","is","supposed","to","be","recorded","."]
code: def observe(self, reward=0.0, terminal=False, parallel=0):
      # Check whether inputs are batched
      if util.is_iterable(x=reward) or (
          isinstance(reward, np.ndarray) and reward.ndim > 0
      ):
          reward = np.asarray(reward)
          num_parallel = reward.shape[0]
          if not isinstance(terminal, np.ndarray) and terminal is False:
              terminal = np.asarray([0 for _ in range(num_parallel)])
          else:
              terminal = np.asarray(terminal)
          if not isinstance(parallel, np.ndarray) and parallel == 0:
              assert num_parallel == self.parallel_interactions
              parallel = np.asarray(list(range(num_parallel)))
          else:
              parallel = np.asarray(parallel)
  
      elif util.is_iterable(x=terminal) or (
          isinstance(terminal, np.ndarray) and terminal.ndim > 0
      ):
          terminal = np.asarray(
              terminal, dtype=util.np_dtype(dtype="int")
          )
          num_parallel = terminal.shape[0]
          if not isinstance(reward, np.ndarray) and reward == 0.0:
              reward = np.asarray([0.0 for _ in range(num_parallel)])
          else:
              reward = np.asarray(reward)
          if not isinstance(parallel, np.ndarray) and parallel == 0:
              assert num_parallel == self.parallel_interactions
              parallel = np.asarray(list(range(num_parallel)))
          else:
              parallel = np.asarray(parallel)
  
      elif util.is_iterable(x=parallel) or (
          isinstance(parallel, np.ndarray) and parallel.ndim > 0
      ):
          parallel = np.asarray(parallel)
          num_parallel = parallel.shape[0]
          if not isinstance(reward, np.ndarray) and reward == 0.0:
              reward = np.asarray([0.0 for _ in range(num_parallel)])
          else:
              reward = np.asarray(reward)
          if not isinstance(terminal, np.ndarray) and terminal is False:
              terminal = np.asarray([0 for _ in range(num_parallel)])
          else:
              terminal = np.asarray(terminal)
  
      else:
          reward = np.asarray([float(reward)])
          terminal = np.asarray([int(terminal)])
          parallel = np.asarray([int(parallel)])
          num_parallel = 1
  
      # Check whether shapes/lengths are consistent
      if parallel.shape[0] == 0:
          raise TensorforceError.value(
              name="Agent.observe",
              argument="len(parallel)",
              value=parallel.shape[0],
              hint="= 0",
          )
      if reward.shape != parallel.shape:
          raise TensorforceError.value(
              name="Agent.observe",
              argument="len(reward)",
              value=reward.shape,
              hint="!= parallel length",
          )
      if terminal.shape != parallel.shape:
          raise TensorforceError.value(
              name="Agent.observe",
              argument="len(terminal)",
              value=terminal.shape,
              hint="!= parallel length",
          )
  
      # Convert terminal to int if necessary
      if terminal.dtype is util.np_dtype(dtype="bool"):
          zeros = np.zeros_like(
              terminal, dtype=util.np_dtype(dtype="int")
          )
          ones = np.ones_like(
              terminal, dtype=util.np_dtype(dtype="int")
          )
          terminal = np.where(terminal, ones, zeros)
  
      # Check whether current timesteps are not completed
      if self.timestep_completed[parallel].any():
          raise TensorforceError(
              message="Calling agent.observe must be preceded by agent.act."
          )
      self.timestep_completed[parallel] = True
  
      # Check whether episode is too long
      self.timestep_counter[parallel] += 1
      if (
          self.max_episode_timesteps is not None
          and np.logical_and(
              terminal == 0,
              self.timestep_counter[parallel]
              > self.max_episode_timesteps,
          ).any()
      ):
          raise TensorforceError(
              message="Episode longer than max_episode_timesteps."
          )
      self.timestep_counter[parallel] = np.where(
          terminal > 0, 0, self.timestep_counter[parallel]
      )
  
      if self.recorder is None:
          pass
  
      elif self.num_episodes < self.recorder.get("start", 0):
          # Increment num_episodes
          for t in terminal.tolist():
              if t > 0:
                  self.num_episodes += 1
  
      else:
          # Store values per parallel interaction
          for p, t, r in zip(
              parallel.tolist(), terminal.tolist(), reward.tolist()
          ):
              # Buffer inputs
              self.buffers["terminal"][p].append(t)
              self.buffers["reward"][p].append(r)
  
              # Continue if not terminal
              if t == 0:
                  continue
              self.num_episodes += 1
  
              # Buffered terminal/reward inputs
              for name in self.states_spec:
                  self.recorded["states"][name].append(
                      np.stack(self.buffers["states"][name][p], axis=0)
                  )
                  self.buffers["states"][name][p].clear()
              for name, spec in self.actions_spec.items():
                  self.recorded["actions"][name].append(
                      np.stack(self.buffers["actions"][name][p], axis=0)
                  )
                  self.buffers["actions"][name][p].clear()
              self.recorded["terminal"].append(
                  np.array(
                      self.buffers["terminal"][p],
                      dtype=self.terminal_spec.np_type(),
                  )
              )
              self.buffers["terminal"][p].clear()
              self.recorded["reward"].append(
                  np.array(
                      self.buffers["reward"][p],
                      dtype=self.reward_spec.np_type(),
                  )
              )
              self.buffers["reward"][p].clear()
  
              # Check whether recording step
              if (
                  self.num_episodes - self.recorder.get("start", 0)
              ) % self.recorder.get("frequency", 1) != 0:
                  continue
  
              # Manage recorder directory
              directory = self.recorder["directory"]
              if os.path.isdir(directory):
                  files = sorted(
                      f
                      for f in os.listdir(directory)
                      if os.path.isfile(os.path.join(directory, f))
                      and os.path.splitext(f)[1] == ".npz"
                  )
              else:
                  os.makedirs(directory)
                  files = list()
              max_traces = self.recorder.get("max-traces")
              if max_traces is not None and len(files) > max_traces - 1:
                  for filename in files[: -max_traces + 1]:
                      filename = os.path.join(directory, filename)
                      os.remove(filename)
  
              # Write recording file
              filename = os.path.join(
                  directory,
                  "trace-{:09d}.npz".format(self.num_episodes - 1),
              )
              # time.strftime('%Y%m%d-%H%M%S')
              kwargs = self.recorded.fmap(
                  function=np.concatenate, cls=ArrayDict
              ).items()
              np.savez_compressed(file=filename, **dict(kwargs))
  
              # Clear recorded values
              for recorded in self.recorded.values():
                  recorded.clear()
  
      if self._is_agent:
          return reward, terminal, parallel
      else:
          return 0
  
file_path: tensorforce/tensorforce/agents/recorder.py
imports: import os
  from collections import OrderedDict
  import numpy
  from tensorforce import TensorforceError, util
  from tensorforce.core import ArrayDict, ListDict, TensorSpec, TensorsSpec
label_desc: Use image node_id 7 for calling the Recorder obj's underlying member method code with example usage: obj.observe(reward, terminal, parallel) and returns: reward,  terminal,  parallel, int