code: def _ft_iterator(
      self,
      v,
      min_chunk_size,
      from_pos,
      to_pos,
      dt_unit,
      ft_feature,
      coldtypedic,
      transfer_features,
      max_pairs,
      verboseprint,
      logfile,
      hdf_key,
  ):
      # fast track feature references
      ftf = ft_feature[0]
      ftt = ft_feature[1]
  
      # if hdf, find requested features
      if isinstance(v, pd.HDFStore):
          v_is_hdf = True
          rf = [
              transfer_features,
              CreatorFunction.c_input_features,
              CreatorFunction.s_input_features,
          ]
          rf = set(
              [
                  feature[:-2]
                  if feature.endswith("_s") or feature.endswith("_t")
                  else feature
                  for feature in _flatten(rf)
              ]
          )
          rf.remove("ft_feature")
          rf.add(ftf)
      else:
          v_is_hdf = False
  
      # number of nodes
      if v_is_hdf:
          N = v.get_storer(hdf_key).nrows
      else:
          N = len(v)
  
      # iteration parameters
      n = min_chunk_size
      i = from_pos
      if to_pos is None:
          to_pos = N
          to_pos_default = True
      else:
          assert to_pos <= N, (
              "the given to_pos parameter is too large, "
              "{} > len(v)={}".format(to_pos, N)
          )
          to_pos_default = False
  
      assert from_pos < N, (
          "the given from_pos argument is too large, "
          "{} >= len(v)".format(from_pos)
      )
      assert from_pos < to_pos, "to_pos must be larger than from_pos"
  
      # cumulatively count the generated edges
      cum_edges = 0
  
      # for testing / logging
      self._triggered = {
          "large_enough": 0,
          "increase_p1d": 0,
          "increased_leq_N": 0,
          "increased_end_of_table": 0,
          "gap_cases": 0,
      }
  
      # create ei list to append to
      ei_list = []
      # start iteration
      while i < to_pos:
          # measure time per iteration
          starttime = datetime.now()
  
          # select partial fast track feature
          if v_is_hdf:
              vi = v.select(
                  hdf_key, start=i, stop=i + n, columns=[ftf]
              )[ftf].values
          else:
              vi = v.iloc[i : i + n][ftf].values
  
          # compute max p1 difference of vi, if p1d <= ftt, increase chunk size
          ftf_first = vi[0]
          ftf_last = vi[-1]
          if dt_unit is None:
              p1d = ftf_last - ftf_first
          else:
              p1d = (ftf_last - ftf_first) / np.timedelta64(1, dt_unit)
  
          # ================================================================
          # case 1) min_chunk_size large enough
          if p1d > ftt:
              # for testing
              self._triggered["large_enough"] += 1
  
              # take one more node here, since 'trapez' always takes away the
              # last node
              if v_is_hdf:
                  vi = v.select(
                      hdf_key, start=i, stop=i + n + 1, columns=rf
                  )
              else:
                  vi = v.iloc[i : i + n + 1]
  
              ei, ns = _ft_create_ei(
                  self,
                  vi,
                  ft_feature,
                  dt_unit,
                  coldtypedic,
                  transfer_features,
                  max_pairs,
                  verboseprint,
                  logfile,
                  symmetry="trapez",
              )
  
              ei_list.extend(ei)
  
              cum_edges += _count_edges(ei)
              verboseprint(" processed sources:", [ns])
              verboseprint(" mapped with targets:", [len(vi) - 1])
              verboseprint(" pos interval:", [i, i + ns])
              verboseprint(" nr of nodes (total):", [N])
              verboseprint(
                  " ft_feature of last source:",
                  [vi.at[vi.iloc[ns - 1].name, ftf]],
              )
              verboseprint(
                  " nr of edges:",
                  [_count_edges(ei)],
                  ", cum nr of edges:",
                  [cum_edges],
              )
              verboseprint(" copied rs: {}".format(_copied_rs(ei)))
  
              i += ns
  
          # ================================================================
          # case 2) increase vi, s.t. p1d > ftt
          # when increasing, form only pairs with p1d <= ftt, no excessive ones
          else:
              # for testing
              self._triggered["increase_p1d"] += 1
  
              verboseprint(
                  "min_chunk_size too small, increasing partial v.."
              )
  
              # again, include the first node with ftf > ftf_first + ftt to
              # pass to _ft_create_ei, which will use it and then get rid of it.
              if dt_unit is None:
                  if v_is_hdf:
                      where = "{} <= {}".format(ftf, ftf_first + ftt)
                      upto = (
                          v.select_as_coordinates(
                              hdf_key, where=where, start=i
                          )[-1]
                          + 2
                      )
                  else:
                      upto = (
                          i
                          + np.searchsorted(
                              v[ftf].values[i:],
                              ftf_first + ftt,
                              side="right",
                          )
                          + 1
                      )
              else:
                  if v_is_hdf:
                      # is there a better way then converting to timestamp?
                      ts = pd.Timestamp(
                          ftf_first + np.timedelta64(ftt, dt_unit)
                      )
                      where = "{} <= {!r}".format(ftf, ts)
                      upto = (
                          v.select_as_coordinates(
                              hdf_key, where=where, start=i
                          )[-1]
                          + 2
                      )
                  else:
                      ts = ftf_first + np.timedelta64(ftt, dt_unit)
                      upto = (
                          i
                          + np.searchsorted(
                              v[ftf].values[i:], ts, side="right"
                          )
                          + 1
                      )
  
              if upto <= N:
                  # for testing
                  self._triggered["increased_leq_N"] += 1
  
                  if v_is_hdf:
                      vi = v.select(
                          hdf_key, start=i, stop=upto, columns=rf
                      )
                  else:
                      vi = v.iloc[i:upto]
  
                  ei, ns = _ft_create_ei(
                      self,
                      vi,
                      ft_feature,
                      dt_unit,
                      coldtypedic,
                      transfer_features,
                      max_pairs,
                      verboseprint,
                      logfile,
                      symmetry="trapez",
                  )
  
                  ei_list.extend(ei)
  
                  cum_edges += _count_edges(ei)
                  verboseprint(" processed sources:", [ns])
                  verboseprint(" mapped with targets:", [upto - i - 1])
                  verboseprint(" pos interval:", [i, i + ns])
                  verboseprint(" nr of nodes (total):", [N])
                  verboseprint(
                      " ft_feature of last source:",
                      [vi.at[vi.iloc[ns - 1].name, ftf]],
                  )
                  verboseprint(
                      " nr of edges:",
                      [_count_edges(ei)],
                      ", cum nr of edges:",
                      [cum_edges],
                  )
                  verboseprint(" copied rs: {}".format(_copied_rs(ei)))
  
                  i += ns
  
              # ============================================================
              # case 3) end of table, compute upper triangle matrix
              else:
                  # for testing
                  self._triggered["increased_end_of_table"] += 1
  
                  if v_is_hdf:
                      vi = v.select(hdf_key, start=i, columns=rf)
                  else:
                      vi = v.iloc[i:]
  
                  ei, ns = _ft_create_ei(
                      self,
                      vi,
                      ft_feature,
                      dt_unit,
                      coldtypedic,
                      transfer_features,
                      max_pairs,
                      verboseprint,
                      logfile,
                      symmetry="triangle",
                  )
  
                  ei_list.extend(ei)
  
                  cum_edges += _count_edges(ei)
                  verboseprint(
                      "# LAST",
                      [len(vi)],
                      "EVENTS PROCESSED (END OF TABLE)",
                  )
                  verboseprint(
                      "# ====================================================="
                  )
                  verboseprint(" processed sources:", [ns])
                  verboseprint(" mapped with targets:", [ns])
                  verboseprint(" pos interval:", [i, i + len(vi)])
                  verboseprint(" nr of nodes (total):", [N])
                  verboseprint(
                      " ft_feature of last source:",
                      [vi.at[vi.iloc[-1].name, ftf]],
                  )
                  verboseprint(
                      " nr of edges:",
                      [_count_edges(ei)],
                      ", cum nr of edges:",
                      [cum_edges],
                  )
                  verboseprint(" copied rs: {}".format(_copied_rs(ei)))
  
                  i += ns
  
          timediff = datetime.now() - starttime
          verboseprint(
              " computation time:",
              "\ts =",
              int(timediff.total_seconds()),
              "\tms =",
              str(timediff.microseconds / 1000.0)[:6],
              "\n",
          )
          # logging
          if logfile:
              with open(logfile, "a") as log:
                  print(
                      "{}\t{:.3f}".format(
                          _count_edges(ei), timediff.total_seconds()
                      ),
                      file=log,
                  )
  
      # BUG: concatenating changes dtypes of float16 and float32
      # to float64 if there is an empty frame in ei_list!
      # concat ei_list
      e = pd.concat(ei_list, ignore_index=True, copy=False)
  
      # set indices
      e.set_index(["s", "t"], inplace=True)
  
      # delete excessive sources (only return sources up to to_pos)
      # PERFORMANCE (look for better solution, not 'isin'...)
      if to_pos_default is False:
          if v_is_hdf:
              indices = v.select_column(
                  hdf_key, "index", start=from_pos, stop=to_pos
              ).values
          else:
              indices = v.iloc[from_pos:to_pos].index.values
  
          s = e.index.get_level_values(level=0)
          e = e.loc[s.isin(indices)]
  
      return e
  
file_path: deepgraph/deepgraph/iterators_and_indexers.py
imports: from datetime import datetime
  from itertools import chain
  import numpy
  import pandas
  from deepgraph._triu_indices import _triu_indices, _union_of_indices, _reduce_triu_indices
  from deepgraph.connector_selector_implementations import _ft_connector, _ft_selector
  from deepgraph.connectors_and_selectors import CreatorFunction, Selector, Connector
  from deepgraph.utils import _flatten, _is_array_like, _merge_dicts
label_desc: Use image node_id 6 for calling a global function with example usage: _ft_iterator(v, min_chunk_size, from_pos, to_pos, dt_unit, ft_feature, coldtypedic, transfer_features, max_pairs, verboseprint, logfile, hdf_key) and returns: e