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: Listen for incoming commands.
  
  Call :meth:`receive` in a loop and call ``callback`` for each command,
  until ``stop_event`` is set, or a Terminate command is received.
  All commands will go into callback, including Terminate command.
  
  It usually runs in a separate thread.
  
  Parameters
  ----------
  callback
      A callback function that takes a :class:`TunerIncomingCommand` as argument.
      It's not expected to return anything.
  stop_event
      A threading event that can be used to stop the loop.
docstr_tok: ["Listen","for","incoming","commands",".","Call",":","meth",":","`","receive","`","in","a","loop","and","call","``","callback","``","for","each","command",",","until","``","stop_event","``","is","set",",","or","a","Terminate","command","is","received",".","All","commands","will","go","into","callback",",","including","Terminate","command",".","It","usually","runs","in","a","separate","thread",".","Parameters","--","--","--","--","--","callback","A","callback","function","that","takes","a",":","class",":","`","TunerIncomingCommand","`","as","argument",".","It","'s","not","expected","to","return","anything",".","stop_event","A","threading","event","that","can","be","used","to","stop","the","loop","."]
code: def listen(self, stop_event: Event) -> None:
      """Listen for incoming commands.
  
      Call :meth:`receive` in a loop and call ``callback`` for each command,
      until ``stop_event`` is set, or a Terminate command is received.
      All commands will go into callback, including Terminate command.
  
      It usually runs in a separate thread.
  
      Parameters
      ----------
      callback
          A callback function that takes a :class:`TunerIncomingCommand` as argument.
          It's not expected to return anything.
      stop_event
          A threading event that can be used to stop the loop.
      """
      while not stop_event.is_set():
          received = self.receive()
          for callback in self._callbacks[received.command_type]:
              callback(received)
  
          # Two ways to stop the loop:
          # 1. The received command is a Terminate command, which is triggered by a NNI manager stop.
          # 2. The stop_event is set from another thread (possibly main thread), which could be an engine shutdown.
          if received.command_type == CommandType.Terminate:
              _logger.debug(
                  "Received command type is terminate. Stop listening."
              )
              stop_event.set()
  
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 4 for calling the TunerCommandChannel obj's underlying member method code with example usage: obj.listen(stop_event) without return types