code: def compute(
      self,
      model_outs: List[Dict[str, torch.Tensor]],
      nsteps: int,
      batch_window_size: int,
      timestamps: Optional[List[float]],
  ) -> TestRecMetricOutput:
      assert timestamps is not None
      (
          lifetime_states,
          window_states,
          local_lifetime_states,
          local_window_states,
      ) = (
          {task_info.name: {} for task_info in self._rec_tasks}
          for _ in range(4)
      )
      for i in range(WARMUP_STEPS, nsteps):
          for task_info in self._rec_tasks:
              local_states = {
                  "num_samples": torch.tensor(
                      model_outs[i][task_info.label_name].shape[-1],
                      dtype=torch.long,
                  ),
                  "time_lapse": torch.tensor(
                      timestamps[i] - timestamps[i - 1],
                      dtype=torch.double,
                  ),
              }
              self._aggregate(
                  local_lifetime_states[task_info.name], local_states
              )
              if nsteps - batch_window_size <= i:
                  self._aggregate(
                      local_window_states[task_info.name], local_states
                  )
  
      for task_info in self._rec_tasks:
          aggregated_lifetime_state = {}
          for k, v in local_lifetime_states[task_info.name].items():
              aggregated_lifetime_state[k] = [
                  torch.zeros_like(v) for _ in range(self.world_size)
              ]
              dist.all_gather(aggregated_lifetime_state[k], v)
          lifetime_states[task_info.name] = self._reduce(
              aggregated_lifetime_state
          )
  
          aggregated_window_state = {}
          for k, v in local_window_states[task_info.name].items():
              aggregated_window_state[k] = [
                  torch.zeros_like(v) for _ in range(self.world_size)
              ]
              dist.all_gather(aggregated_window_state[k], v)
          window_states[task_info.name] = self._reduce(
              aggregated_window_state
          )
  
      lifetime_metrics = {}
      window_metrics = {}
      local_lifetime_metrics = {}
      local_window_metrics = {}
      for task_info in self._rec_tasks:
          lifetime_metrics[task_info.name] = self._compute(
              lifetime_states[task_info.name]
          )
          window_metrics[task_info.name] = self._compute(
              window_states[task_info.name]
          )
          local_lifetime_metrics[task_info.name] = self._compute(
              local_lifetime_states[task_info.name]
          )
          local_window_metrics[task_info.name] = self._compute(
              local_window_states[task_info.name]
          )
      return (
          lifetime_metrics,
          window_metrics,
          local_lifetime_metrics,
          local_window_metrics,
      )
  
imports: import unittest
  from functools import partial, update_wrapper
  from typing import Callable, Dict, List, Optional, Tuple, Type
  import torch
  import torch.distributed
  from torchrec.metrics.metrics_config import DefaultTaskInfo
  from torchrec.metrics.model_utils import parse_task_model_outputs
  from torchrec.metrics.rec_metric import RecComputeMode, RecMetric, RecTaskInfo
  from torchrec.metrics.test_utils import gen_test_batch, gen_test_tasks, metric_test_helper, rec_metric_value_test_launcher, TestMetric
  from torchrec.metrics.tower_qps import TowerQPSMetric
label_desc: Use image node_id 5 for calling the TestTowerQPSMetric obj's underlying member method code with example usage: obj.compute(model_outs, nsteps, batch_window_size, timestamps) and returns: lifetime_metrics,  window_metrics,  local_lifetime_metrics,  local_window_metrics