code: def setup(self) -> None:
      if self.is_local:
          self.num_ray_nodes = 1
          spark_cores = self._get_spark_local_cores()
          if self.ray_node_cpu_cores:
              ray_node_cpu_cores = int(self.ray_node_cpu_cores)
              if ray_node_cpu_cores > spark_cores:
                  warnings.warn(
                      "ray_node_cpu_cores is larger than available Spark cores, "
                      "make sure there are enough resources on your machine"
                  )
              self.ray_node_cpu_cores = ray_node_cpu_cores
          else:
              self.ray_node_cpu_cores = spark_cores
      # For Spark local mode, directly call ray.init() and ray.shutdown().
      # ray.shutdown() would clear up all the ray related processes.
      # Ray Manager is only needed for Spark cluster mode to monitor ray processes.
      else:
          if self.sc.getConf().contains("spark.executor.cores"):
              executor_cores = int(
                  self.sc.getConf().get("spark.executor.cores")
              )
          else:
              executor_cores = None
          if self.ray_node_cpu_cores:
              ray_node_cpu_cores = int(self.ray_node_cpu_cores)
              if executor_cores and ray_node_cpu_cores > executor_cores:
                  warnings.warn(
                      "ray_node_cpu_cores is larger than Spark executor cores, "
                      "make sure there are enough resources on your cluster"
                  )
              self.ray_node_cpu_cores = ray_node_cpu_cores
          elif executor_cores:
              self.ray_node_cpu_cores = executor_cores
          else:
              invalidInputError(
                  False,
                  "spark.executor.cores not detected in the SparkContext, you need"
                  " to manually specify num_ray_nodes and ray_node_cpu_cores for"
                  " RayOnSparkContext to start ray services",
              )
          if self.sc.getConf().contains("spark.executor.instances"):
              num_executors = int(
                  self.sc.getConf().get("spark.executor.instances")
              )
          elif self.sc.getConf().contains("spark.cores.max"):
              import math
  
              num_executors = math.floor(
                  int(self.sc.getConf().get("spark.cores.max"))
                  / self.ray_node_cpu_cores
              )
          else:
              num_executors = None
          if self.num_ray_nodes:
              num_ray_nodes = int(self.num_ray_nodes)
              if num_executors and num_ray_nodes > num_executors:
                  warnings.warn(
                      "num_ray_nodes is larger than the number of Spark executors, "
                      "make sure there are enough resources on your cluster"
                  )
              self.num_ray_nodes = num_ray_nodes
          elif num_executors:
              self.num_ray_nodes = num_executors
          else:
              invalidInputError(
                  False,
                  "spark.executor.cores not detected in the SparkContext, you need"
                  " to manually specify num_ray_nodes and ray_node_cpu_cores for"
                  " RayOnSparkContext to start ray services",
              )
  
          from bigdl.dllib.utils.utils import detect_python_location
  
          self.python_loc = os.environ.get(
              "PYSPARK_PYTHON", detect_python_location()
          )
          self.ray_service = RayServiceFuncGenerator(
              python_loc=self.python_loc,
              redis_port=self.redis_port,
              redis_password=self.redis_password,
              ray_node_cpu_cores=self.ray_node_cpu_cores,
              object_store_memory=self.object_store_memory,
              verbose=self.verbose,
              env=self.env,
              include_webui=self.include_webui,
              extra_params=self.extra_params,
              system_config=self.system_config,
          )
      self.total_cores = self.num_ray_nodes * self.ray_node_cpu_cores
  
imports: import os
  import re
  import subprocess
  import time
  import uuid
  import random
  import warnings
  import tempfile
  import filelock
  import multiprocessing
  from packaging import version
  from bigdl.orca.ray.process import session_execute, ProcessMonitor
  from bigdl.orca.ray.utils import is_local
  from bigdl.orca.ray.utils import resource_to_bytes
  from bigdl.orca.ray.utils import get_parent_pid
  from bigdl.dllib.utils.log4Error import invalidInputError
  from typing import TYPE_CHECKING
  from typing import Any, Dict, Optional, List
label_desc: Use image node_id 2 for calling the RayOnSparkContext obj's underlying member method code with example usage: obj.setup() without return types