docstr: Returns False iff non_terminated_nodes contains any pods in the RayCluster's
  workersToDelete lists.
  
  Explanation:
  If there are any workersToDelete which are non-terminated,
  we should wait for the operator to do its job and delete those
  pods. Therefore, we back off the autoscaler update.
  
  If, on the other hand, all of the workersToDelete have already been cleaned up,
  then we patch away the workersToDelete lists and return True.
  In the future, we may consider having the operator clean up workersToDelete
  on it own:
  https://github.com/ray-project/kuberay/issues/733
  
  Note (Dmitri):
  It is stylistically bad that this function has a side effect.
docstr_tok: ["Returns","False","iff","non_terminated_nodes","contains","any","pods","in","the","RayCluster's","workersToDelete","lists",".","Explanation",":","If","there","are","any","workersToDelete","which","are","non-terminated",",","we","should","wait","for","the","operator","to","do","its","job","and","delete","those","pods",".","Therefore",",","we","back","off","the","autoscaler","update",".","If",",","on","the","other","hand",",","all","of","the","workersToDelete","have","already","been","cleaned","up",",","then","we","patch","away","the","workersToDelete","lists","and","return","True",".","In","the","future",",","we","may","consider","having","the","operator","clean","up","workersToDelete","on","it","own",":","https",":","\/\/github.com\/ray-project\/kuberay\/issues\/733","Note","(","Dmitri",")",":","It","is","stylistically","bad","that","this","function","has","a","side","effect","."]
code: def safe_to_scale(self) -> bool:
      """Returns False iff non_terminated_nodes contains any pods in the RayCluster's
      workersToDelete lists.
  
      Explanation:
      If there are any workersToDelete which are non-terminated,
      we should wait for the operator to do its job and delete those
      pods. Therefore, we back off the autoscaler update.
  
      If, on the other hand, all of the workersToDelete have already been cleaned up,
      then we patch away the workersToDelete lists and return True.
      In the future, we may consider having the operator clean up workersToDelete
      on it own:
      https://github.com/ray-project/kuberay/issues/733
  
      Note (Dmitri):
      It is stylistically bad that this function has a side effect.
      """
      # Get the list of nodes.
      node_set = set(self.node_data_dict.keys())
      worker_groups = self._raycluster["spec"].get(
          "workerGroupSpecs", []
      )
  
      # Accumulates the indices of worker groups with non-empty workersToDelete
      non_empty_worker_group_indices = []
  
      for group_index, worker_group in enumerate(worker_groups):
          workersToDelete = worker_group.get("scaleStrategy", {}).get(
              "workersToDelete", []
          )
          if workersToDelete:
              non_empty_worker_group_indices.append(group_index)
          for worker in workersToDelete:
              if worker in node_set:
                  # The operator hasn't removed this worker yet. Abort
                  # the autoscaler update.
                  logger.warning(
                      f"Waiting for operator to remove worker {worker}."
                  )
                  return False
  
      # All required workersToDelete have been removed.
      # Clean up the workersToDelete field.
      patch_payload = []
      for group_index in non_empty_worker_group_indices:
          patch = worker_delete_patch(group_index, workers_to_delete=[])
          patch_payload.append(patch)
      if patch_payload:
          logger.info("Cleaning up workers to delete.")
          logger.info(f"Submitting patch {patch_payload}.")
          self._submit_raycluster_patch(patch_payload)
  
      # It's safe to proceed with the autoscaler update.
      return True
  
imports: import json
  import logging
  import os
  from collections import defaultdict
  from typing import Any, Dict, List, Optional, Tuple
  import requests
  from ray.autoscaler._private.constants import WORKER_LIVENESS_CHECK_KEY, WORKER_RPC_DRAIN_KEY
  from ray.autoscaler._private.util import NodeID, NodeIP, NodeKind, NodeStatus, NodeType
  from ray.autoscaler.batching_node_provider import BatchingNodeProvider, NodeData, ScaleRequest
  from ray.autoscaler.tags import NODE_KIND_HEAD, NODE_KIND_WORKER, STATUS_UP_TO_DATE, STATUS_UPDATE_FAILED, TAG_RAY_USER_NODE_TYPE
label_desc: Use image node_id 4 for calling the KubeRayNodeProvider obj's underlying member method code with example usage: obj.safe_to_scale() and returns: True, False