class_docstr: Emotion Task 
docstr: map filename w/o suffix to segment samples
  {hxx/xxx/abc:[ [200, 300], ...]}
docstr_tok: ["map","filename","w\/o","suffix","to","segment","samples","{","hxx\/xxx\/abc",":","[","[","200",",","300","]",",","...","]","}"]
code: def class_features(self, segments_path):
      """map filename w/o suffix to segment samples
      {hxx/xxx/abc:[ [200, 300], ...]}
      """
      self._postive_segs = defaultdict(list)
      self._postive_segs_in_sec = defaultdict(list)
  
      # pylint: disable=invalid-name
      def seconds_to_samples(segs_list, pos_example=True, sr=8000):
          """
          Args:
             segs_list: list of tuple (min, max), unit seconds
             pos_example:
                if True, POS segments in Positive sentence
                else, NEG segments in Positive sentence
          Return:
          segs in samples
          """
  
          def _convert(second_range):
              """second_range, tuple(min, max), unit sencods"""
              low, hight = list(
                  map(
                      functools.partial(librosa.time_to_samples, sr=sr),
                      second_range,
                  )
              )
              return (low, hight)
  
          segs_list = list(map(_convert, segs_list))
  
          assert (
              pos_example
          ), "Now only use POS segment in POSITIVE sentece"
          if pos_example:
              # positive segment in positive sentence
              return segs_list
  
          # negtive segment in positive sentence
          res = []
          flat_list = (
              [0]
              + list(itertools.chain.from_iterable(segs_list))
              + [None]
          )
          for _ in range(0, len(flat_list), 2):
              # if flat_list[0] != flat_list[1]:
              # pylint: disable=invalid-name
              s, e = flat_list[0], flat_list[1]
              if s != e:
                  res.append((s, e))
          return res
  
      def load_segments(segments_file):
          """load TextGrid file"""
          if self._postive_segs:
              logging.debug(
                  "TextGrid has been loaded: {}".format(segments_file)
              )
              return
  
          if segments_file:
              logging.debug(
                  "TextGrid loading: {}".format(segments_file)
              )
              for segment_file in segments_file:
                  with open(
                      segment_file
                  ) as f:  # pylint: disable=invalid-name
                      for line in f:
                          filename, segs = line.split(maxsplit=1)
                          # using abspath of filename, without file suffix
                          filename = os.path.splitext(filename)[0]
                          segs = list(
                              map(ast.literal_eval, segs.split())
                          )
                          self._postive_segs_in_sec[filename].extend(
                              segs
                          )
                          # second -> sample
                          segs = seconds_to_samples(segs)
                          self._postive_segs[filename].extend(segs)
              # logging.info(self._postive_segs)
  
      def extract_segments(filename, segments_file):
          """
          Brief:
            extract segments from filename with TextGrid file
            if `filename` in TextGrid file, then extract segments
            else load hole file samples
          Args:
            filename: wavfile name, abspath
            segments_file: TextGrid path
          Return :
              list of segments index (min, max), unit sample
          """
          # load TextGrid file
          load_segments(segments_file)
  
          seg_flag = False  # whther using segment or whole file
          segs = []
          # using abspath without file suffix
          filename = os.path.splitext(filename)[0]
          if filename in self._postive_segs:
              # segement wave file
              # now only positive example has TextGrid file
              segs_index = self._postive_segs[filename]
              segs.extend(segs_index)
              seg_flag = True
          else:
              # return whole file
              segs.append((0, None))
              seg_flag = False
          logging.debug(
              "filename: {} segs_index:{} seg_or_whole:{}".format(
                  filename, segs, seg_flag
              )
          )
          return segs, seg_flag
  
      # pylint: disable=too-many-locals,invalid-name
      def segment_indexes(
          filename, segments_file, clip_size=3, stride=0.5, sr=8000
      ):
          """generate clip index in samples"""
          duration = self.get_duration(
              filename=filename, sr=self._sample_rate
          )
          (
              clip_samples,
              stride_samples,
              total_samples,
          ) = librosa.time_to_samples(
              [clip_size, clip_size * stride, duration], sr=sr
          )
  
          segs_index, _ = extract_segments(filename, segments_file)
          seg_index = []
  
          # pylint: disable=invalid-name
          for st, ed in segs_index:
              if ed is None:
                  ed = librosa.time_to_samples(duration, sr=sr)
                  logging.debug("file end: {} {}".format(duration, ed))
              seg_len = ed - st
  
              sub_segs = []
              if seg_len > clip_samples:
                  nstride = int(
                      (seg_len - clip_samples) / stride_samples + 1
                  )
                  # (strart, end, npad), only support right hand pad
                  sub_segs = [
                      (
                          i * stride_samples,
                          i * stride_samples + clip_samples,
                          0,
                      )
                      for i in range(nstride)
                  ]
                  seg_index.extend(sub_segs)
              else:
                  gap = clip_samples - seg_len
                  if st - gap >= 0:
                      assert ed - (st - gap) == clip_samples
                      sub_segs.append((st - gap, ed, 0))
                  if ed + gap <= total_samples:
                      assert ed + gap - st == clip_samples
                      sub_segs.append((st, ed + gap, 0))
                  else:
                      npad = ed + gap - total_samples
                      assert ed + gap - st == clip_samples
                      sub_segs.append((st, ed + gap, npad))
                  seg_index.extend(sub_segs)
              logging.debug(
                  "Input segs: {}, Output segs: {}".format(
                      (st, ed), sub_segs
                  )
              )
          return seg_index
  
      def _generate_online():
          """generate clips of file"""
          class_file = self._class_file_train
          # file, duration, class
          for filename, duration, label in class_file:
              del duration
              # [ (start, end, npad)...]
              seg_idxs = segment_indexes(
                  filename,
                  segments_path,
                  clip_size=self._clip_size,
                  stride=self._stride,
                  sr=self._sample_rate,
              )
  
              for clip_id, seg in enumerate(seg_idxs):
                  example = (filename, label, seg, clip_id)
                  self._train_meta.append(example)
                  self._train_label_example[label].append(example)
  
      self._train_meta = []
      self._train_label_example = defaultdict(list)
  
      _generate_online()
  
imports: import re
  import ast
  import os
  import copy
  import threading
  import itertools
  import functools
  import random
  from collections import defaultdict
  import librosa
  import numpy
  import pandas
  import delta.compat
  from absl import logging
  from sklearn.model_selection import train_test_split
  from delta import utils
  from delta.data import feat
  from delta.utils.register import registers
  from delta.data.task.base_speech_task import SpeechTask
label_desc: Use image node_id 19 for calling the SpeechClsTask obj's underlying member method code with example usage: obj.class_features(segments_path) and returns: res, segs,  seg_flag, seg_index, low,  hight, segs_list