class_docstr: A channel to communicate with NNI manager.
  
  Each NNI experiment has a channel URL for tuner/assessor/strategy algorithm.
  The channel can only be connected once, so for each Python side :class:`~nni.experiment.Experiment` object,
  there should be exactly one corresponding ``TunerCommandChannel`` instance.
  
  :meth:`connect` must be invoked before sending or receiving data.
  
  The constructor does not have side effect so ``TunerCommandChannel`` can be created anywhere.
  But :meth:`connect` requires an initialized NNI manager, or otherwise the behavior is unpredictable.
  
  :meth:`_send` and :meth:`_receive` are underscore-prefixed because their signatures are scheduled to change by v3.0.
  
  Parameters
  ----------
  url
      The command channel URL.
      For now it must be like ``"ws://localhost:8080/tuner"`` or ``"ws://localhost:8080/url-prefix/tuner"``.
class_docstr_tok: ["A","channel","to","communicate","with","NNI","manager",".","Each","NNI","experiment","has","a","channel","URL","for","tuner\/assessor\/strategy","algorithm",".","The","channel","can","only","be","connected","once",",","so","for","each","Python","side",":","class",":","`","~nni.experiment.Experiment","`","object",",","there","should","be","exactly","one","corresponding","``","TunerCommandChannel","``","instance",".",":","meth",":","`","connect","`","must","be","invoked","before","sending","or","receiving","data",".","The","constructor","does","not","have","side","effect","so","``","TunerCommandChannel","``","can","be","created","anywhere",".","But",":","meth",":","`","connect","`","requires","an","initialized","NNI","manager",",","or","otherwise","the","behavior","is","unpredictable",".",":","meth",":","`","_send","`","and",":","meth",":","`","_receive","`","are","underscore-prefixed","because","their","signatures","are","scheduled","to","change","by","v3.0",".","Parameters","--","--","--","--","--","url","The","command","channel","URL",".","For","now","it","must","be","like","``","``","ws",":","\/\/localhost:8080\/tuner","''","``","or","``","``","ws",":","\/\/localhost:8080\/url-prefix\/tuner","''","``","."]
docstr: Receives a command from NNI manager.
docstr_tok: ["Receives","a","command","from","NNI","manager","."]
code: def receive(self) -> TunerIncomingCommand:
      """Receives a command from NNI manager."""
      command_type, data = self._receive()
      if data:
          data = load(data)
  
      # NOTE: Only handles the commands that are used by NAS.
      # It uses somewhat hacky way to convert the data received from NNI manager
      # to a semantic command.
      if command_type is None:
          # This shouldn't happen. Only for robustness.
          _logger.warning("Received command is empty. Terminating...")
          return Terminate()
      elif command_type == CommandType.Terminate:
          return Terminate()
      elif command_type == CommandType.Initialize:
          if not isinstance(data, dict):
              raise TypeError(
                  f"Initialize command data must be a dict, but got {type(data)}"
              )
          return Initialize(data)
      elif command_type == CommandType.RequestTrialJobs:
          if not isinstance(data, int):
              raise TypeError(
                  f"RequestTrialJobs command data must be an integer, but got {type(data)}"
              )
          return RequestTrialJobs(data)
      elif command_type == CommandType.UpdateSearchSpace:
          if not isinstance(data, dict):
              raise TypeError(
                  f"UpdateSearchSpace command data must be a dict, but got {type(data)}"
              )
          return UpdateSearchSpace(data)
      elif command_type == CommandType.ReportMetricData:
          if not isinstance(data, dict):
              raise TypeError(
                  f"ReportMetricData command data must be a dict, but got {type(data)}"
              )
          if "value" in data:
              data["value"] = load(data["value"])
          return ReportMetricData(**data)
      elif command_type == CommandType.TrialEnd:
          if not isinstance(data, dict):
              raise TypeError(
                  f"TrialEnd command data must be a dict, but got {type(data)}"
              )
          # For some reason, only one parameter (I guess the first one) shows up in the data.
          # But a trial technically is associated with multiple parameters.
          parameter_id = load(data["hyper_params"])["parameter_id"]
          return TrialEnd(
              trial_job_id=data["trial_job_id"],
              parameter_ids=[parameter_id],
              event=data["event"],
          )
      else:
          raise ValueError(f"Unknown command type: {command_type}")
  
imports: from __future__ import annotations
  import logging
  import os
  from collections import defaultdict
  from threading import Event
  from typing import Any, Callable
  from nni.common.serializer import dump, load, PayloadTooLarge
  from nni.runtime.command_channel.websocket import WsChannelClient
  from nni.typehint import Parameters
  from .command_type import CommandType, TunerIncomingCommand, Initialize, RequestTrialJobs, UpdateSearchSpace, ReportMetricData, TrialEnd, Terminate
label_desc: Use image node_id 8 for calling the TunerCommandChannel obj's underlying member method code with example usage: obj.receive() and returns: Terminate, Terminate, Initialize, RequestTrialJobs, UpdateSearchSpace, ReportMetricData, TrialEnd