class_docstr: Ego4d video/audio/imu dataset for the moments benchmark:
  `<https://ego4d-data.org/docs/benchmarks/episodic-memory/>`
  
  This dataset handles the parsing of frames, loading and clip sampling for the
  videos.
  
  IO utilizing :code:`iopath.common.file_io.PathManager` to support
  non-local storage uri's.
docstr: Args:
      annotation_path (str):
          Path or URI to Ego4d moments annotations json (ego4d.json). Download via:
          `<https://github.com/facebookresearch/Ego4d/blob/main/ego4d/cli/README.md>`
  
      metadata_path (str):
          Path or URI to primary Ego4d metadata json (moments.json). Download via:
          `<https://github.com/facebookresearch/Ego4d/blob/main/ego4d/cli/README.md>`
  
      split (Optional[str]): train/val/test
  
      decode_audio (bool): If True, decode audio from video.
  
      imu (bool): If True, load IMU data.
  
      clip_sampler (ClipSampler):
          A standard PTV ClipSampler. By default, if not specified, `MomentsClipSampler`
  
      video_sampler (VideoSampler):
          A standard PTV VideoSampler.
  
      transform (Optional[Callable[[Dict[str, Any]], Any]]):
          This callable is evaluated on the clip output before the clip is returned.
          It can be used for user-defined preprocessing and augmentations to the clips.
  
              The clip input is a dictionary with the following format:
                  {{
                      'video': <video_tensor>,
                      'audio': <audio_tensor>,
                      'imu': <imu_tensor>,
                      'start_time': <float>,
                      'stop_time': <float>
                  }}
  
          If transform is None, the raw clip output in the above format is
          returned unmodified.
  
      decoder (str): Defines what type of decoder used to decode a video within
          `LabeledVideoDataset`.
  
      filtered_labels (List[str]):
          Optional list of moments labels to filter samples for training.
  
      window_sec (int): minimum window size in s
  
      audio_transform_type: melspectrogram / spectrogram / mfcc
  
      imu_path (Optional[str]):
          Path to the ego4d IMU csv file.  Required if imu=True.
  
      label_id_map / label_id_map_path:
          A map of moments labels to consistent integer ids.  If specified as a path
          we expect a vanilla .json dict[str, int].  Exactly one must be specified.
  
      video_path_override ((str) -> str):
          An override for video paths, given the video_uid, to support downsampled/etc
          videos.
  
      video_path_handler (VideoPathHandler):
          Primarily provided as an override for `CachedVideoPathHandler`
  
  Example Usage:
      Ego4dMomentsDataset(
          annotation_path="~/ego4d_data/v1/annotations/moments.json",
          metadata_path="~/ego4d_data/v1/ego4d.json",
          split="train",
          decode_audio=True,
          imu=False,
      )
