code: def select_topk(
      g,
      k,
      weight,
      nodes=None,
      edge_dir="in",
      ascending=False,
      copy_ndata=True,
      copy_edata=True,
      output_device=None,
  ):
      """Select the neighboring edges with k-largest (or k-smallest) weights of the given
      nodes and return the induced subgraph.
  
      For each node, a number of inbound (or outbound when ``edge_dir == 'out'``) edges
      with the largest (or smallest when ``ascending == True``) weights will be chosen.
      The graph returned will then contain all the nodes in the original graph, but only
      the sampled edges.
  
      Node/edge features are not preserved. The original IDs of
      the sampled edges are stored as the `dgl.EID` feature in the returned graph.
  
      Parameters
      ----------
      g : DGLGraph
          The graph.  Must be on CPU.
      k : int or dict[etype, int]
          The number of edges to be selected for each node on each edge type.
  
          This argument can take a single int or a dictionary of edge types and ints.
          If a single int is given, DGL will select this number of edges for each node for
          every edge type.
  
          If -1 is given for a single edge type, all the neighboring edges with that edge
          type will be selected.
      weight : str
          Feature name of the weights associated with each edge.  The feature should have only
          one element for each edge.  The feature can be either int32/64 or float32/64.
      nodes : tensor or dict, optional
          Node IDs to sample neighbors from.
  
          This argument can take a single ID tensor or a dictionary of node types and ID tensors.
          If a single tensor is given, the graph must only have one type of nodes.
  
          If None, DGL will select the edges for all nodes.
      edge_dir : str, optional
          Determines whether to sample inbound or outbound edges.
  
          Can take either ``in`` for inbound edges or ``out`` for outbound edges.
      ascending : bool, optional
          If True, DGL will return edges with k-smallest weights instead of
          k-largest weights.
      copy_ndata: bool, optional
          If True, the node features of the new graph are copied from
          the original graph. If False, the new graph will not have any
          node features.
  
          (Default: True)
      copy_edata: bool, optional
          If True, the edge features of the new graph are copied from
          the original graph.  If False, the new graph will not have any
          edge features.
  
          (Default: True)
      output_device : Framework-specific device context object, optional
          The output device.  Default is the same as the input graph.
  
      Returns
      -------
      DGLGraph
          A sampled subgraph containing only the sampled neighboring edges.  It is on CPU.
  
      Notes
      -----
      If :attr:`copy_ndata` or :attr:`copy_edata` is True, same tensors are used as
      the node or edge features of the original graph and the new graph.
      As a result, users should avoid performing in-place operations
      on the node features of the new graph to avoid feature corruption.
  
      Examples
      --------
      >>> g = dgl.graph(([0, 0, 1, 1, 2, 2], [1, 2, 0, 1, 2, 0]))
      >>> g.edata['weight'] = torch.FloatTensor([0, 1, 0, 1, 0, 1])
      >>> sg = dgl.sampling.select_topk(g, 1, 'weight')
      >>> sg.edges(order='eid')
      (tensor([2, 1, 0]), tensor([0, 1, 2]))
      """
      # Rectify nodes to a dictionary
      if nodes is None:
          nodes = {
              ntype: F.astype(F.arange(0, g.num_nodes(ntype)), g.idtype)
              for ntype in g.ntypes
          }
      elif not isinstance(nodes, dict):
          if len(g.ntypes) > 1:
              raise DGLError(
                  "Must specify node type when the graph is not homogeneous."
              )
          nodes = {g.ntypes[0]: nodes}
      assert g.device == F.cpu(), "Graph must be on CPU."
  
      # Parse nodes into a list of NDArrays.
      nodes = utils.prepare_tensor_dict(g, nodes, "nodes")
      device = utils.context_of(nodes)
      nodes_all_types = []
      for ntype in g.ntypes:
          if ntype in nodes:
              nodes_all_types.append(F.to_dgl_nd(nodes[ntype]))
          else:
              nodes_all_types.append(nd.array([], ctx=nd.cpu()))
  
      if not isinstance(k, dict):
          k_array = [int(k)] * len(g.etypes)
      else:
          if len(k) != len(g.etypes):
              raise DGLError(
                  "K value must be specified for each edge type "
                  "if a dict is provided."
              )
          k_array = [None] * len(g.etypes)
          for etype, value in k.items():
              k_array[g.get_etype_id(etype)] = value
      k_array = F.to_dgl_nd(F.tensor(k_array, dtype=F.int64))
  
      weight_arrays = []
      for etype in g.canonical_etypes:
          if weight in g.edges[etype].data:
              weight_arrays.append(
                  F.to_dgl_nd(g.edges[etype].data[weight])
              )
          else:
              raise DGLError(
                  'Edge weights "{}" do not exist for relation graph "{}".'.format(
                      weight, etype
                  )
              )
  
      subgidx = _CAPI_DGLSampleNeighborsTopk(
          g._graph,
          nodes_all_types,
          k_array,
          edge_dir,
          weight_arrays,
          bool(ascending),
      )
      induced_edges = subgidx.induced_edges
      ret = DGLGraph(subgidx.graph, g.ntypes, g.etypes)
  
      # handle features
      if copy_ndata:
          node_frames = utils.extract_node_subframes(g, device)
          utils.set_new_frames(ret, node_frames=node_frames)
  
      if copy_edata:
          edge_frames = utils.extract_edge_subframes(g, induced_edges)
          utils.set_new_frames(ret, edge_frames=edge_frames)
      return ret if output_device is None else ret.to(output_device)
  
file_path: dgl/python/dgl/sampling/neighbor.py
imports: import os
  import torch
  from ..None import backend, ndarray, utils
  from .._ffi.function import _init_api
  from ..base import DGLError, EID
  from ..heterograph import DGLBlock, DGLGraph
  from .utils import EidExcluder
label_desc: Use image node_id 7 for calling a global function with example usage: select_topk(g, k, weight, nodes, edge_dir, ascending, copy_ndata, copy_edata, output_device) and returns: unknown