docstr: Returns node resource list and node type counts.
  
  Counts the running nodes, pending nodes.
  Args:
       nodes: Existing nodes.
       pending_nodes: Pending nodes.
  Returns:
       node_resources: a list of running + pending resources.
           E.g., [{"CPU": 4}, {"GPU": 2}].
       node_type_counts: running + pending workers per node type.
docstr_tok: ["Returns","node","resource","list","and","node","type","counts",".","Counts","the","running","nodes",",","pending","nodes",".","Args",":","nodes",":","Existing","nodes",".","pending_nodes",":","Pending","nodes",".","Returns",":","node_resources",":","a","list","of","running","+","pending","resources",".","E.g.",",","[","{","``","CPU","''",":","4","}",",","{","``","GPU","''",":","2","}","]",".","node_type_counts",":","running","+","pending","workers","per","node","type","."]
code: def calculate_node_resources(
      self,
      nodes: List[NodeID],
      pending_nodes: Dict[NodeID, int],
      unused_resources_by_ip: Dict[str, ResourceDict],
  ) -> (List[ResourceDict], Dict[NodeType, int]):
      """Returns node resource list and node type counts.
  
      Counts the running nodes, pending nodes.
      Args:
           nodes: Existing nodes.
           pending_nodes: Pending nodes.
      Returns:
           node_resources: a list of running + pending resources.
               E.g., [{"CPU": 4}, {"GPU": 2}].
           node_type_counts: running + pending workers per node type.
      """
  
      node_resources = []
      node_type_counts = collections.defaultdict(int)
  
      def add_node(node_type, available_resources=None):
          if node_type not in self.node_types:
              # We should not get here, but if for some reason we do, log an
              # error and skip the errant node_type.
              logger.error(
                  f"Missing entry for node_type {node_type} in "
                  f"cluster config: {self.node_types} under entry "
                  "available_node_types. This node's resources will be "
                  "ignored. If you are using an unmanaged node, manually "
                  f"set the {TAG_RAY_NODE_KIND} tag to "
                  f'"{NODE_KIND_UNMANAGED}" in your cloud provider\'s '
                  "management console."
              )
              return None
          # Careful not to include the same dict object multiple times.
          available = copy.deepcopy(
              self.node_types[node_type]["resources"]
          )
          # If available_resources is None this might be because the node is
          # no longer pending, but the raylet hasn't sent a heartbeat to gcs
          # yet.
          if available_resources is not None:
              available = copy.deepcopy(available_resources)
  
          node_resources.append(available)
          node_type_counts[node_type] += 1
  
      for node_id in nodes:
          tags = self.provider.node_tags(node_id)
          if TAG_RAY_USER_NODE_TYPE in tags:
              node_type = tags[TAG_RAY_USER_NODE_TYPE]
              ip = self.provider.internal_ip(node_id)
              available_resources = unused_resources_by_ip.get(ip)
              add_node(node_type, available_resources)
  
      for node_type, count in pending_nodes.items():
          for _ in range(count):
              add_node(node_type)
  
      return node_resources, node_type_counts
  
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 9 for calling the ResourceDemandScheduler obj's underlying member method code with example usage: obj.calculate_node_resources(nodes, pending_nodes, unused_resources_by_ip) and returns: node_resources,  node_type_counts, None