docstr: Given resource demands, return node types to add to the cluster.
  
  This method:
      (1) calculates the resources present in the cluster.
      (2) calculates the remaining nodes to add to respect min_workers
          constraint per node type.
      (3) for each strict spread placement group, reserve space on
          available nodes and launch new nodes if necessary.
      (4) calculates the unfulfilled resource bundles.
      (5) calculates which nodes need to be launched to fulfill all
          the bundle requests, subject to max_worker constraints.
  
  Args:
      nodes: List of existing nodes in the cluster.
      launching_nodes: Summary of node types currently being launched.
      resource_demands: Vector of resource demands from the scheduler.
      unused_resources_by_ip: Mapping from ip to available resources.
      pending_placement_groups: Placement group demands.
      max_resources_by_ip: Mapping from ip to static node resources.
      ensure_min_cluster_size: Try to ensure the cluster can fit at least
          this set of resources. This differs from resources_demands in
          that we don't take into account existing usage.
  
      node_availability_summary: A snapshot of the current
          NodeAvailabilitySummary.
  
  Returns:
      Dict of count to add for each node type, and residual of resources
      that still cannot be fulfilled.
docstr_tok: ["Given","resource","demands",",","return","node","types","to","add","to","the","cluster",".","This","method",":","(","1",")","calculates","the","resources","present","in","the","cluster",".","(","2",")","calculates","the","remaining","nodes","to","add","to","respect","min_workers","constraint","per","node","type",".","(","3",")","for","each","strict","spread","placement","group",",","reserve","space","on","available","nodes","and","launch","new","nodes","if","necessary",".","(","4",")","calculates","the","unfulfilled","resource","bundles",".","(","5",")","calculates","which","nodes","need","to","be","launched","to","fulfill","all","the","bundle","requests",",","subject","to","max_worker","constraints",".","Args",":","nodes",":","List","of","existing","nodes","in","the","cluster",".","launching_nodes",":","Summary","of","node","types","currently","being","launched",".","resource_demands",":","Vector","of","resource","demands","from","the","scheduler",".","unused_resources_by_ip",":","Mapping","from","ip","to","available","resources",".","pending_placement_groups",":","Placement","group","demands",".","max_resources_by_ip",":","Mapping","from","ip","to","static","node","resources",".","ensure_min_cluster_size",":","Try","to","ensure","the","cluster","can","fit","at","least","this","set","of","resources",".","This","differs","from","resources_demands","in","that","we","do","n't","take","into","account","existing","usage",".","node_availability_summary",":","A","snapshot","of","the","current","NodeAvailabilitySummary",".","Returns",":","Dict","of","count","to","add","for","each","node","type",",","and","residual","of","resources","that","still","can","not","be","fulfilled","."]
code: def get_nodes_to_launch(
      self,
      nodes: List[NodeID],
      launching_nodes: Dict[NodeType, int],
      resource_demands: List[ResourceDict],
      unused_resources_by_ip: Dict[NodeIP, ResourceDict],
      pending_placement_groups: List[PlacementGroupTableData],
      max_resources_by_ip: Dict[NodeIP, ResourceDict],
      ensure_min_cluster_size: List[ResourceDict],
      node_availability_summary: NodeAvailabilitySummary,
  ) -> (Dict[NodeType, int], List[ResourceDict]):
      """Given resource demands, return node types to add to the cluster.
  
      This method:
          (1) calculates the resources present in the cluster.
          (2) calculates the remaining nodes to add to respect min_workers
              constraint per node type.
          (3) for each strict spread placement group, reserve space on
              available nodes and launch new nodes if necessary.
          (4) calculates the unfulfilled resource bundles.
          (5) calculates which nodes need to be launched to fulfill all
              the bundle requests, subject to max_worker constraints.
  
      Args:
          nodes: List of existing nodes in the cluster.
          launching_nodes: Summary of node types currently being launched.
          resource_demands: Vector of resource demands from the scheduler.
          unused_resources_by_ip: Mapping from ip to available resources.
          pending_placement_groups: Placement group demands.
          max_resources_by_ip: Mapping from ip to static node resources.
          ensure_min_cluster_size: Try to ensure the cluster can fit at least
              this set of resources. This differs from resources_demands in
              that we don't take into account existing usage.
  
          node_availability_summary: A snapshot of the current
              NodeAvailabilitySummary.
  
      Returns:
          Dict of count to add for each node type, and residual of resources
          that still cannot be fulfilled.
      """
      utilization_scorer = partial(
          self.utilization_scorer,
          node_availability_summary=node_availability_summary,
      )
      self._update_node_resources_from_runtime(
          nodes, max_resources_by_ip
      )
  
      node_resources: List[ResourceDict]
      node_type_counts: Dict[NodeType, int]
      node_resources, node_type_counts = self.calculate_node_resources(
          nodes, launching_nodes, unused_resources_by_ip
      )
  
      logger.debug("Cluster resources: {}".format(node_resources))
      logger.debug("Node counts: {}".format(node_type_counts))
      # Step 2: add nodes to add to satisfy min_workers for each type
      (
          node_resources,
          node_type_counts,
          adjusted_min_workers,
      ) = _add_min_workers_nodes(
          node_resources,
          node_type_counts,
          self.node_types,
          self.max_workers,
          self.head_node_type,
          ensure_min_cluster_size,
          utilization_scorer=utilization_scorer,
      )
  
      # Step 3: get resource demands of placement groups and return the
      # groups that should be strictly spread.
      logger.debug(
          f"Placement group demands: {pending_placement_groups}"
      )
      # TODO(Clark): Refactor placement group bundle demands such that their placement
      # group provenance is mantained, since we need to keep an accounting of the
      # cumulative CPU cores allocated as fulfilled during bin packing in order to
      # ensure that a placement group's cumulative allocation is under the placement
      # group's max CPU fraction per node. Without this, and placement group with many
      # bundles might not be schedulable, but will fail to trigger scale-up since the
      # max CPU fraction is properly applied to the cumulative bundle requests for a
      # single node.
      #
      # placement_group_demand_vector: List[Tuple[List[ResourceDict], double]]
      #
      # bin_pack_residual() can keep it's packing priority; we just need to account
      # for (1) the running CPU allocation for the bundle's placement group for that
      # particular node, and (2) the max CPU cores allocatable for a single placement
      # group for that particular node.
      (
          placement_group_demand_vector,
          strict_spreads,
      ) = placement_groups_to_resource_demands(pending_placement_groups)
      # Place placement groups demand vector at the beginning of the resource
      # demands vector to make it consistent (results in the same types of
      # nodes to add) with pg_demands_nodes_max_launch_limit calculated later
      resource_demands = (
          placement_group_demand_vector + resource_demands
      )
  
      (
          spread_pg_nodes_to_add,
          node_resources,
          node_type_counts,
      ) = self.reserve_and_allocate_spread(
          strict_spreads,
          node_resources,
          node_type_counts,
          utilization_scorer,
      )
  
      # Calculate the nodes to add for bypassing max launch limit for
      # placement groups and spreads.
      unfulfilled_placement_groups_demands, _ = get_bin_pack_residual(
          node_resources,
          placement_group_demand_vector,
      )
      # Add 1 to account for the head node.
      max_to_add = self.max_workers + 1 - sum(node_type_counts.values())
      pg_demands_nodes_max_launch_limit, _ = get_nodes_for(
          self.node_types,
          node_type_counts,
          self.head_node_type,
          max_to_add,
          unfulfilled_placement_groups_demands,
          utilization_scorer=utilization_scorer,
      )
      placement_groups_nodes_max_limit = {
          node_type: spread_pg_nodes_to_add.get(node_type, 0)
          + pg_demands_nodes_max_launch_limit.get(node_type, 0)
          for node_type in self.node_types
      }
  
      # Step 4/5: add nodes for pending tasks, actors, and non-strict spread
      # groups
      unfulfilled, _ = get_bin_pack_residual(
          node_resources, resource_demands
      )
      logger.debug("Resource demands: {}".format(resource_demands))
      logger.debug("Unfulfilled demands: {}".format(unfulfilled))
      nodes_to_add_based_on_demand, final_unfulfilled = get_nodes_for(
          self.node_types,
          node_type_counts,
          self.head_node_type,
          max_to_add,
          unfulfilled,
          utilization_scorer=utilization_scorer,
      )
      logger.debug("Final unfulfilled: {}".format(final_unfulfilled))
      # Merge nodes to add based on demand and nodes to add based on
      # min_workers constraint. We add them because nodes to add based on
      # demand was calculated after the min_workers constraint was respected.
      total_nodes_to_add = {}
  
      for node_type in self.node_types:
          nodes_to_add = (
              adjusted_min_workers.get(node_type, 0)
              + spread_pg_nodes_to_add.get(node_type, 0)
              + nodes_to_add_based_on_demand.get(node_type, 0)
          )
          if nodes_to_add > 0:
              total_nodes_to_add[node_type] = nodes_to_add
  
      # Limit the number of concurrent launches
      total_nodes_to_add = (
          self._get_concurrent_resource_demand_to_launch(
              total_nodes_to_add,
              unused_resources_by_ip.keys(),
              nodes,
              launching_nodes,
              adjusted_min_workers,
              placement_groups_nodes_max_limit,
          )
      )
  
      logger.debug("Node requests: {}".format(total_nodes_to_add))
      return total_nodes_to_add, final_unfulfilled
  
