PaddleX/paddlex/repo_apis/PaddleTS_api/ts_fc/config.py

# copyright (c) 2024 PaddlePaddle Authors. All Rights Reserve.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import os
import yaml

from ..ts_base.config import BaseTSConfig
from ....utils.misc import abspath


class LongForecastConfig(BaseTSConfig):
    """Long Forecast Config"""

    def update_input_len(self, seq_len: int):
        """
        upadte the input sequence length

        Args:
            seq_len (int): input length

        Raises:
            TypeError: if seq_len is not dict, raising TypeError
        """
        if "seq_len" not in self:
            raise RuntimeError(
                "Not able to update seq_len, because no seq_len config was found."
            )
        self.set_val("seq_len", seq_len)

    def update_predict_len(self, predict_len: int):
        """
        updaet the predict sequence length

        Args:
            predict_len (int): predict length

        Raises:
            RuntimeError: if predict_len is not set, raising RuntimeError
        """
        if "predict_len" not in self:
            raise RuntimeError(
                "Not able to update predict_len, because no predict_len config was found."
            )
        self.set_val("predict_len", predict_len)

    def update_sampling_stride(self, sampling_stride: int):
        """
        updaet the sampling stride of sequence to reduce the training time

        Args:
            sampling_stride (int): sampling rate

        Raises:
            RuntimeError: if sampling stride is not set, raising RuntimeError
        """
        if "sampling_stride" not in self:
            raise RuntimeError(
                "Not able to update sampling_stride, because no sampling_stride config was found."
            )
        self.set_val("sampling_stride", sampling_stride)

    def update_dataset(self, dataset_dir: str, dataset_type: str = None):
        """
        upadte the dataset

        Args:
            dataset_dir (str): dataset root path
            dataset_type (str, optional): type to set for dataset. Default='TSDataset'
        """
        if dataset_type is None:
            dataset_type = "TSDataset"
        dataset_dir = abspath(dataset_dir)
        ds_cfg = self._make_custom_dataset_config(dataset_dir)
        self.update(ds_cfg)

    def update_basic_info(self, info_params: dict):
        """
        update basic info including time_col, freq, target_cols.

        Args:
            info_params (dict): upadte basic info

        Raises:
            TypeError: if info_params is not dict, raising TypeError
        """
        if isinstance(info_params, dict):
            self.update({"info_params": info_params})
        else:
            raise TypeError("`info_params` must be dict.")

    def update_patience(self, patience: int):
        """
        update patience.

        Args:
            patience (int): upadte patience

        Raises:
            RuntimeError: if patience is not found, raising RuntimeError
        """
        if "patience" not in self.model["model_cfg"]:
            raise RuntimeError(
                "Not able to update patience, because no patience config was found."
            )
        self.model["model_cfg"]["patience"] = patience

    def _make_custom_dataset_config(self, dataset_root_path: str):
        """construct the dataset config that meets the format requirements

        Args:
            dataset_root_path (str): the root directory of dataset.

        Returns:
            dict: the dataset config.
        """
        ds_cfg = {
            "dataset": {
                "name": "TSDataset",
                "dataset_root": dataset_root_path,
                "train_path": os.path.join(dataset_root_path, "train.csv"),
                "val_path": os.path.join(dataset_root_path, "val.csv"),
            },
        }

        return ds_cfg