text: import io
  import os
  import os.path as osp
  import shutil
  
  import mmcv
  import numpy as np
  from mmcv.fileio import FileClient
  
  from ...utils import get_random_string, get_shm_dir, get_thread_id
  from ..registry import PIPELINES
  
  
  @PIPELINES.register_module()
  class SampleFrames(object):
      """Sample frames from the video.
  
      Required keys are "filename", added or modified keys are "total_frames",
      "frame_inds", "frame_interval" and "num_clips".
  
      Args:
          clip_len (int): Frames of each sampled output clip.
          frame_interval (int): Temporal interval of adjacent sampled frames.
              Default: 1.
          num_clips (int): Number of clips to be sampled. Default: 1.
          start_index (int): Specify a start index for frames in consideration of
              different filename format. However, when taking videos as input,
              it should be set to 0, since frames loaded from videos count
              from 0. Default: 1.
          temporal_jitter (bool): Whether to apply temporal jittering.
              Default: False.
          twice_sample (bool): Whether to use twice sample when testing.
              If set to True, it will sample frames with and without fixed shift,
              which is commonly used for testing in TSM model. Default: False.
          out_of_bound_opt (str): The way to deal with out of bounds frame
              indexes. Available options are 'loop', 'repeat_last'.
              Default: 'loop'.
          test_mode (bool): Store True when building test or validation dataset.
              Default: False.
      """
  
      def __init__(self,
                   clip_len,
                   frame_interval=1,
                   num_clips=1,
                   start_index=1,
                   temporal_jitter=False,
                   twice_sample=False,
                   out_of_bound_opt='loop',
                   test_mode=False):
  
          self.clip_len = clip_len
          self.frame_interval = frame_interval
          self.num_clips = num_clips
          self.start_index = start_index
          self.temporal_jitter = temporal_jitter
          self.twice_sample = twice_sample
          self.out_of_bound_opt = out_of_bound_opt
          self.test_mode = test_mode
          assert self.out_of_bound_opt in ['loop', 'repeat_last']
  
      def _get_train_clips(self, num_frames):
          """Get clip offsets in train mode.
  
          It will calculate the average interval for selected frames,
          and randomly shift them within offsets between [0, avg_interval].
          If the total number of frames is smaller than clips num or origin
          frames length, it will return all zero indices.
  
          Args:
              num_frames (int): Total number of frame in the video.
  
          Returns:
              np.ndarray: Sampled frame indices in train mode.
          """
          ori_clip_len = self.clip_len * self.frame_interval
          avg_interval = (num_frames - ori_clip_len + 1) // self.num_clips
  
          if avg_interval > 0:
              base_offsets = np.arange(self.num_clips) * avg_interval
              clip_offsets = base_offsets + np.random.randint(
                  avg_interval, size=self.num_clips)
          elif num_frames > max(self.num_clips, ori_clip_len):
              clip_offsets = np.sort(
                  np.random.randint(
                      num_frames - ori_clip_len + 1, size=self.num_clips))
          elif avg_interval == 0:
              ratio = (num_frames - ori_clip_len + 1.0) / self.num_clips
              clip_offsets = np.around(np.arange(self.num_clips) * ratio)
          else:
              clip_offsets = np.zeros((self.num_clips, ))
  
          return clip_offsets
  
      def _get_test_clips(self, num_frames):
          """Get clip offsets in test mode.
  
          Calculate the average interval for selected frames, and shift them
          fixedly by avg_interval/2. If set twice_sample True, it will sample
          frames together without fixed shift. If the total number of frames is
          not enough, it will return all zero indices.
  
          Args:
              num_frames (int): Total number of frame in the video.
  
          Returns:
              np.ndarray: Sampled frame indices in test mode.
          """
          ori_clip_len = self.clip_len * self.frame_interval
          avg_interval = (num_frames - ori_clip_len + 1) / float(self.num_clips)
          if num_frames > ori_clip_len - 1:
              base_offsets = np.arange(self.num_clips) * avg_interval
              clip_offsets = (base_offsets + avg_interval / 2.0).astype(np.int32)
              if self.twice_sample:
                  clip_offsets = np.concatenate([clip_offsets, base_offsets])
          else:
              clip_offsets = np.zeros((self.num_clips, ))
          return clip_offsets
  
      def _sample_clips(self, num_frames):
          """Choose clip offsets for the video in a given mode.
  
          Args:
              num_frames (int): Total number of frame in the video.
  
          Returns:
              np.ndarray: Sampled frame indices.
          """
          if self.test_mode:
              clip_offsets = self._get_test_clips(num_frames)
          else:
              clip_offsets = self._get_train_clips(num_frames)
  
          return clip_offsets
  
      def __call__(self, results):
          """Perform the SampleFrames loading.
  
          Args:
              results (dict): The resulting dict to be modified and passed
                  to the next transform in pipeline.
          """
          if 'total_frames' not in results:
              # TODO: find a better way to get the total frames number for video
              video_reader = mmcv.VideoReader(results['filename'])
              total_frames = len(video_reader)
              results['total_frames'] = total_frames
          else:
              total_frames = results['total_frames']
  
          clip_offsets = self._sample_clips(total_frames)
          frame_inds = clip_offsets[:, None] + np.arange(
              self.clip_len)[None, :] * self.frame_interval
          frame_inds = np.concatenate(frame_inds)
  
          if self.temporal_jitter:
              perframe_offsets = np.random.randint(
                  self.frame_interval, size=len(frame_inds))
              frame_inds += perframe_offsets
  
          frame_inds = frame_inds.reshape((-1, self.clip_len))
          if self.out_of_bound_opt == 'loop':
              frame_inds = np.mod(frame_inds, total_frames)
          elif self.out_of_bound_opt == 'repeat_last':
              safe_inds = frame_inds < total_frames
              unsafe_inds = 1 - safe_inds
              last_ind = np.max(safe_inds * frame_inds, axis=1)
              new_inds = (safe_inds * frame_inds + (unsafe_inds.T * last_ind).T)
              frame_inds = new_inds
          else:
              raise ValueError('Illegal out_of_bound option.')
          frame_inds = np.concatenate(frame_inds) + self.start_index
          results['frame_inds'] = frame_inds.astype(np.int)
          results['clip_len'] = self.clip_len
          results['frame_interval'] = self.frame_interval
          results['num_clips'] = self.num_clips
          return results
  
  
  @PIPELINES.register_module()
  class DenseSampleFrames(SampleFrames):
      """Select frames from the video by dense sample strategy.
  
      Required keys are "filename", added or modified keys are "total_frames",
      "frame_inds", "frame_interval" and "num_clips".
  
      Args:
          clip_len (int): Frames of each sampled output clip.
          frame_interval (int): Temporal interval of adjacent sampled frames.
              Default: 1.
          num_clips (int): Number of clips to be sampled. Default: 1.
          start_index (int): Specify a start index for frames in consideration of
              different filename format. Default: 1.
          sample_range (int): Total sample range for dense sample.
              Default: 64.
          num_sample_positions (int): Number of sample start positions, Which is
              only used in test mode. Default: 10.
          temporal_jitter (bool): Whether to apply temporal jittering.
              Default: False.
          test_mode (bool): Store True when building test or validation dataset.
              Default: False.
      """
  
      def __init__(self,
                   clip_len,
                   frame_interval=1,
                   num_clips=1,
                   start_index=1,
                   sample_range=64,
                   num_sample_positions=10,
                   temporal_jitter=False,
                   out_of_bound_opt='loop',
                   test_mode=False):
          super().__init__(
              clip_len,
              frame_interval,
              num_clips,
              start_index,
              temporal_jitter,
              out_of_bound_opt=out_of_bound_opt,
              test_mode=test_mode)
          self.sample_range = sample_range
          self.num_sample_positions = num_sample_positions
  
      def _get_train_clips(self, num_frames):
          """Get clip offsets by dense sample strategy in train mode.
  
          It will calculate a sample position and sample interval and set
          start index 0 when sample_pos == 1 or randomly choose from
          [0, sample_pos - 1]. Then it will shift the start index by each
          base offset.
  
          Args:
              num_frames (int): Total number of frame in the video.
  
          Returns:
              np.ndarray: Sampled frame indices in train mode.
          """
          sample_position = max(1, 1 + num_frames - self.sample_range)
          interval = self.sample_range // self.num_clips
          start_idx = 0 if sample_position == 1 else np.random.randint(
              0, sample_position - 1)
          base_offsets = np.arange(self.num_clips) * interval
          clip_offsets = (base_offsets + start_idx) % num_frames
          return clip_offsets
  
      def _get_test_clips(self, num_frames):
          """Get clip offsets by dense sample strategy in test mode.
  
          It will calculate a sample position and sample interval and evenly
          sample several start indexes as start positions between
          [0, sample_position-1]. Then it will shift each start index by the
          base offsets.
  
          Args:
              num_frames (int): Total number of frame in the video.
  
          Returns:
              np.ndarray: Sampled frame indices in train mode.
          """
          sample_position = max(1, 1 + num_frames - self.sample_range)
          interval = self.sample_range // self.num_clips
          start_list = np.linspace(
              0, sample_position - 1, num=self.num_sample_positions, dtype=int)
          base_offsets = np.arange(self.num_clips) * interval
          clip_offsets = list()
          for start_idx in start_list:
              clip_offsets.extend((base_offsets + start_idx) % num_frames)
          clip_offsets = np.array(clip_offsets)
          return clip_offsets
  
  
  @PIPELINES.register_module()
  class PyAVInit(object):
      """Using pyav to initialize the video.
  
      PyAV: https://github.com/mikeboers/PyAV
  
      Required keys are "filename",
      added or modified keys are "video_reader", and "total_frames".
  
      Args:
          io_backend (str): io backend where frames are store.
              Default: 'disk'.
          kwargs (dict): Args for file client.
      """
  
      def __init__(self, io_backend='disk', **kwargs):
          self.io_backend = io_backend
          self.kwargs = kwargs
          self.file_client = None
  
      def __call__(self, results):
          """Perform the PyAV initiation.
  
          Args:
              results (dict): The resulting dict to be modified and passed
                  to the next transform in pipeline.
          """
          try:
              import av
          except ImportError:
              raise ImportError('Please run "conda install av -c conda-forge" '
                                'or "pip install av" to install PyAV first.')
  
          if self.file_client is None:
              self.file_client = FileClient(self.io_backend, **self.kwargs)
  
          file_obj = io.BytesIO(self.file_client.get(results['filename']))
          container = av.open(file_obj)
  
          results['video_reader'] = container
          results['total_frames'] = container.streams.video[0].frames
  
          return results
  
  
  @PIPELINES.register_module()
  class PyAVDecode(object):
      """Using pyav to decode the video.
  
      PyAV: https://github.com/mikeboers/PyAV
  
      Required keys are "video_reader" and "frame_inds",
      added or modified keys are "imgs", "img_shape" and "original_shape".
  
      Args:
          multi_thread (bool): If set to True, it will apply multi
              thread processing. Default: False.
      """
  
      def __init__(self, multi_thread=False):
          self.multi_thread = multi_thread
  
      def __call__(self, results):
          """Perform the PyAV loading.
  
          Args:
              results (dict): The resulting dict to be modified and passed
                  to the next transform in pipeline.
          """
          container = results['video_reader']
          imgs = list()
  
          if self.multi_thread:
              container.streams.video[0].thread_type = 'AUTO'
          if results['frame_inds'].ndim != 1:
              results['frame_inds'] = np.squeeze(results['frame_inds'])
  
          # set max indice to make early stop
          max_inds = max(results['frame_inds'])
          i = 0
          for frame in container.decode(video=0):
              if i > max_inds + 1:
                  break
              imgs.append(frame.to_rgb().to_ndarray())
              i += 1
  
          results['video_reader'] = None
          del container
  
          # the available frame in pyav may be less than its length,
          # which may raise error
          results['imgs'] = [imgs[i % len(imgs)] for i in results['frame_inds']]
  
          results['original_shape'] = imgs[0].shape[:2]
          results['img_shape'] = imgs[0].shape[:2]
  
          return results
  
      def __repr__(self):
          repr_str = self.__class__.__name__
          repr_str += f'(multi_thread={self.multi_thread})'
          return repr_str
  
  
  @PIPELINES.register_module()
  class DecordInit(object):
      """Using decord to initialize the video_reader.
  
      Decord: https://github.com/dmlc/decord
  
      Required keys are "filename",
      added or modified keys are "video_reader" and "total_frames".
      """
  
      def __init__(self, io_backend='disk', num_threads=1, **kwargs):
          self.io_backend = io_backend
          self.num_threads = num_threads
          self.kwargs = kwargs
          self.file_client = None
  
      def __call__(self, results):
          """Perform the PyAV loading.
  
          Args:
              results (dict): The resulting dict to be modified and passed
                  to the next transform in pipeline.
          """
          try:
              import decord
          except ImportError:
              raise ImportError(
                  'Please run "pip install decord" to install Decord first.')
  
          if self.file_client is None:
              self.file_client = FileClient(self.io_backend, **self.kwargs)
  
          file_obj = io.BytesIO(self.file_client.get(results['filename']))
          container = decord.VideoReader(file_obj, num_threads=self.num_threads)
          results['video_reader'] = container
          results['total_frames'] = len(container)
          return results
  
  
  @PIPELINES.register_module()
  class DecordDecode(object):
      """Using decord to decode the video.
  
      Decord: https://github.com/dmlc/decord
  
      Required keys are "video_reader", "filename" and "frame_inds",
      added or modified keys are "imgs" and "original_shape".
      """
  
      def __init__(self, **kwargs):
          pass
  
      def __call__(self, results):
          """Perform the Decord loading.
  
          Args:
              results (dict): The resulting dict to be modified and passed
                  to the next transform in pipeline.
          """
          container = results['video_reader']
  
          if results['frame_inds'].ndim != 1:
              results['frame_inds'] = np.squeeze(results['frame_inds'])
  
          frame_inds = results['frame_inds']
          # Generate frame index mapping in order
          frame_dict = {
              idx: container[idx].asnumpy()
              for idx in np.unique(frame_inds)
          }
  
          imgs = [frame_dict[idx] for idx in frame_inds]
  
          results['video_reader'] = None
          del container
  
          results['imgs'] = imgs
          results['original_shape'] = imgs[0].shape[:2]
          results['img_shape'] = imgs[0].shape[:2]
  
          return results
  
  
  @PIPELINES.register_module()
  class OpenCVInit(object):
      """Using OpenCV to initalize the video_reader.
  
      Required keys are "filename", added or modified keys are "new_path",
      "video_reader" and "total_frames".
      """
  
      def __init__(self, io_backend='disk', **kwargs):
          self.io_backend = io_backend
          self.kwargs = kwargs
          self.file_client = None
          random_string = get_random_string()
          thread_id = get_thread_id()
          self.tmp_folder = osp.join(get_shm_dir(),
                                     f'{random_string}_{thread_id}')
          os.mkdir(self.tmp_folder)
  
      def __call__(self, results):
          """Perform the OpenCV initiation.
  
          Args:
              results (dict): The resulting dict to be modified and passed
                  to the next transform in pipeline.
          """
          if self.io_backend == 'disk':
              new_path = results['filename']
          else:
              if self.file_client is None:
                  self.file_client = FileClient(self.io_backend, **self.kwargs)
  
              thread_id = get_thread_id()
              # save the file of same thread at the same place
              new_path = osp.join(self.tmp_folder, f'tmp_{thread_id}.mp4')
              with open(new_path, 'wb') as f:
                  f.write(self.file_client.get(results['filename']))
  
          container = mmcv.VideoReader(new_path)
          results['new_path'] = new_path
          results['video_reader'] = container
          results['total_frames'] = len(container)
  
          return results
  
      def __del__(self):
          shutil.rmtree(self.tmp_folder)
  
  
  @PIPELINES.register_module()
  class OpenCVDecode(object):
      """Using OpenCV to decode the video.
  
      Required keys are "video_reader", "filename" and "frame_inds", added
      or modified keys are "imgs", "img_shape" and "original_shape".
      """
  
      def __init__(self):
          pass
  
      def __call__(self, results):
          """Perform the OpenCV loading.
  
          Args:
              results (dict): The resulting dict to be modified and passed
                  to the next transform in pipeline.
          """
          container = results['video_reader']
          imgs = list()
  
          if results['frame_inds'].ndim != 1:
              results['frame_inds'] = np.squeeze(results['frame_inds'])
  
          for frame_ind in results['frame_inds']:
              cur_frame = container[frame_ind]
              # last frame may be None in OpenCV
              while isinstance(cur_frame, type(None)):
                  frame_ind -= 1
                  cur_frame = container[frame_ind]
              imgs.append(cur_frame)
  
          results['video_reader'] = None
          del container
  
          imgs = np.array(imgs)
          # The default channel order of OpenCV is BGR, thus we change it to RGB
          imgs = imgs[:, :, :, ::-1]
          results['imgs'] = list(imgs)
          results['original_shape'] = imgs[0].shape[:2]
          results['img_shape'] = imgs[0].shape[:2]
  
          return results
  
  
  @PIPELINES.register_module()
  class FrameSelector(object):
      """Select raw frames with given indices.
  
      Required keys are "frame_dir", "filename_tmpl" and "frame_inds",
      added or modified keys are "imgs", "img_shape" and "original_shape".
  
      Args:
          io_backend (str): IO backend where frames are stored. Default: 'disk'.
          decoding_backend (str): Backend used for image decoding.
              Default: 'cv2'.
          kwargs (dict, optional): Arguments for FileClient.
      """
  
      def __init__(self, io_backend='disk', decoding_backend='cv2', **kwargs):
          self.io_backend = io_backend
          self.decoding_backend = decoding_backend
          self.kwargs = kwargs
          self.file_client = None
  
      def __call__(self, results):
          """Perform the FrameSelector selecting given indices.
  
          Args:
              results (dict): The resulting dict to be modified and passed
                  to the next transform in pipeline.
          """
          mmcv.use_backend(self.decoding_backend)
  
          directory = results['frame_dir']
          filename_tmpl = results['filename_tmpl']
          modality = results['modality']
  
          if self.file_client is None:
              self.file_client = FileClient(self.io_backend, **self.kwargs)
  
          imgs = list()
  
          if results['frame_inds'].ndim != 1:
              results['frame_inds'] = np.squeeze(results['frame_inds'])
  
          for frame_idx in results['frame_inds']:
              if modality == 'RGB':
                  filepath = osp.join(directory, filename_tmpl.format(frame_idx))
                  img_bytes = self.file_client.get(filepath)
                  # Get frame with channel order RGB directly.
                  cur_frame = mmcv.imfrombytes(img_bytes, channel_order='rgb')
                  imgs.append(cur_frame)
              elif modality == 'Flow':
                  x_filepath = osp.join(directory,
                                        filename_tmpl.format('x', frame_idx))
                  y_filepath = osp.join(directory,
                                        filename_tmpl.format('y', frame_idx))
                  x_img_bytes = self.file_client.get(x_filepath)
                  x_frame = mmcv.imfrombytes(x_img_bytes, flag='grayscale')
                  y_img_bytes = self.file_client.get(y_filepath)
                  y_frame = mmcv.imfrombytes(y_img_bytes, flag='grayscale')
                  imgs.extend([x_frame, y_frame])
              else:
                  raise NotImplementedError
  
          results['imgs'] = imgs
          results['original_shape'] = imgs[0].shape[:2]
          results['img_shape'] = imgs[0].shape[:2]
  
          return results
  
  
  @PIPELINES.register_module()
  class LoadLocalizationFeature(object):
      """Load Video features for localizer with given video_name list.
  
      Required keys are "video_name" and "data_prefix",
      added or modified keys are "raw_feature".
  
      Args:
          raw_feature_ext (str): Raw feature file extension.  Default: '.csv'.
      """
  
      def __init__(self, raw_feature_ext='.csv'):
          valid_raw_feature_ext = ('.csv', )
          if raw_feature_ext not in valid_raw_feature_ext:
              raise NotImplementedError
          self.raw_feature_ext = raw_feature_ext
  
      def __call__(self, results):
          """Perform the LoadLocalizationFeature loading.
  
          Args:
              results (dict): The resulting dict to be modified and passed
                  to the next transform in pipeline.
          """
          video_name = results['video_name']
          data_prefix = results['data_prefix']
  
          data_path = osp.join(data_prefix, video_name + self.raw_feature_ext)
          raw_feature = np.loadtxt(
              data_path, dtype=np.float32, delimiter=',', skiprows=1)
  
          results['raw_feature'] = np.transpose(raw_feature, (1, 0))
  
          return results
  
  
  @PIPELINES.register_module()
  class GenerateLocalizationLabels(object):
      """Load video label for localizer with given video_name list.
  
      Required keys are "duration_frame", "duration_second",
      "feature_frame", "annotations", added or modified keys are
      "gt_bbox".
      """
  
      def __call__(self, results):
          """Perform the GenerateLocalizationLabels loading.
  
          Args:
              results (dict): The resulting dict to be modified and passed
                  to the next transform in pipeline.
          """
          video_frame = results['duration_frame']
          video_second = results['duration_second']
          feature_frame = results['feature_frame']
          corrected_second = float(feature_frame) / video_frame * video_second
          annotations = results['annotations']
  
          gt_bbox = []
  
          for annotation in annotations:
              current_start = max(
                  min(1, annotation['segment'][0] / corrected_second), 0)
              current_end = max(
                  min(1, annotation['segment'][1] / corrected_second), 0)
              gt_bbox.append([current_start, current_end])
  
          gt_bbox = np.array(gt_bbox)
          results['gt_bbox'] = gt_bbox
          return results
  
  
  @PIPELINES.register_module()
  class LoadProposals(object):
      """Loading proposals with given proposal results.
  
      Required keys are "video_name"
      added or modified keys are 'bsp_feature', 'tmin', 'tmax',
      'tmin_score', 'tmax_score' and 'reference_temporal_iou'.
  
      Args:
          top_k (int): The top k proposals to be loaded.
          pgm_proposals_dir (str): Directory to load proposals.
          pgm_features_dir (str): Directory to load proposal features.
          proposal_ext (str): Proposal file extension. Default: '.csv'.
          feature_ext (str): Feature file extension. Default: '.npy'.
      """
  
      def __init__(self,
                   top_k,
                   pgm_proposals_dir,
                   pgm_features_dir,
                   proposal_ext='.csv',
                   feature_ext='.npy'):
          self.top_k = top_k
          self.pgm_proposals_dir = pgm_proposals_dir
          self.pgm_features_dir = pgm_features_dir
          valid_proposal_ext = ('.csv', )
          if proposal_ext not in valid_proposal_ext:
              raise NotImplementedError
          self.proposal_ext = proposal_ext
          valid_feature_ext = ('.npy', )
          if feature_ext not in valid_feature_ext:
              raise NotImplementedError
          self.feature_ext = feature_ext
  
      def __call__(self, results):
          """Perform the LoadProposals loading.
  
          Args:
              results (dict): The resulting dict to be modified and passed
                  to the next transform in pipeline.
          """
          video_name = results['video_name']
          proposal_path = osp.join(self.pgm_proposals_dir,
                                   video_name + self.proposal_ext)
          if self.proposal_ext == '.csv':
              pgm_proposals = np.loadtxt(
                  proposal_path, dtype=np.float32, delimiter=',', skiprows=1)
  
          pgm_proposals = np.array(pgm_proposals[:self.top_k])
          tmin = pgm_proposals[:, 0]
          tmax = pgm_proposals[:, 1]
          tmin_score = pgm_proposals[:, 2]
          tmax_score = pgm_proposals[:, 3]
          reference_temporal_iou = pgm_proposals[:, 5]
  
          feature_path = osp.join(self.pgm_features_dir,
                                  video_name + self.feature_ext)
          if self.feature_ext == '.npy':
              bsp_feature = np.load(feature_path).astype(np.float32)
  
          bsp_feature = bsp_feature[:self.top_k, :]
  
          results['bsp_feature'] = bsp_feature
          results['tmin'] = tmin
          results['tmax'] = tmax
          results['tmin_score'] = tmin_score
          results['tmax_score'] = tmax_score
          results['reference_temporal_iou'] = reference_temporal_iou
  
          return results
  