code: def test_forecast_tcmf_xshards(self):
      from zoo.orca import OrcaContext
      import zoo.orca.data.pandas
      import pandas as pd
  
      OrcaContext.pandas_read_backend = "pandas"
  
      def preprocessing(df, id_name, y_name):
          id = df.index
          data = df.to_numpy()
          result = dict({id_name: id, y_name: data})
          return result
  
      def postprocessing(pred_results, output_dt_col_name):
          id_arr = pred_results["id"]
          pred_results = pred_results["prediction"]
          pred_results = np.concatenate(
              (np.expand_dims(id_arr, axis=1), pred_results), axis=1
          )
          final_df = pd.DataFrame(
              pred_results, columns=["id"] + output_dt_col_name
          )
          final_df.id = final_df.id.astype("int")
          final_df = final_df.set_index("id")
          final_df.columns.name = "datetime"
          final_df = (
              final_df.unstack()
              .reset_index()
              .rename({0: "prediction"}, axis=1)
          )
          return final_df
  
      def get_pred(d):
          return d["prediction"]
  
      with tempfile.NamedTemporaryFile() as temp:
          data = np.random.rand(300, 480)
          df = pd.DataFrame(data)
          df.to_csv(temp.name)
          shard = zoo.orca.data.pandas.read_csv(temp.name)
      shard.cache()
      shard_train = shard.transform_shard(preprocessing, "id", "data")
      with self.assertRaises(Exception) as context:
          self.model.fit(shard_train)
      self.assertTrue(
          "key `y` doesn't exist in x" in str(context.exception)
      )
      shard_train = shard.transform_shard(preprocessing, "cid", "y")
      with self.assertRaises(Exception) as context:
          self.model.fit(shard_train)
      self.assertTrue(
          "key `id` doesn't exist in x" in str(context.exception)
      )
      with self.assertRaises(Exception) as context:
          self.model.is_xshards_distributed()
      self.assertTrue(
          "You should run fit before calling is_xshards_distributed()"
          in str(context.exception)
      )
      shard_train = shard.transform_shard(preprocessing, "id", "y")
      self.model.fit(shard_train, **self.fit_params)
      assert self.model.is_xshards_distributed()
      with self.assertRaises(Exception) as context:
          self.model.fit(shard_train)
      self.assertTrue(
          "This model has already been fully trained"
          in str(context.exception)
      )
      with self.assertRaises(Exception) as context:
          self.model.fit_incremental(shard_train)
      self.assertTrue(
          "NotImplementedError" in context.exception.__class__.__name__
      )
      with tempfile.TemporaryDirectory() as tempdirname:
          self.model.save(tempdirname + "/model")
          loaded_model = TCMFForecaster.load(
              tempdirname + "/model", is_xshards_distributed=True
          )
      horizon = np.random.randint(1, 50)
      yhat_shard_origin = self.model.predict(horizon=horizon)
      yhat_list_origin = yhat_shard_origin.collect()
      yhat_list_origin = list(map(get_pred, yhat_list_origin))
      yhat_shard = loaded_model.predict(horizon=horizon)
      yhat_list = yhat_shard.collect()
      yhat_list = list(map(get_pred, yhat_list))
      yhat_origin = np.concatenate(yhat_list_origin)
      yhat = np.concatenate(yhat_list)
      assert yhat.shape == (300, horizon)
      np.testing.assert_equal(yhat, yhat_origin)
      output_dt_col_name = pd.date_range(
          start="2020-05-01", periods=horizon, freq="H"
      ).to_list()
      yhat_df_shards = yhat_shard.transform_shard(
          postprocessing, output_dt_col_name
      )
      final_df_list = yhat_df_shards.collect()
      final_df = pd.concat(final_df_list)
      final_df.sort_values("datetime", inplace=True)
      assert final_df.shape == (300 * horizon, 3)
      OrcaContext.pandas_read_backend = "spark"
  
file_path: analytics-zoo/pyzoo/test/zoo/chronos/forecaster/test_tcmf_forecaster.py
imports: import pytest
  import numpy
  from zoo.chronos.forecaster.tcmf_forecaster import TCMFForecaster
  from unittest import TestCase
  import tempfile
  import pandas
label_desc: Use image node_id 6 for calling the TestChronosModelTCMFForecaster obj's underlying member method code with example usage: obj.test_forecast_tcmf_xshards() and returns: result, final_df, d