Disjoint-time-based dataset class

cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset `dataclass`

Bases: CesnetDataset

This class is used for disjoint-time-based returning of data. Can be created by using get_dataset with parameter dataset_type = DatasetType.DISJOINT_TIME_BASED.

Disjoint-time-based means batch size affects number of returned times in one batch and each set can have different time series. Which time series are returned does not change. Additionally it supports sliding window.

The dataset provides multiple ways to access the data:

Iterable PyTorch DataLoader: For batch processing.
Pandas DataFrame: For loading the entire training, validation or test set at once.
Numpy array: For loading the entire training, validation or test set at once.
See loading data for more details.

The dataset is stored in a PyTables database. The internal TimeBasedDataset, SplittedDataset, TimeBasedInitializerDataset classes (used only when calling set_dataset_config_and_initialize) act as wrappers that implement the PyTorch Dataset interface. These wrappers are compatible with PyTorch’s DataLoader, providing efficient parallel data loading.

The dataset configuration is done through the DisjointTimeBasedConfig class.

Intended usage:

Create an instance of the dataset with the desired data root by calling get_dataset. This will download the dataset if it has not been previously downloaded and return instance of dataset.
Create an instance of DisjointTimeBasedConfig and set it using set_dataset_config_and_initialize. This initializes the dataset, including data splitting (train/validation/test), fitting transformers (if needed), selecting features, and more. This is cached for later use.
Use get_train_dataloader/get_train_df/get_train_numpy to get training data for chosen model.
Validate the model and perform the hyperparameter optimalization on get_val_dataloader/get_val_df/get_val_numpy.
Evaluate the model on get_test_dataloader/get_test_df/get_test_numpy.

Alternatively you can use load_benchmark

Call load_benchmark with the desired benchmark. You can use your own saved benchmark or you can use already built-in one. This will download the dataset and annotations (if available) if they have not been previously downloaded.
Retrieve the initialized dataset using get_initialized_dataset. This will provide a dataset that is ready to use.
Use get_train_dataloader/get_train_df/get_train_numpy to get training data for chosen model.
Validate the model and perform the hyperparameter optimalization on get_val_dataloader/get_val_df/get_val_numpy.
Evaluate the model on get_test_dataloader/get_test_df/get_test_numpy.

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

@dataclass
class DisjointTimeBasedCesnetDataset(CesnetDataset):
    """This class is used for disjoint-time-based returning of data. Can be created by using [`get_dataset`](reference_cesnet_database.md#cesnet_tszoo.datasets.databases.cesnet_database.CesnetDatabase.get_dataset) with parameter `dataset_type` = `DatasetType.DISJOINT_TIME_BASED`.

    Disjoint-time-based means batch size affects number of returned times in one batch and each set can have different time series. Which time series are returned does not change. Additionally it supports sliding window.

    The dataset provides multiple ways to access the data:

    - **Iterable PyTorch DataLoader**: For batch processing.
    - **Pandas DataFrame**: For loading the entire training, validation or test set at once.
    - **Numpy array**: For loading the entire training, validation or test set at once. 
    - See [loading data][loading-data] for more details.

    The dataset is stored in a [PyTables](https://www.pytables.org/) database. The internal `TimeBasedDataset`, `SplittedDataset`, `TimeBasedInitializerDataset` classes (used only when calling [`set_dataset_config_and_initialize`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.set_dataset_config_and_initialize)) act as wrappers that implement the PyTorch [`Dataset`](https://pytorch.org/docs/stable/data.html#torch.utils.data.Dataset) 
    interface. These wrappers are compatible with PyTorch’s [`DataLoader`](https://pytorch.org/docs/stable/data.html#torch.utils.data.DataLoader), providing efficient parallel data loading. 

    The dataset configuration is done through the [`DisjointTimeBasedConfig`](reference_disjoint_time_based_config.md#references.DisjointTimeBasedConfig) class.       

    **Intended usage:**

    1. Create an instance of the dataset with the desired data root by calling [`get_dataset`](reference_cesnet_database.md#cesnet_tszoo.datasets.databases.cesnet_database.CesnetDatabase.get_dataset). This will download the dataset if it has not been previously downloaded and return instance of dataset.
    2. Create an instance of [`DisjointTimeBasedConfig`](reference_disjoint_time_based_config.md#references.DisjointTimeBasedConfig) and set it using [`set_dataset_config_and_initialize`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.set_dataset_config_and_initialize). 
       This initializes the dataset, including data splitting (train/validation/test), fitting transformers (if needed), selecting features, and more. This is cached for later use.
    3. Use [`get_train_dataloader`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_train_dataloader)/[`get_train_df`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_train_df)/[`get_train_numpy`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_train_numpy) to get training data for chosen model.
    4. Validate the model and perform the hyperparameter optimalization on [`get_val_dataloader`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_val_dataloader)/[`get_val_df`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_val_df)/[`get_val_numpy`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_val_numpy).
    5. Evaluate the model on [`get_test_dataloader`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_test_dataloader)/[`get_test_df`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_test_df)/[`get_test_numpy`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_test_numpy).  

    Alternatively you can use [`load_benchmark`][cesnet_tszoo.benchmarks.load_benchmark]

    1. Call [`load_benchmark`][cesnet_tszoo.benchmarks.load_benchmark] with the desired benchmark. You can use your own saved benchmark or you can use already built-in one. This will download the dataset and annotations (if available) if they have not been previously downloaded.
    2. Retrieve the initialized dataset using [`get_initialized_dataset`](reference_benchmarks.md#cesnet_tszoo.benchmarks.Benchmark.get_initialized_dataset). This will provide a dataset that is ready to use.
    3. Use [`get_train_dataloader`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_train_dataloader)/[`get_train_df`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_train_df)/[`get_train_numpy`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_train_numpy) to get training data for chosen model.
    4. Validate the model and perform the hyperparameter optimalization on [`get_val_dataloader`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_val_dataloader)/[`get_val_df`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_val_df)/[`get_val_numpy`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_val_numpy).
    5. Evaluate the model on [`get_test_dataloader`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_test_dataloader)/[`get_test_df`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_test_df)/[`get_test_numpy`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.get_test_numpy).          
    """

    dataset_config: Optional[DisjointTimeBasedConfig] = field(default=None, init=False)
    """Configuration of the dataset."""

    train_dataset: Optional[DisjointTimeBasedSplittedDataset] = field(default=None, init=False)
    """Training set as a `SplittedDataset` instance wrapping multiple `TimeBasedDataset` that wrap the PyTables database."""
    val_dataset: Optional[DisjointTimeBasedSplittedDataset] = field(default=None, init=False)
    """Validation set as a `SplittedDataset` instance wrapping multiple `TimeBasedDataset` that wrap the PyTables database."""
    test_dataset: Optional[DisjointTimeBasedSplittedDataset] = field(default=None, init=False)
    """Test set as a `SplittedDataset` instance wrapping multiple `TimeBasedDataset` that wrap the PyTables database. """

    train_dataloader: Optional[DisjointTimeBasedDataloader] = field(default=None, init=False)
    """Iterable PyTorch [`DataLoader`](https://pytorch.org/docs/stable/data.html#torch.utils.data.DataLoader) for training set."""
    val_dataloader: Optional[DisjointTimeBasedDataloader] = field(default=None, init=False)
    """Iterable PyTorch [`DataLoader`](https://pytorch.org/docs/stable/data.html#torch.utils.data.DataLoader) for validation set."""
    test_dataloader: Optional[DisjointTimeBasedDataloader] = field(default=None, init=False)
    """Iterable PyTorch [`DataLoader`](https://pytorch.org/docs/stable/data.html#torch.utils.data.DataLoader) for test set. """

    dataloader_factory: DisjointTimeBasedDataloaderFactory = field(default=DisjointTimeBasedDataloaderFactory(), init=False)
    """Factory used to create DisjointTimeBasedDataloader.  """

    dataset_type: DatasetType = field(default=DatasetType.DISJOINT_TIME_BASED, init=False)

    _export_config_copy: Optional[DisjointTimeBasedConfig] = field(default=None, init=False)

    def __post_init__(self):
        super().__post_init__()

        self.logger.info("Dataset is disjoint_time_based. Use cesnet_tszoo.configs.DisjointTimeBasedConfig")

    def set_dataset_config_and_initialize(self, dataset_config: DisjointTimeBasedConfig, display_config_details: Optional[Literal["text", "diagram"]] = "text", workers: int | Literal["config"] = "config") -> None:
        """
        Initialize training set, validation set, test set etc.. This method must be called before any data can be accessed. It is required for the final initialization of [`dataset_config`](reference_disjoint_time_based_config.md#references.DisjointTimeBasedConfig).

        The following configuration attributes are used during initialization:

        Dataset config | Description
        -------------- | -----------
        `init_workers` | Specifies the number of workers to use for initialization. Applied when `workers` = "config".
        `partial_fit_initialized_transformers` | Determines whether initialized transformers should be partially fitted on the training data.
        `nan_threshold` | Filters out time series with missing values exceeding the specified threshold.

        Parameters:
            dataset_config: Desired configuration of the dataset.
            display_config_details: Flag indicating whether and how to display the configuration values after initialization. `Default: text`  
            workers: The number of workers to use during initialization. `Default: "config"`  
        """

        assert dataset_config is not None, "Used dataset_config cannot be None."
        assert isinstance(dataset_config, DisjointTimeBasedConfig), f"This config is used for dataset of type '{dataset_config.dataset_type}'. Meanwhile this dataset is of type '{self.metadata.dataset_type}'."

        super(DisjointTimeBasedCesnetDataset, self).set_dataset_config_and_initialize(dataset_config, display_config_details, workers)

    def apply_transformer(self, transform_with: type | list[Transformer] | np.ndarray[Transformer] | TransformerType | Transformer | Literal["min_max_scaler", "standard_scaler", "max_abs_scaler", "log_transformer", "l2_normalizer"] | None | Literal["config"] = "config",
                          partial_fit_initialized_transformers: bool | Literal["config"] = "config", workers: int | Literal["config"] = "config") -> None:
        """Used for updating transformer and relevant configurations set in config.
        Set parameter to `config` to keep it as it is config.
        If exception is thrown during set, no changes are made.

        Affects following configuration:

        Dataset config | Description
        -------------- | -----------
        `transform_with` | Defines the transformer to transform the dataset.
        `partial_fit_initialized_transformers` | If `True`, partial fitting on train set is performed when using initialized transformers.

        Parameters:
            transform_with: Defines the transformer to transform the dataset. `Defaults: config`.  
            partial_fit_initialized_transformers: If `True`, partial fitting on train set is performed when using initiliazed transformers. `Defaults: config`.  
            workers: How many workers to use when setting new transformer. `Defaults: config`.      
        """

        if self.dataset_config is None or not self.dataset_config.is_initialized:
            raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before updating transformer values.")

        self.update_dataset_config_and_initialize(transform_with=transform_with, partial_fit_initialized_transformers=partial_fit_initialized_transformers, workers=workers)

    def update_dataset_config_and_initialize(self,
                                             default_values: list[Number] | npt.NDArray[np.number] | dict[str, Number] | Number | Literal["default"] | None | Literal["config"] = "config",
                                             sliding_window_size: int | None | Literal["config"] = "config",
                                             sliding_window_prediction_size: int | None | Literal["config"] = "config",
                                             sliding_window_step: int | Literal["config"] = "config",
                                             set_shared_size: float | int | Literal["config"] = "config",
                                             train_batch_size: int | Literal["config"] = "config",
                                             val_batch_size: int | Literal["config"] = "config",
                                             test_batch_size: int | Literal["config"] = "config",
                                             preprocess_order: list[str, type] | Literal["config"] = "config",
                                             fill_missing_with: type | FillerType | Literal["mean_filler", "forward_filler", "linear_interpolation_filler"] | None | Literal["config"] = "config",
                                             transform_with: type | list[Transformer] | np.ndarray[Transformer] | TransformerType | Transformer | Literal["min_max_scaler", "standard_scaler", "max_abs_scaler", "log_transformer", "l2_normalizer"] | None | Literal["config"] = "config",
                                             handle_anomalies_with: type | AnomalyHandlerType | Literal["z-score", "interquartile_range"] | None | Literal["config"] = "config",
                                             partial_fit_initialized_transformers: bool | Literal["config"] = "config",
                                             train_workers: int | Literal["config"] = "config",
                                             val_workers: int | Literal["config"] = "config",
                                             test_workers: int | Literal["config"] = "config",
                                             init_workers: int | Literal["config"] = "config",
                                             workers: int | Literal["config"] = "config",
                                             display_config_details: Optional[Literal["text", "diagram"]] = None):
        """Used for updating selected configurations set in config.
        Set parameter to `config` to keep it as it is config.
        If exception is thrown during set, no changes are made.

        Can affect following configuration:

        Dataset config | Description
        -------------- | -----------
        `default_values` | Default values for missing data, applied before fillers. Can set one value for all features or specify for each feature.
        `sliding_window_size` | Number of times in one window. Impacts dataloader behavior. Refer to relevant config for details.
        `sliding_window_prediction_size` | Number of times to predict from sliding_window_size. Refer to relevant config for details.
        `sliding_window_step` | Number of times to move by after each window. Refer to relevant config for details.
        `set_shared_size` | How much times should time periods share. Order of sharing is training set < validation set < test set. Refer to relevant config for details.
        `train_batch_size` | Number of samples per batch for train set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
        `val_batch_size` | Number of samples per batch for val set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
        `test_batch_size` | Number of samples per batch for test set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
        `preprocess_order` | Used order of when preprocesses are applied. Can be also used to add/remove custom handlers.
        `fill_missing_with` | Defines how to fill missing values in the dataset.
        `transform_with` | Defines the transformer to transform the dataset.
        `handle_anomalies_with` | Defines the anomaly handler to handle anomalies in the train set.
        `partial_fit_initialized_transformers` | If `True`, partial fitting on train set is performed when using initialized transformers.
        `train_workers` | Number of workers for loading training data.
        `val_workers` | Number of workers for loading validation data.
        `test_workers` | Number of workers for loading test data.
        `init_workers` | Number of workers for dataset configuration.


        Parameters:
            default_values: Default values for missing data, applied before fillers. `Defaults: config`.  
            sliding_window_size: Number of times in one window. `Defaults: config`.
            sliding_window_prediction_size: Number of times to predict from sliding_window_size. `Defaults: config`.
            sliding_window_step: Number of times to move by after each window. `Defaults: config`.
            set_shared_size: How much times should time periods share. `Defaults: config`.            
            train_batch_size: Number of samples per batch for train set. `Defaults: config`.
            val_batch_size: Number of samples per batch for val set. `Defaults: config`.
            test_batch_size: Number of samples per batch for test set. `Defaults: config`. 
            preprocess_order: Used order of when preprocesses are applied. Can be also used to add/remove custom handlers. `Defaults: config`.                  
            fill_missing_with: Defines how to fill missing values in the dataset. `Defaults: config`. 
            transform_with: Defines the transformer to transform the dataset. `Defaults: config`. 
            handle_anomalies_with: Defines the anomaly handler to handle anomalies in the train set. `Defaults: config`. 
            partial_fit_initialized_transformers: If `True`, partial fitting on train set is performed when using initiliazed transformers. `Defaults: config`.    
            train_workers: Number of workers for loading training data. `Defaults: config`.
            val_workers: Number of workers for loading validation data. `Defaults: config`.
            test_workers: Number of workers for loading test data. `Defaults: config`.
            init_workers: Number of workers for dataset configuration. `Defaults: config`.                          
            workers: How many workers to use when updating configuration. `Defaults: config`.  
            display_config_details: Whether config details should be displayed after configuration. `Defaults: False`. 
        """

        config_editor = DisjointTimeBasedConfigEditor(self._export_config_copy,
                                                      default_values,
                                                      train_batch_size,
                                                      val_batch_size,
                                                      test_batch_size,
                                                      preprocess_order,
                                                      fill_missing_with,
                                                      transform_with,
                                                      handle_anomalies_with,
                                                      "config",
                                                      partial_fit_initialized_transformers,
                                                      train_workers,
                                                      val_workers,
                                                      test_workers,
                                                      init_workers,
                                                      sliding_window_size,
                                                      sliding_window_prediction_size,
                                                      sliding_window_step,
                                                      set_shared_size
                                                      )

        self._update_dataset_config_and_initialize(config_editor, workers, display_config_details)

    def get_data_about_set(self, about: SplitType | Literal["train", "val", "test"]) -> dict:
        """
        Retrieve data related to the specified set.

        Parameters:
            about: Specifies the set to retrieve data about.

        Returned dictionary contains:

        - **ts_ids:** Ids of time series in `about` set.
        - **TimeFormat.ID_TIME:** Times in `about` set, where time format is `TimeFormat.ID_TIME`.
        - **TimeFormat.DATETIME:** Times in `about` set, where time format is `TimeFormat.DATETIME`.
        - **TimeFormat.UNIX_TIME:** Times in `about` set, where time format is `TimeFormat.UNIX_TIME`.
        - **TimeFormat.SHIFTED_UNIX_TIME:** Times in `about` set, where time format is `TimeFormat.SHIFTED_UNIX_TIME`.

        Returns:
            Returns dictionary with details about set.
        """
        if self.dataset_config is None or not self.dataset_config.is_initialized:
            raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before getting data about set.")

        about = SplitType(about)

        time_period = None
        time_series = None

        result = {}

        if about == SplitType.TRAIN:
            if not self.dataset_config.has_train():
                raise ValueError("Train split is not used.")
            time_period = self.dataset_config.train_time_period
            time_series = self.dataset_config.train_ts
        elif about == SplitType.VAL:
            if not self.dataset_config.has_val():
                raise ValueError("Val split is not used.")
            time_period = self.dataset_config.val_time_period
            time_series = self.dataset_config.val_ts
        elif about == SplitType.TEST:
            if not self.dataset_config.has_test():
                raise ValueError("Test split is not used.")
            time_period = self.dataset_config.test_time_period
            time_series = self.dataset_config.test_ts
        else:
            raise ValueError("Specified about parameter is not supported.")

        datetime_temp = np.array([datetime.fromtimestamp(time, timezone.utc) for time in self.metadata.time_indices[TIME_COLUMN_NAME][time_period[ID_TIME_COLUMN_NAME]]])

        result["ts_ids"] = time_series.copy()
        result[TimeFormat.ID_TIME] = time_period[ID_TIME_COLUMN_NAME].copy()
        result[TimeFormat.DATETIME] = datetime_temp.copy()
        result[TimeFormat.UNIX_TIME] = self.metadata.time_indices[TIME_COLUMN_NAME][time_period[ID_TIME_COLUMN_NAME]].copy()
        result[TimeFormat.SHIFTED_UNIX_TIME] = self.metadata.time_indices[TIME_COLUMN_NAME][time_period[ID_TIME_COLUMN_NAME]] - self.metadata.time_indices[TIME_COLUMN_NAME][0]

        return result

    def set_sliding_window(self, sliding_window_size: int | None | Literal["config"] = "config", sliding_window_prediction_size: int | None | Literal["config"] = "config",
                           sliding_window_step: int | None | Literal["config"] = "config", set_shared_size: float | int | Literal["config"] = "config", workers: int | Literal["config"] = "config") -> None:
        """Used for updating sliding window related values set in config.
        Set parameter to `config` to keep it as it is config.
        If exception is thrown during set, no changes are made.

        Affects following configuration: 

        Dataset config | Description
        -------------- | -----------
        `sliding_window_size` | Number of times in one window. Impacts dataloader behavior. Refer to relevant config for details.
        `sliding_window_prediction_size` | Number of times to predict from sliding_window_size. Refer to relevant config for details.
        `sliding_window_step` | Number of times to move by after each window. Refer to relevant config for details.
        `set_shared_size` | How much times should time periods share. Order of sharing is training set < validation set < test set. Refer to relevant config for details.

        Parameters:
            sliding_window_size: Number of times in one window. `Defaults: config`.
            sliding_window_prediction_size: Number of times to predict from sliding_window_size. `Defaults: config`.
            sliding_window_step: Number of times to move by after each window. `Defaults: config`.
            set_shared_size: How much times should time periods share. `Defaults: config`.
            workers: How many workers to use when setting new sliding window values. `Defaults: config`.  
        """

        if self.dataset_config is None or not self.dataset_config.is_initialized:
            raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before updating sliding window values.")

        self.update_dataset_config_and_initialize(sliding_window_size=sliding_window_size, sliding_window_prediction_size=sliding_window_prediction_size, sliding_window_step=sliding_window_step, set_shared_size=set_shared_size, workers=workers)
        self.logger.info("Sliding window values has been changed successfuly.")

    def set_batch_sizes(self, train_batch_size: int | Literal["config"] = "config", val_batch_size: int | Literal["config"] = "config", test_batch_size: int | Literal["config"] = "config") -> None:
        """Used for updating batch sizes set in config.
        Set parameter to `config` to keep it as it is config.
        If exception is thrown during set, no changes are made.

        Affects following configuration: 

        Dataset config | Description
        -------------- | -----------
        `train_batch_size` | Number of samples per batch for train set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
        `val_batch_size` | Number of samples per batch for val set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
        `test_batch_size` | Number of samples per batch for test set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.

        Parameters:
            train_batch_size: Number of samples per batch for train set. `Defaults: config`.
            val_batch_size: Number of samples per batch for val set. `Defaults: config`.
            test_batch_size: Number of samples per batch for test set. `Defaults: config`.
        """

        if self.dataset_config is None or not self.dataset_config.is_initialized:
            raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before updating batch sizes.")

        self.update_dataset_config_and_initialize(train_batch_size=train_batch_size, val_batch_size=val_batch_size, test_batch_size=test_batch_size, workers="config")
        self.logger.info("Batch sizes has been changed successfuly.")

    def set_workers(self, train_workers: int | Literal["config"] = "config", val_workers: int | Literal["config"] = "config",
                    test_workers: int | Literal["config"] = "config", init_workers: int | Literal["config"] = "config") -> None:
        """Used for updating workers set in config.
        Set parameter to `config` to keep it as it is config.
        If exception is thrown during set, no changes are made.

        Affects following configuration: 

        Dataset config | Description
        -------------- | -----------
        `train_workers` | Number of workers for loading training data.
        `val_workers` | Number of workers for loading validation data.
        `test_workers` | Number of workers for loading test data.
        `init_workers` | Number of workers for dataset configuration.

        Parameters:
            train_workers: Number of workers for loading training data. `Defaults: config`.
            val_workers: Number of workers for loading validation data. `Defaults: config`.
            test_workers: Number of workers for loading test data. `Defaults: config`.
            init_workers: Number of workers for dataset configuration. `Defaults: config`.            
        """

        if self.dataset_config is None or not self.dataset_config.is_initialized:
            raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before updating workers.")

        self.update_dataset_config_and_initialize(train_workers=train_workers, val_workers=val_workers, test_workers=test_workers, init_workers=init_workers, workers="config")
        self.logger.info("Workers has been changed successfuly.")

    def _initialize_datasets(self) -> None:
        """Called in [`set_dataset_config_and_initialize`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.set_dataset_config_and_initialize), this method initializes the set datasets (train, validation, test and all). """

        if self.dataset_config.has_train():
            load_config = DisjointTimeLoadConfig(self.dataset_config, SplitType.TRAIN)
            self.train_dataset = DisjointTimeBasedSplittedDataset(self.metadata.dataset_path, self.metadata.data_table_path, load_config, self.dataset_config.train_workers)

            self.logger.debug("train_dataset initiliazed.")

        if self.dataset_config.has_val():
            load_config = DisjointTimeLoadConfig(self.dataset_config, SplitType.VAL)
            self.val_dataset = DisjointTimeBasedSplittedDataset(self.metadata.dataset_path, self.metadata.data_table_path, load_config, self.dataset_config.val_workers)

            self.logger.debug("val_dataset initiliazed.")

        if self.dataset_config.has_test():
            load_config = DisjointTimeLoadConfig(self.dataset_config, SplitType.TEST)
            self.test_dataset = DisjointTimeBasedSplittedDataset(self.metadata.dataset_path, self.metadata.data_table_path, load_config, self.dataset_config.test_workers)
            self.logger.debug("test_dataset initiliazed.")

    def _initialize_transformers_and_details(self, workers: int) -> None:
        """
        Called in [`set_dataset_config_and_initialize`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.set_dataset_config_and_initialize). 

        Goes through data to validate time series against `nan_threshold`, fit/partial fit `transformers`, fit `anomaly handlers` and prepare `fillers`.
        """

        if self.dataset_config.has_train():

            self.__initialize_config_for_train_set(workers)

            self.logger.debug("Train set updated: %s time series left.", len(self.dataset_config.train_ts))

        if self.dataset_config.has_val():
            init_config = DisjointTimeDatasetInitConfig(self.dataset_config, SplitType.VAL, PreprocessOrderGroup([]))

            ts_ids_to_take = self.__initialize_config_for_non_fit_sets(init_config, workers, "val")
            self.dataset_config.val_ts = self.dataset_config.val_ts[ts_ids_to_take]
            self.dataset_config.val_ts_row_ranges = self.dataset_config.val_ts_row_ranges[ts_ids_to_take]
            self.dataset_config._update_preprocess_order_supported_ids(self.dataset_config.val_preprocess_order, ts_ids_to_take)

            self.logger.debug("Val set updated: %s time series left.", len(self.dataset_config.val_ts))

        if self.dataset_config.has_test():
            init_config = DisjointTimeDatasetInitConfig(self.dataset_config, SplitType.TEST, PreprocessOrderGroup([]))

            ts_ids_to_take = self.__initialize_config_for_non_fit_sets(init_config, workers, "test")
            self.dataset_config.test_ts = self.dataset_config.test_ts[ts_ids_to_take]
            self.dataset_config.test_ts_row_ranges = self.dataset_config.test_ts_row_ranges[ts_ids_to_take]
            self.dataset_config._update_preprocess_order_supported_ids(self.dataset_config.test_preprocess_order, ts_ids_to_take)

            self.logger.debug("Test set updated: %s time series left.", len(self.dataset_config.test_ts))

        self.logger.info("Dataset initialization complete. Configuration updated.")

    def __initialize_config_for_train_set(self, workers: int) -> None:
        """Initializes config for provided time series. """

        self.logger.info("Updating config for train set and fitting values.")

        is_first_cycle = True

        ts_ids_to_take = []

        groups = self.dataset_config._get_train_preprocess_init_order_groups()
        for i, group in enumerate(groups):
            ts_ids_to_take = []
            self.logger.info("Starting fitting cycle %s/%s.", i + 1, len(groups))

            init_config = DisjointTimeDatasetInitConfig(self.dataset_config, SplitType.TRAIN, group)
            init_dataset = DisjointTimeBasedInitializerDataset(self.metadata.dataset_path, self.metadata.data_table_path, init_config)

            sampler = SequentialSampler(init_dataset)
            dataloader = DataLoader(init_dataset, num_workers=workers, collate_fn=self._collate_fn, worker_init_fn=DisjointTimeBasedInitializerDataset.worker_init_fn, persistent_workers=False, sampler=sampler)

            if workers == 0:
                init_dataset.pytables_worker_init()

            for ts_id, data in enumerate(tqdm(dataloader, total=len(init_config.ts_row_ranges))):

                init_dataset_return: InitDatasetReturn = data[0]

                if init_dataset_return.is_under_nan_threshold:
                    ts_ids_to_take.append(ts_id)

                    # updates inner preprocessors passed from InitDataset
                    fitted_inner_index = 0
                    for inner_preprocess_order in group.preprocess_inner_orders:
                        if inner_preprocess_order.should_be_fitted:
                            inner_preprocess_order.holder.update_instance(init_dataset_return.preprocess_fitted_instances[fitted_inner_index].instance, ts_id)
                            fitted_inner_index += 1

                    # updates outer preprocessors based on passed train data from InitDataset
                    for outer_preprocess_order in group.preprocess_outer_orders:
                        if outer_preprocess_order.should_be_fitted:
                            outer_preprocess_order.holder.fit(init_dataset_return.train_data, ts_id)

                        if outer_preprocess_order.can_be_applied:
                            init_dataset_return.train_data = outer_preprocess_order.holder.apply(init_dataset_return.train_data, ts_id)

            if workers == 0:
                init_dataset.cleanup()

            # Update config based on filtered time series
            if is_first_cycle:

                if len(ts_ids_to_take) == 0:
                    raise ValueError("No valid time series left in train set after applying nan_threshold.")

                self.dataset_config.train_ts_row_ranges = self.dataset_config.train_ts_row_ranges[ts_ids_to_take]
                self.dataset_config.train_ts = self.dataset_config.train_ts[ts_ids_to_take]
                self.dataset_config._update_preprocess_order_supported_ids(self.dataset_config.train_preprocess_order, ts_ids_to_take)
                self.logger.debug("invalid ts_ids removed: %s time series left.", len(ts_ids_to_take))

                is_first_cycle = False

    def __initialize_config_for_non_fit_sets(self, init_config: DisjointTimeDatasetInitConfig, workers: int, set_name: str) -> np.ndarray:
        """Initializes config for provided time series without fitting. """
        init_dataset = DisjointTimeBasedInitializerDataset(self.metadata.dataset_path, self.metadata.data_table_path, init_config)

        sampler = SequentialSampler(init_dataset)
        dataloader = DataLoader(init_dataset, num_workers=workers, collate_fn=self._collate_fn, worker_init_fn=DisjointTimeBasedInitializerDataset.worker_init_fn, persistent_workers=False, sampler=sampler)

        if workers == 0:
            init_dataset.pytables_worker_init()

        ts_ids_to_take = []

        self.logger.info("Updating config for %s set.", set_name)
        for i, data in enumerate(tqdm(dataloader, total=len(init_config.ts_row_ranges))):
            init_dataset_return: InitDatasetReturn = data[0]

            if init_dataset_return.is_under_nan_threshold:
                ts_ids_to_take.append(i)

        if workers == 0:
            init_dataset.cleanup()

        if len(ts_ids_to_take) == 0:
            raise ValueError(f"No valid time series left in {set_name} set after applying nan_threshold.")

        return ts_ids_to_take

    def _update_export_config_copy(self) -> None:
        """
        Called at the end of [`set_dataset_config_and_initialize`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.set_dataset_config_and_initialize) or when changing config values. 

        Updates values of config used for saving config.
        """
        self._export_config_copy.database_name = self.metadata.database_name

        self._export_config_copy.train_ts = self.dataset_config.train_ts.copy() if self.dataset_config.has_train() else None
        self._export_config_copy.val_ts = self.dataset_config.val_ts.copy() if self.dataset_config.has_val() else None
        self._export_config_copy.test_ts = self.dataset_config.test_ts.copy() if self.dataset_config.has_test() else None

        self._export_config_copy.sliding_window_size = self.dataset_config.sliding_window_size
        self._export_config_copy.sliding_window_prediction_size = self.dataset_config.sliding_window_prediction_size
        self._export_config_copy.sliding_window_step = self.dataset_config.sliding_window_step
        self._export_config_copy.set_shared_size = self.dataset_config.set_shared_size

        super(DisjointTimeBasedCesnetDataset, self)._update_export_config_copy()

    def _get_singular_time_series_dataset(self, parent_dataset: DisjointTimeBasedSplittedDataset, ts_id: int) -> DisjointTimeBasedSplittedDataset:
        """Returns dataset for single time series """

        temp = np.where(np.isin(parent_dataset.load_config.ts_row_ranges[self.metadata.ts_id_name], [ts_id]))[0]

        if len(temp) == 0:
            raise ValueError(f"ts_id {ts_id} was not found in valid time series for this set. Available time series are: {parent_dataset.load_config.ts_row_ranges[self.metadata.ts_id_name]}")

        time_series_position = temp[0]

        split_load_config = parent_dataset.load_config.create_split_copy(slice(time_series_position, time_series_position + 1))

        dataset = DisjointTimeBasedSplittedDataset(self.metadata.dataset_path, self.metadata.data_table_path, split_load_config, 0)
        self.logger.debug("Singular time series dataset initiliazed.")

        return dataset

    def _get_data_for_plot(self, ts_id: int, feature_indices: np.ndarray[int], time_format: TimeFormat) -> tuple[np.ndarray, np.ndarray]:
        """Dataset type specific retrieval of data. """

        train_id_result, val_id_result, test_id_result = None, None, None

        if (self.dataset_config.has_train()):
            train_id_result = np.argwhere(np.isin(self.dataset_config.train_ts, ts_id)).ravel()
        if (self.dataset_config.has_val()):
            val_id_result = np.argwhere(np.isin(self.dataset_config.val_ts, ts_id)).ravel()
        if (self.dataset_config.has_test()):
            test_id_result = np.argwhere(np.isin(self.dataset_config.test_ts, ts_id)).ravel()

        data = None
        time_period = None

        if self.dataset_config.has_train() and len(train_id_result) > 0:
            data = self.__get_ts_data_for_plot(self.train_dataset, ts_id, feature_indices)
            time_period = self.get_data_about_set(SplitType.TRAIN)[time_format]
            self.logger.debug("Valid ts_id found: %d", train_id_result[0])

        elif self.dataset_config.has_val() and len(val_id_result) > 0:
            data = self.__get_ts_data_for_plot(self.val_dataset, ts_id, feature_indices)
            time_period = self.get_data_about_set(SplitType.VAL)[time_format]
            self.logger.debug("Valid ts_id found: %d", val_id_result[0])

        elif self.dataset_config.has_test() and len(test_id_result) > 0:
            data = self.__get_ts_data_for_plot(self.test_dataset, ts_id, feature_indices)
            time_period = self.get_data_about_set(SplitType.TEST)[time_format]
            self.logger.debug("Valid ts_id found: %d", test_id_result[0])
        else:
            raise ValueError(f"Invalid ts_id '{ts_id}'. The provided ts_id is not found in the available time series IDs.", self.dataset_config.train_ts, self.dataset_config.val_ts, self.dataset_config.test_ts)

        return data, time_period

    def __get_ts_data_for_plot(self, dataset: DisjointTimeBasedSplittedDataset, ts_id: int, feature_indices: list[int]):
        dataset = self._get_singular_time_series_dataset(dataset, ts_id)

        dataloader = self.dataloader_factory.create_dataloader(dataset, self.dataset_config, 0, True, None)

        temp_data = dataset_loaders.create_numpy_from_dataloader(dataloader, np.array([ts_id]), dataset.load_config.time_format, dataset.load_config.include_time, DatasetType.TIME_BASED, True)

        if (dataset.load_config.time_format == TimeFormat.DATETIME and dataset.load_config.include_time):
            temp_data = temp_data[0]

        temp_data = temp_data[0][:, feature_indices]

        return temp_data