code: def __init__(
      self,
      annotation_path: str,
      metadata_path: str,
      split: Optional[str] = None,
      decode_audio: bool = True,
      imu: bool = False,
      clip_sampler: Optional[ClipSampler] = None,
      video_sampler: Type[
          torch.utils.data.Sampler
      ] = torch.utils.data.SequentialSampler,
      transform: Optional[
          Callable[[Dict[str, Any]], Dict[str, Any]]
      ] = None,
      decoder: str = "pyav",
      filtered_labels: Optional[List[str]] = None,
      window_sec: int = 10,
      audio_transform_type: str = "melspectrogram",
      imu_path: str = None,
      label_id_map: Optional[Dict[str, int]] = None,
      label_id_map_path: Optional[str] = None,
      video_path_override: Optional[Callable[[str], str]] = None,
      video_path_handler: Optional[VideoPathHandler] = None,
      eligible_video_uids: Optional[Set[str]] = None,
  ) -> None:
      """
      Args:
          annotation_path (str):
              Path or URI to Ego4d moments annotations json (ego4d.json). Download via:
              `<https://github.com/facebookresearch/Ego4d/blob/main/ego4d/cli/README.md>`
  
          metadata_path (str):
              Path or URI to primary Ego4d metadata json (moments.json). Download via:
              `<https://github.com/facebookresearch/Ego4d/blob/main/ego4d/cli/README.md>`
  
          split (Optional[str]): train/val/test
  
          decode_audio (bool): If True, decode audio from video.
  
          imu (bool): If True, load IMU data.
  
          clip_sampler (ClipSampler):
              A standard PTV ClipSampler. By default, if not specified, `MomentsClipSampler`
  
          video_sampler (VideoSampler):
              A standard PTV VideoSampler.
  
          transform (Optional[Callable[[Dict[str, Any]], Any]]):
              This callable is evaluated on the clip output before the clip is returned.
              It can be used for user-defined preprocessing and augmentations to the clips.
  
                  The clip input is a dictionary with the following format:
                      {{
                          'video': <video_tensor>,
                          'audio': <audio_tensor>,
                          'imu': <imu_tensor>,
                          'start_time': <float>,
                          'stop_time': <float>
                      }}
  
              If transform is None, the raw clip output in the above format is
              returned unmodified.
  
          decoder (str): Defines what type of decoder used to decode a video within
              `LabeledVideoDataset`.
  
          filtered_labels (List[str]):
              Optional list of moments labels to filter samples for training.
  
          window_sec (int): minimum window size in s
  
          audio_transform_type: melspectrogram / spectrogram / mfcc
  
          imu_path (Optional[str]):
              Path to the ego4d IMU csv file.  Required if imu=True.
  
          label_id_map / label_id_map_path:
              A map of moments labels to consistent integer ids.  If specified as a path
              we expect a vanilla .json dict[str, int].  Exactly one must be specified.
  
          video_path_override ((str) -> str):
              An override for video paths, given the video_uid, to support downsampled/etc
              videos.
  
          video_path_handler (VideoPathHandler):
              Primarily provided as an override for `CachedVideoPathHandler`
  
      Example Usage:
          Ego4dMomentsDataset(
              annotation_path="~/ego4d_data/v1/annotations/moments.json",
              metadata_path="~/ego4d_data/v1/ego4d.json",
              split="train",
              decode_audio=True,
              imu=False,
          )
      """
  
      assert annotation_path
      assert metadata_path
      assert split in [
          "train",
          "val",
          "test",
      ], f"Split '{split}' not supported for ego4d"
      self.split: str = split
      self.training: bool = split == "train"
      self.window_sec = window_sec
      self._transform_source = transform
      self.decode_audio = decode_audio
      self.audio_transform_type = audio_transform_type
      assert (label_id_map is not None) ^ (
          label_id_map_path is not None
      ), f"Either label_id_map or label_id_map_path required ({label_id_map_path} / {label_id_map})"  # noqa
  
      self.video_means = (0.45, 0.45, 0.45)
      self.video_stds = (0.225, 0.225, 0.225)
      self.video_crop_size = 224
      self.video_min_short_side_scale = 256
      self.video_max_short_side_scale = 320
  
      try:
          with g_pathmgr.open(metadata_path, "r") as f:
              metadata = json.load(f)
      except Exception:
          raise FileNotFoundError(
              f"{metadata_path} must be a valid metadata json for Ego4D"
          )
  
      self.video_metadata_map: Dict[str, Any] = {
          x["video_uid"]: x for x in metadata["videos"]
      }
  
      if not g_pathmgr.isfile(annotation_path):
          raise FileNotFoundError(f"{annotation_path} not found.")
  
      try:
          with g_pathmgr.open(annotation_path, "r") as f:
              moments_annotations = json.load(f)
      except Exception:
          raise FileNotFoundError(
              f"{annotation_path} must be json for Ego4D dataset"
          )
  
      self.label_name_id_map: Dict[str, int]
      if label_id_map:
          self.label_name_id_map = label_id_map
      else:
          self.label_name_id_map = get_label_id_map(label_id_map_path)
          assert self.label_name_id_map
  
      self.num_classes: int = len(self.label_name_id_map)
      log.info(f"Label Classes: {self.num_classes}")
  
      self.imu_data: Optional[Ego4dImuDataBase] = None
      if imu:
          assert imu_path, "imu_path not provided"
          self.imu_data = Ego4dImuData(imu_path)
  
      video_uids = set()
      clip_uids = set()
      clip_video_map = {}
      labels = set()
      labels_bypassed = set()
      cnt_samples_bypassed = 0
      cnt_samples_bypassed_labels = 0
      samples = []
  
      for vid in moments_annotations["videos"]:
          video_uid = vid["video_uid"]
          video_uids.add(video_uid)
          vsplit = vid["split"]
          if split and vsplit != split:
              continue
          # If IMU, filter videos without IMU
          if self.imu_data and not self.imu_data.has_imu(video_uid):
              continue
          if (
              eligible_video_uids
              and video_uid not in eligible_video_uids
          ):
              continue
          for clip in vid["clips"]:
              clip_uid = clip["clip_uid"]
              clip_uids.add(clip_uid)
              clip_video_map[clip_uid] = video_uid
              clip_start_sec = clip["video_start_sec"]
              clip_end_sec = clip["video_end_sec"]
              for vann in clip["annotations"]:
                  for lann in vann["labels"]:
                      label = lann["label"]
                      labels.add(label)
                      start = lann["start_time"]
                      end = lann["end_time"]
                      # remove sample with same timestamp
                      if start == end:
                          continue
                      start_video = lann["video_start_time"]
                      end_video = lann["video_end_time"]
                      assert end_video >= start_video
  
                      if (
                          abs(start_video - (clip_start_sec + start))
                          > 0.5
                      ):
                          log.warning(
                              f"Suspect clip/video start mismatch: clip: {clip_start_sec:.2f} + {start:.2f} video: {start_video:.2f}"  # noqa
                          )
  
                      # filter annotation base on the existing label map
                      if (
                          filtered_labels
                          and label not in filtered_labels
                      ):
                          cnt_samples_bypassed += 1
                          labels_bypassed.add(label)
                          continue
                      metadata = self.video_metadata_map[video_uid]
  
                      if metadata["is_stereo"]:
                          cnt_samples_bypassed += 1
                          continue
  
                      if video_path_override:
                          video_path = video_path_override(video_uid)
                      else:
                          video_path = metadata["manifold_path"]
                      if not video_path:
                          cnt_samples_bypassed += 1
                          log.error(
                              "Bypassing invalid video_path: {video_uid}"
                          )
                          continue
  
                      sample = {
                          "clip_uid": clip_uid,
                          "video_uid": video_uid,
                          "duration": metadata["duration_sec"],
                          "clip_video_start_sec": clip_start_sec,
                          "clip_video_end_sec": clip_end_sec,
                          "labels": [label],
                          "label_video_start_sec": start_video,
                          "label_video_end_sec": end_video,
                          "video_path": video_path,
                      }
                      assert (
                          sample["label_video_end_sec"]
                          > sample["label_video_start_sec"]
                      )
  
                      if self.label_name_id_map:
                          if label in self.label_name_id_map:
                              sample[
                                  "labels_id"
                              ] = self.label_name_id_map[label]
                          else:
                              cnt_samples_bypassed_labels += 1
                              continue
                      else:
                          log.error("Missing label_name_id_map")
                      samples.append(sample)
  
      self.cnt_samples: int = len(samples)
  
      log.info(
          f"Loaded {self.cnt_samples} samples. Bypass: {cnt_samples_bypassed} Label Lookup Bypass: {cnt_samples_bypassed_labels}"  # noqa
      )
  
      for sample in samples:
          assert (
              "labels_id" in sample
          ), f"init: Sample missing labels_id: {sample}"
  
      if not clip_sampler:
          clip_sampler = MomentsClipSampler(self.window_sec)
  
      super().__init__(
          [(x["video_path"], x) for x in samples],
          clip_sampler,
          video_sampler,
          transform=self._transform_mm,
          decode_audio=decode_audio,
          decoder=decoder,
      )
  
      if video_path_handler:
          self.video_path_handler = video_path_handler
  
