code: def calc_perplexity(
      output_dir: str,
      batch_size: int,
      dtype: str,
      ngpu: int,
      seed: int,
      num_workers: int,
      log_level: Union[int, str],
      data_path_and_name_and_type: Sequence[Tuple[str, str, str]],
      key_file: Optional[str],
      train_config: Optional[str],
      model_file: Optional[str],
      log_base: Optional[float],
      allow_variable_data_keys: bool,
  ):
      assert check_argument_types()
      logging.basicConfig(
          level=log_level,
          format="%(asctime)s (%(module)s:%(lineno)d) %(levelname)s: %(message)s",
      )
  
      if ngpu >= 1:
          device = "cuda"
      else:
          device = "cpu"
  
      # 1. Set random-seed
      set_all_random_seed(seed)
  
      # 2. Build LM
      model, train_args = LMTask.build_model_from_file(
          train_config, model_file, device
      )
      # Wrape model to make model.nll() data-parallel
      wrapped_model = ForwardAdaptor(model, "nll")
      wrapped_model.to(dtype=getattr(torch, dtype)).eval()
      logging.info(f"Model:\n{model}")
  
      # 3. Build data-iterator
      loader = LMTask.build_streaming_iterator(
          data_path_and_name_and_type,
          dtype=dtype,
          batch_size=batch_size,
          key_file=key_file,
          num_workers=num_workers,
          preprocess_fn=LMTask.build_preprocess_fn(train_args, False),
          collate_fn=LMTask.build_collate_fn(train_args, False),
          allow_variable_data_keys=allow_variable_data_keys,
          inference=True,
      )
  
      # 4. Start for-loop
      with DatadirWriter(output_dir) as writer:
          total_nll = 0.0
          total_ntokens = 0
          for keys, batch in loader:
              assert isinstance(batch, dict), type(batch)
              assert all(isinstance(s, str) for s in keys), keys
              _bs = len(next(iter(batch.values())))
              assert len(keys) == _bs, f"{len(keys)} != {_bs}"
  
              with torch.no_grad():
                  batch = to_device(batch, device)
                  if ngpu <= 1:
                      # NOTE(kamo): data_parallel also should work with ngpu=1,
                      # but for debuggability it's better to keep this block.
                      nll, lengths = wrapped_model(**batch)
                  else:
                      nll, lengths = data_parallel(
                          wrapped_model,
                          (),
                          range(ngpu),
                          module_kwargs=batch,
                      )
  
              assert _bs == len(nll) == len(lengths), (
                  _bs,
                  len(nll),
                  len(lengths),
              )
              # nll: (B, L) -> (B,)
              nll = nll.detach().cpu().numpy().sum(1)
              # lengths: (B,)
              lengths = lengths.detach().cpu().numpy()
              total_nll += nll.sum()
              total_ntokens += lengths.sum()
  
              for key, _nll, ntoken in zip(keys, nll, lengths):
                  if log_base is None:
                      utt_ppl = np.exp(_nll / ntoken)
                  else:
                      utt_ppl = log_base ** (
                          _nll / ntoken / np.log(log_base)
                      )
  
                  # Write PPL of each utts for debugging or analysis
                  writer["utt2ppl"][key] = str(utt_ppl)
                  writer["utt2ntokens"][key] = str(ntoken)
  
          if log_base is None:
              ppl = np.exp(total_nll / total_ntokens)
          else:
              ppl = log_base ** (
                  total_nll / total_ntokens / np.log(log_base)
              )
  
          with (Path(output_dir) / "ppl").open(
              "w", encoding="utf-8"
          ) as f:
              f.write(f"{ppl}\n")
          with (Path(output_dir) / "base").open(
              "w", encoding="utf-8"
          ) as f:
              if log_base is None:
                  _log_base = np.e
              else:
                  _log_base = log_base
              f.write(f"{_log_base}\n")
          logging.info(f"PPL={ppl}")
  
file_path: espnet/espnet2/bin/lm_calc_perplexity.py
imports: import argparse
  import logging
  import sys
  from pathlib import Path
  from typing import Optional, Sequence, Tuple, Union
  import numpy
  import torch
  from torch.nn.parallel import data_parallel
  from typeguard import check_argument_types
  from espnet2.fileio.datadir_writer import DatadirWriter
  from espnet2.tasks.lm import LMTask
  from espnet2.torch_utils.device_funcs import to_device
  from espnet2.torch_utils.forward_adaptor import ForwardAdaptor
  from espnet2.torch_utils.set_all_random_seed import set_all_random_seed
  from espnet2.utils import config_argparse
  from espnet2.utils.types import float_or_none, str2bool, str2triple_str, str_or_none
  from espnet.utils.cli_utils import get_commandline_args
label_desc: Use image node_id 1 for calling a global function with example usage: calc_perplexity(output_dir, batch_size, dtype, ngpu, seed, num_workers, log_level, data_path_and_name_and_type, key_file, train_config, model_file, log_base, allow_variable_data_keys) without return types