dataset_config `class-attribute` `instance-attribute`

dataset_config: Optional[DisjointTimeBasedConfig] = field(default=None, init=False)

Configuration of the dataset.

train_dataset `class-attribute` `instance-attribute`

train_dataset: Optional[DisjointTimeBasedSplittedDataset] = field(default=None, init=False)

Training set as a SplittedDataset instance wrapping multiple TimeBasedDataset that wrap the PyTables database.

val_dataset `class-attribute` `instance-attribute`

val_dataset: Optional[DisjointTimeBasedSplittedDataset] = field(default=None, init=False)

Validation set as a SplittedDataset instance wrapping multiple TimeBasedDataset that wrap the PyTables database.

test_dataset `class-attribute` `instance-attribute`

test_dataset: Optional[DisjointTimeBasedSplittedDataset] = field(default=None, init=False)

Test set as a SplittedDataset instance wrapping multiple TimeBasedDataset that wrap the PyTables database.

train_dataloader `class-attribute` `instance-attribute`

train_dataloader: Optional[DisjointTimeBasedDataloader] = field(default=None, init=False)

Iterable PyTorch DataLoader for training set.

val_dataloader `class-attribute` `instance-attribute`

val_dataloader: Optional[DisjointTimeBasedDataloader] = field(default=None, init=False)

Iterable PyTorch DataLoader for validation set.

test_dataloader `class-attribute` `instance-attribute`

test_dataloader: Optional[DisjointTimeBasedDataloader] = field(default=None, init=False)

Iterable PyTorch DataLoader for test set.

dataloader_factory `class-attribute` `instance-attribute`

dataloader_factory: DisjointTimeBasedDataloaderFactory = field(default=DisjointTimeBasedDataloaderFactory(), init=False)

Factory used to create DisjointTimeBasedDataloader.

dataset_type `class-attribute` `instance-attribute`

dataset_type: DatasetType = field(default=DISJOINT_TIME_BASED, init=False)

_export_config_copy `class-attribute` `instance-attribute`

_export_config_copy: Optional[DisjointTimeBasedConfig] = field(default=None, init=False)

metadata `instance-attribute`

metadata: DatasetMetadata

Holds various metadata used in dataset for its creation, loading data and similar.

all_dataset `class-attribute` `instance-attribute`

all_dataset: Optional[Dataset] = field(default=None, init=False)

All set as a BaseDataset instance wrapping the PyTables database.

all_dataloader `class-attribute` `instance-attribute`

all_dataloader: Optional[DataLoader] = field(default=None, init=False)

Iterable PyTorch DataLoader for all set.

related_to `class-attribute` `instance-attribute`

related_to: Optional[str] = field(default=None, init=False)

Name of file with relevant results to used benchmark.

_collate_fn `class-attribute` `instance-attribute`

_collate_fn: Optional[Callable] = field(default=None, init=False)

init

__init__(metadata: DatasetMetadata) -> None

__post_init__

__post_init__()

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def __post_init__(self):
    super().__post_init__()

    self.logger.info("Dataset is disjoint_time_based. Use cesnet_tszoo.configs.DisjointTimeBasedConfig")

set_dataset_config_and_initialize

set_dataset_config_and_initialize(dataset_config: DisjointTimeBasedConfig, display_config_details: Optional[Literal['text', 'diagram']] = 'text', workers: int | Literal['config'] = 'config') -> None

Initialize training set, validation set, test set etc.. This method must be called before any data can be accessed. It is required for the final initialization of dataset_config.

The following configuration attributes are used during initialization:

Dataset config	Description
`init_workers`	Specifies the number of workers to use for initialization. Applied when `workers` = "config".
`partial_fit_initialized_transformers`	Determines whether initialized transformers should be partially fitted on the training data.
`nan_threshold`	Filters out time series with missing values exceeding the specified threshold.

Parameters:

Name	Type	Description	Default
`dataset_config`	`DisjointTimeBasedConfig`	Desired configuration of the dataset.	required
`display_config_details`	`Optional[Literal['text', 'diagram']]`	Flag indicating whether and how to display the configuration values after initialization. `Default: text`	`'text'`
`workers`	`int \| Literal['config']`	The number of workers to use during initialization. `Default: "config"`	`'config'`

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def set_dataset_config_and_initialize(self, dataset_config: DisjointTimeBasedConfig, display_config_details: Optional[Literal["text", "diagram"]] = "text", workers: int | Literal["config"] = "config") -> None:
    """
    Initialize training set, validation set, test set etc.. This method must be called before any data can be accessed. It is required for the final initialization of [`dataset_config`](reference_disjoint_time_based_config.md#references.DisjointTimeBasedConfig).

    The following configuration attributes are used during initialization:

    Dataset config | Description
    -------------- | -----------
    `init_workers` | Specifies the number of workers to use for initialization. Applied when `workers` = "config".
    `partial_fit_initialized_transformers` | Determines whether initialized transformers should be partially fitted on the training data.
    `nan_threshold` | Filters out time series with missing values exceeding the specified threshold.

    Parameters:
        dataset_config: Desired configuration of the dataset.
        display_config_details: Flag indicating whether and how to display the configuration values after initialization. `Default: text`  
        workers: The number of workers to use during initialization. `Default: "config"`  
    """

    assert dataset_config is not None, "Used dataset_config cannot be None."
    assert isinstance(dataset_config, DisjointTimeBasedConfig), f"This config is used for dataset of type '{dataset_config.dataset_type}'. Meanwhile this dataset is of type '{self.metadata.dataset_type}'."

    super(DisjointTimeBasedCesnetDataset, self).set_dataset_config_and_initialize(dataset_config, display_config_details, workers)

apply_transformer

apply_transformer(transform_with: type | list[Transformer] | ndarray[Transformer] | TransformerType | Transformer | Literal['min_max_scaler', 'standard_scaler', 'max_abs_scaler', 'log_transformer', 'l2_normalizer'] | None | Literal['config'] = 'config', partial_fit_initialized_transformers: bool | Literal['config'] = 'config', workers: int | Literal['config'] = 'config') -> None

Used for updating transformer and relevant configurations set in config. Set parameter to config to keep it as it is config. If exception is thrown during set, no changes are made.

Affects following configuration:

Dataset config	Description
`transform_with`	Defines the transformer to transform the dataset.
`partial_fit_initialized_transformers`	If `True`, partial fitting on train set is performed when using initialized transformers.

Parameters:

