docstr: Update static node type resources with runtime resources
  
  This will update the cached static node type resources with the runtime
  resources. Because we can not know the exact autofilled memory or
  object_store_memory from config file.
docstr_tok: ["Update","static","node","type","resources","with","runtime","resources","This","will","update","the","cached","static","node","type","resources","with","the","runtime","resources",".","Because","we","can","not","know","the","exact","autofilled","memory","or","object_store_memory","from","config","file","."]
code: def _update_node_resources_from_runtime(
      self,
      nodes: List[NodeID],
      max_resources_by_ip: Dict[NodeIP, ResourceDict],
  ):
      """Update static node type resources with runtime resources
  
      This will update the cached static node type resources with the runtime
      resources. Because we can not know the exact autofilled memory or
      object_store_memory from config file.
      """
      need_update = len(self.node_types) != len(
          self.node_resource_updated
      )
  
      if not need_update:
          return
      for node_id in nodes:
          tags = self.provider.node_tags(node_id)
  
          if TAG_RAY_USER_NODE_TYPE not in tags:
              continue
  
          node_type = tags[TAG_RAY_USER_NODE_TYPE]
          if (
              node_type in self.node_resource_updated
              or node_type not in self.node_types
          ):
              # continue if the node type has been updated or is not an known
              # node type
              continue
          ip = self.provider.internal_ip(node_id)
          runtime_resources = max_resources_by_ip.get(ip)
          if runtime_resources:
              runtime_resources = copy.deepcopy(runtime_resources)
              resources = self.node_types[node_type].get(
                  "resources", {}
              )
              for key in [
                  "CPU",
                  "GPU",
                  "memory",
                  "object_store_memory",
              ]:
                  if key in runtime_resources:
                      resources[key] = runtime_resources[key]
              self.node_types[node_type]["resources"] = resources
  
              node_kind = tags[TAG_RAY_NODE_KIND]
              if node_kind == NODE_KIND_WORKER:
                  # Here, we do not record the resources have been updated
                  # if it is the head node kind. Because it need be updated
                  # by worker kind runtime resource. The most difference
                  # between head and worker is the memory resources. The head
                  # node needs to configure redis memory which is not needed
                  # for worker nodes.
                  self.node_resource_updated.add(node_type)
  
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 6 for calling the ResourceDemandScheduler obj's underlying member method code with example usage: obj._update_node_resources_from_runtime(nodes, max_resources_by_ip) without return types