class_docstr: Duck-typed version of PyG's GraphStore and FeatureStore.
class_docstr_tok: ["Duck-typed","version","of","PyG","'s","GraphStore","and","FeatureStore","."]
docstr: This function takes edge information and uses it to construct
  a cugraph Graph.  It determines the numerical edge type by
  sorting the keys of the input dictionary
  (the canonical edge types).
  
  Parameters
  ----------
  edge_info: Dict[Tuple[str, str, str], List[TensorType]] (Required)
      Input edge info dictionary, where keys are the canonical
      edge type and values are the edge index (src/dst).
  
  multi_gpu: bool (Optional, default=False)
      Whether to construct a single-GPU or multi-GPU cugraph Graph.
      Defaults to a single-GPU graph.
  
  order: str (CSC or CSR)
      Essentially whether to reverse edges so that the cuGraph
      sampling algorithm operates on the CSC matrix instead of
      the CSR matrix.  Should nearly always be CSC unless there
      is a specific expectation of reverse sampling, or correctness
      testing is being performed.
  
  Returns
  -------
  A newly-constructed directed cugraph.MultiGraph object.
docstr_tok: ["This","function","takes","edge","information","and","uses","it","to","construct","a","cugraph","Graph",".","It","determines","the","numerical","edge","type","by","sorting","the","keys","of","the","input","dictionary","(","the","canonical","edge","types",")",".","Parameters","--","--","--","--","--","edge_info",":","Dict","[","Tuple","[","str",",","str",",","str","]",",","List","[","TensorType","]","]","(","Required",")","Input","edge","info","dictionary",",","where","keys","are","the","canonical","edge","type","and","values","are","the","edge","index","(","src\/dst",")",".","multi_gpu",":","bool","(","Optional",",","default=False",")","Whether","to","construct","a","single-GPU","or","multi-GPU","cugraph","Graph",".","Defaults","to","a","single-GPU","graph",".","order",":","str","(","CSC","or","CSR",")","Essentially","whether","to","reverse","edges","so","that","the","cuGraph","sampling","algorithm","operates","on","the","CSC","matrix","instead","of","the","CSR","matrix",".","Should","nearly","always","be","CSC","unless","there","is","a","specific","expectation","of","reverse","sampling",",","or","correctness","testing","is","being","performed",".","Returns","--","--","--","-","A","newly-constructed","directed","cugraph.MultiGraph","object","."]
code: def __construct_graph(
      self,
      edge_info: Dict[Tuple[str, str, str], List[TensorType]],
      multi_gpu: bool = False,
      order: str = "CSC",
  ) -> cugraph.MultiGraph:
      """
      This function takes edge information and uses it to construct
      a cugraph Graph.  It determines the numerical edge type by
      sorting the keys of the input dictionary
      (the canonical edge types).
  
      Parameters
      ----------
      edge_info: Dict[Tuple[str, str, str], List[TensorType]] (Required)
          Input edge info dictionary, where keys are the canonical
          edge type and values are the edge index (src/dst).
  
      multi_gpu: bool (Optional, default=False)
          Whether to construct a single-GPU or multi-GPU cugraph Graph.
          Defaults to a single-GPU graph.
  
      order: str (CSC or CSR)
          Essentially whether to reverse edges so that the cuGraph
          sampling algorithm operates on the CSC matrix instead of
          the CSR matrix.  Should nearly always be CSC unless there
          is a specific expectation of reverse sampling, or correctness
          testing is being performed.
  
      Returns
      -------
      A newly-constructed directed cugraph.MultiGraph object.
      """
  
      # Ensure the original dict is not modified.
      edge_info_cg = {}
  
      if order != "CSR" and order != "CSC":
          raise ValueError("Order must be either CSC (default) or CSR!")
  
      # Iterate over the keys in sorted order so that the created
      # numerical types correspond to the lexicographic order
      # of the keys, which is critical to converting the numeric
      # keys back to canonical edge types later.
      # FIXME don't always convert to host arrays (#3383)
      for pyg_can_edge_type in sorted(edge_info.keys()):
          src_type, _, dst_type = pyg_can_edge_type
          srcs, dsts = edge_info[pyg_can_edge_type]
  
          src_offset = np.searchsorted(
              self.__vertex_type_offsets["type"], src_type
          )
          srcs_t = srcs + int(
              self.__vertex_type_offsets["start"][src_offset]
          )
          if isinstance(srcs_t, torch.Tensor):
              srcs_t = srcs_t.cpu()
          else:
              if isinstance(srcs_t, dask_cudf.Series):
                  srcs_t = srcs_t.compute()
              if isinstance(srcs_t, cudf.Series):
                  srcs_t = srcs_t.values_host
  
          dst_offset = np.searchsorted(
              self.__vertex_type_offsets["type"], dst_type
          )
          dsts_t = dsts + int(
              self.__vertex_type_offsets["start"][dst_offset]
          )
          if isinstance(dsts_t, torch.Tensor):
              dsts_t = dsts_t.cpu()
          else:
              if isinstance(dsts_t, dask_cudf.Series):
                  dsts_t = dsts_t.compute()
              if isinstance(dsts_t, cudf.Series):
                  dsts_t = dsts_t.values_host
  
          edge_info_cg[pyg_can_edge_type] = (srcs_t, dsts_t)
  
      na_src = np.concatenate(
          [
              edge_info_cg[pyg_can_edge_type][0]
              for pyg_can_edge_type in sorted(edge_info_cg.keys())
          ]
      )
  
      na_dst = np.concatenate(
          [
              edge_info_cg[pyg_can_edge_type][1]
              for pyg_can_edge_type in sorted(edge_info_cg.keys())
          ]
      )
  
      et_offsets = self.__edge_type_offsets
      na_etp = np.concatenate(
          [
              np.full(
                  int(
                      et_offsets["stop"][i] - et_offsets["start"][i] + 1
                  ),
                  i,
                  dtype="int32",
              )
              for i in range(len(self.__edge_type_offsets["start"]))
          ]
      )
  
      vertex_dtype = na_src.dtype
  
      if multi_gpu:
          client = distributed.get_client()
          nworkers = len(client.scheduler_info()["workers"])
          npartitions = nworkers * 4
  
          src_dar = self.__dask_array_from_numpy(na_src, npartitions)
          del na_src
  
          dst_dar = self.__dask_array_from_numpy(na_dst, npartitions)
          del na_dst
  
          etp_dar = self.__dask_array_from_numpy(na_etp, npartitions)
          del na_etp
  
          df = dd.from_dask_array(etp_dar, columns=["etp"])
          df["src"] = dst_dar if order == "CSC" else src_dar
          df["dst"] = src_dar if order == "CSC" else dst_dar
  
          del src_dar
          del dst_dar
          del etp_dar
  
          if df.etp.dtype != "int32":
              raise ValueError("Edge type must be int32!")
  
          # Ensure the dataframe is constructed on each partition
          # instead of adding additional synchronization head from potential
          # host to device copies.
          def get_empty_df():
              return cudf.DataFrame(
                  {
                      "etp": cudf.Series([], dtype="int32"),
                      "src": cudf.Series([], dtype=vertex_dtype),
                      "dst": cudf.Series([], dtype=vertex_dtype),
                  }
              )
  
          # Have to check for empty partitions and handle them appropriately
          df = df.persist()
          df = df.map_partitions(
              lambda f: cudf.DataFrame.from_pandas(f)
              if len(f) > 0
              else get_empty_df(),
              meta=get_empty_df(),
          ).reset_index(
              drop=True
          )  # should be ok for dask
      else:
          df = pandas.DataFrame(
              {
                  "src": pandas.Series(na_dst)
                  if order == "CSC"
                  else pandas.Series(na_src),
                  "dst": pandas.Series(na_src)
                  if order == "CSC"
                  else pandas.Series(na_dst),
                  "etp": pandas.Series(na_etp),
              }
          )
          df = cudf.from_pandas(df)
          df.reset_index(drop=True, inplace=True)
  
      graph = cugraph.MultiGraph(directed=True)
      if multi_gpu:
          graph.from_dask_cudf_edgelist(
              df,
              source="src",
              destination="dst",
              edge_type="etp",
          )
          distributed.get_client().publish_dataset(cugraph_graph=graph)
      else:
          graph.from_cudf_edgelist(
              df,
              source="src",
              destination="dst",
              edge_type="etp",
          )
  
      del df
      return graph
  
imports: from typing import Optional, Tuple, Any, Union, List, Dict
  from enum import Enum, auto
  from dataclasses import dataclass
  from collections import defaultdict
  from itertools import chain
  from functools import cached_property
  import numpy
  import cupy
  import pandas
  import cudf
  import cugraph
  import warnings
  import dask.array
  import dask.dataframe
  import dask.distributed
  import dask_cudf
  from cugraph.utilities.utils import import_optional, MissingModule
label_desc: Use image node_id 6 for calling the CuGraphStore obj's underlying member method code with example usage: obj.__construct_graph(edge_info, multi_gpu, order) and returns: graph, cudf