docstr: Handle forecasting functions
docstr_tok: ["Handle","forecasting","functions"]
code: def handle_forecasting_function(self):
      """Handle forecasting functions"""
      aggregated_batch_list = []
      child = self.children[0]
      for batch in child.exec():
          aggregated_batch_list.append(batch)
      aggregated_batch = Batch.concat(aggregated_batch_list, copy=False)
      aggregated_batch.drop_column_alias()
  
      arg_map = {arg.key: arg.value for arg in self.node.metadata}
      if not self.node.impl_path:
          impl_path = (
              Path(f"{self.function_dir}/forecast.py")
              .absolute()
              .as_posix()
          )
      else:
          impl_path = self.node.impl_path.absolute().as_posix()
      library = "statsforecast"
      supported_libraries = ["statsforecast", "neuralforecast"]
  
      if "horizon" not in arg_map.keys():
          raise ValueError(
              "Horizon must be provided while creating function of type FORECASTING"
          )
      try:
          horizon = int(arg_map["horizon"])
      except Exception as e:
          err_msg = f"{str(e)}. HORIZON must be integral."
          logger.error(err_msg)
          raise FunctionIODefinitionError(err_msg)
  
      if "library" in arg_map.keys():
          try:
              assert arg_map["library"].lower() in supported_libraries
          except Exception:
              err_msg = (
                  "EvaDB currently supports "
                  + str(supported_libraries)
                  + " only."
              )
              logger.error(err_msg)
              raise FunctionIODefinitionError(err_msg)
  
          library = arg_map["library"].lower()
  
      """
          The following rename is needed for statsforecast/neuralforecast, which requires the column name to be the following:
          - The unique_id (string, int or category) represents an identifier for the series.
          - The ds (datestamp) column should be of a format expected by Pandas, ideally YYYY-MM-DD for a date or YYYY-MM-DD HH:MM:SS for a timestamp.
          - The y (numeric) represents the measurement we wish to forecast.
          For reference: https://nixtla.github.io/statsforecast/docs/getting-started/getting_started_short.html
          """
      aggregated_batch.rename(columns={arg_map["predict"]: "y"})
      if "time" in arg_map.keys():
          aggregated_batch.rename(columns={arg_map["time"]: "ds"})
      if "id" in arg_map.keys():
          aggregated_batch.rename(columns={arg_map["id"]: "unique_id"})
      if "conf" in arg_map.keys():
          try:
              conf = round(arg_map["conf"])
          except Exception:
              err_msg = "Confidence must be a number."
              logger.error(err_msg)
              raise FunctionIODefinitionError(err_msg)
      else:
          conf = 90
  
      if conf > 100:
          err_msg = "Confidence must <= 100."
          logger.error(err_msg)
          raise FunctionIODefinitionError(err_msg)
  
      data = aggregated_batch.frames
      if "unique_id" not in list(data.columns):
          data["unique_id"] = [1 for x in range(len(data))]
  
      if "ds" not in list(data.columns):
          data["ds"] = [x + 1 for x in range(len(data))]
  
      """
              Set or infer data frequency
          """
  
      if (
          "frequency" not in arg_map.keys()
          or arg_map["frequency"] == "auto"
      ):
          arg_map["frequency"] = pd.infer_freq(data["ds"])
      frequency = arg_map["frequency"]
      if frequency is None:
          raise RuntimeError(
              f"Can not infer the frequency for {self.node.name}. Please explicitly set it."
          )
  
      season_dict = {  # https://pandas.pydata.org/docs/user_guide/timeseries.html#timeseries-offset-aliases
          "H": 24,
          "M": 12,
          "Q": 4,
          "SM": 24,
          "BM": 12,
          "BMS": 12,
          "BQ": 4,
          "BH": 24,
      }
  
      new_freq = (
          frequency.split("-")[0] if "-" in frequency else frequency
      )  # shortens longer frequencies like Q-DEC
      season_length = (
          season_dict[new_freq] if new_freq in season_dict else 1
      )
  
      """
              Neuralforecast implementation
          """
      if library == "neuralforecast":
          try_to_import_neuralforecast()
          from neuralforecast import NeuralForecast
          from neuralforecast.auto import (
              AutoDeepAR,
              AutoFEDformer,
              AutoInformer,
              AutoNBEATS,
              AutoNHITS,
              AutoPatchTST,
              AutoTFT,
          )
  
          # from neuralforecast.auto import AutoAutoformer as AutoAFormer
          from neuralforecast.losses.pytorch import MQLoss
          from neuralforecast.models import (
              NBEATS,
              NHITS,
              TFT,
              DeepAR,
              FEDformer,
              Informer,
              PatchTST,
          )
  
          # from neuralforecast.models import Autoformer as AFormer
  
          model_dict = {
              "AutoNBEATS": AutoNBEATS,
              "AutoNHITS": AutoNHITS,
              "NBEATS": NBEATS,
              "NHITS": NHITS,
              "PatchTST": PatchTST,
              "AutoPatchTST": AutoPatchTST,
              "DeepAR": DeepAR,
              "AutoDeepAR": AutoDeepAR,
              "FEDformer": FEDformer,
              "AutoFEDformer": AutoFEDformer,
              # "AFormer": AFormer,
              # "AutoAFormer": AutoAFormer,
              "Informer": Informer,
              "AutoInformer": AutoInformer,
              "TFT": TFT,
              "AutoTFT": AutoTFT,
          }
  
          if "model" not in arg_map.keys():
              arg_map["model"] = "TFT"
  
          if "auto" not in arg_map.keys() or (
              arg_map["auto"].lower()[0] == "t"
              and "auto" not in arg_map["model"].lower()
          ):
              arg_map["model"] = "Auto" + arg_map["model"]
  
          try:
              model_here = model_dict[arg_map["model"]]
          except Exception:
              err_msg = "Supported models: " + str(model_dict.keys())
              logger.error(err_msg)
              raise FunctionIODefinitionError(err_msg)
          model_args = {}
  
          if "auto" not in arg_map["model"].lower():
              model_args["input_size"] = 2 * horizon
              model_args["early_stop_patience_steps"] = 20
          else:
              model_args_config = {
                  "input_size": 2 * horizon,
                  "early_stop_patience_steps": 20,
              }
  
          if len(data.columns) >= 4:
              exogenous_columns = [
                  x
                  for x in list(data.columns)
                  if x not in ["ds", "y", "unique_id"]
              ]
              if "auto" not in arg_map["model"].lower():
                  model_args["hist_exog_list"] = exogenous_columns
              else:
                  model_args_config[
                      "hist_exog_list"
                  ] = exogenous_columns
  
          if "auto" in arg_map["model"].lower():
  
              def get_optuna_config(trial):
                  return model_args_config
  
              model_args["config"] = get_optuna_config
              model_args["backend"] = "optuna"
  
          model_args["h"] = horizon
          model_args["loss"] = MQLoss(level=[conf])
  
          model = NeuralForecast(
              [model_here(**model_args)],
              freq=new_freq,
          )
  
      # """
      #     Statsforecast implementation
      # """
      else:
          if (
              "auto" in arg_map.keys()
              and arg_map["auto"].lower()[0] != "t"
          ):
              raise RuntimeError(
                  "Statsforecast implementation only supports automatic hyperparameter optimization. Please set AUTO to true."
              )
          try_to_import_statsforecast()
          from statsforecast import StatsForecast
          from statsforecast.models import (
              AutoARIMA,
              AutoCES,
              AutoETS,
              AutoTheta,
          )
  
          model_dict = {
              "AutoARIMA": AutoARIMA,
              "AutoCES": AutoCES,
              "AutoETS": AutoETS,
              "AutoTheta": AutoTheta,
          }
  
          if "model" not in arg_map.keys():
              arg_map["model"] = "ARIMA"
  
          if "auto" not in arg_map["model"].lower():
              arg_map["model"] = "Auto" + arg_map["model"]
  
          try:
              model_here = model_dict[arg_map["model"]]
          except Exception:
              err_msg = "Supported models: " + str(model_dict.keys())
              logger.error(err_msg)
              raise FunctionIODefinitionError(err_msg)
  
          model = StatsForecast(
              [model_here(season_length=season_length)], freq=new_freq
          )
  
      data["ds"] = pd.to_datetime(data["ds"])
  
      model_save_dir_name = (
          library + "_" + arg_map["model"] + "_" + new_freq
          if "statsforecast" in library
          else library
          + "_"
          + str(conf)
          + "_"
          + arg_map["model"]
          + "_"
          + new_freq
      )
      if len(data.columns) >= 4 and library == "neuralforecast":
          model_save_dir_name += "_exogenous_" + str(
              sorted(exogenous_columns)
          )
  
      model_dir = os.path.join(
          self.db.catalog().get_configuration_catalog_value(
              "model_dir"
          ),
          "tsforecasting",
          model_save_dir_name,
          str(hashlib.sha256(data.to_string().encode()).hexdigest()),
      )
      Path(model_dir).mkdir(parents=True, exist_ok=True)
  
      model_save_name = "horizon" + str(horizon) + ".pkl"
  
      model_path = os.path.join(model_dir, model_save_name)
  
      existing_model_files = sorted(
          os.listdir(model_dir),
          key=lambda x: int(x.split("horizon")[1].split(".pkl")[0]),
      )
      existing_model_files = [
          x
          for x in existing_model_files
          if int(x.split("horizon")[1].split(".pkl")[0]) >= horizon
      ]
      if len(existing_model_files) == 0:
          logger.info("Training, please wait...")
          for column in data.columns:
              if column != "ds" and column != "unique_id":
                  data[column] = data.apply(
                      lambda x: self.convert_to_numeric(x[column]),
                      axis=1,
                  )
          rmses = []
          if library == "neuralforecast":
              cuda_devices_here = "0"
              if "CUDA_VISIBLE_DEVICES" in os.environ:
                  cuda_devices_here = os.environ[
                      "CUDA_VISIBLE_DEVICES"
                  ].split(",")[0]
  
              with set_env(CUDA_VISIBLE_DEVICES=cuda_devices_here):
                  model.fit(df=data, val_size=horizon)
                  model.save(model_path, overwrite=True)
                  if (
                      "metrics" in arg_map
                      and arg_map["metrics"].lower()[0] == "t"
                  ):
                      crossvalidation_df = model.cross_validation(
                          df=data, val_size=horizon
                      )
                      for uid in crossvalidation_df.unique_id.unique():
                          crossvalidation_df_here = crossvalidation_df[
                              crossvalidation_df.unique_id == uid
                          ]
                          rmses.append(
                              root_mean_squared_error(
                                  crossvalidation_df_here.y,
                                  crossvalidation_df_here[
                                      arg_map["model"] + "-median"
                                  ],
                              )
                              / np.mean(crossvalidation_df_here.y)
                          )
                          mean_rmse = np.mean(rmses)
                          with open(model_path + "_rmse", "w") as f:
                              f.write(str(mean_rmse) + "\n")
          else:
              # The following lines of code helps eliminate the math error encountered in statsforecast when only one datapoint is available in a time series
              for col in data["unique_id"].unique():
                  if len(data[data["unique_id"] == col]) == 1:
                      data = data._append(
                          [data[data["unique_id"] == col]],
                          ignore_index=True,
                      )
  
              model.fit(df=data[["ds", "y", "unique_id"]])
              hypers = ""
              if "arima" in arg_map["model"].lower():
                  from statsforecast.arima import arima_string
  
                  hypers += arima_string(model.fitted_[0, 0].model_)
              f = open(model_path, "wb")
              pickle.dump(model, f)
              f.close()
              if (
                  "metrics" not in arg_map
                  or arg_map["metrics"].lower()[0] == "t"
              ):
                  crossvalidation_df = model.cross_validation(
                      df=data[["ds", "y", "unique_id"]],
                      h=horizon,
                      step_size=24,
                      n_windows=1,
                  ).reset_index()
                  for uid in crossvalidation_df.unique_id.unique():
                      crossvalidation_df_here = crossvalidation_df[
                          crossvalidation_df.unique_id == uid
                      ]
                      rmses.append(
                          root_mean_squared_error(
                              crossvalidation_df_here.y,
                              crossvalidation_df_here[arg_map["model"]],
                          )
                          / np.mean(crossvalidation_df_here.y)
                      )
                  mean_rmse = np.mean(rmses)
                  with open(model_path + "_rmse", "w") as f:
                      f.write(str(mean_rmse) + "\n")
                      f.write(hypers + "\n")
      elif not Path(model_path).exists():
          model_path = os.path.join(model_dir, existing_model_files[-1])
      io_list = self._resolve_function_io(None)
      data["ds"] = data.ds.astype(str)
      metadata_here = [
          FunctionMetadataCatalogEntry("model_name", arg_map["model"]),
          FunctionMetadataCatalogEntry("model_path", model_path),
          FunctionMetadataCatalogEntry(
              "predict_column_rename", arg_map.get("predict", "y")
          ),
          FunctionMetadataCatalogEntry(
              "time_column_rename", arg_map.get("time", "ds")
          ),
          FunctionMetadataCatalogEntry(
              "id_column_rename", arg_map.get("id", "unique_id")
          ),
          FunctionMetadataCatalogEntry("horizon", horizon),
          FunctionMetadataCatalogEntry("library", library),
          FunctionMetadataCatalogEntry("conf", conf),
      ]
  
      return (
          self.node.name,
          impl_path,
          self.node.function_type,
          io_list,
          metadata_here,
      )
  
