class_docstr: Sweep controller class.
  
  Internal datastructures on the sweep object to coordinate local controller with
  cloud controller.
  
  Data structures:
      controller: {
          schedule: [
              { id: SCHEDULE_ID
                data: {param1: val1, param2: val2}},
          ]
          earlystop: [RUN_ID, ...]
      scheduler:
          scheduled: [
              { id: SCHEDULE_ID
                runid: RUN_ID},
          ]
  
  `controller` is only updated by the client
  `scheduler` is only updated by the cloud backend
  
  Protocols:
      Scheduling a run:
      - client controller adds a schedule entry on the controller.schedule list
      - cloud backend notices the new entry and creates a run with the parameters
      - cloud backend adds a scheduled entry on the scheduler.scheduled list
      - client controller notices that the run has been scheduled and removes it from
        controller.schedule list
  
  Current implementation details:
      - Runs are only schedule if there are no other runs scheduled.
class_docstr_tok: ["Sweep","controller","class",".","Internal","datastructures","on","the","sweep","object","to","coordinate","local","controller","with","cloud","controller",".","Data","structures",":","controller",":","{","schedule",":","[","{","id",":","SCHEDULE_ID","data",":","{","param1",":","val1",",","param2",":","val2","}","}",",","]","earlystop",":","[","RUN_ID",",","...","]","scheduler",":","scheduled",":","[","{","id",":","SCHEDULE_ID","runid",":","RUN_ID","}",",","]","`","controller","`","is","only","updated","by","the","client","`","scheduler","`","is","only","updated","by","the","cloud","backend","Protocols",":","Scheduling","a","run",":","-","client","controller","adds","a","schedule","entry","on","the","controller.schedule","list","-","cloud","backend","notices","the","new","entry","and","creates","a","run","with","the","parameters","-","cloud","backend","adds","a","scheduled","entry","on","the","scheduler.scheduled","list","-","client","controller","notices","that","the","run","has","been","scheduled","and","removes","it","from","controller.schedule","list","Current","implementation","details",":","-","Runs","are","only","schedule","if","there","are","no","other","runs","scheduled","."]
code: def __init__(
      self, sweep_id_or_config=None, entity=None, project=None
  ):
      # sweep id configured in constuctor
      self._sweep_id: Optional[str] = None
  
      # configured parameters
      # Configuration to be created
      self._create: Dict = {}
      # Custom search
      self._custom_search: Optional[
          Callable[
              [Union[dict, sweeps.SweepConfig], List[sweeps.SweepRun]],
              Optional[sweeps.SweepRun],
          ]
      ] = None
      # Custom stopping
      self._custom_stopping: Optional[
          Callable[
              [Union[dict, sweeps.SweepConfig], List[sweeps.SweepRun]],
              List[sweeps.SweepRun],
          ]
      ] = None
      # Program function (used for future jupyter support)
      self._program_function = None
  
      # The following are updated every sweep step
      # raw sweep object (dict of strings)
      self._sweep_obj = None
      # parsed sweep config (dict)
      self._sweep_config: Optional[
          Union[dict, sweeps.SweepConfig]
      ] = None
      # sweep metric used to optimize (str or None)
      self._sweep_metric: Optional[str] = None
      # list of _Run objects
      self._sweep_runs: Optional[List[sweeps.SweepRun]] = None
      # dictionary mapping name of run to run object
      self._sweep_runs_map: Optional[Dict[str, sweeps.SweepRun]] = None
      # scheduler dict (read only from controller) - used as feedback from the server
      self._scheduler: Optional[Dict] = None
      # controller dict (write only from controller) - used to send commands to server
      self._controller: Optional[Dict] = None
      # keep track of controller dict from previous step
      self._controller_prev_step: Optional[Dict] = None
  
      # Internal
      # Keep track of whether the sweep has been started
      self._started: bool = False
      # indicate whether there is more to schedule
      self._done_scheduling: bool = False
      # indicate whether the sweep needs to be created
      self._defer_sweep_creation: bool = False
      # count of logged lines since last status
      self._logged: int = 0
      # last status line printed
      self._laststatus: str = ""
      # keep track of logged actions for print_actions()
      self._log_actions: List[Tuple[str, str]] = []
      # keep track of logged debug for print_debug()
      self._log_debug: List[str] = []
  
      # all backend commands use internal api
      environ = os.environ
      if entity:
          env.set_entity(entity, env=environ)
      if project:
          env.set_project(project, env=environ)
      self._api = InternalApi(environ=environ)
  
      if isinstance(sweep_id_or_config, str):
          self._sweep_id = sweep_id_or_config
      elif isinstance(sweep_id_or_config, dict) or isinstance(
          sweep_id_or_config, sweeps.SweepConfig
      ):
          self._create = sweeps.SweepConfig(sweep_id_or_config)
  
          # check for custom search and or stopping functions
          for config_key, controller_attr in zip(
              ["method", "early_terminate"],
              ["_custom_search", "_custom_stopping"],
          ):
              if callable(
                  config_key in self._create
                  and self._create[config_key]
              ):
                  setattr(
                      self, controller_attr, self._create[config_key]
                  )
                  self._create[config_key] = "custom"
  
          self._sweep_id = self.create(from_dict=True)
      elif sweep_id_or_config is None:
          self._defer_sweep_creation = True
          return
      else:
          raise ControllerError("Unhandled sweep controller type")
      sweep_obj = self._sweep_object_read_from_backend()
      if sweep_obj is None:
          raise ControllerError("Can not find sweep")
      self._sweep_obj = sweep_obj
  
imports: import json
  import os
  import random
  import string
  import time
  from typing import Callable, Dict, List, Optional, Tuple, Union
  import yaml
  from wandb import env
  from wandb.apis import InternalApi
  from wandb.sdk import wandb_sweep
  from wandb.sdk.launch.sweeps.utils import handle_sweep_config_violations, sweep_config_err_text_from_jsonschema_violations
  from wandb.util import get_module
label_desc: Use image node_id 1 to create a new _WandbController object  with example: obj = _WandbController(sweep_id_or_config, entity, project)