class_docstr: Class for generating images from a Stable Diffusion checkpoint trained using Determined
  AI. Initialize with no arguments in order to run plain Stable Diffusion without any trained
  textual inversion embeddings. Can optionally be run on a Determined cluster through the
  .generate_on_cluster() method for large-scale generation. Only intended for use with a GPU.
docstr: Creates a DetSDTextualInversionPipeline instance on the cluster, drawing hyperparameters
  and other needed information from the Determined master, and then generates images. Expects
  the `hyperparameters` section of the config to be broken into the following sections:
  - `batch_size`: The batch size to use for generation.
  - `main_process_generator_seed`: an integer specifying the seed used by the chief worker for
  generation. Other workers add their process index to this value.
  - `save_freq`: an integer specifying how often to write images to tensorboard.
  - `pipeline`: containing all __init__ args
  - `uuids`: a (possibly empty) array of any checkpoint UUIDs which are to be loaded into
  the pipeline.
  - `local_checkpoint_paths`: a (possibly empty) array of any checkpoint paths which are to be
  loaded into the pipeline.
  - `call_kwargs`: all arguments which are to be passed to the `__call__` method.
docstr_tok: ["Creates","a","DetSDTextualInversionPipeline","instance","on","the","cluster",",","drawing","hyperparameters","and","other","needed","information","from","the","Determined","master",",","and","then","generates","images",".","Expects","the","`","hyperparameters","`","section","of","the","config","to","be","broken","into","the","following","sections",":","-","`","batch_size","`",":","The","batch","size","to","use","for","generation",".","-","`","main_process_generator_seed","`",":","an","integer","specifying","the","seed","used","by","the","chief","worker","for","generation",".","Other","workers","add","their","process","index","to","this","value",".","-","`","save_freq","`",":","an","integer","specifying","how","often","to","write","images","to","tensorboard",".","-","`","pipeline","`",":","containing","all","__init__","args","-","`","uuids","`",":","a","(","possibly","empty",")","array","of","any","checkpoint","UUIDs","which","are","to","be","loaded","into","the","pipeline",".","-","`","local_checkpoint_paths","`",":","a","(","possibly","empty",")","array","of","any","checkpoint","paths","which","are","to","be","loaded","into","the","pipeline",".","-","`","call_kwargs","`",":","all","arguments","which","are","to","be","passed","to","the","`","__call__","`","method","."]
code: def generate_on_cluster(cls) -> None:
      """Creates a DetSDTextualInversionPipeline instance on the cluster, drawing hyperparameters
      and other needed information from the Determined master, and then generates images. Expects
      the `hyperparameters` section of the config to be broken into the following sections:
      - `batch_size`: The batch size to use for generation.
      - `main_process_generator_seed`: an integer specifying the seed used by the chief worker for
      generation. Other workers add their process index to this value.
      - `save_freq`: an integer specifying how often to write images to tensorboard.
      - `pipeline`: containing all __init__ args
      - `uuids`: a (possibly empty) array of any checkpoint UUIDs which are to be loaded into
      the pipeline.
      - `local_checkpoint_paths`: a (possibly empty) array of any checkpoint paths which are to be
      loaded into the pipeline.
      - `call_kwargs`: all arguments which are to be passed to the `__call__` method.
      """
      info = det.get_cluster_info()
      assert (
          info is not None
      ), "generate_on_cluster() must be called on a Determined cluster."
      hparams = info.trial.hparams
      trial_id = info.trial.trial_id
      latest_checkpoint = info.latest_checkpoint
  
      # Extract relevant groups from hparams.
      batch_size = hparams["batch_size"]
      main_process_generator_seed = hparams[
          "main_process_generator_seed"
      ]
      save_freq = hparams["save_freq"]
      pipeline_init_kwargs = hparams["pipeline"]
      uuid_list = hparams["uuids"]
      local_checkpoint_paths_list = hparams["local_checkpoint_paths"]
      call_kwargs = hparams["call_kwargs"]
  
      logger = logging.getLogger(__name__)
  
      # Get the distributed context, as needed.
      try:
          distributed = (
              det.core.DistributedContext.from_torch_distributed()
          )
      except KeyError:
          distributed = None
  
      with det.core.init(
          distributed=distributed,
          tensorboard_mode=det.core.TensorboardMode.MANUAL,
      ) as core_context:
          # Get worker data.
          process_index = core_context.distributed.get_rank()
          is_main_process = process_index == 0
          local_process_index = (
              core_context.distributed.get_local_rank()
          )
          is_local_main_process = local_process_index == 0
  
          device = (
              f"cuda:{process_index}"
              if distributed is not None
              else "cuda"
          )
          pipeline_init_kwargs["device"] = device
  
          # Instantiate the pipeline and load in any checkpoints by uuid.
          pipeline = cls(**pipeline_init_kwargs)
          # Only the local chief worker performs the download.
          if uuid_list:
              if is_local_main_process:
                  paths = pipeline.load_from_uuids(uuid_list)
              else:
                  paths = None
              paths = core_context.distributed.broadcast_local(paths)
              if not is_local_main_process:
                  for path in paths:
                      pipeline.load_from_checkpoint_dir(path)
  
          if local_checkpoint_paths_list:
              for path_str in local_checkpoint_paths_list:
                  path = pathlib.Path(path_str)
                  pipeline.load_from_checkpoint_dir(
                      checkpoint_dir=path.parent,
                      learned_embeddings_filename=path.name,
                  )
  
          # Create the Tensorboard writer.
          tb_dir = core_context.train.get_tensorboard_path()
          tb_writer = SummaryWriter(log_dir=tb_dir)
          # Include relevant __call__ args in the tensorboard tag.
          important_generation_params = (
              "prompt",
              "guidance_scale",
              "num_inference_steps",
          )
          tb_tag = "/".join(
              [
                  f"{k}: {call_kwargs[k]}"
                  for k in important_generation_params
              ]
          )
  
          # Use unique seeds, to avoid repeated images, and add the corresponding generator to the
          # call_kwargs.
          seed = main_process_generator_seed + process_index
          generator = torch.Generator(
              device=pipeline.device
          ).manual_seed(seed)
          call_kwargs["generator"] = generator
          # Add seed information to the tensorboard tag.
          tb_tag += f"/seed: {seed}"
          # Update the call_kwargs with the batch size, if needed.
          if batch_size > 1:
              call_kwargs["prompt"] = [
                  call_kwargs["prompt"]
              ] * batch_size
  
          pipeline.steps_completed = 0
          pipeline.num_generated_imgs = 0
          pipeline.image_history = []
  
          # Restore from a checkpoint, if necessary.
          if latest_checkpoint is not None:
              pipeline._restore_latest_checkpoint(
                  core_context=core_context,
                  latest_checkpoint=latest_checkpoint,
                  generator=generator,
                  trial_id=trial_id,
              )
              if is_main_process:
                  logger.info(
                      f"Resumed from checkpoint at step {pipeline.steps_completed}"
                  )
  
          if is_main_process:
              logger.info(
                  "--------------- Generating Images ---------------"
              )
  
          # There will be a single op of len max_length, as defined in the searcher config.
          for op in core_context.searcher.operations():
              while pipeline.steps_completed < op.length:
                  pipeline.image_history.extend(
                      pipeline(**call_kwargs).images
                  )
                  pipeline.steps_completed += 1
  
                  # Write to tensorboard and checkpoint at the specified frequency.
                  if (
                      pipeline.steps_completed % save_freq == 0
                      or pipeline.steps_completed == op.length
                  ):
                      pipeline._write_tb_imgs(
                          core_context=core_context,
                          tb_writer=tb_writer,
                          tb_tag=tb_tag,
                      )
  
                      # Checkpointing.
                      devices_and_generators = (
                          core_context.distributed.gather(
                              (device, generator.get_state())
                          )
                      )
                      if is_main_process:
                          logger.info(
                              f"Saving at step {pipeline.steps_completed}"
                          )
                          # Save the state of the generators as the checkpoint.
                          pipeline._save(
                              core_context=core_context,
                              devices_and_generators=devices_and_generators,
                              trial_id=trial_id,
                          )
                          op.report_progress(pipeline.steps_completed)
  
                      # Only preempt after a checkpoint has been saved.
                      if core_context.preempt.should_preempt():
                          return
  
              if is_main_process:
                  # Report zero upon completion.
                  op.report_completed(0)
  
imports: import json
  import logging
  import os
  import pathlib
  from typing import Any, Dict, List, Literal, Optional, Sequence, Tuple, Union
  import torch
  from detsd import defaults, utils
  from diffusers.pipelines.stable_diffusion import StableDiffusionPipeline, StableDiffusionPipelineOutput
  from torch.utils.tensorboard import SummaryWriter
  from torchvision.transforms.functional import pil_to_tensor
  import determined
  from determined.experimental import client
label_desc: Use image node_id 2 for calling the DetSDTextualInversionPipeline obj's underlying member method code with example usage: obj.generate_on_cluster(cls) without return types