imports: import contextlib
  import hashlib
  import locale
  import os
  import pickle
  import re
  import time
  from pathlib import Path
  from typing import Dict, List
  import numpy
  import pandas
  from evadb.catalog.catalog_utils import get_metadata_properties
  from evadb.catalog.models.function_catalog import FunctionCatalogEntry
  from evadb.catalog.models.function_io_catalog import FunctionIOCatalogEntry
  from evadb.catalog.models.function_metadata_catalog import FunctionMetadataCatalogEntry
  from evadb.configuration.constants import DEFAULT_SKLEARN_TRAIN_MODEL, DEFAULT_TRAIN_REGRESSION_METRIC, DEFAULT_TRAIN_TIME_LIMIT, DEFAULT_XGBOOST_TASK, SKLEARN_SUPPORTED_MODELS, EvaDB_INSTALLATION_DIR
  from evadb.database import EvaDBDatabase
  from evadb.executor.abstract_executor import AbstractExecutor
  from evadb.functions.decorators.utils import load_io_from_function_decorators
  from evadb.models.storage.batch import Batch
  from evadb.plan_nodes.create_function_plan import CreateFunctionPlan
  from evadb.third_party.huggingface.create import gen_hf_io_catalog_entries
  from evadb.utils.errors import FunctionIODefinitionError
  from evadb.utils.generic_utils import load_function_class_from_file, string_comparison_case_insensitive, try_to_import_flaml_automl, try_to_import_ludwig, try_to_import_neuralforecast, try_to_import_statsforecast, try_to_import_torch, try_to_import_ultralytics
  from evadb.utils.logging_manager import logger
label_desc: Use image node_id 8 for calling the CreateFunctionExecutor obj's underlying member method code with example usage: obj.handle_forecasting_function() and returns: self,  impl_path,  self,  io_list,  metadata_here, model_args_config