class_docstr: Forecaster base model for lstm, seq2seq, tcn and nbeats forecasters.
class_docstr_tok: ["Forecaster","base","model","for","lstm",",","seq2seq",",","tcn","and","nbeats","forecasters","."]
docstr: Fit(Train) the forecaster.
  
  :param data: The data support following formats:
  
         | 1. a numpy ndarray tuple (x, y):
         | x's shape is (num_samples, lookback, feature_dim) where lookback and feature_dim
         | should be the same as past_seq_len and input_feature_num.
         | y's shape is (num_samples, horizon, target_dim), where horizon and target_dim
         | should be the same as future_seq_len and output_feature_num.
         |
         | 2. a xshard item:
         | each partition can be a dictionary of {'x': x, 'y': y}, where x and y's shape
         | should follow the shape stated before.
         |
         | 3. pytorch dataloader:
         | the dataloader should return x, y in each iteration with the shape as following:
         | x's shape is (num_samples, lookback, feature_dim) where lookback and feature_dim
         | should be the same as past_seq_len and input_feature_num.
         | y's shape is (num_samples, horizon, target_dim), where horizon and target_dim
         | should be the same as future_seq_len and output_feature_num.
         |
         | 4. A bigdl.chronos.data.tsdataset.TSDataset instance:
         | Forecaster will automatically process the TSDataset.
         | By default, TSDataset will be transformed to a pytorch dataloader,
         | which is memory-friendly while a little bit slower.
         | Users may call `roll` on the TSDataset before calling `fit`
         | Then the training speed will be faster but will consume more memory.
  
  :param validation_data: Validation sample for validation loop. Defaults to 'None'.
         If you do not input data for 'validation_data', the validation_step will be skipped.
         Validation data will be ignored under distributed mode.
         The validation_data support following formats:
  
         | 1. a numpy ndarray tuple (x, y):
         | x's shape is (num_samples, lookback, feature_dim) where lookback and feature_dim
         | should be the same as past_seq_len and input_feature_num.
         | y's shape is (num_samples, horizon, target_dim), where horizon and target_dim
         | should be the same as future_seq_len and output_feature_num.
         |
         | 2. a xshard item:
         | each partition can be a dictionary of {'x': x, 'y': y}, where x and y's shape
         | should follow the shape stated before.
         |
         | 3. pytorch dataloader:
         | the dataloader should return x, y in each iteration with the shape as following:
         | x's shape is (num_samples, lookback, feature_dim) where lookback and feature_dim
         | should be the same as past_seq_len and input_feature_num.
         | y's shape is (num_samples, horizon, target_dim), where horizon and target_dim
         | should be the same as future_seq_len and output_feature_num.
         |
         | 4. A bigdl.chronos.data.tsdataset.TSDataset instance:
         | Forecaster will automatically process the TSDataset.
         | By default, TSDataset will be transformed to a pytorch dataloader,
         | which is memory-friendly while a little bit slower.
         | Users may call `roll` on the TSDataset before calling `fit`
         | Then the training speed will be faster but will consume more memory.
  
  :param epochs: Number of epochs you want to train. The value defaults to 1.
  :param batch_size: Number of batch size you want to train. The value defaults to 32.
         If you input a pytorch dataloader for `data`, the batch_size will follow the
         batch_size setted in `data`.if the forecaster is distributed, the batch_size will be
         evenly distributed to all workers.
  :param validation_mode:  A str represent the operation mode while having 'validation_data'.
         Defaults to 'output'. The validation_mode includes the following types:
  
         | 1. output:
         | If you choose 'output' for validation_mode, it will return a dict that records the
         | average validation loss of each epoch.
         |
         | 2. earlystop:
         | Monitor the val_loss and stop training when it stops improving.
         |
         | 3. best_epoch:
         | Monitor the val_loss. And load the checkpoint of the epoch with the smallest
         | val_loss after the training.
  
  :param earlystop_patience: Number of checks with no improvement after which training will
         be stopped. It takes effect when 'validation_mode' is 'earlystop'. Under the default
         configuration, one check happens after every training epoch.
  :param use_trail_id: choose a internal according to trial_id, which is used only
         in multi-objective search.
  :return: Validation loss if 'validation_data' is not None.