Name	Type	Description	Default
`transform_with`	`type \| list[Transformer] \| ndarray[Transformer] \| TransformerType \| Transformer \| Literal['min_max_scaler', 'standard_scaler', 'max_abs_scaler', 'log_transformer', 'l2_normalizer'] \| None \| Literal['config']`	Defines the transformer to transform the dataset. `Defaults: config`.	`'config'`
`partial_fit_initialized_transformers`	`bool \| Literal['config']`	If `True`, partial fitting on train set is performed when using initiliazed transformers. `Defaults: config`.	`'config'`
`workers`	`int \| Literal['config']`	How many workers to use when setting new transformer. `Defaults: config`.	`'config'`

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def apply_transformer(self, transform_with: type | list[Transformer] | np.ndarray[Transformer] | TransformerType | Transformer | Literal["min_max_scaler", "standard_scaler", "max_abs_scaler", "log_transformer", "l2_normalizer"] | None | Literal["config"] = "config",
                      partial_fit_initialized_transformers: bool | Literal["config"] = "config", workers: int | Literal["config"] = "config") -> None:
    """Used for updating transformer and relevant configurations set in config.
    Set parameter to `config` to keep it as it is config.
    If exception is thrown during set, no changes are made.

    Affects following configuration:

    Dataset config | Description
    -------------- | -----------
    `transform_with` | Defines the transformer to transform the dataset.
    `partial_fit_initialized_transformers` | If `True`, partial fitting on train set is performed when using initialized transformers.

    Parameters:
        transform_with: Defines the transformer to transform the dataset. `Defaults: config`.  
        partial_fit_initialized_transformers: If `True`, partial fitting on train set is performed when using initiliazed transformers. `Defaults: config`.  
        workers: How many workers to use when setting new transformer. `Defaults: config`.      
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before updating transformer values.")

    self.update_dataset_config_and_initialize(transform_with=transform_with, partial_fit_initialized_transformers=partial_fit_initialized_transformers, workers=workers)

update_dataset_config_and_initialize

update_dataset_config_and_initialize(default_values: list[Number] | NDArray[number] | dict[str, Number] | Number | Literal['default'] | None | Literal['config'] = 'config', sliding_window_size: int | None | Literal['config'] = 'config', sliding_window_prediction_size: int | None | Literal['config'] = 'config', sliding_window_step: int | Literal['config'] = 'config', set_shared_size: float | int | Literal['config'] = 'config', train_batch_size: int | Literal['config'] = 'config', val_batch_size: int | Literal['config'] = 'config', test_batch_size: int | Literal['config'] = 'config', preprocess_order: list[str, type] | Literal['config'] = 'config', fill_missing_with: type | FillerType | Literal['mean_filler', 'forward_filler', 'linear_interpolation_filler'] | None | Literal['config'] = 'config', transform_with: type | list[Transformer] | ndarray[Transformer] | TransformerType | Transformer | Literal['min_max_scaler', 'standard_scaler', 'max_abs_scaler', 'log_transformer', 'l2_normalizer'] | None | Literal['config'] = 'config', handle_anomalies_with: type | AnomalyHandlerType | Literal['z-score', 'interquartile_range'] | None | Literal['config'] = 'config', partial_fit_initialized_transformers: bool | Literal['config'] = 'config', train_workers: int | Literal['config'] = 'config', val_workers: int | Literal['config'] = 'config', test_workers: int | Literal['config'] = 'config', init_workers: int | Literal['config'] = 'config', workers: int | Literal['config'] = 'config', display_config_details: Optional[Literal['text', 'diagram']] = None)

Used for updating selected configurations set in config. Set parameter to config to keep it as it is config. If exception is thrown during set, no changes are made.

Can affect following configuration:

Dataset config	Description
`default_values`	Default values for missing data, applied before fillers. Can set one value for all features or specify for each feature.
`sliding_window_size`	Number of times in one window. Impacts dataloader behavior. Refer to relevant config for details.
`sliding_window_prediction_size`	Number of times to predict from sliding_window_size. Refer to relevant config for details.
`sliding_window_step`	Number of times to move by after each window. Refer to relevant config for details.
`set_shared_size`	How much times should time periods share. Order of sharing is training set < validation set < test set. Refer to relevant config for details.
`train_batch_size`	Number of samples per batch for train set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
`val_batch_size`	Number of samples per batch for val set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
`test_batch_size`	Number of samples per batch for test set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
`preprocess_order`	Used order of when preprocesses are applied. Can be also used to add/remove custom handlers.
`fill_missing_with`	Defines how to fill missing values in the dataset.
`transform_with`	Defines the transformer to transform the dataset.
`handle_anomalies_with`	Defines the anomaly handler to handle anomalies in the train set.
`partial_fit_initialized_transformers`	If `True`, partial fitting on train set is performed when using initialized transformers.
`train_workers`	Number of workers for loading training data.
`val_workers`	Number of workers for loading validation data.
`test_workers`	Number of workers for loading test data.
`init_workers`	Number of workers for dataset configuration.

Parameters:

Name	Type	Description	Default
`default_values`	`list[Number] \| NDArray[number] \| dict[str, Number] \| Number \| Literal['default'] \| None \| Literal['config']`	Default values for missing data, applied before fillers. `Defaults: config`.	`'config'`
`sliding_window_size`	`int \| None \| Literal['config']`	Number of times in one window. `Defaults: config`.	`'config'`
`sliding_window_prediction_size`	`int \| None \| Literal['config']`	Number of times to predict from sliding_window_size. `Defaults: config`.	`'config'`
`sliding_window_step`	`int \| Literal['config']`	Number of times to move by after each window. `Defaults: config`.	`'config'`
`set_shared_size`	`float \| int \| Literal['config']`	How much times should time periods share. `Defaults: config`.	`'config'`
`train_batch_size`	`int \| Literal['config']`	Number of samples per batch for train set. `Defaults: config`.	`'config'`
`val_batch_size`	`int \| Literal['config']`	Number of samples per batch for val set. `Defaults: config`.	`'config'`
`test_batch_size`	`int \| Literal['config']`	Number of samples per batch for test set. `Defaults: config`.	`'config'`
`preprocess_order`	`list[str, type] \| Literal['config']`	Used order of when preprocesses are applied. Can be also used to add/remove custom handlers. `Defaults: config`.	`'config'`
`fill_missing_with`	`type \| FillerType \| Literal['mean_filler', 'forward_filler', 'linear_interpolation_filler'] \| None \| Literal['config']`	Defines how to fill missing values in the dataset. `Defaults: config`.	`'config'`
`transform_with`	`type \| list[Transformer] \| ndarray[Transformer] \| TransformerType \| Transformer \| Literal['min_max_scaler', 'standard_scaler', 'max_abs_scaler', 'log_transformer', 'l2_normalizer'] \| None \| Literal['config']`	Defines the transformer to transform the dataset. `Defaults: config`.	`'config'`
`handle_anomalies_with`	`type \| AnomalyHandlerType \| Literal['z-score', 'interquartile_range'] \| None \| Literal['config']`	Defines the anomaly handler to handle anomalies in the train set. `Defaults: config`.	`'config'`
`partial_fit_initialized_transformers`	`bool \| Literal['config']`	If `True`, partial fitting on train set is performed when using initiliazed transformers. `Defaults: config`.	`'config'`
`train_workers`	`int \| Literal['config']`	Number of workers for loading training data. `Defaults: config`.	`'config'`
`val_workers`	`int \| Literal['config']`	Number of workers for loading validation data. `Defaults: config`.	`'config'`
`test_workers`	`int \| Literal['config']`	Number of workers for loading test data. `Defaults: config`.	`'config'`
`init_workers`	`int \| Literal['config']`	Number of workers for dataset configuration. `Defaults: config`.	`'config'`
`workers`	`int \| Literal['config']`	How many workers to use when updating configuration. `Defaults: config`.	`'config'`
`display_config_details`	`Optional[Literal['text', 'diagram']]`	Whether config details should be displayed after configuration. `Defaults: False`.	`None`

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def update_dataset_config_and_initialize(self,
                                         default_values: list[Number] | npt.NDArray[np.number] | dict[str, Number] | Number | Literal["default"] | None | Literal["config"] = "config",
                                         sliding_window_size: int | None | Literal["config"] = "config",
                                         sliding_window_prediction_size: int | None | Literal["config"] = "config",
                                         sliding_window_step: int | Literal["config"] = "config",
                                         set_shared_size: float | int | Literal["config"] = "config",
                                         train_batch_size: int | Literal["config"] = "config",
                                         val_batch_size: int | Literal["config"] = "config",
                                         test_batch_size: int | Literal["config"] = "config",
                                         preprocess_order: list[str, type] | Literal["config"] = "config",
                                         fill_missing_with: type | FillerType | Literal["mean_filler", "forward_filler", "linear_interpolation_filler"] | None | Literal["config"] = "config",
                                         transform_with: type | list[Transformer] | np.ndarray[Transformer] | TransformerType | Transformer | Literal["min_max_scaler", "standard_scaler", "max_abs_scaler", "log_transformer", "l2_normalizer"] | None | Literal["config"] = "config",
                                         handle_anomalies_with: type | AnomalyHandlerType | Literal["z-score", "interquartile_range"] | None | Literal["config"] = "config",
                                         partial_fit_initialized_transformers: bool | Literal["config"] = "config",
                                         train_workers: int | Literal["config"] = "config",
                                         val_workers: int | Literal["config"] = "config",
                                         test_workers: int | Literal["config"] = "config",
                                         init_workers: int | Literal["config"] = "config",
                                         workers: int | Literal["config"] = "config",
                                         display_config_details: Optional[Literal["text", "diagram"]] = None):
    """Used for updating selected configurations set in config.
    Set parameter to `config` to keep it as it is config.
    If exception is thrown during set, no changes are made.

    Can affect following configuration:

    Dataset config | Description
    -------------- | -----------
    `default_values` | Default values for missing data, applied before fillers. Can set one value for all features or specify for each feature.
    `sliding_window_size` | Number of times in one window. Impacts dataloader behavior. Refer to relevant config for details.
    `sliding_window_prediction_size` | Number of times to predict from sliding_window_size. Refer to relevant config for details.
    `sliding_window_step` | Number of times to move by after each window. Refer to relevant config for details.
    `set_shared_size` | How much times should time periods share. Order of sharing is training set < validation set < test set. Refer to relevant config for details.
    `train_batch_size` | Number of samples per batch for train set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
    `val_batch_size` | Number of samples per batch for val set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
    `test_batch_size` | Number of samples per batch for test set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
    `preprocess_order` | Used order of when preprocesses are applied. Can be also used to add/remove custom handlers.
    `fill_missing_with` | Defines how to fill missing values in the dataset.
    `transform_with` | Defines the transformer to transform the dataset.
    `handle_anomalies_with` | Defines the anomaly handler to handle anomalies in the train set.
    `partial_fit_initialized_transformers` | If `True`, partial fitting on train set is performed when using initialized transformers.
    `train_workers` | Number of workers for loading training data.
    `val_workers` | Number of workers for loading validation data.
    `test_workers` | Number of workers for loading test data.
    `init_workers` | Number of workers for dataset configuration.


    Parameters:
        default_values: Default values for missing data, applied before fillers. `Defaults: config`.  
        sliding_window_size: Number of times in one window. `Defaults: config`.
        sliding_window_prediction_size: Number of times to predict from sliding_window_size. `Defaults: config`.
        sliding_window_step: Number of times to move by after each window. `Defaults: config`.
        set_shared_size: How much times should time periods share. `Defaults: config`.            
        train_batch_size: Number of samples per batch for train set. `Defaults: config`.
        val_batch_size: Number of samples per batch for val set. `Defaults: config`.
        test_batch_size: Number of samples per batch for test set. `Defaults: config`. 
        preprocess_order: Used order of when preprocesses are applied. Can be also used to add/remove custom handlers. `Defaults: config`.                  
        fill_missing_with: Defines how to fill missing values in the dataset. `Defaults: config`. 
        transform_with: Defines the transformer to transform the dataset. `Defaults: config`. 
        handle_anomalies_with: Defines the anomaly handler to handle anomalies in the train set. `Defaults: config`. 
        partial_fit_initialized_transformers: If `True`, partial fitting on train set is performed when using initiliazed transformers. `Defaults: config`.    
        train_workers: Number of workers for loading training data. `Defaults: config`.
        val_workers: Number of workers for loading validation data. `Defaults: config`.
        test_workers: Number of workers for loading test data. `Defaults: config`.
        init_workers: Number of workers for dataset configuration. `Defaults: config`.                          
        workers: How many workers to use when updating configuration. `Defaults: config`.  
        display_config_details: Whether config details should be displayed after configuration. `Defaults: False`. 
    """

    config_editor = DisjointTimeBasedConfigEditor(self._export_config_copy,
                                                  default_values,
                                                  train_batch_size,
                                                  val_batch_size,
                                                  test_batch_size,
                                                  preprocess_order,
                                                  fill_missing_with,
                                                  transform_with,
                                                  handle_anomalies_with,
                                                  "config",
                                                  partial_fit_initialized_transformers,
                                                  train_workers,
                                                  val_workers,
                                                  test_workers,
                                                  init_workers,
                                                  sliding_window_size,
                                                  sliding_window_prediction_size,
                                                  sliding_window_step,
                                                  set_shared_size
                                                  )

    self._update_dataset_config_and_initialize(config_editor, workers, display_config_details)

get_data_about_set

get_data_about_set(about: SplitType | Literal['train', 'val', 'test']) -> dict

Retrieve data related to the specified set.

Parameters:

Name	Type	Description	Default
`about`	`SplitType \| Literal['train', 'val', 'test']`	Specifies the set to retrieve data about.	required

Returned dictionary contains:

ts_ids: Ids of time series in about set.
TimeFormat.ID_TIME: Times in about set, where time format is TimeFormat.ID_TIME.
TimeFormat.DATETIME: Times in about set, where time format is TimeFormat.DATETIME.
TimeFormat.UNIX_TIME: Times in about set, where time format is TimeFormat.UNIX_TIME.
TimeFormat.SHIFTED_UNIX_TIME: Times in about set, where time format is TimeFormat.SHIFTED_UNIX_TIME.

Returns:

Type	Description
`dict`	Returns dictionary with details about set.

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def get_data_about_set(self, about: SplitType | Literal["train", "val", "test"]) -> dict:
    """
    Retrieve data related to the specified set.

    Parameters:
        about: Specifies the set to retrieve data about.

    Returned dictionary contains:

    - **ts_ids:** Ids of time series in `about` set.
    - **TimeFormat.ID_TIME:** Times in `about` set, where time format is `TimeFormat.ID_TIME`.
    - **TimeFormat.DATETIME:** Times in `about` set, where time format is `TimeFormat.DATETIME`.
    - **TimeFormat.UNIX_TIME:** Times in `about` set, where time format is `TimeFormat.UNIX_TIME`.
    - **TimeFormat.SHIFTED_UNIX_TIME:** Times in `about` set, where time format is `TimeFormat.SHIFTED_UNIX_TIME`.

    Returns:
        Returns dictionary with details about set.
    """
    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before getting data about set.")

    about = SplitType(about)

    time_period = None
    time_series = None

    result = {}

    if about == SplitType.TRAIN:
        if not self.dataset_config.has_train():
            raise ValueError("Train split is not used.")
        time_period = self.dataset_config.train_time_period
        time_series = self.dataset_config.train_ts
    elif about == SplitType.VAL:
        if not self.dataset_config.has_val():
            raise ValueError("Val split is not used.")
        time_period = self.dataset_config.val_time_period
        time_series = self.dataset_config.val_ts
    elif about == SplitType.TEST:
        if not self.dataset_config.has_test():
            raise ValueError("Test split is not used.")
        time_period = self.dataset_config.test_time_period
        time_series = self.dataset_config.test_ts
    else:
        raise ValueError("Specified about parameter is not supported.")

    datetime_temp = np.array([datetime.fromtimestamp(time, timezone.utc) for time in self.metadata.time_indices[TIME_COLUMN_NAME][time_period[ID_TIME_COLUMN_NAME]]])

    result["ts_ids"] = time_series.copy()
    result[TimeFormat.ID_TIME] = time_period[ID_TIME_COLUMN_NAME].copy()
    result[TimeFormat.DATETIME] = datetime_temp.copy()
    result[TimeFormat.UNIX_TIME] = self.metadata.time_indices[TIME_COLUMN_NAME][time_period[ID_TIME_COLUMN_NAME]].copy()
    result[TimeFormat.SHIFTED_UNIX_TIME] = self.metadata.time_indices[TIME_COLUMN_NAME][time_period[ID_TIME_COLUMN_NAME]] - self.metadata.time_indices[TIME_COLUMN_NAME][0]

    return result

set_sliding_window

set_sliding_window(sliding_window_size: int | None | Literal['config'] = 'config', sliding_window_prediction_size: int | None | Literal['config'] = 'config', sliding_window_step: int | None | Literal['config'] = 'config', set_shared_size: float | int | Literal['config'] = 'config', workers: int | Literal['config'] = 'config') -> None

Used for updating sliding window related values set in config. Set parameter to config to keep it as it is config. If exception is thrown during set, no changes are made.

Affects following configuration:

Dataset config	Description
`sliding_window_size`	Number of times in one window. Impacts dataloader behavior. Refer to relevant config for details.
`sliding_window_prediction_size`	Number of times to predict from sliding_window_size. Refer to relevant config for details.
`sliding_window_step`	Number of times to move by after each window. Refer to relevant config for details.
`set_shared_size`	How much times should time periods share. Order of sharing is training set < validation set < test set. Refer to relevant config for details.

Parameters:

Name	Type	Description	Default
`sliding_window_size`	`int \| None \| Literal['config']`	Number of times in one window. `Defaults: config`.	`'config'`
`sliding_window_prediction_size`	`int \| None \| Literal['config']`	Number of times to predict from sliding_window_size. `Defaults: config`.	`'config'`
`sliding_window_step`	`int \| None \| Literal['config']`	Number of times to move by after each window. `Defaults: config`.	`'config'`
`set_shared_size`	`float \| int \| Literal['config']`	How much times should time periods share. `Defaults: config`.	`'config'`
`workers`	`int \| Literal['config']`	How many workers to use when setting new sliding window values. `Defaults: config`.	`'config'`

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def set_sliding_window(self, sliding_window_size: int | None | Literal["config"] = "config", sliding_window_prediction_size: int | None | Literal["config"] = "config",
                       sliding_window_step: int | None | Literal["config"] = "config", set_shared_size: float | int | Literal["config"] = "config", workers: int | Literal["config"] = "config") -> None:
    """Used for updating sliding window related values set in config.
    Set parameter to `config` to keep it as it is config.
    If exception is thrown during set, no changes are made.

    Affects following configuration: 

    Dataset config | Description
    -------------- | -----------
    `sliding_window_size` | Number of times in one window. Impacts dataloader behavior. Refer to relevant config for details.
    `sliding_window_prediction_size` | Number of times to predict from sliding_window_size. Refer to relevant config for details.
    `sliding_window_step` | Number of times to move by after each window. Refer to relevant config for details.
    `set_shared_size` | How much times should time periods share. Order of sharing is training set < validation set < test set. Refer to relevant config for details.

    Parameters:
        sliding_window_size: Number of times in one window. `Defaults: config`.
        sliding_window_prediction_size: Number of times to predict from sliding_window_size. `Defaults: config`.
        sliding_window_step: Number of times to move by after each window. `Defaults: config`.
        set_shared_size: How much times should time periods share. `Defaults: config`.
        workers: How many workers to use when setting new sliding window values. `Defaults: config`.  
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before updating sliding window values.")

    self.update_dataset_config_and_initialize(sliding_window_size=sliding_window_size, sliding_window_prediction_size=sliding_window_prediction_size, sliding_window_step=sliding_window_step, set_shared_size=set_shared_size, workers=workers)
    self.logger.info("Sliding window values has been changed successfuly.")

set_batch_sizes

set_batch_sizes(train_batch_size: int | Literal['config'] = 'config', val_batch_size: int | Literal['config'] = 'config', test_batch_size: int | Literal['config'] = 'config') -> None

Used for updating batch sizes set in config. Set parameter to config to keep it as it is config. If exception is thrown during set, no changes are made.

Affects following configuration:

Dataset config	Description
`train_batch_size`	Number of samples per batch for train set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
`val_batch_size`	Number of samples per batch for val set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
`test_batch_size`	Number of samples per batch for test set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.

Parameters:

Name	Type	Description	Default
`train_batch_size`	`int \| Literal['config']`	Number of samples per batch for train set. `Defaults: config`.	`'config'`
`val_batch_size`	`int \| Literal['config']`	Number of samples per batch for val set. `Defaults: config`.	`'config'`
`test_batch_size`	`int \| Literal['config']`	Number of samples per batch for test set. `Defaults: config`.	`'config'`

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def set_batch_sizes(self, train_batch_size: int | Literal["config"] = "config", val_batch_size: int | Literal["config"] = "config", test_batch_size: int | Literal["config"] = "config") -> None:
    """Used for updating batch sizes set in config.
    Set parameter to `config` to keep it as it is config.
    If exception is thrown during set, no changes are made.

    Affects following configuration: 

    Dataset config | Description
    -------------- | -----------
    `train_batch_size` | Number of samples per batch for train set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
    `val_batch_size` | Number of samples per batch for val set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
    `test_batch_size` | Number of samples per batch for test set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.

    Parameters:
        train_batch_size: Number of samples per batch for train set. `Defaults: config`.
        val_batch_size: Number of samples per batch for val set. `Defaults: config`.
        test_batch_size: Number of samples per batch for test set. `Defaults: config`.
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before updating batch sizes.")

    self.update_dataset_config_and_initialize(train_batch_size=train_batch_size, val_batch_size=val_batch_size, test_batch_size=test_batch_size, workers="config")
    self.logger.info("Batch sizes has been changed successfuly.")

set_workers

set_workers(train_workers: int | Literal['config'] = 'config', val_workers: int | Literal['config'] = 'config', test_workers: int | Literal['config'] = 'config', init_workers: int | Literal['config'] = 'config') -> None

Used for updating workers set in config. Set parameter to config to keep it as it is config. If exception is thrown during set, no changes are made.

Affects following configuration:

Dataset config	Description
`train_workers`	Number of workers for loading training data.
`val_workers`	Number of workers for loading validation data.
`test_workers`	Number of workers for loading test data.
`init_workers`	Number of workers for dataset configuration.

Parameters:

Name	Type	Description	Default
`train_workers`	`int \| Literal['config']`	Number of workers for loading training data. `Defaults: config`.	`'config'`
`val_workers`	`int \| Literal['config']`	Number of workers for loading validation data. `Defaults: config`.	`'config'`
`test_workers`	`int \| Literal['config']`	Number of workers for loading test data. `Defaults: config`.	`'config'`
`init_workers`	`int \| Literal['config']`	Number of workers for dataset configuration. `Defaults: config`.	`'config'`

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def set_workers(self, train_workers: int | Literal["config"] = "config", val_workers: int | Literal["config"] = "config",
                test_workers: int | Literal["config"] = "config", init_workers: int | Literal["config"] = "config") -> None:
    """Used for updating workers set in config.
    Set parameter to `config` to keep it as it is config.
    If exception is thrown during set, no changes are made.

    Affects following configuration: 

    Dataset config | Description
    -------------- | -----------
    `train_workers` | Number of workers for loading training data.
    `val_workers` | Number of workers for loading validation data.
    `test_workers` | Number of workers for loading test data.
    `init_workers` | Number of workers for dataset configuration.

    Parameters:
        train_workers: Number of workers for loading training data. `Defaults: config`.
        val_workers: Number of workers for loading validation data. `Defaults: config`.
        test_workers: Number of workers for loading test data. `Defaults: config`.
        init_workers: Number of workers for dataset configuration. `Defaults: config`.            
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before updating workers.")

    self.update_dataset_config_and_initialize(train_workers=train_workers, val_workers=val_workers, test_workers=test_workers, init_workers=init_workers, workers="config")
    self.logger.info("Workers has been changed successfuly.")

_initialize_datasets

_initialize_datasets() -> None

Called in set_dataset_config_and_initialize, this method initializes the set datasets (train, validation, test and all).

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def _initialize_datasets(self) -> None:
    """Called in [`set_dataset_config_and_initialize`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.set_dataset_config_and_initialize), this method initializes the set datasets (train, validation, test and all). """

    if self.dataset_config.has_train():
        load_config = DisjointTimeLoadConfig(self.dataset_config, SplitType.TRAIN)
        self.train_dataset = DisjointTimeBasedSplittedDataset(self.metadata.dataset_path, self.metadata.data_table_path, load_config, self.dataset_config.train_workers)

        self.logger.debug("train_dataset initiliazed.")

    if self.dataset_config.has_val():
        load_config = DisjointTimeLoadConfig(self.dataset_config, SplitType.VAL)
        self.val_dataset = DisjointTimeBasedSplittedDataset(self.metadata.dataset_path, self.metadata.data_table_path, load_config, self.dataset_config.val_workers)

        self.logger.debug("val_dataset initiliazed.")

    if self.dataset_config.has_test():
        load_config = DisjointTimeLoadConfig(self.dataset_config, SplitType.TEST)
        self.test_dataset = DisjointTimeBasedSplittedDataset(self.metadata.dataset_path, self.metadata.data_table_path, load_config, self.dataset_config.test_workers)
        self.logger.debug("test_dataset initiliazed.")

_initialize_transformers_and_details

_initialize_transformers_and_details(workers: int) -> None

Called in set_dataset_config_and_initialize.

