code: def __init__(
      self,
      scheduler_addr,
      scheduler_port,
      worker_addrs,
      nthreads=0,
      n_workers=None,
      ssh_username=None,
      ssh_port=22,
      ssh_private_key=None,
      nohost=False,
      logdir=None,
      remote_python=None,
      memory_limit=None,
      worker_port=None,
      nanny_port=None,
      remote_dask_worker="distributed.cli.dask_worker",
      local_directory=None,
      **kwargs,
  ):
      self.scheduler_addr = scheduler_addr
      self.scheduler_port = scheduler_port
      self.nthreads = nthreads
      nprocs = kwargs.pop("nprocs", None)
      if kwargs:
          raise TypeError(
              f"__init__() got an unexpected keyword argument {', '.join(kwargs.keys())}"
          )
      if nprocs is not None and n_workers is not None:
          raise ValueError(
              "Both nprocs and n_workers were specified. Use n_workers only."
          )
      elif nprocs is not None:
          warnings.warn(
              "The nprocs argument will be removed in a future release. It has been "
              "renamed to n_workers.",
              FutureWarning,
          )
          n_workers = nprocs
      elif n_workers is None:
          n_workers = 1
  
      self.n_workers = n_workers
  
      self.ssh_username = ssh_username
      self.ssh_port = ssh_port
      self.ssh_private_key = ssh_private_key
  
      self.nohost = nohost
  
      self.remote_python = remote_python
  
      self.memory_limit = memory_limit
      self.worker_port = worker_port
      self.nanny_port = nanny_port
      self.remote_dask_worker = remote_dask_worker
      self.local_directory = local_directory
  
      # Generate a universal timestamp to use for log files
      import datetime
  
      if logdir is not None:
          logdir = os.path.join(
              logdir,
              "dask-ssh_"
              + datetime.datetime.now().strftime("%Y-%m-%d_%H:%M:%S"),
          )
          print(
              bcolors.WARNING + "Output will be redirected to logfiles "
              'stored locally on individual worker nodes under "{logdir}".'.format(
                  logdir=logdir
              )
              + bcolors.ENDC
          )
      self.logdir = logdir
  
      # Keep track of all running threads
      self.threads = []
  
      # Start the scheduler node
      self.scheduler = start_scheduler(
          logdir,
          scheduler_addr,
          scheduler_port,
          ssh_username,
          ssh_port,
          ssh_private_key,
          remote_python,
      )
  
      # Start worker nodes
      self.workers = []
      for addr in worker_addrs:
          self.add_worker(addr)
  
imports: from __future__ import annotations
  import logging
  import os
  import socket
  import sys
  import traceback
  import warnings
  from queue import Queue
  from threading import Thread
  from time import sleep
  from tlz import merge
  from tornado import gen
  from distributed.metrics import time
label_desc: Use image node_id 1 to create a new SSHCluster object  with example: obj = SSHCluster(scheduler_addr, scheduler_port, worker_addrs, nthreads, n_workers, ssh_username, ssh_port, ssh_private_key, nohost, logdir, remote_python, memory_limit, worker_port, nanny_port, remote_dask_worker, local_directory)