code: def weakly_connected_components(input_graph):
      """
      Generate the Weakly Connected Components and attach a component label to
      each vertex.
  
      Parameters
      ----------
      input_graph : cugraph.Graph
          The graph descriptor should contain the connectivity information
          and weights. The adjacency list will be computed if not already
          present.
          The current implementation only supports undirected graphs.
  
      Returns
      -------
      result : dask_cudf.DataFrame
          GPU distributed data frame containing 2 dask_cudf.Series
  
      ddf['vertex']: dask_cudf.Series
          Contains the vertex identifiers
      ddf['labels']: dask_cudf.Series
          Contains the wcc labels
  
      Examples
      --------
      >>> import cugraph.dask as dcg
      >>> import dask_cudf
      >>> # ... Init a DASK Cluster
      >>> #    see https://docs.rapids.ai/api/cugraph/stable/dask-cugraph.html
      >>> # Download dataset from https://github.com/rapidsai/cugraph/datasets/..
      >>> chunksize = dcg.get_chunksize(datasets_path / "karate.csv")
      >>> ddf = dask_cudf.read_csv(datasets_path / "karate.csv",
      ...                          chunksize=chunksize, delimiter=" ",
      ...                          names=["src", "dst", "value"],
      ...                          dtype=["int32", "int32", "float32"])
      >>> dg = cugraph.Graph(directed=False)
      >>> dg.from_dask_cudf_edgelist(ddf, source='src', destination='dst',
      ...                            edge_attr='value')
      >>> result = dcg.weakly_connected_components(dg)
  
      """
  
      if input_graph.is_directed():
          raise ValueError("input graph must be undirected")
  
      # Initialize dask client
      client = default_client()
  
      do_expensive_check = False
  
      result = [
          client.submit(
              _call_plc_wcc,
              Comms.get_session_id(),
              input_graph._plc_graph[w],
              do_expensive_check,
              workers=[w],
              allow_other_workers=False,
          )
          for w in Comms.get_workers()
      ]
  
      wait(result)
  
      cudf_result = [
          client.submit(convert_to_cudf, cp_arrays)
          for cp_arrays in result
      ]
  
      wait(cudf_result)
  
      ddf = dask_cudf.from_delayed(cudf_result).persist()
      wait(ddf)
      # Wait until the inactive futures are released
      wait(
          [
              (r.release(), c_r.release())
              for r, c_r in zip(result, cudf_result)
          ]
      )
  
      if input_graph.renumbered:
          ddf = input_graph.unrenumber(ddf, "vertex")
  
      return ddf
  
file_path: cugraph/python/cugraph/cugraph/dask/components/connectivity.py
imports: from dask.distributed import wait, default_client
  import cugraph.dask.comms.comms
  import dask_cudf
  import cudf
  from pylibcugraph import ResourceHandle
  from pylibcugraph import weakly_connected_components
label_desc: Use image node_id 3 for calling a global function with example usage: weakly_connected_components(input_graph) and returns: ddf