code: def test_explicit(self):
      # FEATURES.MD
      self.start_tests(name="explicit")
  
      with TemporaryDirectory() as directory:
          update = dict(unit="episodes", batch_size=1)
          agent, environment = self.prepare(
              memory=50,
              update=update,
              config=dict(
                  device="CPU",
                  eager_mode=False,
                  create_debug_assertions=True,
                  tf_log_level=20,
              ),
          )
          states = environment.reset()
  
          # save: default checkpoint format
          weights0 = agent.model.policy.network.layers[0][
              1
          ].weights.numpy()
          agent.save(directory=directory)
          actions = agent.act(states=states)
          states, terminal, reward = environment.execute(
              actions=actions
          )
          agent.observe(terminal=terminal, reward=reward)
          self.assertEqual(agent.timesteps, 1)
          agent.close()
          self.finished_test()
  
          # load: only directory
          agent = Agent.load(
              directory=directory, environment=environment
          )
          x = agent.model.policy.network.layers[0][1].weights.numpy()
          self.assertTrue(np.allclose(x, weights0))
          self.assertEqual(agent.timesteps, 0)
          self.finished_test()
  
          # one timestep
          actions = agent.act(states=states)
          states, terminal, reward = environment.execute(
              actions=actions
          )
  
          # save: numpy format, append timesteps
          agent.save(
              directory=directory, format="numpy", append="timesteps"
          )
          agent.close()
          self.finished_test()
  
          # load: numpy format and directory
          agent = Agent.load(
              directory=directory,
              format="numpy",
              environment=environment,
          )
          x = agent.model.policy.network.layers[0][1].weights.numpy()
          self.assertTrue(np.allclose(x, weights0))
          self.assertEqual(agent.timesteps, 1)
          self.finished_test()
  
          # one timestep
          actions = agent.act(states=states)
          states, terminal, reward = environment.execute(
              actions=actions
          )
          agent.observe(terminal=terminal, reward=reward)
  
          # save: numpy format, append timesteps
          agent.save(
              directory=directory, format="numpy", append="timesteps"
          )
          agent.close()
          self.finished_test()
  
          # load: numpy format and directory
          agent = Agent.load(
              directory=directory,
              format="numpy",
              environment=environment,
          )
          x = agent.model.policy.network.layers[0][1].weights.numpy()
          self.assertTrue(np.allclose(x, weights0))
          self.assertEqual(agent.timesteps, 2)
          self.finished_test()
  
          # one episode
          while not terminal:
              actions = agent.act(states=states)
              states, terminal, reward = environment.execute(
                  actions=actions
              )
              agent.observe(terminal=terminal, reward=reward)
  
          # save: hdf5 format, filename, append episodes
          weights1 = agent.model.policy.network.layers[0][
              1
          ].weights.numpy()
          self.assertTrue(not np.allclose(weights1, weights0))
          self.assertEqual(agent.episodes, 1)
          agent.save(
              directory=directory,
              filename="agent2",
              format="hdf5",
              append="episodes",
          )
          agent.close()
          self.finished_test()
  
          # env close
          environment.close()
  
          # differing agent config: update, parallel_interactions
          # TODO: episode length, others?
          environment = Environment.create(
              environment=self.environment_spec()
          )
  
          # load: filename (hdf5 format implicit)
          update["batch_size"] = 2
          agent = Agent.load(
              directory=directory,
              filename="agent2",
              environment=environment,
              update=update,
              parallel_interactions=2,
          )
          x = agent.model.policy.network.layers[0][1].weights.numpy()
          self.assertTrue(np.allclose(x, weights1))
          self.assertEqual(agent.episodes, 1)
          agent.close()
          self.finished_test()
  
          # load: tensorflow format (filename explicit)
          # TODO: parallel_interactions=2 should be possible, but problematic if all variables are
          # saved in checkpoint format
          agent = Agent.load(
              directory=directory,
              format="checkpoint",
              environment=environment,
              update=update,
              parallel_interactions=1,
          )
          x = agent.model.policy.network.layers[0][1].weights.numpy()
          self.assertTrue(np.allclose(x, weights0))
          self.assertEqual(agent.timesteps, 0)
          self.assertEqual(agent.episodes, 0)
          agent.close()
          self.finished_test()
  
          # load: numpy format, full filename including timesteps suffix
          agent = Agent.load(
              directory=directory,
              filename="agent-1",
              format="numpy",
              environment=environment,
              update=update,
              parallel_interactions=2,
          )
          x = agent.model.policy.network.layers[0][1].weights.numpy()
          self.assertTrue(np.allclose(x, weights0))
          self.assertEqual(agent.timesteps, 1)
          self.assertEqual(agent.episodes, 0)
          self.finished_test()
  
          # three episodes (due to batch_size change, mismatch with loaded internal last_update)
          for _ in range(3):
              states = environment.reset()
              terminal = False
              while not terminal:
                  actions = agent.act(states=states)
                  states, terminal, reward = environment.execute(
                      actions=actions
                  )
                  agent.observe(terminal=terminal, reward=reward)
          self.assertEqual(agent.updates, 1)
  
          # save: saved-model format, append updates
          agent.save(
              directory=directory,
              format="saved-model",
              append="updates",
          )
          agent.close()
  
          # saved-model functions
          def batch(x):
              return np.expand_dims(x, axis=0)
  
          def unbatch(x):
              if isinstance(x, tf.Tensor):
                  x = x.numpy()
              if x.shape == (1,):
                  return x.item()
              else:
                  return np.squeeze(x, axis=0)
  
          def recursive_map(function, dictionary):
              mapped = dict()
              for key, value in dictionary.items():
                  if isinstance(value, dict):
                      mapped[key] = recursive_map(function, value)
                  else:
                      mapped[key] = function(value)
              return mapped
  
          # load: saved-model format
          agent = tf.saved_model.load(
              export_dir=os.path.join(directory, "agent-1")
          )
  
          # one episode
          states = environment.reset()
          internals = agent.initial_internals()
          internals = recursive_map(batch, internals)
          terminal = False
          while not terminal:
              states = dict(states)
              auxiliaries = dict(
                  int_action1=dict(
                      mask=batch(states.pop("int_action1_mask"))
                  ),
                  int_action2=dict(
                      mask=batch(states.pop("int_action2_mask"))
                  ),
                  int_action3=dict(
                      mask=batch(states.pop("int_action3_mask"))
                  ),
              )
              states = recursive_map(batch, states)
              actions_internals = agent.act(
                  states, internals, auxiliaries, False
              )
              actions = actions_internals["actions"]
              internals = actions_internals["internals"]
              actions = recursive_map(unbatch, actions)
              states, terminal, _ = environment.execute(actions=actions)
  
          environment.close()
  
          # saved-model format with singleton state/action, no internals, no masking
          policy = dict(
              network=dict(type="auto", size=8, depth=1, rnn=False)
          )
          update = dict(unit="episodes", batch_size=1)
          baseline = dict(
              network=dict(type="auto", size=7, depth=1, rnn=False)
          )
          agent, environment = self.prepare(
              states=dict(
                  type="float", shape=(), min_value=1.0, max_value=2.0
              ),
              actions=dict(
                  type="float", shape=(), min_value=1.0, max_value=2.0
              ),
              policy=policy,
              update=update,
              baseline=baseline,
              config=dict(
                  device="CPU",
                  eager_mode=False,
                  create_debug_assertions=True,
                  tf_log_level=20,
              ),
          )
  
          # one episode
          states = environment.reset()
          terminal = False
          while not terminal:
              actions = agent.act(states=states)
              states, terminal, reward = environment.execute(
                  actions=actions
              )
              agent.observe(terminal=terminal, reward=reward)
          self.assertEqual(agent.updates, 1)
  
          # save: saved-model format, append updates
          agent.save(
              directory=directory,
              format="saved-model",
              append="updates",
          )
          agent.close()
  
          # load: saved-model format
          agent = tf.saved_model.load(
              export_dir=os.path.join(directory, "agent-1")
          )
  
          # one episode
          states = environment.reset()
          terminal = False
          while not terminal:
              states = batch(states)
              actions = agent.act(states, True)
              actions = unbatch(actions)
              states, terminal, _ = environment.execute(actions=actions)
  
          environment.close()
  
          files = set(os.listdir(path=directory))
          self.assertTrue(
              files
              == {
                  "agent.json",
                  "agent-1",
                  "agent-1.data-00000-of-00001",
                  "agent-1.index",
                  "agent-1.npz",
                  "agent2.json",
                  "agent-2.npz",
                  "agent2-1.hdf5",
                  "checkpoint",
              }
          )
          files = set(
              os.listdir(path=os.path.join(directory, "agent-1"))
          )
          self.assertTrue(
              files == {"assets", "saved_model.pb", "variables"}
          )
          files = set(
              os.listdir(
                  path=os.path.join(directory, "agent-1", "variables")
              )
          )
          self.assertTrue(
              files
              == {"variables.data-00000-of-00001", "variables.index"}
          )
  
      self.finished_test()
  
file_path: tensorforce/test/test_saving.py
imports: import os
  from tempfile import TemporaryDirectory
  import unittest
  import numpy
  import tensorflow
  from tensorforce import Agent, Environment, Runner
  from test.unittest_base import UnittestBase
label_desc: Use image node_id 2 for calling the TestSaving obj's underlying member method code with example usage: obj.test_explicit() and returns: np, mapped, x, np