imports: import csv
  import json
  import logging
  import os
  from bisect import bisect_left
  from collections import defaultdict
  from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Type
  import numpy
  import torch
  import torch.autograd.profiler
  import torch.utils.data
  import torchaudio
  from iopath.common.file_io import g_pathmgr
  from pytorchvideo.data import LabeledVideoDataset
  from pytorchvideo.data.clip_sampling import ClipSampler
  from pytorchvideo.data.ego4d.utils import Ego4dImuDataBase, get_label_id_map, MomentsClipSampler
  from pytorchvideo.data.utils import get_logger
  from pytorchvideo.data.video import VideoPathHandler
  from pytorchvideo.transforms import ApplyTransformToKey, Div255, Normalize, RandomShortSideScale, ShortSideScale
  from torchvision.transforms import CenterCrop, Compose, RandomCrop, RandomHorizontalFlip
label_desc: Use image node_id 1 to create a new Ego4dMomentsDataset object from inherited base classes: LabeledVideoDataset with example: obj = Ego4dMomentsDataset(annotation_path, metadata_path, split, decode_audio, imu, clip_sampler, video_sampler, transform, decoder, filtered_labels, window_sec, audio_transform_type, imu_path, label_id_map, label_id_map_path, video_path_override, video_path_handler, eligible_video_uids)