code: def format_entities(
      events: DataframeLike,
      entity_types: List[str],
      defs: HyperBindings,
      direct: bool,
      drop_na: bool,
      engine: Engine,
      npartitions: Optional[int],
      chunksize: Optional[int],
      debug: bool = False,
  ) -> DataframeLike:
      logger.debug("@format_entities :: %s", entity_types)
      logger.debug("dtypes: %s", events.dtypes)
  
      cat_lookup = make_reverse_lookup(defs.categories)
      logger.debug(
          "@format_entities cat_lookup [ %s ] => [ %s ]",
          defs.categories,
          cat_lookup,
      )
  
      mt_df = mt_nodes(defs, events, entity_types, direct, engine)
      logger.debug("mt_df :: %s", mt_df.dtypes)
  
      entity_dfs = [
          format_entities_from_col(
              defs,
              cat_lookup,
              drop_na,
              engine,
              col_name,
              events[[col_name]],
              mt_df,
              debug,
          )
          for col_name in entity_types
      ]
      if debug and (engine in [Engine.DASK, Engine.DASK_CUDF]):
          entity_dfs = [df.persist() for df in entity_dfs]
          for df in entity_dfs:
              logger.debug(
                  "format_entities sub df dtypes: %s", df.dtypes
              )
              if df.dtypes.to_dict() != entity_dfs[0].dtypes.to_dict():
                  logger.error("MISMATCHES")
                  d1 = df.dtypes.to_dict()
                  d2 = entity_dfs[0].dtypes.to_dict()
                  for k, v in d1.items():
                      if k not in d2:
                          logger.error(
                              "key %s (::%s) missing in df_0", k, v
                          )
                      elif d2[k] != v:
                          logger.error("%s:%s <> %s:%s", k, v, k, d2[k])
                  for k, v in d2.items():
                      if k not in d1:
                          logger.error(
                              "key %s (::%s) missing in df_i", k, v
                          )
              logger.debug("entity_df: %s", df.compute())
  
      df = concat(entity_dfs, engine, debug).drop_duplicates(
          [defs.node_id]
      )
      if debug and (engine in [Engine.DASK, Engine.DASK_CUDF]):
          df = df.persist()
          df.compute()
          logger.debug("////format_entities")
  
      return df
  
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 13 for calling a global function with example usage: format_entities(events, entity_types, defs, direct, drop_na, engine, npartitions, chunksize, debug) and returns: df