Goes through data to validate time series against nan_threshold, fit/partial fit transformers, fit anomaly handlers and prepare fillers.

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def _initialize_transformers_and_details(self, workers: int) -> None:
    """
    Called in [`set_dataset_config_and_initialize`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.set_dataset_config_and_initialize). 

    Goes through data to validate time series against `nan_threshold`, fit/partial fit `transformers`, fit `anomaly handlers` and prepare `fillers`.
    """

    if self.dataset_config.has_train():

        self.__initialize_config_for_train_set(workers)

        self.logger.debug("Train set updated: %s time series left.", len(self.dataset_config.train_ts))

    if self.dataset_config.has_val():
        init_config = DisjointTimeDatasetInitConfig(self.dataset_config, SplitType.VAL, PreprocessOrderGroup([]))

        ts_ids_to_take = self.__initialize_config_for_non_fit_sets(init_config, workers, "val")
        self.dataset_config.val_ts = self.dataset_config.val_ts[ts_ids_to_take]
        self.dataset_config.val_ts_row_ranges = self.dataset_config.val_ts_row_ranges[ts_ids_to_take]
        self.dataset_config._update_preprocess_order_supported_ids(self.dataset_config.val_preprocess_order, ts_ids_to_take)

        self.logger.debug("Val set updated: %s time series left.", len(self.dataset_config.val_ts))

    if self.dataset_config.has_test():
        init_config = DisjointTimeDatasetInitConfig(self.dataset_config, SplitType.TEST, PreprocessOrderGroup([]))

        ts_ids_to_take = self.__initialize_config_for_non_fit_sets(init_config, workers, "test")
        self.dataset_config.test_ts = self.dataset_config.test_ts[ts_ids_to_take]
        self.dataset_config.test_ts_row_ranges = self.dataset_config.test_ts_row_ranges[ts_ids_to_take]
        self.dataset_config._update_preprocess_order_supported_ids(self.dataset_config.test_preprocess_order, ts_ids_to_take)

        self.logger.debug("Test set updated: %s time series left.", len(self.dataset_config.test_ts))

    self.logger.info("Dataset initialization complete. Configuration updated.")

__initialize_config_for_train_set

__initialize_config_for_train_set(workers: int) -> None

Initializes config for provided time series.

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def __initialize_config_for_train_set(self, workers: int) -> None:
    """Initializes config for provided time series. """

    self.logger.info("Updating config for train set and fitting values.")

    is_first_cycle = True

    ts_ids_to_take = []

    groups = self.dataset_config._get_train_preprocess_init_order_groups()
    for i, group in enumerate(groups):
        ts_ids_to_take = []
        self.logger.info("Starting fitting cycle %s/%s.", i + 1, len(groups))

        init_config = DisjointTimeDatasetInitConfig(self.dataset_config, SplitType.TRAIN, group)
        init_dataset = DisjointTimeBasedInitializerDataset(self.metadata.dataset_path, self.metadata.data_table_path, init_config)

        sampler = SequentialSampler(init_dataset)
        dataloader = DataLoader(init_dataset, num_workers=workers, collate_fn=self._collate_fn, worker_init_fn=DisjointTimeBasedInitializerDataset.worker_init_fn, persistent_workers=False, sampler=sampler)

        if workers == 0:
            init_dataset.pytables_worker_init()

        for ts_id, data in enumerate(tqdm(dataloader, total=len(init_config.ts_row_ranges))):

            init_dataset_return: InitDatasetReturn = data[0]

            if init_dataset_return.is_under_nan_threshold:
                ts_ids_to_take.append(ts_id)

                # updates inner preprocessors passed from InitDataset
                fitted_inner_index = 0
                for inner_preprocess_order in group.preprocess_inner_orders:
                    if inner_preprocess_order.should_be_fitted:
                        inner_preprocess_order.holder.update_instance(init_dataset_return.preprocess_fitted_instances[fitted_inner_index].instance, ts_id)
                        fitted_inner_index += 1

                # updates outer preprocessors based on passed train data from InitDataset
                for outer_preprocess_order in group.preprocess_outer_orders:
                    if outer_preprocess_order.should_be_fitted:
                        outer_preprocess_order.holder.fit(init_dataset_return.train_data, ts_id)

                    if outer_preprocess_order.can_be_applied:
                        init_dataset_return.train_data = outer_preprocess_order.holder.apply(init_dataset_return.train_data, ts_id)

        if workers == 0:
            init_dataset.cleanup()

        # Update config based on filtered time series
        if is_first_cycle:

            if len(ts_ids_to_take) == 0:
                raise ValueError("No valid time series left in train set after applying nan_threshold.")

            self.dataset_config.train_ts_row_ranges = self.dataset_config.train_ts_row_ranges[ts_ids_to_take]
            self.dataset_config.train_ts = self.dataset_config.train_ts[ts_ids_to_take]
            self.dataset_config._update_preprocess_order_supported_ids(self.dataset_config.train_preprocess_order, ts_ids_to_take)
            self.logger.debug("invalid ts_ids removed: %s time series left.", len(ts_ids_to_take))

            is_first_cycle = False

__initialize_config_for_non_fit_sets

__initialize_config_for_non_fit_sets(init_config: DisjointTimeDatasetInitConfig, workers: int, set_name: str) -> np.ndarray

Initializes config for provided time series without fitting.

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def __initialize_config_for_non_fit_sets(self, init_config: DisjointTimeDatasetInitConfig, workers: int, set_name: str) -> np.ndarray:
    """Initializes config for provided time series without fitting. """
    init_dataset = DisjointTimeBasedInitializerDataset(self.metadata.dataset_path, self.metadata.data_table_path, init_config)

    sampler = SequentialSampler(init_dataset)
    dataloader = DataLoader(init_dataset, num_workers=workers, collate_fn=self._collate_fn, worker_init_fn=DisjointTimeBasedInitializerDataset.worker_init_fn, persistent_workers=False, sampler=sampler)

    if workers == 0:
        init_dataset.pytables_worker_init()

    ts_ids_to_take = []

    self.logger.info("Updating config for %s set.", set_name)
    for i, data in enumerate(tqdm(dataloader, total=len(init_config.ts_row_ranges))):
        init_dataset_return: InitDatasetReturn = data[0]

        if init_dataset_return.is_under_nan_threshold:
            ts_ids_to_take.append(i)

    if workers == 0:
        init_dataset.cleanup()

    if len(ts_ids_to_take) == 0:
        raise ValueError(f"No valid time series left in {set_name} set after applying nan_threshold.")

    return ts_ids_to_take

_update_export_config_copy

_update_export_config_copy() -> None

Called at the end of set_dataset_config_and_initialize or when changing config values.

Updates values of config used for saving config.

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def _update_export_config_copy(self) -> None:
    """
    Called at the end of [`set_dataset_config_and_initialize`](reference_disjoint_time_based_cesnet_dataset.md#cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset.set_dataset_config_and_initialize) or when changing config values. 

    Updates values of config used for saving config.
    """
    self._export_config_copy.database_name = self.metadata.database_name

    self._export_config_copy.train_ts = self.dataset_config.train_ts.copy() if self.dataset_config.has_train() else None
    self._export_config_copy.val_ts = self.dataset_config.val_ts.copy() if self.dataset_config.has_val() else None
    self._export_config_copy.test_ts = self.dataset_config.test_ts.copy() if self.dataset_config.has_test() else None

    self._export_config_copy.sliding_window_size = self.dataset_config.sliding_window_size
    self._export_config_copy.sliding_window_prediction_size = self.dataset_config.sliding_window_prediction_size
    self._export_config_copy.sliding_window_step = self.dataset_config.sliding_window_step
    self._export_config_copy.set_shared_size = self.dataset_config.set_shared_size

    super(DisjointTimeBasedCesnetDataset, self)._update_export_config_copy()

_get_singular_time_series_dataset

_get_singular_time_series_dataset(parent_dataset: DisjointTimeBasedSplittedDataset, ts_id: int) -> DisjointTimeBasedSplittedDataset

Returns dataset for single time series

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def _get_singular_time_series_dataset(self, parent_dataset: DisjointTimeBasedSplittedDataset, ts_id: int) -> DisjointTimeBasedSplittedDataset:
    """Returns dataset for single time series """

    temp = np.where(np.isin(parent_dataset.load_config.ts_row_ranges[self.metadata.ts_id_name], [ts_id]))[0]

    if len(temp) == 0:
        raise ValueError(f"ts_id {ts_id} was not found in valid time series for this set. Available time series are: {parent_dataset.load_config.ts_row_ranges[self.metadata.ts_id_name]}")

    time_series_position = temp[0]

    split_load_config = parent_dataset.load_config.create_split_copy(slice(time_series_position, time_series_position + 1))

    dataset = DisjointTimeBasedSplittedDataset(self.metadata.dataset_path, self.metadata.data_table_path, split_load_config, 0)
    self.logger.debug("Singular time series dataset initiliazed.")

    return dataset

_get_data_for_plot

_get_data_for_plot(ts_id: int, feature_indices: ndarray[int], time_format: TimeFormat) -> tuple[np.ndarray, np.ndarray]

Dataset type specific retrieval of data.

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def _get_data_for_plot(self, ts_id: int, feature_indices: np.ndarray[int], time_format: TimeFormat) -> tuple[np.ndarray, np.ndarray]:
    """Dataset type specific retrieval of data. """

    train_id_result, val_id_result, test_id_result = None, None, None

    if (self.dataset_config.has_train()):
        train_id_result = np.argwhere(np.isin(self.dataset_config.train_ts, ts_id)).ravel()
    if (self.dataset_config.has_val()):
        val_id_result = np.argwhere(np.isin(self.dataset_config.val_ts, ts_id)).ravel()
    if (self.dataset_config.has_test()):
        test_id_result = np.argwhere(np.isin(self.dataset_config.test_ts, ts_id)).ravel()

    data = None
    time_period = None

    if self.dataset_config.has_train() and len(train_id_result) > 0:
        data = self.__get_ts_data_for_plot(self.train_dataset, ts_id, feature_indices)
        time_period = self.get_data_about_set(SplitType.TRAIN)[time_format]
        self.logger.debug("Valid ts_id found: %d", train_id_result[0])

    elif self.dataset_config.has_val() and len(val_id_result) > 0:
        data = self.__get_ts_data_for_plot(self.val_dataset, ts_id, feature_indices)
        time_period = self.get_data_about_set(SplitType.VAL)[time_format]
        self.logger.debug("Valid ts_id found: %d", val_id_result[0])

    elif self.dataset_config.has_test() and len(test_id_result) > 0:
        data = self.__get_ts_data_for_plot(self.test_dataset, ts_id, feature_indices)
        time_period = self.get_data_about_set(SplitType.TEST)[time_format]
        self.logger.debug("Valid ts_id found: %d", test_id_result[0])
    else:
        raise ValueError(f"Invalid ts_id '{ts_id}'. The provided ts_id is not found in the available time series IDs.", self.dataset_config.train_ts, self.dataset_config.val_ts, self.dataset_config.test_ts)

    return data, time_period

__get_ts_data_for_plot

__get_ts_data_for_plot(dataset: DisjointTimeBasedSplittedDataset, ts_id: int, feature_indices: list[int])

Source code in cesnet_tszoo\datasets\disjoint_time_based_cesnet_dataset.py

def __get_ts_data_for_plot(self, dataset: DisjointTimeBasedSplittedDataset, ts_id: int, feature_indices: list[int]):
    dataset = self._get_singular_time_series_dataset(dataset, ts_id)

    dataloader = self.dataloader_factory.create_dataloader(dataset, self.dataset_config, 0, True, None)

    temp_data = dataset_loaders.create_numpy_from_dataloader(dataloader, np.array([ts_id]), dataset.load_config.time_format, dataset.load_config.include_time, DatasetType.TIME_BASED, True)

    if (dataset.load_config.time_format == TimeFormat.DATETIME and dataset.load_config.include_time):
        temp_data = temp_data[0]

    temp_data = temp_data[0][:, feature_indices]

    return temp_data

get_train_dataloader

get_train_dataloader(ts_id: int | None = None, workers: int | Literal['config'] = 'config', **kwargs) -> DataLoader

Returns a PyTorch DataLoader for training set.

The DataLoader is created on the first call and cached for subsequent use.
The cached dataloader is cleared when either get_train_df or get_train_numpy is called.

The structure of the returned batch depends on the time_format and whether sliding_window_size is used:

When sliding_window_size is used:
- With time_format == TimeFormat.DATETIME and included time:
  - np.ndarray of shape (num_time_series, times - 1, features)
  - np.ndarray of shape (num_time_series, 1, features)
  - np.ndarray of times with shape (times - 1)
  - np.ndarray of time with shape (1)
- When time_format != TimeFormat.DATETIME or time is not included:
  - np.ndarray of shape (num_time_series, times - 1, features)
  - np.ndarray of shape (num_time_series, 1, features)
When sliding_window_size is not used:
- With time_format == TimeFormat.DATETIME and included time:
  - np.ndarray of shape (num_time_series, times, features)
  - np.ndarray of time with shape (times)
- When time_format != TimeFormat.DATETIME or time is not included:
  - np.ndarray of shape (num_time_series, times, features)

The DataLoader is configured with the following config attributes:

Dataset config	Description
`train_batch_size`	Number of samples per batch. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
`sliding_window_size`	Available only for time-based datasets. Modifies the shape of the returned data.
`sliding_window_prediction_size`	Available only for time-based datasets. Modifies the shape of the returned data.
`sliding_window_step`	Available only for time-based datasets. Number of times to move by after each window.
`train_workers`	Specifies the number of workers to use for loading train data. Applied when `workers` = "config".
`train_dataloader_order`	Available only for series-based datasets. Whether to load train data in sequential or random order.
`random_state`	Seed for loading train data in random order.

Parameters:

Name	Type	Description	Default
`workers`	`int \| Literal['config']`	The number of workers to use for loading train data. `Default: "config"`	`'config'`
`ts_id`	`int \| None`	Specifies time series to take. If None returns all time series as normal. `Default: "None"`	`None`

Returns:

