class_docstr: A controller/agent that populates a Launch RunQueue from a hyperparameter sweep.
class_docstr_tok: ["A","controller\/agent","that","populates","a","Launch","RunQueue","from","a","hyperparameter","sweep","."]
code: def __init__(
      self,
      api: "Api",
      *args: Optional[Any],
      polling_sleep: Optional[float] = None,
      sweep_id: Optional[str] = None,
      entity: Optional[str] = None,
      project: Optional[str] = None,
      project_queue: Optional[str] = None,
      num_workers: Optional[Union[int, str]] = None,
      **kwargs: Optional[Any],
  ):
      from wandb.apis.public import Api as PublicApi
  
      self._api = api
      self._public_api = PublicApi()
      self._entity = (
          entity
          or os.environ.get("WANDB_ENTITY")
          or api.settings("entity")
          or api.default_entity
      )
      self._project = (
          project
          or os.environ.get("WANDB_PROJECT")
          or api.settings("project")
      )
      self._sweep_id: str = sweep_id or "empty-sweep-id"
      self._state: SchedulerState = SchedulerState.PENDING
  
      # Make sure the provided sweep_id corresponds to a valid sweep
      try:
          resp = self._api.sweep(
              sweep_id, "{}", entity=self._entity, project=self._project
          )
          if resp.get("state") == SchedulerState.CANCELLED.name:
              self._state = SchedulerState.CANCELLED
          self._sweep_config = yaml.safe_load(resp["config"])
          self._num_runs_launched: int = self._get_num_runs_launched(
              resp["runs"]
          )
          if self._num_runs_launched > 0:
              wandb.termlog(
                  f"{LOG_PREFIX}Found {self._num_runs_launched} previous valid runs for sweep {self._sweep_id}"
              )
      except Exception as e:
          raise SchedulerError(
              f"{LOG_PREFIX}Exception when finding sweep ({sweep_id}) {e}"
          )
  
      # Scheduler may receive additional kwargs which will be piped into the launch command
      self._kwargs: Dict[str, Any] = kwargs
  
      # Dictionary of the runs being managed by the scheduler
      self._runs: Dict[str, SweepRun] = {}
      # Threading lock to ensure thread-safe access to the runs dictionary
      self._threading_lock: threading.Lock = threading.Lock()
      self._polling_sleep = polling_sleep or DEFAULT_POLLING_SLEEP
      self._project_queue = project_queue
      # Optionally run multiple workers in (pseudo-)parallel. Workers do not
      # actually run training workloads, they simply send heartbeat messages
      # (emulating a real agent) and add new runs to the launch queue. The
      # launch agent is the one that actually runs the training workloads.
      self._workers: Dict[int, _Worker] = {}
  
      # Init wandb scheduler run
      self._wandb_run = self._init_wandb_run()
  
      # Grab params from scheduler wandb run config
      num_workers = num_workers or self._wandb_run.config.get(
          "scheduler", {}
      ).get("num_workers")
      self._num_workers = (
          int(num_workers) if str(num_workers).isdigit() else 8
      )
      self._settings_config: Dict[
          str, Any
      ] = self._wandb_run.config.get("settings", {})
  
imports: import asyncio
  import base64
  import logging
  import os
  import socket
  import threading
  import time
  import traceback
  from abc import ABC, abstractmethod
  from dataclasses import dataclass
  from enum import Enum
  from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Tuple, Union
  import click
  import yaml
  import wandb
  from wandb.errors import CommError
  from wandb.sdk.launch._launch_add import launch_add
  from wandb.sdk.launch.errors import LaunchError
  from wandb.sdk.launch.sweeps import SchedulerError
  from wandb.sdk.launch.sweeps.utils import create_sweep_command_args, make_launch_sweep_entrypoint
  from wandb.sdk.launch.utils import event_loop_thread_exec
  from wandb.sdk.lib.runid import generate_id
label_desc: Use image node_id 1 to create a new Scheduler object from inherited base classes: ABC with example: obj = Scheduler(api)