imports: import collections
  import copy
  import logging
  import os
  from abc import abstractmethod
  from functools import partial
  from typing import Callable, Dict, List, Optional, Tuple
  import ray
  from ray._private.gcs_utils import PlacementGroupTableData
  from ray.autoscaler._private.constants import AUTOSCALER_CONSERVE_GPU_NODES, AUTOSCALER_UTILIZATION_SCORER_KEY
  from ray.autoscaler._private.loader import load_function_or_class
  from ray.autoscaler._private.node_provider_availability_tracker import NodeAvailabilitySummary
  from ray.autoscaler._private.util import NodeID, NodeIP, NodeType, NodeTypeConfigDict, ResourceDict, is_placement_group_resource
  from ray.autoscaler.node_provider import NodeProvider
  from ray.autoscaler.tags import NODE_KIND_HEAD, NODE_KIND_UNMANAGED, NODE_KIND_WORKER, TAG_RAY_NODE_KIND, TAG_RAY_USER_NODE_TYPE
  from ray.core.generated.common_pb2 import PlacementStrategy
label_desc: Use image node_id 5 for calling the ResourceDemandScheduler obj's underlying member method code with example usage: obj.get_nodes_to_launch(nodes, launching_nodes, resource_demands, unused_resources_by_ip, pending_placement_groups, max_resources_by_ip, ensure_min_cluster_size, node_availability_summary) and returns: total_nodes_to_add,  final_unfulfilled