code: def parse_resource_demands(resource_load_by_shape):
      """Handle the message.resource_load_by_shape protobuf for the demand
      based autoscaling. Catch and log all exceptions so this doesn't
      interfere with the utilization based autoscaler until we're confident
      this is stable. Worker queue backlogs are added to the appropriate
      resource demand vector.
  
      Args:
          resource_load_by_shape (pb2.gcs.ResourceLoad): The resource demands
              in protobuf form or None.
  
      Returns:
          List[ResourceDict]: Waiting bundles (ready and feasible).
          List[ResourceDict]: Infeasible bundles.
      """
      waiting_bundles, infeasible_bundles = [], []
      try:
          for resource_demand_pb in list(
              resource_load_by_shape.resource_demands
          ):
              request_shape = dict(resource_demand_pb.shape)
              for _ in range(
                  resource_demand_pb.num_ready_requests_queued
              ):
                  waiting_bundles.append(request_shape)
              for _ in range(
                  resource_demand_pb.num_infeasible_requests_queued
              ):
                  infeasible_bundles.append(request_shape)
  
              # Infeasible and ready states for tasks are (logically)
              # mutually exclusive.
              if resource_demand_pb.num_infeasible_requests_queued > 0:
                  backlog_queue = infeasible_bundles
              else:
                  backlog_queue = waiting_bundles
              for _ in range(resource_demand_pb.backlog_size):
                  backlog_queue.append(request_shape)
              if (
                  len(waiting_bundles + infeasible_bundles)
                  > AUTOSCALER_MAX_RESOURCE_DEMAND_VECTOR_SIZE
              ):
                  break
      except Exception:
          logger.exception("Failed to parse resource demands.")
  
      return waiting_bundles, infeasible_bundles
  
imports: import argparse
  import json
  import logging
  import os
  import signal
  import sys
  import time
  import traceback
  from collections import Counter
  from dataclasses import asdict
  from typing import Any, Callable, Dict, Optional, Union
  import ray
  import ray._private.ray_constants
  import ray._private.utils
  from ray._private.event.event_logger import get_event_logger
  from ray._private.ray_logging import setup_component_logger
  from ray._raylet import GcsClient
  from ray.autoscaler._private.autoscaler import StandardAutoscaler
  from ray.autoscaler._private.commands import teardown_cluster
  from ray.autoscaler._private.constants import AUTOSCALER_MAX_RESOURCE_DEMAND_VECTOR_SIZE, AUTOSCALER_METRIC_PORT, AUTOSCALER_UPDATE_INTERVAL_S
  from ray.autoscaler._private.event_summarizer import EventSummarizer
  from ray.autoscaler._private.load_metrics import LoadMetrics
  from ray.autoscaler._private.prom_metrics import AutoscalerPrometheusMetrics
  from ray.autoscaler._private.util import format_readonly_node_type
  from ray.core.generated import gcs_pb2
  from ray.core.generated.event_pb2 import Event
  from ray.experimental.internal_kv import _initialize_internal_kv, _internal_kv_del, _internal_kv_get, _internal_kv_initialized, _internal_kv_put
label_desc: Use image node_id 1 for calling a global function with example usage: parse_resource_demands(resource_load_by_shape) and returns: waiting_bundles,  infeasible_bundles