code: def test_checkpointer(tmpdir, device):
      from speechbrain.utils.checkpoints import Checkpointer
  
      class Recoverable(torch.nn.Module):
          def __init__(self, param):
              super().__init__()
              self.param = torch.nn.Parameter(torch.tensor([param]))
  
          def forward(self, x):
              return x * self.param
  
      recoverable = Recoverable(2.0)
      recoverables = {"recoverable": recoverable}
      recoverer = Checkpointer(tmpdir, recoverables)
      recoverable.param.data = torch.tensor([1.0], device=device)
      # Should not be possible since no checkpoint saved yet:
      assert not recoverer.recover_if_possible()
      result = recoverable(10.0)
      # Check that parameter has not been loaded from original value:
      assert recoverable.param.data == torch.tensor(
          [1.0], device=device
      )
  
      ckpt = recoverer.save_checkpoint()
      # Check that the name recoverable has a save file:
      # NOTE: Here assuming .pt filename; if convention changes, change test
      assert (ckpt.path / "recoverable.ckpt").exists()
      # Check that saved checkpoint is found, and location correct:
      assert recoverer.list_checkpoints()[0] == ckpt
      assert recoverer.list_checkpoints()[0].path.parent == tmpdir
      recoverable.param.data = torch.tensor([2.0], device=device)
      recoverer.recover_if_possible()
      # Check that parameter has been loaded immediately:
      assert recoverable.param.data == torch.tensor(
          [1.0], device=device
      )
      result = recoverable(10.0)
      # And result correct
      assert result == 10.0
  
      other = Recoverable(2.0)
      recoverer.add_recoverable("other", other)
      # Check that both objects are now found:
      assert recoverer.recoverables["recoverable"] == recoverable
      assert recoverer.recoverables["other"] == other
      new_ckpt = recoverer.save_checkpoint()
      # Check that now both recoverables have a save file:
      assert (new_ckpt.path / "recoverable.ckpt").exists()
      assert (new_ckpt.path / "other.ckpt").exists()
      assert new_ckpt in recoverer.list_checkpoints()
      recoverable.param.data = torch.tensor([2.0], device=device)
      other.param.data = torch.tensor([10.0], device=device)
      chosen_ckpt = recoverer.recover_if_possible()
      # Should choose newest by default:
      assert chosen_ckpt == new_ckpt
      # Check again that parameters have been loaded immediately:
      assert recoverable.param.data == torch.tensor(
          [1.0], device=device
      )
      assert other.param.data == torch.tensor([2.0], device=device)
      other_result = other(10.0)
      # And again we should have the correct computations:
      assert other_result == 20.0
  
      # Recover from oldest, which does not have "other":
      # This also tests a custom sort
      # Raises by default:
      with pytest.raises(RuntimeError):
          chosen_ckpt = recoverer.recover_if_possible(
              importance_key=lambda x: -x.meta["unixtime"]
          )
      # However this operation may have loaded the first object
      # so let's set the values manually:
      recoverable.param.data = torch.tensor([2.0], device=device)
      other.param.data = torch.tensor([10.0], device=device)
      recoverer.allow_partial_load = True
      chosen_ckpt = recoverer.recover_if_possible(
          importance_key=lambda x: -x.meta["unixtime"]
      )
      # Should have chosen the original:
      assert chosen_ckpt == ckpt
      # And should recover recoverable:
      assert recoverable.param.data == torch.tensor(
          [1.0], device=device
      )
      # But not other:
      other_result = other(10.0)
      assert other.param.data == torch.tensor([10.0], device=device)
      assert other_result == 100.0
  
      # Test saving names checkpoints with meta info, and custom filter
      epoch_ckpt = recoverer.save_checkpoint(
          name="ep1", meta={"loss": 2.0}
      )
      assert "ep1" in epoch_ckpt.path.name
      other.param.data = torch.tensor([2.0], device=device)
      recoverer.save_checkpoint(meta={"loss": 3.0})
      chosen_ckpt = recoverer.recover_if_possible(
          ckpt_predicate=lambda ckpt: "loss" in ckpt.meta,
          importance_key=lambda ckpt: -ckpt.meta["loss"],
      )
      assert chosen_ckpt == epoch_ckpt
      assert other.param.data == torch.tensor([10.0], device=device)
  
file_path: speechbrain/tests/unittests/test_checkpoints.py
imports: import torch
  import pytest
label_desc: Use image node_id 1 for calling a global function with example usage: test_checkpointer(tmpdir, device) and returns: unknown