text: #!/usr/bin/env python3
  # Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved.
  
  """Multi-view test a video classification model."""
  
  import numpy as np
  import torch
  
  import slowfast.utils.checkpoint as cu
  import slowfast.utils.distributed as du
  import slowfast.utils.logging as logging
  import slowfast.utils.misc as misc
  from slowfast.datasets import loader
  from slowfast.models import model_builder
  from slowfast.utils.meters import AVAMeter, TestMeter
  
  logger = logging.get_logger(__name__)
  
  
  def perform_test(test_loader, model, test_meter, cfg):
      """
      For classification:
      Perform mutli-view testing that uniformly samples N clips from a video along
      its temporal axis. For each clip, it takes 3 crops to cover the spatial
      dimension, followed by averaging the softmax scores across all Nx3 views to
      form a video-level prediction. All video predictions are compared to
      ground-truth labels and the final testing performance is logged.
      For detection:
      Perform fully-convolutional testing on the full frames without crop.
      Args:
          test_loader (loader): video testing loader.
          model (model): the pretrained video model to test.
          test_meter (TestMeter): testing meters to log and ensemble the testing
              results.
          cfg (CfgNode): configs. Details can be found in
              slowfast/config/defaults.py
      """
      # Enable eval mode.
      model.eval()
      test_meter.iter_tic()
  
      for cur_iter, (inputs, labels, video_idx, meta) in enumerate(test_loader):
          # Transfer the data to the current GPU device.
          if isinstance(inputs, (list,)):
              for i in range(len(inputs)):
                  inputs[i] = inputs[i].cuda(non_blocking=True)
                  print(inputs[i].size())
          else:
              inputs = inputs.cuda(non_blocking=True)
              print(inputs.size())
  
          # Transfer the data to the current GPU device.
          labels = labels.cuda()
          video_idx = video_idx.cuda()
          for key, val in meta.items():
              if isinstance(val, (list,)):
                  for i in range(len(val)):
                      val[i] = val[i].cuda(non_blocking=True)
              else:
                  meta[key] = val.cuda(non_blocking=True)
  
          if cfg.DETECTION.ENABLE:
              # Compute the predictions.
              preds = model(inputs, meta["boxes"])
  
              preds = preds.cpu()
              ori_boxes = meta["ori_boxes"].cpu()
              metadata = meta["metadata"].cpu()
  
              if cfg.NUM_GPUS > 1:
                  preds = torch.cat(du.all_gather_unaligned(preds), dim=0)
                  ori_boxes = torch.cat(du.all_gather_unaligned(ori_boxes), dim=0)
                  metadata = torch.cat(du.all_gather_unaligned(metadata), dim=0)
  
              test_meter.iter_toc()
              # Update and log stats.
              test_meter.update_stats(
                  preds.detach().cpu(),
                  ori_boxes.detach().cpu(),
                  metadata.detach().cpu(),
              )
              test_meter.log_iter_stats(None, cur_iter)
          else:
              # Perform the forward pass.
              preds = model(inputs)
  
              # Gather all the predictions across all the devices to perform ensemble.
              if cfg.NUM_GPUS > 1:
                  preds, labels, video_idx = du.all_gather(
                      [preds, labels, video_idx]
                  )
  
              test_meter.iter_toc()
              # Update and log stats.
              test_meter.update_stats(
                  preds.detach().cpu(),
                  labels.detach().cpu(),
                  video_idx.detach().cpu(),
              )
              test_meter.log_iter_stats(cur_iter)
  
          test_meter.iter_tic()
  
      # Log epoch stats and print the final testing results.
      test_meter.finalize_metrics()
      test_meter.reset()
  
  
  def test(cfg):
      """
      Perform multi-view testing on the pretrained video model.
      Args:
          cfg (CfgNode): configs. Details can be found in
              slowfast/config/defaults.py
      """
      # Set random seed from configs.
      np.random.seed(cfg.RNG_SEED)
      torch.manual_seed(cfg.RNG_SEED)
  
      # Setup logging format.
      logging.setup_logging()
  
      # Print config.
      logger.info("Test with config:")
      logger.info(cfg)
  
      # Build the video model and print model statistics.
      model = model_builder.build_model(cfg)
      if du.is_master_proc():
          misc.log_model_info(model)
  
      # Load a checkpoint to test if applicable.
      if cfg.TEST.CHECKPOINT_FILE_PATH != "":
          cu.load_checkpoint(
              cfg.TEST.CHECKPOINT_FILE_PATH,
              model,
              cfg.NUM_GPUS > 1,
              None,
              inflation=False,
              convert_from_caffe2=cfg.TEST.CHECKPOINT_TYPE == "caffe2",
          )
      elif cu.has_checkpoint(cfg.OUTPUT_DIR):
          last_checkpoint = cu.get_last_checkpoint(cfg.OUTPUT_DIR)
          cu.load_checkpoint(last_checkpoint, model, cfg.NUM_GPUS > 1)
      elif cfg.TRAIN.CHECKPOINT_FILE_PATH != "":
          # If no checkpoint found in TEST.CHECKPOINT_FILE_PATH or in the current
          # checkpoint folder, try to load checkpint from
          # TRAIN.CHECKPOINT_FILE_PATH and test it.
          cu.load_checkpoint(
              cfg.TRAIN.CHECKPOINT_FILE_PATH,
              model,
              cfg.NUM_GPUS > 1,
              None,
              inflation=False,
              convert_from_caffe2=cfg.TRAIN.CHECKPOINT_TYPE == "caffe2",
          )
      else:
          # raise NotImplementedError("Unknown way to load checkpoint.")
          logger.info("Testing with random initialization. Only for debugging.")
  
      # Create video testing loaders.
      test_loader = loader.construct_loader(cfg, "test")
      logger.info("Testing model for {} iterations".format(len(test_loader)))
  
      if cfg.DETECTION.ENABLE:
          assert cfg.NUM_GPUS == cfg.TEST.BATCH_SIZE
          test_meter = AVAMeter(len(test_loader), cfg, mode="test")
      else:
          assert (
              len(test_loader.dataset)
              % (cfg.TEST.NUM_ENSEMBLE_VIEWS * cfg.TEST.NUM_SPATIAL_CROPS)
              == 0
          )
          # Create meters for multi-view testing.
          test_meter = TestMeter(
              len(test_loader.dataset)
              // (cfg.TEST.NUM_ENSEMBLE_VIEWS * cfg.TEST.NUM_SPATIAL_CROPS),
              cfg.TEST.NUM_ENSEMBLE_VIEWS * cfg.TEST.NUM_SPATIAL_CROPS,
              cfg.MODEL.NUM_CLASSES,
              len(test_loader),
          )
  
      # # Perform multi-view test on the entire dataset.
      perform_test(test_loader, model, test_meter, cfg)
  