code: def fit(
      self,
      data,
      validation_data=None,
      epochs=1,
      batch_size=32,
      validation_mode="output",
      earlystop_patience=1,
      use_trial_id=None,
  ):
      """
      Fit(Train) the forecaster.
  
      :param data: The data support following formats:
  
             | 1. a numpy ndarray tuple (x, y):
             | x's shape is (num_samples, lookback, feature_dim) where lookback and feature_dim
             | should be the same as past_seq_len and input_feature_num.
             | y's shape is (num_samples, horizon, target_dim), where horizon and target_dim
             | should be the same as future_seq_len and output_feature_num.
             |
             | 2. a xshard item:
             | each partition can be a dictionary of {'x': x, 'y': y}, where x and y's shape
             | should follow the shape stated before.
             |
             | 3. pytorch dataloader:
             | the dataloader should return x, y in each iteration with the shape as following:
             | x's shape is (num_samples, lookback, feature_dim) where lookback and feature_dim
             | should be the same as past_seq_len and input_feature_num.
             | y's shape is (num_samples, horizon, target_dim), where horizon and target_dim
             | should be the same as future_seq_len and output_feature_num.
             |
             | 4. A bigdl.chronos.data.tsdataset.TSDataset instance:
             | Forecaster will automatically process the TSDataset.
             | By default, TSDataset will be transformed to a pytorch dataloader,
             | which is memory-friendly while a little bit slower.
             | Users may call `roll` on the TSDataset before calling `fit`
             | Then the training speed will be faster but will consume more memory.
  
      :param validation_data: Validation sample for validation loop. Defaults to 'None'.
             If you do not input data for 'validation_data', the validation_step will be skipped.
             Validation data will be ignored under distributed mode.
             The validation_data support following formats:
  
             | 1. a numpy ndarray tuple (x, y):
             | x's shape is (num_samples, lookback, feature_dim) where lookback and feature_dim
             | should be the same as past_seq_len and input_feature_num.
             | y's shape is (num_samples, horizon, target_dim), where horizon and target_dim
             | should be the same as future_seq_len and output_feature_num.
             |
             | 2. a xshard item:
             | each partition can be a dictionary of {'x': x, 'y': y}, where x and y's shape
             | should follow the shape stated before.
             |
             | 3. pytorch dataloader:
             | the dataloader should return x, y in each iteration with the shape as following:
             | x's shape is (num_samples, lookback, feature_dim) where lookback and feature_dim
             | should be the same as past_seq_len and input_feature_num.
             | y's shape is (num_samples, horizon, target_dim), where horizon and target_dim
             | should be the same as future_seq_len and output_feature_num.
             |
             | 4. A bigdl.chronos.data.tsdataset.TSDataset instance:
             | Forecaster will automatically process the TSDataset.
             | By default, TSDataset will be transformed to a pytorch dataloader,
             | which is memory-friendly while a little bit slower.
             | Users may call `roll` on the TSDataset before calling `fit`
             | Then the training speed will be faster but will consume more memory.
  
      :param epochs: Number of epochs you want to train. The value defaults to 1.
      :param batch_size: Number of batch size you want to train. The value defaults to 32.
             If you input a pytorch dataloader for `data`, the batch_size will follow the
             batch_size setted in `data`.if the forecaster is distributed, the batch_size will be
             evenly distributed to all workers.
      :param validation_mode:  A str represent the operation mode while having 'validation_data'.
             Defaults to 'output'. The validation_mode includes the following types:
  
             | 1. output:
             | If you choose 'output' for validation_mode, it will return a dict that records the
             | average validation loss of each epoch.
             |
             | 2. earlystop:
             | Monitor the val_loss and stop training when it stops improving.
             |
             | 3. best_epoch:
             | Monitor the val_loss. And load the checkpoint of the epoch with the smallest
             | val_loss after the training.
  
      :param earlystop_patience: Number of checks with no improvement after which training will
             be stopped. It takes effect when 'validation_mode' is 'earlystop'. Under the default
             configuration, one check happens after every training epoch.
      :param use_trail_id: choose a internal according to trial_id, which is used only
             in multi-objective search.
      :return: Validation loss if 'validation_data' is not None.
      """
      # input transform
      if isinstance(data, TSDataset):
          data = tsdataset_to_dataloader(
              data,
              batch_size=batch_size,
              lookback=self.data_config["past_seq_len"],
              horizon=self.data_config["future_seq_len"],
              num_processes=self.num_processes,
          )
      if isinstance(data, DataLoader) and self.distributed:
          data = loader_to_creator(data)
      if isinstance(data, tuple) and self.distributed:
          data = np_to_creator(data)
      try:
          from bigdl.orca.data.shard import SparkXShards
  
          if isinstance(data, SparkXShards) and not self.distributed:
              warnings.warn(
                  "Xshards is collected to local since the "
                  "forecaster is non-distribued."
              )
              data = xshard_to_np(data)
          if (
              isinstance(validation_data, SparkXShards)
              and not self.distributed
          ):
              warnings.warn(
                  "Xshards is collected to local since the "
                  "forecaster is non-distribued."
              )
              validation_data = xshard_to_np(validation_data)
      except ImportError:
          pass
  
      invalidOperationError(
          self.internal is not None,
          "The model is not properly built. "
          "Have you set search spaces in arguments? "
          "If so, you need to run tune before fit "
          "to search and build the model.",
      )
  
      # fit on internal
      if self.distributed:
          # for cluster mode
          from bigdl.orca.common import OrcaContext
  
          sc = OrcaContext.get_spark_context().getConf()
          num_nodes = (
              1
              if sc.get("spark.master").startswith("local")
              else int(sc.get("spark.executor.instances"))
          )
          if batch_size % self.workers_per_node != 0:
              invalidInputError(
                  False,
                  "Please make sure that batch_size can be divisible by "
                  "the product of worker_per_node and num_nodes, "
                  f"but 'batch_size' is {batch_size}, 'workers_per_node' "
                  f"is {self.workers_per_node}, 'num_nodes' is {num_nodes}",
              )
          batch_size //= self.workers_per_node * num_nodes
          return self.internal.fit(
              data=data, epochs=epochs, batch_size=batch_size
          )
      else:
          from bigdl.chronos.pytorch import TSTrainer as Trainer
  
          # numpy data shape checking
          if isinstance(data, tuple):
              check_data(data[0], data[1], self.data_config)
  
          # data transformation
          if isinstance(data, tuple):
              data = np_to_dataloader(
                  data, batch_size, self.num_processes
              )
  
          # dataloader change batch_size for multi-process
          if isinstance(data, DataLoader) and self.num_processes:
              data = dataloader_batch_resize(
                  data, batch_size, self.num_processes
              )
  
          # training process
          # forecaster_log_dir is a temp directory for training log
          # validation_ckpt_dir is a temp directory for best checkpoint on validation data
          with TemporaryDirectory() as forecaster_log_dir:
              with TemporaryDirectory() as validation_ckpt_dir:
                  from pytorch_lightning.loggers import CSVLogger
  
                  logger = (
                      False
                      if validation_data is None
                      else CSVLogger(
                          save_dir=forecaster_log_dir,
                          flush_logs_every_n_steps=10,
                          name="forecaster_tmp_log",
                      )
                  )
                  from pytorch_lightning.callbacks import EarlyStopping
  
                  early_stopping = EarlyStopping(
                      "val/loss", patience=earlystop_patience
                  )
                  from pytorch_lightning.callbacks import (
                      ModelCheckpoint,
                  )
  
                  checkpoint_callback = ModelCheckpoint(
                      monitor="val/loss",
                      dirpath=validation_ckpt_dir,
                      filename="best",
                      save_on_train_epoch_end=True,
                  )
                  if validation_mode == "earlystop":
                      callbacks = [early_stopping]
                  elif validation_mode == "best_epoch":
                      callbacks = [checkpoint_callback]
                  else:
                      callbacks = None
                  # Trainer init
                  self.trainer = Trainer(
                      logger=logger,
                      max_epochs=epochs,
                      callbacks=callbacks,
                      enable_checkpointing=self.checkpoint_callback,
                      num_processes=self.num_processes,
                      use_ipex=self.use_ipex,
                      log_every_n_steps=10,
                      distributed_backend=self.local_distributed_backend,
                  )
  
                  # This error is only triggered when the python
                  # interpreter starts additional processes.
                  # num_process=1 and subprocess will be safely started in the main process,
                  # so this error will not be triggered.
                  invalidInputError(
                      is_main_process(),
                      "Make sure new Python interpreters can "
                      "safely import the main module. ",
                      fixMsg="you should use if __name__ == '__main__':, "
                      "otherwise performance will be degraded.",
                  )
  
                  # build internal according to use_trail_id for multi-objective HPO
                  mo_hpo = False
                  if hasattr(self, "tune_trainer"):
                      if self.tune_trainer.hposearcher.objective.mo_hpo:
                          mo_hpo = True
                  if mo_hpo:
                      invalidOperationError(
                          self.tune_trainer.hposearcher.study,
                          "You must tune before fit the model.",
                      )
                      invalidInputError(
                          use_trial_id is not None,
                          "For multibojective HPO, "
                          "you must specify a trial id for fit.",
                      )
                      trial = (
                          self.tune_trainer.hposearcher.study.trials[
                              use_trial_id
                          ]
                      )
                      self.internal = self.tune_internal._model_build(
                          trial
                      )
  
                  # fitting
                  if not validation_data:
                      self.trainer.fit(self.internal, data)
                      self.fitted = True
                  else:
                      if isinstance(validation_data, TSDataset):
                          _rolled = validation_data.numpy_x is None
                          validation_data = validation_data.to_torch_data_loader(
                              batch_size=batch_size,
                              roll=_rolled,
                              lookback=self.data_config["past_seq_len"],
                              horizon=self.data_config[
                                  "future_seq_len"
                              ],
                              feature_col=validation_data.roll_feature,
                              target_col=validation_data.roll_target,
                              shuffle=False,
                          )
                      if isinstance(validation_data, tuple):
                          validation_data = np_to_dataloader(
                              validation_data,
                              batch_size,
                              self.num_processes,
                          )
                      self.trainer.fit(
                          self.internal, data, validation_data
                      )
                      self.fitted = True
                      fit_csv = os.path.join(
                          forecaster_log_dir,
                          "forecaster_tmp_log/version_0/metrics.csv",
                      )
                      best_path = os.path.join(
                          validation_ckpt_dir, "best.ckpt"
                      )
                      fit_out = read_csv(fit_csv)
                      if validation_mode == "best_epoch":
                          self.load(best_path)
                      # modify logger attr in trainer, otherwise predict will report error
                      self.trainer._logger_connector.on_trainer_init(
                          False,
                          self.trainer.flush_logs_every_n_steps,
                          self.trainer.log_every_n_steps,
                          self.trainer.move_metrics_to_cpu,
                      )
                      return fit_out
  
imports: from bigdl.chronos.forecaster.abstract import Forecaster
  from bigdl.chronos.forecaster.utils import *
  from bigdl.chronos.metric.forecast_metrics import Evaluator
  from typing import Optional
  import numpy
  import warnings
  import os
  from tempfile import TemporaryDirectory
  import torch
  from torch.utils.data import TensorDataset, DataLoader
  from .utils_hpo import GenericLightningModule, _format_metric_str, _config_has_search_space
  from bigdl.nano.utils.common import invalidOperationError, invalidInputError
  from bigdl.chronos.data.tsdataset import TSDataset
  from bigdl.chronos.pytorch.context_manager import DummyForecasterContextManager, ForecasterContextManager
  from bigdl.chronos.pytorch.utils import _pytorch_fashion_inference
label_desc: Use image node_id 5 for calling the BasePytorchForecaster obj's underlying member method code with example usage: obj.fit(data, validation_data, epochs, batch_size, validation_mode, earlystop_patience, use_trial_id) and returns: self, fit_out