code: def format_direct_edges(
      engine: Engine,
      events: DataframeLike,
      entity_types,
      defs: HyperBindings,
      edge_shape,
      drop_na: bool,
      drop_edge_attrs: bool,
      debug: bool = False,
  ) -> DataframeLike:
      is_using_categories = len(defs.categories.keys()) > 0
      cat_lookup = make_reverse_lookup(defs.categories)
  
      subframes = []
      for col1 in sorted(edge_shape.keys()):
          for col2 in sorted(edge_shape[col1]):
              fields = list(
                  set(
                      [defs.event_id]
                      + (
                          [x for x in events.columns]
                          if not drop_edge_attrs
                          else [col1, col2]
                      )
                  )
              )
              raw = events[fields]
              if drop_na:
                  raw = raw.dropna(subset=[col1, col2])
              raw = raw.copy()
              if is_using_categories:
                  raw[defs.edge_type] = (
                      col2cat(cat_lookup, col1)
                      + defs.delim
                      + col2cat(cat_lookup, col2)
                  )
                  raw[defs.category] = col1 + defs.delim + col2
              else:
                  raw[defs.edge_type] = col1 + defs.delim + col2
              raw[defs.source] = (
                  col2cat(cat_lookup, col1) + defs.delim
              ) + raw[col1].astype(str).fillna(defs.null_val)
              raw[defs.destination] = (
                  col2cat(cat_lookup, col2) + defs.delim
              ) + raw[col2].astype(str).fillna(defs.null_val)
              if drop_edge_attrs:
                  raw = raw.drop(columns=[col1, col2])
              if debug and (engine in [Engine.DASK, Engine.DASK_CUDF]):
                  raw = raw.persist()
                  raw.compute()
              subframes.append(raw)
  
      if len(subframes):
          result_cols = list(
              set(
                  (
                      [
                          x
                          for x in events.columns.tolist()
                          if not x == defs.node_type
                      ]
                      if not drop_edge_attrs
                      else []
                  )
                  + [
                      defs.edge_type,
                      defs.source,
                      defs.destination,
                      defs.event_id,
                  ]  # noqa: W503
                  + ([defs.category] if is_using_categories else [])
              )
          )  # noqa: W503
          if debug and (engine in [Engine.DASK, Engine.DASK_CUDF]):
              # subframes = [ df.persist() for df in subframes ]
              for df in subframes:
                  logger.debug(
                      "format_direct_edges subdf: %s", df.dtypes
                  )
          out = concat(subframes, engine=engine, debug=debug)[
              result_cols
          ]
          if debug and (engine in [Engine.DASK, Engine.DASK_CUDF]):
              out = out.persist()
              out.compute()
              logger.debug("////format_direct_edges")
          return out
      else:
          return events[:0][[]]
  
imports: from typing import TYPE_CHECKING, Any, Dict, List, Optional
  from .Engine import Engine, DataframeLike, DataframeLocalLike
  import numpy, pandas, pyarrow, sys
  from .util import setup_logger
label_desc: Use image node_id 16 for calling a global function with example usage: format_direct_edges(engine, events, entity_types, defs, edge_shape, drop_na, drop_edge_attrs, debug) and returns: out, events