Type	Description
`DataLoader`	An iterable `DataLoader` containing data from training set.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_train_dataloader(self, ts_id: int | None = None, workers: int | Literal["config"] = "config", **kwargs) -> DataLoader:
    """
    Returns a PyTorch [`DataLoader`](https://pytorch.org/docs/stable/data.html#torch.utils.data.DataLoader) for training set.

    The `DataLoader` is created on the first call and cached for subsequent use. <br/>
    The cached dataloader is cleared when either [`get_train_df`](reference_cesnet_dataset.md#cesnet_tszoo.datasets.cesnet_dataset.CesnetDataset.get_train_df) or [`get_train_numpy`](reference_cesnet_dataset.md#cesnet_tszoo.datasets.cesnet_dataset.CesnetDataset.get_train_numpy) is called.

    The structure of the returned batch depends on the `time_format` and whether `sliding_window_size` is used:

    - When `sliding_window_size` is used:
        - With `time_format` == TimeFormat.DATETIME and included time:
            - `np.ndarray` of shape `(num_time_series, times - 1, features)`
            - `np.ndarray` of shape `(num_time_series, 1, features)`
            - `np.ndarray` of times with shape `(times - 1)`
            - `np.ndarray` of time with shape `(1)`
        - When `time_format` != TimeFormat.DATETIME or time is not included:
            - `np.ndarray` of shape `(num_time_series, times - 1, features)`
            - `np.ndarray` of shape `(num_time_series, 1, features)`
    - When `sliding_window_size` is not used:
        - With `time_format` == TimeFormat.DATETIME and included time:
            - `np.ndarray` of shape `(num_time_series, times, features)`
            - `np.ndarray` of time with shape `(times)`
        - When `time_format` != TimeFormat.DATETIME or time is not included:
            - `np.ndarray` of shape `(num_time_series, times, features)`

    The `DataLoader` is configured with the following config attributes:

    Dataset config | Description
    -------------- | -----------
    `train_batch_size` | Number of samples per batch. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
    `sliding_window_size` | Available only for time-based datasets. Modifies the shape of the returned data.
    `sliding_window_prediction_size` | Available only for time-based datasets. Modifies the shape of the returned data.
    `sliding_window_step` | Available only for time-based datasets. Number of times to move by after each window.
    `train_workers` | Specifies the number of workers to use for loading train data. Applied when `workers` = "config".
    `train_dataloader_order` | Available only for series-based datasets. Whether to load train data in sequential or random order.
    `random_state` | Seed for loading train data in random order.

    Parameters:
        workers: The number of workers to use for loading train data. `Default: "config"` 
        ts_id: Specifies time series to take. If None returns all time series as normal. `Default: "None"`

    Returns:
        An iterable `DataLoader` containing data from training set.          
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to access train_dataloader.")

    if not self.dataset_config.has_train():
        raise ValueError("Dataloader for training set is not available in the dataset configuration.")

    assert self.train_dataset is not None, "The train_dataset must be initialized before accessing data from training set."

    default_kwargs = {'take_all': False, "cache_loader": True}
    kwargs = {**default_kwargs, **kwargs}

    if ts_id is not None:

        if ts_id == self.dataset_config.used_singular_train_time_series and self.train_dataloader is not None:
            self.logger.debug("Returning cached train_dataloader.")
            return self.train_dataloader

        dataset = self._get_singular_time_series_dataset(self.train_dataset, ts_id)
        self.dataset_config.used_singular_train_time_series = ts_id
        if self.train_dataloader:
            del self.train_dataloader
            self.train_dataloader = None
            self.logger.info("Destroyed previous cached train_dataloader.")

        self.dataset_config.used_train_workers = 0
        self.train_dataloader = self.dataloader_factory.create_dataloader(dataset, self.dataset_config, 0, False, self.dataset_config.train_batch_size)
        self.logger.info("Created new cached train_dataloader.")
        return self.train_dataloader
    elif self.dataset_config.used_singular_train_time_series is not None and self.train_dataloader is not None:
        del self.train_dataloader
        self.train_dataloader = None
        self.dataset_config.used_singular_train_time_series = None
        self.logger.info("Destroyed previous cached train_dataloader.")

    if workers == "config":
        workers = self.dataset_config.train_workers

    # If the dataloader is cached and number of used workers did not change, return the cached dataloader
    if self.train_dataloader and kwargs["cache_loader"] and workers == self.dataset_config.used_train_workers:
        self.logger.debug("Returning cached train_dataloader.")
        return self.train_dataloader

    # Update the used workers count
    self.dataset_config.used_train_workers = workers

    # If there's a previously cached dataloader, destroy it
    if self.train_dataloader:
        del self.train_dataloader
        self.train_dataloader = None
        self.logger.info("Destroyed previous cached train_dataloader.")

    # If caching is enabled, create a new cached dataloader
    if kwargs["cache_loader"]:
        self.train_dataloader = self.dataloader_factory.create_dataloader(self.train_dataset, self.dataset_config, workers, kwargs['take_all'], self.dataset_config.train_batch_size, order=self.dataset_config.train_dataloader_order)
        self.logger.info("Created new cached train_dataloader.")
        return self.train_dataloader

    # If caching is disabled, create a new uncached dataloader
    self.logger.debug("Created new uncached train_dataloader.")
    return self.dataloader_factory.create_dataloader(self.train_dataset, self.dataset_config, workers, kwargs['take_all'], self.dataset_config.train_batch_size, order=self.dataset_config.train_dataloader_order)

get_val_dataloader

get_val_dataloader(ts_id: int | None = None, workers: int | Literal['config'] = 'config', **kwargs) -> DataLoader

Returns a PyTorch DataLoader for validation set.

The DataLoader is created on the first call and cached for subsequent use.
The cached dataloader is cleared when either get_val_df or get_val_numpy is called.

The structure of the returned batch depends on the time_format and whether sliding_window_size is used:

When sliding_window_size is used:
- With time_format == TimeFormat.DATETIME and included time:
  - np.ndarray of shape (num_time_series, times - 1, features)
  - np.ndarray of shape (num_time_series, 1, features)
  - np.ndarray of times with shape (times - 1)
  - np.ndarray of time with shape (1)
- When time_format != TimeFormat.DATETIME or time is not included:
  - np.ndarray of shape (num_time_series, times - 1, features)
  - np.ndarray of shape (num_time_series, 1, features)
When sliding_window_size is not used:
- With time_format == TimeFormat.DATETIME and included time:
  - np.ndarray of shape (num_time_series, times, features)
  - np.ndarray of time with shape (times)
- When time_format != TimeFormat.DATETIME or time is not included:
  - np.ndarray of shape (num_time_series, times, features)

The DataLoader is configured with the following config attributes:

Dataset config	Description
`val_batch_size`	Number of samples per batch. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
`sliding_window_size`	Available only for time-based datasets. Modifies the shape of the returned data.
`sliding_window_prediction_size`	Available only for time-based datasets. Modifies the shape of the returned data.
`sliding_window_step`	Available only for time-based datasets. Number of times to move by after each window.
`val_workers`	Specifies the number of workers to use for loading validation data. Applied when `workers` = "config".

Parameters:

Name	Type	Description	Default
`workers`	`int \| Literal['config']`	The number of workers to use for loading validation data. `Default: "config"`	`'config'`
`ts_id`	`int \| None`	Specifies time series to take. If None returns all time series as normal. `Default: "None"`	`None`

Returns:

Type	Description
`DataLoader`	An iterable `DataLoader` containing data from validation set.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_val_dataloader(self, ts_id: int | None = None, workers: int | Literal["config"] = "config", **kwargs) -> DataLoader:
    """
    Returns a PyTorch [`DataLoader`](https://pytorch.org/docs/stable/data.html#torch.utils.data.DataLoader) for validation set.

    The `DataLoader` is created on the first call and cached for subsequent use. <br/>
    The cached dataloader is cleared when either [`get_val_df`](reference_cesnet_dataset.md#cesnet_tszoo.datasets.cesnet_dataset.CesnetDataset.get_val_df) or [`get_val_numpy`](reference_cesnet_dataset.md#cesnet_tszoo.datasets.cesnet_dataset.CesnetDataset.get_val_numpy) is called.

    The structure of the returned batch depends on the `time_format` and whether `sliding_window_size` is used:

    - When `sliding_window_size` is used:
        - With `time_format` == TimeFormat.DATETIME and included time:
            - `np.ndarray` of shape `(num_time_series, times - 1, features)`
            - `np.ndarray` of shape `(num_time_series, 1, features)`
            - `np.ndarray` of times with shape `(times - 1)`
            - `np.ndarray` of time with shape `(1)`
        - When `time_format` != TimeFormat.DATETIME or time is not included:
            - `np.ndarray` of shape `(num_time_series, times - 1, features)`
            - `np.ndarray` of shape `(num_time_series, 1, features)`
    - When `sliding_window_size` is not used:
        - With `time_format` == TimeFormat.DATETIME and included time:
            - `np.ndarray` of shape `(num_time_series, times, features)`
            - `np.ndarray` of time with shape `(times)`
        - When `time_format` != TimeFormat.DATETIME or time is not included:
            - `np.ndarray` of shape `(num_time_series, times, features)`

    The `DataLoader` is configured with the following config attributes:

    Dataset config | Description
    -------------- | -----------
    `val_batch_size` | Number of samples per batch. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
    `sliding_window_size` | Available only for time-based datasets. Modifies the shape of the returned data.
    `sliding_window_prediction_size` | Available only for time-based datasets. Modifies the shape of the returned data.
    `sliding_window_step` | Available only for time-based datasets. Number of times to move by after each window.
    `val_workers` | Specifies the number of workers to use for loading validation data. Applied when `workers` = "config".


    Parameters:
        workers: The number of workers to use for loading validation data. `Default: "config"`  
        ts_id: Specifies time series to take. If None returns all time series as normal. `Default: "None"`

    Returns:
        An iterable `DataLoader` containing data from validation set.        
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to access val_dataloader.")

    if not self.dataset_config.has_val():
        raise ValueError("Dataloader for validation set is not available in the dataset configuration.")

    assert self.val_dataset is not None, "The val_dataset must be initialized before accessing data from validation set."

    default_kwargs = {'take_all': False, "cache_loader": True}
    kwargs = {**default_kwargs, **kwargs}

    if ts_id is not None:

        if ts_id == self.dataset_config.used_singular_val_time_series and self.val_dataloader is not None:
            self.logger.debug("Returning cached val_dataloader.")
            return self.val_dataloader

        dataset = self._get_singular_time_series_dataset(self.val_dataset, ts_id)
        self.dataset_config.used_singular_val_time_series = ts_id
        if self.val_dataloader:
            del self.val_dataloader
            self.val_dataloader = None
            self.logger.info("Destroyed previous cached val_dataloader.")

        self.dataset_config.used_val_workers = 0
        self.val_dataloader = self.dataloader_factory.create_dataloader(dataset, self.dataset_config, 0, False, self.dataset_config.val_batch_size)
        self.logger.info("Created new cached val_dataloader.")
        return self.val_dataloader
    elif self.dataset_config.used_singular_val_time_series is not None and self.val_dataloader is not None:
        del self.val_dataloader
        self.val_dataloader = None
        self.dataset_config.used_singular_val_time_series = None
        self.logger.info("Destroyed previous cached val_dataloader.")

    if workers == "config":
        workers = self.dataset_config.val_workers

    # If the dataloader is cached and number of used workers did not change, return the cached dataloader
    if self.val_dataloader and kwargs["cache_loader"] and workers == self.dataset_config.used_val_workers:
        self.logger.debug("Returning cached val_dataloader.")
        return self.val_dataloader

    # Update the used workers count
    self.dataset_config.used_val_workers = workers

    # If there's a previously cached dataloader, destroy it
    if self.val_dataloader:
        del self.val_dataloader
        self.val_dataloader = None
        self.logger.info("Destroyed previous cached val_dataloader.")

    # If caching is enabled, create a new cached dataloader
    if kwargs["cache_loader"]:
        self.val_dataloader = self.dataloader_factory.create_dataloader(self.val_dataset, self.dataset_config, workers, kwargs['take_all'], self.dataset_config.val_batch_size)
        self.logger.info("Created new cached val_dataloader.")
        return self.val_dataloader

    # If caching is disabled, create a new uncached dataloader
    self.logger.debug("Created new uncached val_dataloader.")
    return self.dataloader_factory.create_dataloader(self.val_dataset, self.dataset_config, workers, kwargs['take_all'], self.dataset_config.val_batch_size)

get_test_dataloader

get_test_dataloader(ts_id: int | None = None, workers: int | Literal['config'] = 'config', **kwargs) -> DataLoader

Returns a PyTorch DataLoader for test set.

The DataLoader is created on the first call and cached for subsequent use.
The cached dataloader is cleared when either get_test_df or get_test_numpy is called.

The structure of the returned batch depends on the time_format and whether sliding_window_size is used:

When sliding_window_size is used:
- With time_format == TimeFormat.DATETIME and included time:
  - np.ndarray of shape (num_time_series, times - 1, features)
  - np.ndarray of shape (num_time_series, 1, features)
  - np.ndarray of times with shape (times - 1)
  - np.ndarray of time with shape (1)
- When time_format != TimeFormat.DATETIME or time is not included:
  - np.ndarray of shape (num_time_series, times - 1, features)
  - np.ndarray of shape (num_time_series, 1, features)
When sliding_window_size is not used:
- With time_format == TimeFormat.DATETIME and included time:
  - np.ndarray of shape (num_time_series, times, features)
  - np.ndarray of time with shape (times)
- When time_format != TimeFormat.DATETIME or time is not included:
  - np.ndarray of shape (num_time_series, times, features)

The DataLoader is configured with the following config attributes:

Dataset config	Description
`test_batch_size`	Number of samples per batch. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
`sliding_window_size`	Available only for time-based datasets. Modifies the shape of the returned data.
`sliding_window_prediction_size`	Available only for time-based datasets. Modifies the shape of the returned data.
`sliding_window_step`	Available only for time-based datasets. Number of times to move by after each window.
`test_workers`	Specifies the number of workers to use for loading test data. Applied when `workers` = "config".

Parameters:

Name	Type	Description	Default
`workers`	`int \| Literal['config']`	The number of workers to use for loading test data. `Default: "config"`	`'config'`
`ts_id`	`int \| None`	Specifies time series to take. If None returns all time series as normal. `Default: "None"`	`None`

Returns:

Type	Description
`DataLoader`	An iterable `DataLoader` containing data from test set.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_test_dataloader(self, ts_id: int | None = None, workers: int | Literal["config"] = "config", **kwargs) -> DataLoader:
    """
    Returns a PyTorch [`DataLoader`](https://pytorch.org/docs/stable/data.html#torch.utils.data.DataLoader) for test set.

    The `DataLoader` is created on the first call and cached for subsequent use. <br/>
    The cached dataloader is cleared when either [`get_test_df`](reference_cesnet_dataset.md#cesnet_tszoo.datasets.cesnet_dataset.CesnetDataset.get_test_df) or [`get_test_numpy`](reference_cesnet_dataset.md#cesnet_tszoo.datasets.cesnet_dataset.CesnetDataset.get_test_numpy) is called.

    The structure of the returned batch depends on the `time_format` and whether `sliding_window_size` is used:

    - When `sliding_window_size` is used:
        - With `time_format` == TimeFormat.DATETIME and included time:
            - `np.ndarray` of shape `(num_time_series, times - 1, features)`
            - `np.ndarray` of shape `(num_time_series, 1, features)`
            - `np.ndarray` of times with shape `(times - 1)`
            - `np.ndarray` of time with shape `(1)`
        - When `time_format` != TimeFormat.DATETIME or time is not included:
            - `np.ndarray` of shape `(num_time_series, times - 1, features)`
            - `np.ndarray` of shape `(num_time_series, 1, features)`
    - When `sliding_window_size` is not used:
        - With `time_format` == TimeFormat.DATETIME and included time:
            - `np.ndarray` of shape `(num_time_series, times, features)`
            - `np.ndarray` of time with shape `(times)`
        - When `time_format` != TimeFormat.DATETIME or time is not included:
            - `np.ndarray` of shape `(num_time_series, times, features)`

    The `DataLoader` is configured with the following config attributes:

    Dataset config | Description
    -------------- | -----------
    `test_batch_size` | Number of samples per batch. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
    `sliding_window_size` | Available only for time-based datasets. Modifies the shape of the returned data.
    `sliding_window_prediction_size` | Available only for time-based datasets. Modifies the shape of the returned data.
    `sliding_window_step` | Available only for time-based datasets. Number of times to move by after each window.
    `test_workers` | Specifies the number of workers to use for loading test data. Applied when `workers` = "config".


    Parameters:
        workers: The number of workers to use for loading test data. `Default: "config"`  
        ts_id: Specifies time series to take. If None returns all time series as normal. `Default: "None"`

    Returns:
        An iterable `DataLoader` containing data from test set.        
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to access test_dataloader.")

    if not self.dataset_config.has_test():
        raise ValueError("Dataloader for test set is not available in the dataset configuration.")

    assert self.test_dataset is not None, "The test_dataset must be initialized before accessing data from test set."

    default_kwargs = {'take_all': False, "cache_loader": True}
    kwargs = {**default_kwargs, **kwargs}

    if ts_id is not None:

        if ts_id == self.dataset_config.used_singular_test_time_series and self.test_dataloader is not None:
            self.logger.debug("Returning cached test_dataloader.")
            return self.test_dataloader

        dataset = self._get_singular_time_series_dataset(self.test_dataset, ts_id)
        self.dataset_config.used_singular_test_time_series = ts_id
        if self.test_dataloader:
            del self.test_dataloader
            self.test_dataloader = None
            self.logger.info("Destroyed previous cached test_dataloader.")

        self.dataset_config.used_test_workers = 0
        self.test_dataloader = self.dataloader_factory.create_dataloader(dataset, self.dataset_config, 0, False, self.dataset_config.test_batch_size)
        self.logger.info("Created new cached test_dataloader.")
        return self.test_dataloader
    elif self.dataset_config.used_singular_test_time_series is not None and self.test_dataloader is not None:
        del self.test_dataloader
        self.test_dataloader = None
        self.dataset_config.used_singular_test_time_series = None
        self.logger.info("Destroyed previous cached test_dataloader.")

    if workers == "config":
        workers = self.dataset_config.test_workers

    # If the dataloader is cached and number of used workers did not change, return the cached dataloader
    if self.test_dataloader and kwargs["cache_loader"] and workers == self.dataset_config.used_test_workers:
        self.logger.debug("Returning cached test_dataloader.")
        return self.test_dataloader

    # Update the used workers count
    self.dataset_config.used_test_workers = workers

    # If there's a previously cached dataloader, destroy it
    if self.test_dataloader:
        del self.test_dataloader
        self.test_dataloader = None
        self.logger.info("Destroyed previous cached test_dataloader.")

    # If caching is enabled, create a new cached dataloader
    if kwargs["cache_loader"]:
        self.test_dataloader = self.dataloader_factory.create_dataloader(self.test_dataset, self.dataset_config, workers, kwargs['take_all'], self.dataset_config.test_batch_size)
        self.logger.info("Created new cached test_dataloader.")
        return self.test_dataloader

    # If caching is disabled, create a new uncached dataloader
    self.logger.debug("Created new uncached test_dataloader.")
    return self.dataloader_factory.create_dataloader(self.test_dataset, self.dataset_config, workers, kwargs['take_all'], self.dataset_config.test_batch_size)

get_all_dataloader

get_all_dataloader(ts_id: int | None = None, workers: int | Literal['config'] = 'config', **kwargs) -> DataLoader

Returns a PyTorch DataLoader for all set.

The DataLoader is created on the first call and cached for subsequent use.
The cached dataloader is cleared when either get_all_df or get_all_numpy is called.

The structure of the returned batch depends on the time_format and whether sliding_window_size is used:

When sliding_window_size is used:
- With time_format == TimeFormat.DATETIME and included time:
  - np.ndarray of shape (num_time_series, times - 1, features)
  - np.ndarray of shape (num_time_series, 1, features)
  - np.ndarray of times with shape (times - 1)
  - np.ndarray of time with shape (1)
- When time_format != TimeFormat.DATETIME or time is not included:
  - np.ndarray of shape (num_time_series, times - 1, features)
  - np.ndarray of shape (num_time_series, 1, features)
When sliding_window_size is not used:
- With time_format == TimeFormat.DATETIME and included time:
  - np.ndarray of shape (num_time_series, times, features)
  - np.ndarray of time with shape (times)
- When time_format != TimeFormat.DATETIME or time is not included:
  - np.ndarray of shape (num_time_series, times, features)

The DataLoader is configured with the following config attributes:

Dataset config	Description
`all_batch_size`	Number of samples per batch. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
`sliding_window_size`	Available only for time-based datasets. Modifies the shape of the returned data.
`sliding_window_prediction_size`	Available only for time-based datasets. Modifies the shape of the returned data.
`sliding_window_step`	Available only for time-based datasets. Number of times to move by after each window.
`all_workers`	Specifies the number of workers to use for loading all data. Applied when `workers` = "config".

Parameters:

Name	Type	Description	Default
`workers`	`int \| Literal['config']`	The number of workers to use for loading all data. `Default: "config"`	`'config'`
`ts_id`	`int \| None`	Specifies time series to take. If None returns all time series as normal. `Default: "None"`	`None`

Returns:

Type	Description
`DataLoader`	An iterable `DataLoader` containing data from all set.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_all_dataloader(self, ts_id: int | None = None, workers: int | Literal["config"] = "config", **kwargs) -> DataLoader:
    """
    Returns a PyTorch [`DataLoader`](https://pytorch.org/docs/stable/data.html#torch.utils.data.DataLoader) for all set.

    The `DataLoader` is created on the first call and cached for subsequent use. <br/>
    The cached dataloader is cleared when either [`get_all_df`](reference_cesnet_dataset.md#cesnet_tszoo.datasets.cesnet_dataset.CesnetDataset.get_all_df) or [`get_all_numpy`](reference_cesnet_dataset.md#cesnet_tszoo.datasets.cesnet_dataset.CesnetDataset.get_all_numpy) is called.

    The structure of the returned batch depends on the `time_format` and whether `sliding_window_size` is used:

    - When `sliding_window_size` is used:
        - With `time_format` == TimeFormat.DATETIME and included time:
            - `np.ndarray` of shape `(num_time_series, times - 1, features)`
            - `np.ndarray` of shape `(num_time_series, 1, features)`
            - `np.ndarray` of times with shape `(times - 1)`
            - `np.ndarray` of time with shape `(1)`
        - When `time_format` != TimeFormat.DATETIME or time is not included:
            - `np.ndarray` of shape `(num_time_series, times - 1, features)`
            - `np.ndarray` of shape `(num_time_series, 1, features)`
    - When `sliding_window_size` is not used:
        - With `time_format` == TimeFormat.DATETIME and included time:
            - `np.ndarray` of shape `(num_time_series, times, features)`
            - `np.ndarray` of time with shape `(times)`
        - When `time_format` != TimeFormat.DATETIME or time is not included:
            - `np.ndarray` of shape `(num_time_series, times, features)`

    The `DataLoader` is configured with the following config attributes:

    Dataset config | Description
    -------------- | -----------
    `all_batch_size` | Number of samples per batch. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.
    `sliding_window_size` | Available only for time-based datasets. Modifies the shape of the returned data.
    `sliding_window_prediction_size` | Available only for time-based datasets. Modifies the shape of the returned data.
    `sliding_window_step` | Available only for time-based datasets. Number of times to move by after each window.
    `all_workers` | Specifies the number of workers to use for loading all data. Applied when `workers` = "config".


    Parameters:
        workers: The number of workers to use for loading all data. `Default: "config"`  
        ts_id: Specifies time series to take. If None returns all time series as normal. `Default: "None"`

    Returns:
        An iterable `DataLoader` containing data from all set.       
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to access all_dataloader.")

    if not self.dataset_config.has_all():
        raise ValueError("Dataloader for all set is not available in the dataset configuration.")

    assert self.all_dataset is not None, "The all_dataset must be initialized before accessing data from all set."

    default_kwargs = {'take_all': False, "cache_loader": True}
    kwargs = {**default_kwargs, **kwargs}

    if ts_id is not None:

        if ts_id == self.dataset_config.used_singular_all_time_series and self.all_dataloader is not None:
            self.logger.debug("Returning cached all_dataloader.")
            return self.all_dataloader

        dataset = self._get_singular_time_series_dataset(self.all_dataset, ts_id)
        self.dataset_config.used_singular_all_time_series = ts_id
        if self.all_dataloader:
            del self.all_dataloader
            self.all_dataloader = None
            self.logger.info("Destroyed previous cached all_dataloader.")

        self.dataset_config.used_all_workers = 0
        self.all_dataloader = self.dataloader_factory.create_dataloader(dataset, self.dataset_config, 0, False, self.dataset_config.all_batch_size)
        self.logger.info("Created new cached all_dataloader.")
        return self.all_dataloader
    elif self.dataset_config.used_singular_all_time_series is not None and self.all_dataloader is not None:
        del self.all_dataloader
        self.all_dataloader = None
        self.dataset_config.used_singular_all_time_series = None
        self.logger.info("Destroyed previous cached all_dataloader.")

    if workers == "config":
        workers = self.dataset_config.all_workers

    # If the dataloader is cached and number of used workers did not change, return the cached dataloader
    if self.all_dataloader and kwargs["cache_loader"] and workers == self.dataset_config.used_all_workers:
        self.logger.debug("Returning cached all_dataloader.")
        return self.all_dataloader

    # Update the used workers count
    self.dataset_config.used_all_workers = workers

    # If there's a previously cached dataloader, destroy it
    if self.all_dataloader:
        del self.all_dataloader
        self.all_dataloader = None
        self.logger.info("Destroyed previous cached all_dataloader.")

    # If caching is enabled, create a new cached dataloader
    if kwargs["cache_loader"]:
        self.all_dataloader = self.dataloader_factory.create_dataloader(self.all_dataset, self.dataset_config, workers, kwargs['take_all'], self.dataset_config.all_batch_size)
        self.logger.info("Created new cached all_dataloader.")
        return self.all_dataloader

    # If caching is disabled, create a new uncached dataloader
    self.logger.debug("Creating new uncached all_dataloader.")
    return self.dataloader_factory.create_dataloader(self.all_dataset, self.dataset_config, workers, kwargs['take_all'], self.dataset_config.all_batch_size)

get_train_df

get_train_df(workers: int | Literal['config'] = 'config', as_single_dataframe: bool = True) -> pd.DataFrame

Creates a Pandas DataFrame containing all the data from training set grouped by time series.

This method uses the train_dataloader with a batch size set to the total number of data in the training set. The cached train_dataloader is cleared during this operation.

Memory usage

The entire training set is loaded into memory, which may lead to high memory usage. If working with large training set, consider using get_train_dataloader instead to handle data in batches.

Parameters:

Name	Type	Description	Default
`workers`	`int \| Literal['config']`	The number of workers to use for loading train data. `Default: "config"`	`'config'`
`as_single_dataframe`	`bool`	Whether to return a single dataframe with all time series combined, or to create separate dataframes for each time series. `Default: True`	`True`

Returns:

Type	Description
`DataFrame`	A single Pandas DataFrame containing all data from training set, or a list of DataFrames (one per time series).

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_train_df(self, workers: int | Literal["config"] = "config", as_single_dataframe: bool = True) -> pd.DataFrame:
    """
    Creates a Pandas [`DataFrame`](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html) containing all the data from training set grouped by time series.

    This method uses the `train_dataloader` with a batch size set to the total number of data in the training set. The cached `train_dataloader` is cleared during this operation.

    !!! warning "Memory usage"
        The entire training set is loaded into memory, which may lead to high memory usage. If working with large training set, consider using `get_train_dataloader` instead to handle data in batches.

    Parameters:
        workers: The number of workers to use for loading train data. `Default: "config"`  
        as_single_dataframe: Whether to return a single dataframe with all time series combined, or to create separate dataframes for each time series. `Default: True` 

    Returns:
        A single Pandas DataFrame containing all data from training set, or a list of DataFrames (one per time series).
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to access train_dataloader.")

    if not self.dataset_config.has_train():
        raise ValueError("Dataloader for training set is not available in the dataset configuration.")

    assert self.train_dataset is not None, "The train_dataset must be initialized before accessing data from training set."

    ts_ids, time_period = self.dataset_config._get_train()

    should_take_all = self.dataset_config.dataset_type != DatasetType.SERIES_BASED

    dataloader = self.get_train_dataloader(workers=workers, take_all=should_take_all, cache_loader=False)
    return self._get_df(dataloader, as_single_dataframe, ts_ids, time_period)

get_val_df

get_val_df(workers: int | Literal['config'] = 'config', as_single_dataframe: bool = True) -> pd.DataFrame

Create a Pandas DataFrame containing all the data from validation set grouped by time series.

This method uses the val_dataloader with a batch size set to the total number of data in the validation set. The cached val_dataloader is cleared during this operation.

Memory usage

The entire validation set is loaded into memory, which may lead to high memory usage. If working with large validation set, consider using get_val_dataloader instead to handle data in batches.

Parameters:

Name	Type	Description	Default
`workers`	`int \| Literal['config']`	The number of workers to use for loading validation data. `Default: "config"`	`'config'`
`as_single_dataframe`	`bool`	Whether to return a single dataframe with all time series combined, or to create separate dataframes for each time series. `Default: True`	`True`

Returns:

Type	Description
`DataFrame`	A single Pandas DataFrame containing all data from validation set, or a list of DataFrames (one per time series).

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_val_df(self, workers: int | Literal["config"] = "config", as_single_dataframe: bool = True) -> pd.DataFrame:
    """
    Create a Pandas [`DataFrame`](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html) containing all the data from validation set grouped by time series.

    This method uses the `val_dataloader` with a batch size set to the total number of data in the validation set. The cached `val_dataloader` is cleared during this operation.

    !!! warning "Memory usage"
        The entire validation set is loaded into memory, which may lead to high memory usage. If working with large validation set, consider using `get_val_dataloader` instead to handle data in batches.

    Parameters:
        workers: The number of workers to use for loading validation data. `Default: "config"`  
        as_single_dataframe: Whether to return a single dataframe with all time series combined, or to create separate dataframes for each time series. `Default: True` 

    Returns:
        A single Pandas DataFrame containing all data from validation set, or a list of DataFrames (one per time series).
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to access val_dataloader.")

    if not self.dataset_config.has_val():
        raise ValueError("Dataloader for validation set is not available in the dataset configuration.")

    assert self.val_dataset is not None, "The val_dataset must be initialized before accessing data from validation set."

    ts_ids, time_period = self.dataset_config._get_val()

    should_take_all = self.dataset_config.dataset_type != DatasetType.SERIES_BASED

    dataloader = self.get_val_dataloader(workers=workers, take_all=should_take_all, cache_loader=False)
    return self._get_df(dataloader, as_single_dataframe, ts_ids, time_period)

get_test_df

get_test_df(workers: int | Literal['config'] = 'config', as_single_dataframe: bool = True) -> pd.DataFrame

Creates a Pandas DataFrame containing all the data from test set grouped by time series.

This method uses the test_dataloader with a batch size set to the total number of data in the test set. The cached test_dataloader is cleared during this operation.

Memory usage

The entire test set is loaded into memory, which may lead to high memory usage. If working with large test set, consider using get_test_dataloader instead to handle data in batches.

Parameters:

Name	Type	Description	Default
`workers`	`int \| Literal['config']`	The number of workers to use for loading test data. `Default: "config"`	`'config'`
`as_single_dataframe`	`bool`	Whether to return a single dataframe with all time series combined, or to create separate dataframes for each time series. `Default: True`	`True`

Returns:

Type	Description
`DataFrame`	A single Pandas DataFrame containing all data from test set, or a list of DataFrames (one per time series).

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_test_df(self, workers: int | Literal["config"] = "config", as_single_dataframe: bool = True) -> pd.DataFrame:
    """
    Creates a Pandas [`DataFrame`](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html) containing all the data from test set grouped by time series.

    This method uses the `test_dataloader` with a batch size set to the total number of data in the test set. The cached `test_dataloader` is cleared during this operation.

    !!! warning "Memory usage"
        The entire test set is loaded into memory, which may lead to high memory usage. If working with large test set, consider using `get_test_dataloader` instead to handle data in batches.

    Parameters:
        workers: The number of workers to use for loading test data. `Default: "config"`  
        as_single_dataframe: Whether to return a single dataframe with all time series combined, or to create separate dataframes for each time series. `Default: True` 

    Returns:
        A single Pandas DataFrame containing all data from test set, or a list of DataFrames (one per time series).
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to access test_dataloader.")

    if not self.dataset_config.has_test():
        raise ValueError("Dataloader for test set is not available in the dataset configuration.")

    assert self.test_dataset is not None, "The test_dataset must be initialized before accessing data from test set."

    ts_ids, time_period = self.dataset_config._get_test()

    should_take_all = self.dataset_config.dataset_type != DatasetType.SERIES_BASED

    dataloader = self.get_test_dataloader(workers=workers, take_all=should_take_all, cache_loader=False)
    return self._get_df(dataloader, as_single_dataframe, ts_ids, time_period)

get_all_df

get_all_df(workers: int | Literal['config'] = 'config', as_single_dataframe: bool = True) -> pd.DataFrame

Creates a Pandas DataFrame containing all the data from all set grouped by time series.

This method uses the all_dataloader with a batch size set to the total number of data in the all set. The cached all_dataloader is cleared during this operation.

Memory usage

The entire all set is loaded into memory, which may lead to high memory usage. If working with large all set, consider using get_all_dataloader instead to handle data in batches.

Parameters:

Name	Type	Description	Default
`workers`	`int \| Literal['config']`	The number of workers to use for loading all data. `Default: "config"`	`'config'`
`as_single_dataframe`	`bool`	Whether to return a single dataframe with all time series combined, or to create separate dataframes for each time series. `Default: True`	`True`

Returns:

Type	Description
`DataFrame`	A single Pandas DataFrame containing all data from all set, or a list of DataFrames (one per time series).

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_all_df(self, workers: int | Literal["config"] = "config", as_single_dataframe: bool = True) -> pd.DataFrame:
    """
    Creates a Pandas [`DataFrame`](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html) containing all the data from all set grouped by time series.

    This method uses the `all_dataloader` with a batch size set to the total number of data in the all set. The cached `all_dataloader` is cleared during this operation.

    !!! warning "Memory usage"
        The entire all set is loaded into memory, which may lead to high memory usage. If working with large all set, consider using `get_all_dataloader` instead to handle data in batches.

    Parameters:
        workers: The number of workers to use for loading all data. `Default: "config"`  
        as_single_dataframe: Whether to return a single dataframe with all time series combined, or to create separate dataframes for each time series. `Default: True` 

    Returns:
        A single Pandas DataFrame containing all data from all set, or a list of DataFrames (one per time series).
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to access all_dataloader.")

    if not self.dataset_config.has_all():
        raise ValueError("Dataloader for all set is not available in the dataset configuration.")

    assert self.all_dataset is not None, "The all_dataset must be initialized before accessing data from all set."

    ts_ids, time_period = self.dataset_config._get_all()

    should_take_all = self.dataset_config.dataset_type != DatasetType.SERIES_BASED

    dataloader = self.get_all_dataloader(workers=workers, take_all=should_take_all, cache_loader=False)
    return self._get_df(dataloader, as_single_dataframe, ts_ids, time_period)

get_train_numpy

get_train_numpy(workers: int | Literal['config'] = 'config') -> np.ndarray

Creates a NumPy array containing all the data from training set grouped by time series, with the shape (num_time_series, num_times, num_features).

This method uses the train_dataloader with a batch size set to the total number of data in the training set. The cached train_dataloader is cleared during this operation.

Memory usage

The entire training set is loaded into memory, which may lead to high memory usage. If working with large training set, consider using get_train_dataloader instead to handle data in batches.

Parameters:

Name	Type	Description	Default
`workers`	`int \| Literal['config']`	The number of workers to use for loading train data. `Default: "config"`	`'config'`

Returns:

Type	Description
`ndarray`	A NumPy array containing all the data in training set with the shape `(num_time_series, num_times, num_features)`.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_train_numpy(self, workers: int | Literal["config"] = "config") -> np.ndarray:
    """
    Creates a NumPy array containing all the data from training set grouped by time series, with the shape `(num_time_series, num_times, num_features)`.

    This method uses the `train_dataloader` with a batch size set to the total number of data in the training set. The cached `train_dataloader` is cleared during this operation.

    !!! warning "Memory usage"
        The entire training set is loaded into memory, which may lead to high memory usage. If working with large training set, consider using `get_train_dataloader` instead to handle data in batches.        

    Parameters:
        workers: The number of workers to use for loading train data. `Default: "config"`  

    Returns:
        A NumPy array containing all the data in training set with the shape `(num_time_series, num_times, num_features)`.
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to access train_dataloader.")

    if not self.dataset_config.has_train():
        raise ValueError("Dataloader for training set is not available in the dataset configuration.")

    assert self.train_dataset is not None, "The train_dataset must be initialized before accessing data from training set."

    ts_ids, time_period = self.dataset_config._get_train()

    should_take_all = self.dataset_config.dataset_type != DatasetType.SERIES_BASED

    dataloader = self.get_train_dataloader(workers=workers, take_all=should_take_all, cache_loader=False)
    return self._get_numpy(dataloader, ts_ids, time_period)

get_val_numpy

get_val_numpy(workers: int | Literal['config'] = 'config') -> np.ndarray

Creates a NumPy array containing all the data from validation set grouped by time series, with the shape (num_time_series, num_times, num_features).

This method uses the val_dataloader with a batch size set to the total number of data in the validation set. The cached val_dataloader is cleared during this operation.

Memory usage

The entire validation set is loaded into memory, which may lead to high memory usage. If working with large validation set, consider using get_val_dataloader instead to handle data in batches.

Parameters:

Name	Type	Description	Default
`workers`	`int \| Literal['config']`	The number of workers to use for loading validation data. `Default: "config"`	`'config'`

Returns:

Type	Description
`ndarray`	A NumPy array containing all the data in validation set with the shape `(num_time_series, num_times, num_features)`.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_val_numpy(self, workers: int | Literal["config"] = "config") -> np.ndarray:
    """
    Creates a NumPy array containing all the data from validation set grouped by time series, with the shape `(num_time_series, num_times, num_features)`.

    This method uses the `val_dataloader` with a batch size set to the total number of data in the validation set. The cached `val_dataloader` is cleared during this operation.

    !!! warning "Memory usage"
        The entire validation set is loaded into memory, which may lead to high memory usage. If working with large validation set, consider using `get_val_dataloader` instead to handle data in batches.        

    Parameters:
        workers: The number of workers to use for loading validation data. `Default: "config"`  

    Returns:
        A NumPy array containing all the data in validation set with the shape `(num_time_series, num_times, num_features)`.
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to access val_dataloader.")

    if not self.dataset_config.has_val():
        raise ValueError("Dataloader for validation set is not available in the dataset configuration.")

    assert self.val_dataset is not None, "The val_dataset must be initialized before accessing data from validation set."

    ts_ids, time_period = self.dataset_config._get_val()

    should_take_all = self.dataset_config.dataset_type != DatasetType.SERIES_BASED

    dataloader = self.get_val_dataloader(workers=workers, take_all=should_take_all, cache_loader=False)
    return self._get_numpy(dataloader, ts_ids, time_period)

get_test_numpy

get_test_numpy(workers: int | Literal['config'] = 'config') -> np.ndarray

Creates a NumPy array containing all the data from test set grouped by time series, with the shape (num_time_series, num_times, num_features).

This method uses the test_dataloader with a batch size set to the total number of data in the test set. The cached test_dataloader is cleared during this operation.

Memory usage

The entire test set is loaded into memory, which may lead to high memory usage. If working with large test set, consider using get_test_dataloader instead to handle data in batches.

Parameters:

Name	Type	Description	Default
`workers`	`int \| Literal['config']`	The number of workers to use for loading test data. `Default: "config"`	`'config'`

Returns:

Type	Description
`ndarray`	A NumPy array containing all the data in test set with the shape `(num_time_series, num_times, num_features)`.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_test_numpy(self, workers: int | Literal["config"] = "config") -> np.ndarray:
    """
    Creates a NumPy array containing all the data from test set grouped by time series, with the shape `(num_time_series, num_times, num_features)`.

    This method uses the `test_dataloader` with a batch size set to the total number of data in the test set. The cached `test_dataloader` is cleared during this operation.

    !!! warning "Memory usage"
        The entire test set is loaded into memory, which may lead to high memory usage. If working with large test set, consider using `get_test_dataloader` instead to handle data in batches.        

    Parameters:
        workers: The number of workers to use for loading test data. `Default: "config"`  

    Returns:
        A NumPy array containing all the data in test set with the shape `(num_time_series, num_times, num_features)`.
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to access test_dataloader.")

    if not self.dataset_config.has_test():
        raise ValueError("Dataloader for test set is not available in the dataset configuration.")

    assert self.test_dataset is not None, "The test_dataset must be initialized before accessing data from test set."

    ts_ids, time_period = self.dataset_config._get_test()

    should_take_all = self.dataset_config.dataset_type != DatasetType.SERIES_BASED

    dataloader = self.get_test_dataloader(workers=workers, take_all=should_take_all, cache_loader=False)
    return self._get_numpy(dataloader, ts_ids, time_period)

get_all_numpy

get_all_numpy(workers: int | Literal['config'] = 'config') -> np.ndarray

Creates a NumPy array containing all the data from all set grouped by time series, with the shape (num_time_series, num_times, num_features).

This method uses the all_dataloader with a batch size set to the total number of data in the all set. The cached all_dataloader is cleared during this operation.

Memory usage

The entire all set is loaded into memory, which may lead to high memory usage. If working with large all set, consider using get_all_dataloader instead to handle data in batches.

Parameters:

Name	Type	Description	Default
`workers`	`int \| Literal['config']`	The number of workers to use for loading all data. `Default: "config"`	`'config'`

Returns:

Type	Description
`ndarray`	A NumPy array containing all the data in all set with the shape `(num_time_series, num_times, num_features)`.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_all_numpy(self, workers: int | Literal["config"] = "config") -> np.ndarray:
    """
    Creates a NumPy array containing all the data from all set grouped by time series, with the shape `(num_time_series, num_times, num_features)`.

    This method uses the `all_dataloader` with a batch size set to the total number of data in the all set. The cached `all_dataloader` is cleared during this operation.

    !!! warning "Memory usage"
        The entire all set is loaded into memory, which may lead to high memory usage. If working with large all set, consider using `get_all_dataloader` instead to handle data in batches.        

    Parameters:
        workers: The number of workers to use for loading all data. `Default: "config"`  

    Returns:
        A NumPy array containing all the data in all set with the shape `(num_time_series, num_times, num_features)`.
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to access all_dataloader.")

    if not self.dataset_config.has_all():
        raise ValueError("Dataloader for all set is not available in the dataset configuration.")

    assert self.all_dataset is not None, "The all_dataset must be initialized before accessing data from all set."

    ts_ids, time_period = self.dataset_config._get_all()

    should_take_all = self.dataset_config.dataset_type != DatasetType.SERIES_BASED

    dataloader = self.get_all_dataloader(workers=workers, take_all=should_take_all, cache_loader=False)
    return self._get_numpy(dataloader, ts_ids, time_period)

_update_dataset_config_and_initialize

_update_dataset_config_and_initialize(config_editor: ConfigEditor, workers: int | Literal['config'] = 'config', display_config_details: Optional[Literal['test', 'diagram']] = None)

Updates config via passed config editor.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def _update_dataset_config_and_initialize(self, config_editor: ConfigEditor, workers: int | Literal["config"] = "config", display_config_details: Optional[Literal["test", "diagram"]] = None):
    """Updates config via passed config editor. """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before updating dataset configuration.")

    if display_config_details is not None:
        display_config_details = DisplayType(display_config_details)

    original_config = deepcopy(self.dataset_config)
    original_export_config = deepcopy(self._export_config_copy)

    try:
        if config_editor.requires_init:
            self.logger.info("Re-initialization is required.")
            config_editor.modify_dataset_config(self._export_config_copy, self.metadata)
            self.set_dataset_config_and_initialize(self._export_config_copy, None, workers)

        else:
            config_editor.modify_dataset_config(self.dataset_config, self.metadata)

    except Exception:
        self.dataset_config = original_config
        self._export_config_copy = original_export_config
        self.logger.error("Error occured, reverting changes.")
        raise

    if self.train_dataloader is not None:
        del self.train_dataloader
        self.train_dataloader = None
        self.logger.info("Destroyed cached train_dataloader.")

    if self.val_dataloader is not None:
        del self.val_dataloader
        self.val_dataloader = None
        self.logger.info("Destroyed cached val_dataloader.")

    if self.test_dataloader is not None:
        del self.test_dataloader
        self.test_dataloader = None
        self.logger.info("Destroyed cached test_dataloader.")

    if self.all_dataloader is not None:
        del self.all_dataloader
        self.all_dataloader = None
        self.logger.info("Destroyed cached all_dataloader.")

    self._update_config_imported_status(None)
    self._update_export_config_copy()

    self.logger.info("Configuration has been changed successfuly.")

    if display_config_details is not None:
        self.summary(display_config_details)

apply_filler

apply_filler(fill_missing_with: type | FillerType | Literal['mean_filler', 'forward_filler', 'linear_interpolation_filler'] | None, workers: int | Literal['config'] = 'config') -> None

Used for updating filler set in config. Set parameter to config to keep it as it is config. If exception is thrown during set, no changes are made.

Affects following configuration:

Dataset config	Description
`fill_missing_with`	Defines how to fill missing values in the dataset.

Parameters:

Name	Type	Description	Default
`fill_missing_with`	`type \| FillerType \| Literal['mean_filler', 'forward_filler', 'linear_interpolation_filler'] \| None`	Defines how to fill missing values in the dataset. `Defaults: config`.	required
`workers`	`int \| Literal['config']`	How many workers to use when setting new filler. `Defaults: config`.	`'config'`

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def apply_filler(self, fill_missing_with: type | FillerType | Literal["mean_filler", "forward_filler", "linear_interpolation_filler"] | None, workers: int | Literal["config"] = "config") -> None:
    """Used for updating filler set in config.
    Set parameter to `config` to keep it as it is config.
    If exception is thrown during set, no changes are made.

    Affects following configuration: 

    Dataset config | Description
    -------------- | -----------
    `fill_missing_with` | Defines how to fill missing values in the dataset.

    Parameters:
        fill_missing_with: Defines how to fill missing values in the dataset. `Defaults: config`.  
        workers: How many workers to use when setting new filler. `Defaults: config`.      
    """
    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before updating filler.")

    self.update_dataset_config_and_initialize(fill_missing_with=fill_missing_with, workers=workers)
    self.logger.info("Filler has been changed successfuly.")

apply_anomaly_handler

apply_anomaly_handler(handle_anomalies_with: type | AnomalyHandlerType | Literal['z-score', 'interquartile_range'] | None | Literal['config'], workers: int | Literal['config'] = 'config') -> None

Used for updating anomaly handler set in config.

Set parameter to config to keep it as it is config.

If exception is thrown during set, no changes are made.

Affects following configuration:

Dataset config	Description
`handle_anomalies_with`	Defines the anomaly handler to handle anomalies in the dataset.

Parameters:

Name	Type	Description	Default
`handle_anomalies_with`	`type \| AnomalyHandlerType \| Literal['z-score', 'interquartile_range'] \| None \| Literal['config']`	Defines the anomaly handler to handle anomalies in the dataset. `Defaults: config`.	required
`workers`	`int \| Literal['config']`	How many workers to use when setting new filler. `Defaults: config`.	`'config'`

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def apply_anomaly_handler(self, handle_anomalies_with: type | AnomalyHandlerType | Literal["z-score", "interquartile_range"] | None | Literal["config"], workers: int | Literal["config"] = "config") -> None:
    """Used for updating anomaly handler set in config.

    Set parameter to `config` to keep it as it is config.

    If exception is thrown during set, no changes are made.

    Affects following configuration:

    Dataset config | Description
    -------------- | -----------
    `handle_anomalies_with` | Defines the anomaly handler to handle anomalies in the dataset.

    Parameters:
        handle_anomalies_with: Defines the anomaly handler to handle anomalies in the dataset. `Defaults: config`.  
        workers: How many workers to use when setting new filler. `Defaults: config`.      
    """
    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before updating anomaly handler.")

    self.update_dataset_config_and_initialize(handle_anomalies_with=handle_anomalies_with, workers=workers)
    self.logger.info("Anomaly handler has been changed successfuly.")

set_default_values

set_default_values(default_values: list[Number] | NDArray[number] | dict[str, Number] | Number | Literal['default'] | None, workers: int | Literal['config'] = 'config') -> None

Used for updating default values set in config. Set parameter to config to keep it as it is config. If exception is thrown during set, no changes are made.

Affects following configuration:

Dataset config	Description
`default_values`	Default values for missing data, applied before fillers. Can set one value for all features or specify for each feature.

Parameters:

Name	Type	Description	Default
`default_values`	`list[Number] \| NDArray[number] \| dict[str, Number] \| Number \| Literal['default'] \| None`	Default values for missing data, applied before fillers. `Defaults: config`.	required
`workers`	`int \| Literal['config']`	How many workers to use when setting new default values. `Defaults: config`.	`'config'`

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def set_default_values(self, default_values: list[Number] | npt.NDArray[np.number] | dict[str, Number] | Number | Literal["default"] | None, workers: int | Literal["config"] = "config") -> None:
    """Used for updating default values set in config.
    Set parameter to `config` to keep it as it is config.
    If exception is thrown during set, no changes are made.

    Affects following configuration:

    Dataset config | Description
    -------------- | -----------
    `default_values` | Default values for missing data, applied before fillers. Can set one value for all features or specify for each feature.

    Parameters:
        default_values: Default values for missing data, applied before fillers. `Defaults: config`.  
        workers: How many workers to use when setting new default values. `Defaults: config`.      
    """
    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before updating default values.")

    self.update_dataset_config_and_initialize(default_values=default_values, workers=workers)
    self.logger.info("Default values has been changed successfuly.")

set_preprocess_order

set_preprocess_order(preprocess_order: list[str, type] | Literal['config'] = 'config', workers: int | Literal['config'] = 'config') -> None

Used for updating preprocess_order set in config. Set parameter to config to keep it as it is config. If exception is thrown during set, no changes are made.

Affects following configuration:

Dataset config	Description
`preprocess_order`	Used order of when preprocesses are applied. Can be also used to add/remove custom handlers.

Parameters:

Name	Type	Description	Default
`preprocess_order`	`list[str, type] \| Literal['config']`	Used order of when preprocesses are applied. Can be also used to add/remove custom handlers. `Defaults: config`.	`'config'`
`workers`	`int \| Literal['config']`	How many workers to use when setting new default values. `Defaults: config`.	`'config'`

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def set_preprocess_order(self, preprocess_order: list[str, type] | Literal["config"] = "config", workers: int | Literal["config"] = "config") -> None:
    """Used for updating preprocess_order set in config.
    Set parameter to `config` to keep it as it is config.
    If exception is thrown during set, no changes are made.

    Affects following configuration: 

    Dataset config | Description
    -------------- | -----------
    `preprocess_order` | Used order of when preprocesses are applied. Can be also used to add/remove custom handlers.

    Parameters:
        preprocess_order: Used order of when preprocesses are applied. Can be also used to add/remove custom handlers. `Defaults: config`.  
        workers: How many workers to use when setting new default values. `Defaults: config`.      
    """
    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized, use set_dataset_config_and_initialize() before updating preprocess order.")

    self.update_dataset_config_and_initialize(preprocess_order=preprocess_order, workers=workers)
    self.logger.info("Preprocess order has been changed successfuly.")

display_dataset_details

display_dataset_details() -> None

Display information about the contents of the dataset.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

    def display_dataset_details(self) -> None:
        """Display information about the contents of the dataset.  """

        to_display = f'''
Dataset details:

    {self.metadata.aggregation}
        Time indices: {range(self.metadata.time_indices[ID_TIME_COLUMN_NAME][0], self.metadata.time_indices[ID_TIME_COLUMN_NAME][-1])}
        Datetime: {(datetime.fromtimestamp(self.metadata.time_indices['time'][0], tz=timezone.utc), datetime.fromtimestamp(self.metadata.time_indices['time'][-1], timezone.utc))}

    {self.metadata.source_type}
        Time series indices: {get_abbreviated_list_string(self.metadata.ts_indices[self.metadata.ts_id_name])}; use 'get_available_ts_indices' for full list
        Features with default values: {self.metadata.default_values}

        Additional data: {list(self.metadata.additional_data.keys())}
        '''

        print(to_display)

summary

summary(display_type: Literal['text', 'diagram']) -> None

Used to display used configurations. Can be displayed as interactive html diagram or text summary.

Parameters:

Name	Type	Description	Default
`display_type`	`Literal['text', 'diagram']`	Whether configuration should be display as diagram or text summary.	required

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def summary(self, display_type: Literal["text", "diagram"]) -> None:
    """Used to display used configurations. Can be displayed as interactive html diagram or text summary.

    Parameters:
        display_type: Whether configuration should be display as diagram or text summary.
    """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to display summary.")

    display_type = DisplayType(display_type)

    if display_type == DisplayType.TEXT:
        print(self.dataset_config)
    elif display_type == DisplayType.DIAGRAM:
        steps = self.dataset_config._get_summary_steps()
        return css_utils.display_summary_diagram(steps)
    else:
        raise NotImplementedError()

save_summary_diagram_as_html

save_summary_diagram_as_html(path: str)

Saves diagram produces from summary method as html file to specified path.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def save_summary_diagram_as_html(self, path: str):
    """Saves diagram produces from `summary` method as html file to specified path. """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to save summary diagram.")

    steps = self.dataset_config._get_summary_steps()
    html = css_utils.get_summary_diagram(steps)

    with open(path, "w", encoding="utf-8") as f:
        f.write(html)

get_feature_names

get_feature_names() -> list[str]

Returns a list of all available feature names in the dataset.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_feature_names(self) -> list[str]:
    """Returns a list of all available feature names in the dataset. """

    return list(self.metadata.features.keys())

get_available_ts_indices

get_available_ts_indices() -> np.ndarray

Returns the available time series indices in this dataset.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_available_ts_indices(self) -> np.ndarray:
    """Returns the available time series indices in this dataset. """
    return self.metadata.ts_indices

get_additional_data

get_additional_data(data_name: str) -> pd.DataFrame

Create a Pandas DataFrame of additional data of data_name.

Parameters:

Name	Type	Description	Default
`data_name`	`str`	Name of additional data to return.	required

Returns:

Type	Description
`DataFrame`	Dataframe of additional data of `data_name`.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_additional_data(self, data_name: str) -> pd.DataFrame:
    """Create a Pandas [`DataFrame`](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html) of additional data of `data_name`.

    Parameters:
        data_name: Name of additional data to return.

    Returns:
        Dataframe of additional data of `data_name`.
    """

    if data_name not in self.metadata.additional_data:
        self.logger.error("%s is not available for this dataset.", data_name)
        raise ValueError(f"{data_name} is not available for this dataset.", f"Possible options are: {self.metadata.additional_data}")

    data = get_additional_data(self.metadata.dataset_path, data_name)
    data_df = pd.DataFrame(data)

    for column, column_type in self.metadata.additional_data[data_name]:
        if column_type == datetime:
            data_df[column] = data_df[column].apply(lambda x: datetime.fromtimestamp(x, tz=timezone.utc))
        else:
            data_df[column] = data_df[column].astype(column_type)

    return data_df

plot

plot(ts_id: int, plot_type: Literal['scatter', 'line'], features: list[str] | str | Literal['config'] = 'config', feature_per_plot: bool = True, time_format: TimeFormat | Literal['config', 'id_time', 'datetime', 'unix_time', 'shifted_unix_time'] = 'config', is_interactive: bool = True) -> None

Displays a graph for the selected ts_id and its features.

The plotting is done using the Plotly library, which provides interactive graphs.

Parameters:

Name	Type	Description	Default
`ts_id`	`int`	The ID of the time series to display.	required
`plot_type`	`Literal['scatter', 'line']`	The type of graph to plot.	required
`features`	`list[str] \| str \| Literal['config']`	The features to display in the plot. `Defaults: "config"`.	`'config'`
`feature_per_plot`	`bool`	Whether each feature should be displayed in a separate plot or combined into one. `Defaults: True`.	`True`
`time_format`	`TimeFormat \| Literal['config', 'id_time', 'datetime', 'unix_time', 'shifted_unix_time']`	The time format to use for the x-axis. `Defaults: "config"`.	`'config'`
`is_interactive`	`bool`	Whether the plot should be interactive (e.g., zoom, hover). `Defaults: True`.	`True`

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def plot(self, ts_id: int, plot_type: Literal["scatter", "line"], features: list[str] | str | Literal["config"] = "config", feature_per_plot: bool = True,
         time_format: TimeFormat | Literal["config", "id_time", "datetime", "unix_time", "shifted_unix_time"] = "config", is_interactive: bool = True) -> None:
    """
    Displays a graph for the selected `ts_id` and its `features`.

    The plotting is done using the [`Plotly`](https://plotly.com/python/) library, which provides interactive graphs.

    Parameters:
        ts_id: The ID of the time series to display.
        plot_type: The type of graph to plot.
        features: The features to display in the plot. `Defaults: "config"`.
        feature_per_plot: Whether each feature should be displayed in a separate plot or combined into one. `Defaults: True`.
        time_format: The time format to use for the x-axis. `Defaults: "config"`.
        is_interactive: Whether the plot should be interactive (e.g., zoom, hover). `Defaults: True`.
    """

    if time_format == "config":

        if self.dataset_config is None or not self.dataset_config.is_initialized:
            raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to plot.")

        time_format = self.dataset_config.time_format
        self.logger.debug("Using time format from dataset configuration: %s", time_format)
    else:
        time_format = TimeFormat(time_format)
        self.logger.debug("Using specified time format: %s", time_format)

    time_series, times, features = self.__get_data_for_plot(ts_id, features, time_format)
    self.logger.debug("Received data for plotting. Time series, times, and features are ready.")

    plots = []

    if feature_per_plot:
        self.logger.debug("Creating individual plots for each feature.")
        fig = make_subplots(rows=len(features), cols=1, shared_xaxes=False, x_title=time_format.value)

        for i, feature in enumerate(features):
            if plot_type == "scatter":
                plot = go.Scatter(x=times, y=time_series[:, i], mode="markers", name=feature, legendgroup=feature)
                self.logger.debug("Creating scatter plot for feature: %s", feature)
            elif plot_type == "line":
                plot = go.Scatter(x=times, y=time_series[:, i], mode="lines", name=feature)
                self.logger.debug("Creating line plot for feature: %s", feature)
            else:
                raise ValueError("Invalid plot type.")

            fig.add_traces(plot, rows=i + 1, cols=1)

        fig.update_layout(height=200 + 120 * len(features), width=2000, autosize=len(features) == 1, showlegend=True)
        self.logger.debug("Created subplots for features: %s.", features)
    else:
        self.logger.debug("Creating a combined plot for all features.")
        for i, feature in enumerate(features):
            if plot_type == "scatter":
                plot = go.Scatter(x=times, y=time_series[:, i], mode="markers", name=feature)
                self.logger.debug("Creating scatter plot for feature: %s", feature)
            elif plot_type == "line":
                plot = go.Scatter(x=times, y=time_series[:, i], mode="lines", name=feature)
                self.logger.debug("Creating line plot for feature: %s", feature)
            else:
                raise ValueError("Invalid plot type.")
            plots.append(plot)

        fig = go.Figure(data=plots)
        fig.update_layout(xaxis_title=time_format.value, showlegend=True, height=200 + 120 * 2)
        self.logger.debug("Created combined plot for features: %s.", features)

    if not is_interactive:
        self.logger.debug("Disabling interactivity for the plot.")
        fig.update_layout(updatemenus=[], dragmode=False, hovermode=False)

    self.logger.debug("Displaying the plot.")
    fig.show()

add_annotation

add_annotation(annotation: str, annotation_group: str, ts_id: int | None, id_time: int | None, enforce_ids: bool = True) -> None

Adds an annotation to the specified annotation_group.

If the provided annotation_group does not exist, it will be created.
At least one of ts_id or id_time must be provided to associate the annotation with time series or/and time point.

Parameters:

Name	Type	Description	Default
`annotation`	`str`	The annotation to be added.	required
`annotation_group`	`str`	The group to which the annotation should be added.	required
`ts_id`	`int \| None`	The time series ID to which the annotation should be added.	required
`id_time`	`int \| None`	The time ID to which the annotation should be added.	required
`enforce_ids`	`bool`	Flag indicating whether the `ts_id` and `id_time` must belong to this dataset. `Default: True`	`True`

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def add_annotation(self, annotation: str, annotation_group: str, ts_id: int | None, id_time: int | None, enforce_ids: bool = True) -> None:
    """ 
    Adds an annotation to the specified `annotation_group`.

    - If the provided `annotation_group` does not exist, it will be created.
    - At least one of `ts_id` or `id_time` must be provided to associate the annotation with time series or/and time point.

    Parameters:
        annotation: The annotation to be added.
        annotation_group: The group to which the annotation should be added.
        ts_id: The time series ID to which the annotation should be added.
        id_time: The time ID to which the annotation should be added.
        enforce_ids: Flag indicating whether the `ts_id` and `id_time` must belong to this dataset. `Default: True`  
    """

    if enforce_ids:
        self._validate_annotation_ids(ts_id, id_time)
    self.annotations.add_annotation(annotation, annotation_group, ts_id, id_time)

    if ts_id is not None and id_time is not None:
        self._update_annotations_imported_status(AnnotationType.BOTH, None)
    elif ts_id is not None and id_time is None:
        self._update_annotations_imported_status(AnnotationType.TS_ID, None)
    elif ts_id is None and id_time is not None:
        self._update_annotations_imported_status(AnnotationType.ID_TIME, None)

remove_annotation

remove_annotation(annotation_group: str, ts_id: int | None, id_time: int | None) -> None

Removes an annotation from the specified annotation_group.

At least one of ts_id or id_time must be provided to associate the annotation with time series or/and time point.

Parameters:

Name	Type	Description	Default
`annotation_group`	`str`	The annotation group from which the annotation should be removed.	required
`ts_id`	`int \| None`	The time series ID from which the annotation should be removed.	required
`id_time`	`int \| None`	The time ID from which the annotation should be removed.	required

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def remove_annotation(self, annotation_group: str, ts_id: int | None, id_time: int | None) -> None:
    """  
    Removes an annotation from the specified `annotation_group`.

    - At least one of `ts_id` or `id_time` must be provided to associate the annotation with time series or/and time point.

    Parameters:
        annotation_group: The annotation group from which the annotation should be removed.
        ts_id: The time series ID from which the annotation should be removed.
        id_time: The time ID from which the annotation should be removed. 
    """

    self.annotations.remove_annotation(annotation_group, ts_id, id_time, False)

    if ts_id is not None and id_time is not None:
        self._update_annotations_imported_status(AnnotationType.BOTH, None)
    elif ts_id is not None and id_time is None:
        self._update_annotations_imported_status(AnnotationType.TS_ID, None)
    elif ts_id is None and id_time is not None:
        self._update_annotations_imported_status(AnnotationType.ID_TIME, None)

add_annotation_group

add_annotation_group(annotation_group: str, on: AnnotationType | Literal['id_time', 'ts_id', 'both'])

Adds a new annotation_group.

Parameters:

Name	Type	Description	Default
`annotation_group`	`str`	The name of the annotation group to be added.	required
`on`	`AnnotationType \| Literal['id_time', 'ts_id', 'both']`	Specifies which part of the data should be annotated. If set to `"both"`, annotations will be applied as if `id_time` and `ts_id` were both set.	required

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def add_annotation_group(self, annotation_group: str, on: AnnotationType | Literal["id_time", "ts_id", "both"]):
    """ 
    Adds a new `annotation_group`.

    Parameters:
        annotation_group: The name of the annotation group to be added.
        on: Specifies which part of the data should be annotated. If set to `"both"`, annotations will be applied as if `id_time` and `ts_id` were both set.
    """
    on = AnnotationType(on)

    self.annotations.add_annotation_group(annotation_group, on, False)

    self._update_annotations_imported_status(on, None)

remove_annotation_group

remove_annotation_group(annotation_group: str, on: AnnotationType | Literal['id_time', 'ts_id', 'both'])

Removes the specified annotation_group.

Parameters:

Name	Type	Description	Default
`annotation_group`	`str`	The name of the annotation group to be removed.	required
`on`	`AnnotationType \| Literal['id_time', 'ts_id', 'both']`	Specifies which part of the data the `annotation_group` should be removed from. If set to `"both"`, annotations will be applied as if `id_time` and `ts_id` were both set.	required

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def remove_annotation_group(self, annotation_group: str, on: AnnotationType | Literal["id_time", "ts_id", "both"]):
    """ 
    Removes the specified `annotation_group`.

    Parameters:
        annotation_group: The name of the annotation group to be removed.
        on: Specifies which part of the data the `annotation_group` should be removed from. If set to `"both"`, annotations will be applied as if `id_time` and `ts_id` were both set.        
    """
    on = AnnotationType(on)

    self.annotations.remove_annotation_group(annotation_group, on, False)

    self._update_annotations_imported_status(on, None)

get_annotations

get_annotations(on: AnnotationType | Literal['id_time', 'ts_id', 'both']) -> pd.DataFrame

Returns the annotations as a Pandas DataFrame.

Parameters:

Name	Type	Description	Default
`on`	`AnnotationType \| Literal['id_time', 'ts_id', 'both']`	Specifies which annotations to return. If set to `"both"`, annotations will be applied as if `id_time` and `ts_id` were both set.	required

Returns:

Type	Description
`DataFrame`	A Pandas DataFrame containing the selected annotations.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_annotations(self, on: AnnotationType | Literal["id_time", "ts_id", "both"]) -> pd.DataFrame:
    """ 
    Returns the annotations as a Pandas [`DataFrame`](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html).

    Parameters:
        on: Specifies which annotations to return. If set to `"both"`, annotations will be applied as if `id_time` and `ts_id` were both set.         

    Returns:
        A Pandas DataFrame containing the selected annotations.      
    """
    on = AnnotationType(on)

    return self.annotations.get_annotations(on, self.metadata.ts_id_name)

import_annotations

import_annotations(identifier: str, enforce_ids: bool = True) -> None

Imports annotations from a CSV file.

First, it attempts to load the built-in annotations, if no built-in annotations with such an identifier exists, it attempts to load a custom annotations from the "data_root"/tszoo/annotations/ directory.

data_root is specified when the dataset is created.

Parameters:

Name	Type	Description	Default
`identifier`	`str`	The name of the CSV file.	required
`enforce_ids`	`bool`	Flag indicating whether the `ts_id` and `id_time` must belong to this dataset. `Default: True`	`True`

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def import_annotations(self, identifier: str, enforce_ids: bool = True) -> None:
    """ 
    Imports annotations from a CSV file.

    First, it attempts to load the built-in annotations, if no built-in annotations with such an identifier exists, it attempts to load a custom annotations from the `"data_root"/tszoo/annotations/` directory.

    `data_root` is specified when the dataset is created.     

    Parameters:
        identifier: The name of the CSV file.     
        enforce_ids: Flag indicating whether the `ts_id` and `id_time` must belong to this dataset. `Default: True`                
    """

    annotations_file_path, is_built_in = get_annotations_path_and_whether_it_is_built_in(identifier, self.metadata.annotations_root, self.logger)

    if is_built_in:
        self.logger.info("Built-in annotations found: %s.", identifier)
        if not os.path.exists(annotations_file_path):
            self.logger.info("Downloading annotations with identifier: %s", identifier)
            annotations_url = f"{ANNOTATIONS_DOWNLOAD_BUCKET}&file={identifier}"  # probably will change annotations bucket... placeholder
            resumable_download(url=annotations_url, file_path=annotations_file_path, silent=False)

        self.logger.debug("Loading annotations from %s", annotations_file_path)
        temp_df = pd.read_csv(annotations_file_path)
        self.logger.debug("Created DataFrame from file: %s", annotations_file_path)
    else:
        self.logger.info("Custom annotations found: %s.", identifier)
        self.logger.debug("Loading annotations from %s", annotations_file_path)
        temp_df = pd.read_csv(annotations_file_path)
        self.logger.debug("Created DataFrame from file: %s", annotations_file_path)

    ts_id_index = None
    time_id_index = None
    on = None

    # Check the columns of the DataFrame to identify the type of annotation
    if self.metadata.ts_id_name in temp_df.columns and ID_TIME_COLUMN_NAME in temp_df.columns:
        self.annotations.clear_time_in_time_series()
        time_id_index = temp_df.columns.tolist().index(ID_TIME_COLUMN_NAME)
        ts_id_index = temp_df.columns.tolist().index(self.metadata.ts_id_name)
        on = AnnotationType.BOTH
        self.logger.info("Annotations detected as %s (both %s and id_time)", AnnotationType.BOTH, self.metadata.ts_id_name)

    elif self.metadata.ts_id_name in temp_df.columns:
        self.annotations.clear_time_series()
        ts_id_index = temp_df.columns.tolist().index(self.metadata.ts_id_name)
        on = AnnotationType.TS_ID
        self.logger.info("Annotations detected as %s (%s only)", AnnotationType.TS_ID, self.metadata.ts_id_name)

    elif ID_TIME_COLUMN_NAME in temp_df.columns:
        self.annotations.clear_time()
        time_id_index = temp_df.columns.tolist().index(ID_TIME_COLUMN_NAME)
        on = AnnotationType.ID_TIME
        self.logger.info("Annotations detected as %s (%s only)", AnnotationType.ID_TIME, ID_TIME_COLUMN_NAME)

    else:
        raise ValueError(f"Could not find {self.metadata.ts_id_name} and {ID_TIME_COLUMN_NAME} in the imported CSV.")

    # Process each row in the DataFrame and add annotations
    for row in temp_df.itertuples(False):
        for i, _ in enumerate(temp_df.columns):
            if i == time_id_index or i == ts_id_index:
                continue

            ts_id = None
            if ts_id_index is not None:
                ts_id = row[ts_id_index]

            id_time = None
            if time_id_index is not None:
                id_time = row[time_id_index]

            self.add_annotation(row[i], temp_df.columns[i], ts_id, id_time, enforce_ids)

    self._update_annotations_imported_status(on, identifier)
    self.logger.info("Successfully imported annotations from %s", annotations_file_path)

import_config

import_config(identifier: str, display_config_details: Optional[Literal['text', 'diagram']] = 'text', workers: int | Literal['config'] = 'config') -> None

Import the dataset_config from a pickle file and initializes the dataset. Config type must correspond to dataset type.

First, it attempts to load the built-in config, if no built-in config with such an identifier exists, it attempts to load a custom config from the "data_root"/tszoo/configs/ directory.

data_root is specified when the dataset is created.

The following configuration attributes are used during initialization:

Dataset config	Description
`init_workers`	Specifies the number of workers to use for initialization. Applied when `workers` = "config".
`partial_fit_initialized_transformers`	Determines whether initialized transformers should be partially fitted on the training data.
`nan_threshold`	Filters out time series with missing values exceeding the specified threshold.

Parameters:

Name	Type	Description	Default
`identifier`	`str`	Name of the pickle file.	required
`display_config_details`	`Optional[Literal['text', 'diagram']]`	Flag indicating whether to display the configuration values after initialization. `Default: True`	`'text'`
`workers`	`int \| Literal['config']`	The number of workers to use during initialization. `Default: "config"`	`'config'`

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def import_config(self, identifier: str, display_config_details: Optional[Literal["text", "diagram"]] = "text", workers: int | Literal["config"] = "config") -> None:
    """ 
    Import the dataset_config from a pickle file and initializes the dataset. Config type must correspond to dataset type.

    First, it attempts to load the built-in config, if no built-in config with such an identifier exists, it attempts to load a custom config from the `"data_root"/tszoo/configs/` directory.

    `data_root` is specified when the dataset is created.       

    The following configuration attributes are used during initialization:

    Dataset config | Description
    -------------- | -----------
    `init_workers` | Specifies the number of workers to use for initialization. Applied when `workers` = "config".
    `partial_fit_initialized_transformers` | Determines whether initialized transformers should be partially fitted on the training data.
    `nan_threshold` | Filters out time series with missing values exceeding the specified threshold.

    Parameters:
        identifier: Name of the pickle file.
        display_config_details: Flag indicating whether to display the configuration values after initialization. `Default: True` 
        workers: The number of workers to use during initialization. `Default: "config"`  
    """

    if display_config_details is not None:
        display_config_details = DisplayType(display_config_details)

    # Load config
    config = load_config(identifier, self.metadata.configs_root, self.metadata.database_name, self.metadata.source_type, self.metadata.aggregation, self.logger)

    self.logger.info("Initializing dataset configuration with the imported config.")
    self.set_dataset_config_and_initialize(config, display_config_details, workers)

    self._update_config_imported_status(identifier)
    self.logger.info("Successfully used config with identifier %s", identifier)

save_annotations

save_annotations(identifier: str, on: AnnotationType | Literal['id_time', 'ts_id', 'both'], force_write: bool = False) -> None

Saves the annotations as a CSV file.

The file will be saved to a path determined by the data_root specified when the dataset was created.

The annotations will be saved under the directory data_root/tszoo/annotations/.

Parameters:

Name	Type	Description	Default
`identifier`	`str`	The name of the CSV file.	required
`on`	`AnnotationType \| Literal['id_time', 'ts_id', 'both']`	What annotation type should be saved. If set to `"both"`, annotations will be applied as if `id_time` and `ts_id` were both set.	required
`force_write`	`bool`	If set to `True`, will overwrite any existing files with the same name. `Default: False`	`False`

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def save_annotations(self, identifier: str, on: AnnotationType | Literal["id_time", "ts_id", "both"], force_write: bool = False) -> None:
    """ 
    Saves the annotations as a CSV file.

    The file will be saved to a path determined by the `data_root` specified when the dataset was created.

    The annotations will be saved under the directory `data_root/tszoo/annotations/`.

    Parameters:
        identifier: The name of the CSV file.
        on: What annotation type should be saved. If set to `"both"`, annotations will be applied as if `id_time` and `ts_id` were both set.   
        force_write: If set to `True`, will overwrite any existing files with the same name. `Default: False`               
    """

    if exists_built_in_annotations(identifier):
        raise ValueError("Built-in annotations with this identifier already exists. Choose another identifier.")

    on = AnnotationType(on)

    temp_df = self.get_annotations(on)

    # Ensure the annotations root directory exists, creating it if necessary
    if not os.path.exists(self.metadata.annotations_root):
        os.makedirs(self.metadata.annotations_root)
        self.logger.info("Created annotations directory at %s", self.metadata.annotations_root)

    path = os.path.join(self.metadata.annotations_root, f"{identifier}.csv")

    if os.path.exists(path) and not force_write:
        raise ValueError(f"Annotations already exist at {path}. Set force_write=True to overwrite.")
    self.logger.debug("Annotations CSV file path: %s", path)

    temp_df.to_csv(path, index=False)

    self._update_annotations_imported_status(on, identifier)
    self.logger.info("Annotations successfully saved to %s", path)

save_config

save_config(identifier: str, create_with_details_file: bool = True, force_write: bool = False, **kwargs) -> None

Saves the config as a pickle file.

The file will be saved to a path determined by the data_root specified when the dataset was created. The config will be saved under the directory data_root/tszoo/configs/.

Parameters:

Name	Type	Description	Default
`identifier`	`str`	The name of the pickle file.	required
`create_with_details_file`	`bool`	Whether to export the config along with a readable text file that provides details. `Defaults: True`.	`True`
`force_write`	`bool`	If set to `True`, will overwrite any existing files with the same name. `Default: False`	`False`

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def save_config(self, identifier: str, create_with_details_file: bool = True, force_write: bool = False, **kwargs) -> None:
    """ 
    Saves the config as a pickle file.

    The file will be saved to a path determined by the `data_root` specified when the dataset was created. 
    The config will be saved under the directory `data_root/tszoo/configs/`.

    Parameters:
        identifier: The name of the pickle file.
        create_with_details_file: Whether to export the config along with a readable text file that provides details. `Defaults: True`. 
        force_write: If set to `True`, will overwrite any existing files with the same name. `Default: False`            
    """

    default_kwargs = {'hard_force': False}
    kwargs = {**default_kwargs, **kwargs}

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to save config.")

    if not kwargs["hard_force"] and exists_built_in_config(identifier):
        raise ValueError("Built-in config with this identifier already exists. Choose another identifier.")

    # Ensure the config directory exists
    if not os.path.exists(self.metadata.configs_root):
        os.makedirs(self.metadata.configs_root)
        self.logger.info("Created config directory at %s", self.metadata.configs_root)

    path_pickle = os.path.join(self.metadata.configs_root, f"{identifier}.pickle")
    path_details = os.path.join(self.metadata.configs_root, f"{identifier}.txt")

    if os.path.exists(path_pickle) and not force_write:
        raise ValueError(f"Config at path {path_pickle} already exists. Set force_write=True to overwrite.")
    self.logger.debug("Config pickle path: %s", path_pickle)

    if create_with_details_file:
        if os.path.exists(path_details) and not force_write:
            raise ValueError(f"Config details at path {path_details} already exists. Set force_write=True to overwrite.")
        self.logger.debug("Config details path: %s", path_details)

    if not self.dataset_config.filler_factory.creates_built_in:
        self.logger.warning("You are using a custom filler. Ensure the config is distributed with the source code of the filler.")

    if not self.dataset_config.anomaly_handler_factory.creates_built_in:
        self.logger.warning("You are using a custom anomaly handler. Ensure the config is distributed with the source code of the anomaly handler.")

    if not self.dataset_config.transformer_factory.creates_built_in:
        self.logger.warning("You are using a custom transformer. Ensure the config is distributed with the source code of the transformer.")

    if len(self.dataset_config.preprocess_order) != len(MANDATORY_PREPROCESSES_ORDER):
        self.logger.warning("You are using at least one custom handler. Ensure the config is distributed with the source code of every custom handler.")

    pickle_dump(self._export_config_copy, path_pickle)
    self.logger.info("Config pickle saved to %s", path_pickle)

    if create_with_details_file:
        with open(path_details, "w", encoding="utf-8") as file:
            file.write(str(self.dataset_config))
        self.logger.info("Config details saved to %s", path_details)

    self._update_config_imported_status(identifier)
    self.dataset_config.export_update_needed = False
    self.logger.info("Config successfully saved")

save_benchmark

save_benchmark(identifier: str, force_write: bool = False, **kwargs) -> None

Saves the benchmark as a YAML file.

The benchmark, along with any associated annotations and config files, will be saved in a path determined by the data_root specified when creating the dataset. The default save path for benchmark is "data_root/tszoo/benchmarks/".

If you are using imported annotations or config (whether custom or built-in), their file names will be set in the benchmark file. If new annotations or config are created during the process, their filenames will be derived from the provided identifier and set in the benchmark file.

Parameters:

Name	Type	Description	Default
`identifier`	`str`	The name of the YAML file.	required
`force_write`	`bool`	If set to `True`, will overwrite any existing files with the same name. `Default: False`	`False`

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def save_benchmark(self, identifier: str, force_write: bool = False, **kwargs) -> None:
    """ 
    Saves the benchmark as a YAML file.

    The benchmark, along with any associated annotations and config files, will be saved in a path determined by the `data_root` specified when creating the dataset. 
    The default save path for benchmark is `"data_root/tszoo/benchmarks/"`.

    If you are using imported `annotations` or `config` (whether custom or built-in), their file names will be set in the `benchmark` file. 
    If new `annotations` or `config` are created during the process, their filenames will be derived from the provided `identifier` and set in the `benchmark` file.

    Parameters:
        identifier: The name of the YAML file.
        force_write: If set to `True`, will overwrite any existing files with the same name. `Default: False`            
    """

    default_kwargs = {'hard_force': False}
    kwargs = {**default_kwargs, **kwargs}

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting to save benchmark.")

    if not kwargs["hard_force"] and exists_built_in_benchmark(identifier):
        raise ValueError("Built-in benchmark with this identifier already exists. Choose another identifier.")

    # Determine annotation names based on the available annotations and whether the annotations were imported
    if len(self.annotations.time_series_annotations) > 0:
        annotations_ts_name = self.imported_annotations_ts_identifier if self.imported_annotations_ts_identifier is not None else f"{identifier}_{AnnotationType.TS_ID.value}"
    else:
        annotations_ts_name = None

    if len(self.annotations.time_annotations) > 0:
        annotations_time_name = self.imported_annotations_time_identifier if self.imported_annotations_time_identifier is not None else f"{identifier}_{AnnotationType.ID_TIME.value}"
    else:
        annotations_time_name = None

    if len(self.annotations.time_in_series_annotations) > 0:
        annotations_both_name = self.imported_annotations_both_identifier if self.imported_annotations_both_identifier is not None else f"{identifier}_{AnnotationType.BOTH.value}"
    else:
        annotations_both_name = None

    # Use the imported identifier if available and update is not necessary, otherwise default to the current identifier
    config_name = self.dataset_config.import_identifier if (self.dataset_config.import_identifier is not None and not self.dataset_config.export_update_needed) else identifier

    export_benchmark = ExportBenchmark(self.metadata.database_name,
                                       self.metadata.source_type.value,
                                       self.metadata.aggregation.value,
                                       self.metadata.dataset_type.value,
                                       config_name,
                                       annotations_ts_name,
                                       annotations_time_name,
                                       annotations_both_name,
                                       related_results_identifier=self.related_to,
                                       version=version.config_and_benchmarks_current_version)

    # If the config was not imported, save it
    if self.dataset_config.import_identifier is None or self.dataset_config.export_update_needed:
        self.save_config(export_benchmark.config_identifier, force_write=force_write, hard_force=kwargs["hard_force"])
    else:
        self.logger.info("Using already existing config with identifier: %s", self.dataset_config.import_identifier)

    # Save ts_id annotations if available and not previously imported
    if self.imported_annotations_ts_identifier is None and len(self.annotations.time_series_annotations) > 0:
        self.save_annotations(export_benchmark.annotations_ts_identifier, AnnotationType.TS_ID, force_write=force_write)
    elif self.imported_annotations_ts_identifier is not None:
        self.logger.info("Using already existing annotations with identifier: %s; type: %s", self.imported_annotations_ts_identifier, AnnotationType.TS_ID)

    # Save id_time annotations if available and not previously imported
    if self.imported_annotations_time_identifier is None and len(self.annotations.time_annotations) > 0:
        self.save_annotations(export_benchmark.annotations_time_identifier, AnnotationType.ID_TIME, force_write=force_write)
    elif self.imported_annotations_time_identifier is not None:
        self.logger.info("Using already existing annotations with identifier: %s; type: %s", self.imported_annotations_time_identifier, AnnotationType.ID_TIME)

    # Save both annotations if available and not previously imported
    if self.imported_annotations_both_identifier is None and len(self.annotations.time_in_series_annotations) > 0:
        self.save_annotations(export_benchmark.annotations_both_identifier, AnnotationType.BOTH, force_write=force_write)
    elif self.imported_annotations_both_identifier is not None:
        self.logger.info("Using already existing annotations with identifier: %s; type: %s", self.imported_annotations_both_identifier, AnnotationType.BOTH)

    # Ensure the benchmark directory exists
    if not os.path.exists(self.metadata.benchmarks_root):
        os.makedirs(self.metadata.benchmarks_root)
        self.logger.info("Created benchmarks directory at %s", self.metadata.benchmarks_root)

    benchmark_path = os.path.join(self.metadata.benchmarks_root, f"{identifier}.yaml")

    if os.path.exists(benchmark_path) and not force_write:
        self.logger.error("Benchmark file already exists at %s", benchmark_path)
        raise ValueError(f"Benchmark at path {benchmark_path} already exists. Set force_write=True to overwrite.")
    self.logger.debug("Benchmark YAML file path: %s", benchmark_path)

    yaml_dump(export_benchmark.to_dict(), benchmark_path)
    self.logger.info("Benchmark successfully saved to %s", benchmark_path)

get_transformers

get_transformers() -> np.ndarray[Transformer] | Transformer | None

Returns used transformers from config.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def get_transformers(self) -> np.ndarray[Transformer] | Transformer | None:
    """Returns used transformers from config. """
    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before attempting get transformers.")

    for i, preprocess_type in enumerate(self.dataset_config.preprocess_order):
        if preprocess_type == PreprocessType.TRANSFORMING:
            holder: TransformerHolder = self.dataset_config.train_preprocess_order[i].holder
            return holder.transformers

    return None

check_errors

check_errors() -> None

Validates whether the dataset is corrupted.

Raises an exception if corrupted.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def check_errors(self) -> None:
    """
    Validates whether the dataset is corrupted. 

    Raises an exception if corrupted.
    """

    dataset, _ = load_database(self.metadata.dataset_path)

    try:
        node_iter = dataset.walk_nodes()

        # Process each node in the dataset
        for node in node_iter:
            if isinstance(node, tb.Table):

                iter_by = min(LOADING_WARNING_THRESHOLD, len(node))
                iters_done = 0

                # Process the node in chunks to avoid memory issues
                while iters_done < len(node):
                    iter_by = min(LOADING_WARNING_THRESHOLD, len(node) - iters_done)
                    _ = node[iters_done: iters_done + iter_by]  # Fetch the data in chunks
                    iters_done += iter_by

                self.logger.info("Table '%s' checked successfully. (%d rows processed)", node._v_pathname, len(node))

        self.logger.info("Dataset check completed with no errors found.")

    except Exception as e:
        self.logger.error("Error encountered during dataset check: %s", str(e))

    finally:
        dataset.close()
        self.logger.debug("Dataset connection closed.")

__get_data_for_plot

__get_data_for_plot(ts_id: int, features: list[str] | str, time_format: TimeFormat) -> tuple[np.ndarray, np.ndarray, list[str]]

Returns prepared data for plotting.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def __get_data_for_plot(self, ts_id: int, features: list[str] | str, time_format: TimeFormat) -> tuple[np.ndarray, np.ndarray, list[str]]:
    """Returns prepared data for plotting. """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before getting data for plotting.")

    features_indices = []

    if features == "config":
        features = deepcopy(self.dataset_config.features_to_take_without_ids)
        features_indices = np.arange(len(features))
        self.logger.debug("Features set from dataset config: %s", features)
    else:
        if isinstance(features, str):
            features = [features]

        if len(features) == 0:
            raise ValueError("No features specified to plot. Please provide valid features.")
        if len(set(features)) != len(features):
            raise ValueError("Duplicate features detected. All features must be unique.")

        for feature in features:
            if feature not in self.dataset_config.features_to_take_without_ids:
                raise ValueError(f"Feature '{feature}' is not valid. It is not present in the dataset configuration.", self.dataset_config.features_to_take_without_ids)

            index_in_config_features = self.dataset_config.features_to_take_without_ids.index(feature)
            features_indices.append(index_in_config_features)

    real_feature_indices = np.array(self.dataset_config.indices_of_features_to_take_no_ids)[features_indices]
    real_feature_indices = real_feature_indices.astype(int)

    time_series, time_period = self._get_data_for_plot(ts_id, real_feature_indices, time_format)
    self.logger.debug("Time series data and corresponding time values retrieved.")

    return time_series, time_period, features

_validate_annotation_ids

_validate_annotation_ids(ts_id: int | None, id_time: int | None) -> None

Validates whether the ts_id and id_time belong to this dataset.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def _validate_annotation_ids(self, ts_id: int | None, id_time: int | None) -> None:
    """Validates whether the `ts_id` and `id_time` belong to this dataset. """

    assert ts_id is not None or id_time is not None, "Either ts_id or id_time must be provided."

    # Handle when id_time is provided
    if id_time is not None:
        time_indices = self.metadata.time_indices
        if id_time < time_indices[ID_TIME_COLUMN_NAME][0] or id_time > time_indices[ID_TIME_COLUMN_NAME][-1]:
            valid_range = range(time_indices[ID_TIME_COLUMN_NAME][0], time_indices[ID_TIME_COLUMN_NAME][-1])
            raise ValueError(f"id_time {id_time} does not fall within the valid range for {self.metadata.aggregation}. "
                             f"Valid id_time range: {valid_range}.")

    # Handle when ts_id is provided
    if ts_id is not None:
        ts_indices = self.metadata.ts_indices[self.metadata.ts_id_name]

        if ts_id not in ts_indices:
            valid_ts_range = self.metadata.ts_indices[self.metadata.ts_id_name]
            raise ValueError(f"ts_id {ts_id} does not exist in the available range for {self.metadata.source_type}. "
                             f"Valid ts_id values: {valid_ts_range}.")

_get_df

_get_df(dataloader: DataLoader, as_single_dataframe: bool, ts_ids: ndarray, time_period: ndarray) -> pd.DataFrame

Returns all data from the DataLoader as a Pandas DataFrame.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def _get_df(self, dataloader: DataLoader, as_single_dataframe: bool, ts_ids: np.ndarray, time_period: np.ndarray) -> pd.DataFrame:
    """Returns all data from the DataLoader as a Pandas [`DataFrame`](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html). """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before getting DataFrame.")

    total_samples = len(ts_ids) * len(time_period)
    if total_samples >= LOADING_WARNING_THRESHOLD:
        self.logger.warning("The dataset contains %d samples (%d time series × %d times). Consider using get_*_dataloader() for batch loading.", total_samples, len(ts_ids), len(time_period))

    if as_single_dataframe:
        self.logger.debug("Returning a single DataFrame with all features for all time series.")
        return dataset_loaders.create_single_df_from_dataloader(
            dataloader,
            ts_ids,
            self.dataset_config.features_to_take,
            self.dataset_config.time_format,
            self.dataset_config.include_ts_id,
            self.dataset_config.include_time,
            self.dataset_config.dataset_type,
            True
        )
    else:
        self.logger.debug("Returning multiple DataFrames, one per time series.")
        return dataset_loaders.create_multiple_df_from_dataloader(
            dataloader,
            ts_ids,
            self.dataset_config.features_to_take,
            self.dataset_config.time_format,
            self.dataset_config.include_ts_id,
            self.dataset_config.include_time,
            self.dataset_config.dataset_type,
            True
        )

_get_numpy

_get_numpy(dataloader: DataLoader, ts_ids: ndarray, time_period: ndarray) -> np.ndarray

Returns all data from the DataLoader as a NumPy ndarray.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def _get_numpy(self, dataloader: DataLoader, ts_ids: np.ndarray, time_period: np.ndarray) -> np.ndarray:
    """Returns all data from the DataLoader as a NumPy `ndarray`. """

    if self.dataset_config is None or not self.dataset_config.is_initialized:
        raise ValueError("Dataset is not initialized. Please call set_dataset_config_and_initialize() before getting Numpy array.")

    total_samples = len(ts_ids) * len(time_period)
    if total_samples >= LOADING_WARNING_THRESHOLD:
        self.logger.warning("The dataset contains %d samples (%d time series × %d times). Consider using get_*_dataloader() for batch loading.", total_samples, len(ts_ids), len(time_period))

    self.logger.debug("Creating numpy array from dataloader.")
    return dataset_loaders.create_numpy_from_dataloader(
        dataloader,
        ts_ids,
        self.dataset_config.time_format,
        self.dataset_config.include_time,
        self.dataset_config.dataset_type,
        True
    )

_clear

_clear() -> None

Clears set data. Mainly called when initializing new config.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def _clear(self) -> None:
    """Clears set data. Mainly called when initializing new config. """
    self.train_dataset = None
    self.train_dataloader = None
    self.val_dataset = None
    self.val_dataloader = None
    self.test_dataset = None
    self.test_dataloader = None
    self.all_dataset = None
    self.all_dataloader = None
    self.dataset_config = None
    self.logger.debug("Dataset attributes had been cleared. ")

_update_annotations_imported_status

_update_annotations_imported_status(on: AnnotationType, identifier: str)

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def _update_annotations_imported_status(self, on: AnnotationType, identifier: str):
    if on == AnnotationType.TS_ID:
        self.imported_annotations_ts_identifier = identifier
    elif on == AnnotationType.ID_TIME:
        self.imported_annotations_time_identifier = identifier
    elif on == AnnotationType.BOTH:
        self.imported_annotations_both_identifier = identifier

_update_config_imported_status

_update_config_imported_status(identifier: str) -> None

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def _update_config_imported_status(self, identifier: str) -> None:
    self.dataset_config.import_identifier = identifier
    self._export_config_copy.import_identifier = identifier

_validate_config_for_dataset

_validate_config_for_dataset(config: DatasetConfig) -> bool

Validates whether config is supposed to be used for this dataset.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py

def _validate_config_for_dataset(self, config: DatasetConfig) -> bool:
    """Validates whether config is supposed to be used for this dataset. """

    if config.database_name != self.metadata.database_name:
        self.logger.error("This config is not compatible with the current dataset. Difference in database name between config and this dataset.")
        raise ValueError("This config is not compatible with the current dataset.", f"config.database_name == {config.database_name} and dataset.database_name == {self.metadata.database_name}")

    if config.dataset_type != self.metadata.dataset_type:
        self.logger.error("This config is not compatible with the current dataset. Difference in is_series_based between config and this dataset.")
        raise ValueError("This config is not compatible with the current dataset.", f"config.dataset_type == {config.dataset_type} and dataset.dataset_type == {self.metadata.dataset_type}")

    if config.aggregation != self.metadata.aggregation:
        self.logger.error("This config is not compatible with the current dataset. Difference in aggregation type between config and this dataset.")
        raise ValueError("This config is not compatible with the current dataset.", f"config.aggregation == {config.aggregation} and dataset.aggregation == {self.metadata.aggregation}")

    if config.source_type != self.metadata.source_type:
        self.logger.error("This config is not compatible with the current dataset. Difference in source type between config and this dataset.")
        raise ValueError("This config is not compatible with the current dataset.", f"config.source_type == {config.source_type} and dataset.source_type == {self.metadata.source_type}")

Disjoint-time-based dataset class

cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset dataclass

dataset_config class-attribute instance-attribute

train_dataset class-attribute instance-attribute

val_dataset class-attribute instance-attribute

test_dataset class-attribute instance-attribute

train_dataloader class-attribute instance-attribute

val_dataloader class-attribute instance-attribute

test_dataloader class-attribute instance-attribute

dataloader_factory class-attribute instance-attribute

dataset_type class-attribute instance-attribute

_export_config_copy class-attribute instance-attribute

metadata instance-attribute

all_dataset class-attribute instance-attribute

all_dataloader class-attribute instance-attribute

related_to class-attribute instance-attribute

_collate_fn class-attribute instance-attribute

__init__

__post_init__

set_dataset_config_and_initialize

apply_transformer

update_dataset_config_and_initialize

get_data_about_set

set_sliding_window

set_batch_sizes

set_workers

_initialize_datasets

_initialize_transformers_and_details

__initialize_config_for_train_set

__initialize_config_for_non_fit_sets

_update_export_config_copy

_get_singular_time_series_dataset

_get_data_for_plot

__get_ts_data_for_plot

get_train_dataloader

get_val_dataloader

get_test_dataloader

get_all_dataloader

get_train_df

get_val_df

get_test_df

get_all_df

get_train_numpy

get_val_numpy

get_test_numpy

get_all_numpy

_update_dataset_config_and_initialize

apply_filler

apply_anomaly_handler

set_default_values

set_preprocess_order

display_dataset_details

summary

save_summary_diagram_as_html

get_feature_names

get_available_ts_indices

get_additional_data

plot

add_annotation

remove_annotation

add_annotation_group

remove_annotation_group

get_annotations

import_annotations

import_config

save_annotations

save_config

save_benchmark

get_transformers

check_errors

__get_data_for_plot

_validate_annotation_ids

_get_df

_get_numpy

_clear

_update_annotations_imported_status

_update_config_imported_status

_validate_config_for_dataset

cesnet_tszoo.datasets.disjoint_time_based_cesnet_dataset.DisjointTimeBasedCesnetDataset `dataclass`

dataset_config `class-attribute` `instance-attribute`

train_dataset `class-attribute` `instance-attribute`

val_dataset `class-attribute` `instance-attribute`

test_dataset `class-attribute` `instance-attribute`

train_dataloader `class-attribute` `instance-attribute`

val_dataloader `class-attribute` `instance-attribute`

test_dataloader `class-attribute` `instance-attribute`

dataloader_factory `class-attribute` `instance-attribute`

dataset_type `class-attribute` `instance-attribute`

_export_config_copy `class-attribute` `instance-attribute`

metadata `instance-attribute`

all_dataset `class-attribute` `instance-attribute`

all_dataloader `class-attribute` `instance-attribute`

related_to `class-attribute` `instance-attribute`

_collate_fn `class-attribute` `instance-attribute`

init