code: def __init__(
      self,
      *,
      model_creator: Union[Callable[[Dict], "Module"], None],
      optimizer_creator: Union[
          Callable[["Module", Dict], "Optimizer"], None
      ] = None,
      loss_creator: Union[
          "Loss", Callable[[Dict], "Loss"], None
      ] = None,
      metrics: Union["Metric", List["Metric"], None] = None,
      scheduler_creator: Optional[
          Callable[[Dict], "LRScheduler"]
      ] = None,
      config: Dict = None,
      use_tqdm: bool = False,
      backend: str = "ray",
      workers_per_node: int = 1,
      sync_stats: bool = True,
      log_level: int = logging.INFO,
  ):
      if config is not None and "batch_size" in config:
          invalidInputError(
              False,
              "Please do not specify batch_size in config. Input batch_size in the"
              " fit/evaluate/predict function of the estimator instead.",
          )
  
      # todo remove ray_ctx to run on workers
      ray_ctx = OrcaRayContext.get()
      if not isinstance(
          model_creator, types.FunctionType
      ):  # Torch model is also callable.
          invalidInputError(
              False, "Must provide a function for model_creator."
          )
  
      self.model_creator = model_creator
      self.optimizer_creator = optimizer_creator
      self.loss_creator = loss_creator
      self.scheduler_creator = scheduler_creator
      self.sync_stats = sync_stats
      self.use_tqdm = use_tqdm
      self.backend = backend
      self.workers_per_node = workers_per_node
  
      self.config = {} if config is None else config
      worker_config = copy.copy(self.config)
      self.setup_params = dict(
          model_creator=self.model_creator,
          optimizer_creator=self.optimizer_creator,
          loss_creator=self.loss_creator,
          scheduler_creator=self.scheduler_creator,
          config=worker_config,
          metrics=metrics,
          sync_stats=sync_stats,
          log_level=log_level,
      )
      self.setup(
          params=self.setup_params,
          backend=self.backend,
          runner_cls=PytorchRayWorker,
          workers_per_node=self.workers_per_node,
      )
  
imports: import types
  import copy
  import logging
  import math
  from bigdl.orca.data.ray_xshards import RayXShards
  from bigdl.orca.learn.pytorch.pytorch_ray_worker import PytorchRayWorker
  from bigdl.orca.learn.utils import maybe_dataframe_to_xshards, dataframe_to_xshards, convert_predict_xshards_to_dataframe, update_predict_xshards, process_xshards_of_pandas_dataframe, add_predict_to_pd_xshards
  from bigdl.orca.ray import OrcaRayContext
  from bigdl.orca.learn.pytorch.core.base_ray_estimator import BaseRayEstimator
  from bigdl.orca.learn.pytorch.utils import process_stats, check_for_failure
  from bigdl.orca.learn.pytorch.callbacks.maincallback import make_only_mainCallback
  from bigdl.orca.learn.pytorch.callbacks.tqdm import TqdmCallback, is_tqdm_exists
  from bigdl.orca.learn.pytorch.callbacks.maxsteps import MaxstepsCallback
  import ray
  from bigdl.dllib.utils.log4Error import invalidInputError
  from typing import TYPE_CHECKING, Union, Optional, Callable, Dict, List
label_desc: Use image node_id 1 to create a new PyTorchRayEstimator object from inherited base classes: BaseRayEstimator with example: obj = PyTorchRayEstimator()