Skip to content

Series-based dataset class

cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset dataclass

Bases: CesnetDataset

This class is used for series-based returning of data. Can be created by using get_dataset with parameter is_series_based = True.

Series-based means batch size affects number of returned time series in one batch. Which times for each time series are returned does not change.

The dataset provides multiple ways to access the data:

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

The dataset is stored in a PyTables database. The internal SeriesBasedDataset and SeriesBasedInitializerDataset 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 SeriesBasedConfig class.

Intended usage:

  1. 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.
  2. Create an instance of SeriesBasedConfig and set it using set_dataset_config_and_initialize. This initializes the dataset, including data splitting (train/validation/test), fitting scalers (if needed), selecting features, and more. This is cached for later use.
  3. Use get_train_dataloader/get_train_df/get_train_numpy to get training data for chosen model.
  4. Validate the model and perform the hyperparameter optimalization on get_val_dataloader/get_val_df/get_val_numpy.
  5. Evaluate the model on get_test_dataloader/get_test_df/get_test_numpy.

Alternatively you can use load_benchmark

  1. 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.
  2. Retrieve the initialized dataset using get_initialized_dataset. This will provide a dataset that is ready to use.
  3. Use get_train_dataloader/get_train_df/get_train_numpy to get training data for chosen model.
  4. Validate the model and perform the hyperparameter optimalization on get_val_dataloader/get_val_df/get_val_numpy.
  5. Evaluate the model on get_test_dataloader/get_test_df/get_test_numpy.

Parameters:

Name Type Description Default
database_name str

Name of the database.

 required
dataset_path str

Path to the dataset file.

required
configs_root str

Path to the folder where configurations are saved.

 required
benchmarks_root str

Path to the folder where benchmarks are saved.

 required
annotations_root str

Path to the folder where annotations are saved.

 required
source_type SourceType

The source type of the dataset.

 required
aggregation AgreggationType

The aggregation type for the selected source type.

 required
ts_id_name str

Name of the id used for time series.

 required
default_values dict

Default values for each available feature.

 required
additional_data dict[str, tuple]

Available small datasets. Can get them by calling get_additional_data with their name.

 required

Attributes:

Name Type Description
time_indices

Available time IDs for the dataset.
ts_indices

Available time series IDs for the dataset.
annotations

Annotations for the selected dataset.
logger

Logger for displaying information.

imported_annotations_ts_identifier

Identifier for the imported annotations of type AnnotationType.TS_ID.
imported_annotations_time_identifier

Identifier for the imported annotations of type AnnotationType.ID_TIME.
imported_annotations_both_identifier

Identifier for the imported annotations of type AnnotationType.BOTH.

The following attributes are initialized when set_dataset_config_and_initialize is called.

Attributes:

Name Type Description
dataset_config Optional[SeriesBasedConfig]

Configuration of the dataset.
train_dataset Optional[SeriesBasedDataset]

Training set as a SeriesBasedDataset instance wrapping the PyTables database.
val_dataset Optional[SeriesBasedDataset]

Validation set as a SeriesBasedDataset instance wrapping the PyTables database.
test_dataset Optional[SeriesBasedDataset]

Test set as a SeriesBasedDataset instance wrapping the PyTables database.
all_dataset Optional[SeriesBasedDataset]

All set as a SeriesBasedDataset instance wrapping the PyTables database.

train_dataloader Optional[DataLoader]

Iterable PyTorch DataLoader for training set.
val_dataloader Optional[DataLoader]

Iterable PyTorch DataLoader for validation set.
test_dataloader Optional[DataLoader]

Iterable PyTorch DataLoader for test set.
all_dataloader Optional[DataLoader]

Iterable PyTorch DataLoader for all set.
Source code in cesnet_tszoo\datasets\series_based_cesnet_dataset.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
@dataclass
class SeriesBasedCesnetDataset(CesnetDataset):
    """
    This class is used for series-based returning of data. Can be created by using [`get_dataset`][cesnet_tszoo.datasets.cesnet_database.CesnetDatabase.get_dataset] with parameter `is_series_based` = `True`.

    Series-based means batch size affects number of returned time series in one batch. Which times for each time series are returned does not change.

    The dataset provides multiple ways to access the data:

    - **Iterable PyTorch DataLoader**: For batch processing.
    - **Pandas DataFrame**: For loading the entire training, validation, test or all set at once.
    - **Numpy array**: For loading the entire training, validation, test or all 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 `SeriesBasedDataset` and `SeriesBasedInitializerDataset` classes (used only when calling [`set_dataset_config_and_initialize`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.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 [`SeriesBasedConfig`][cesnet_tszoo.configs.series_based_config.SeriesBasedConfig] class.     

    **Intended usage:**

    1. Create an instance of the dataset with the desired data root by calling [`get_dataset`][cesnet_tszoo.datasets.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 [`SeriesBasedConfig`][cesnet_tszoo.configs.series_based_config.SeriesBasedConfig] and set it using [`set_dataset_config_and_initialize`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.set_dataset_config_and_initialize]. 
       This initializes the dataset, including data splitting (train/validation/test), fitting scalers (if needed), selecting features, and more. This is cached for later use.
    3. Use [`get_train_dataloader`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_train_dataloader]/[`get_train_df`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_train_df]/[`get_train_numpy`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_train_numpy] to get training data for chosen model.
    4. Validate the model and perform the hyperparameter optimalization on [`get_val_dataloader`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_val_dataloader]/[`get_val_df`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_val_df]/[`get_val_numpy`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_val_numpy].
    5. Evaluate the model on [`get_test_dataloader`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_test_dataloader]/[`get_test_df`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_test_df]/[`get_test_numpy`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.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`][cesnet_tszoo.benchmarks.Benchmark.get_initialized_dataset]. This will provide a dataset that is ready to use.
    3. Use [`get_train_dataloader`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_train_dataloader]/[`get_train_df`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_train_df]/[`get_train_numpy`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_train_numpy] to get training data for chosen model.
    4. Validate the model and perform the hyperparameter optimalization on [`get_val_dataloader`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_val_dataloader]/[`get_val_df`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_val_df]/[`get_val_numpy`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_val_numpy].
    5. Evaluate the model on [`get_test_dataloader`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_test_dataloader]/[`get_test_df`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_test_df]/[`get_test_numpy`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_test_numpy].       

    Parameters:
        database_name: Name of the database.
        dataset_path: Path to the dataset file.     
        configs_root: Path to the folder where configurations are saved.
        benchmarks_root: Path to the folder where benchmarks are saved.
        annotations_root: Path to the folder where annotations are saved.
        source_type: The source type of the dataset.
        aggregation: The aggregation type for the selected source type.
        ts_id_name: Name of the id used for time series.
        default_values: Default values for each available feature.
        additional_data: Available small datasets. Can get them by calling [`get_additional_data`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.get_additional_data] with their name.

    Attributes:
        time_indices: Available time IDs for the dataset.
        ts_indices: Available time series IDs for the dataset.
        annotations: Annotations for the selected dataset.
        logger: Logger for displaying information.  
        imported_annotations_ts_identifier: Identifier for the imported annotations of type `AnnotationType.TS_ID`.
        imported_annotations_time_identifier: Identifier for the imported annotations of type `AnnotationType.ID_TIME`.
        imported_annotations_both_identifier: Identifier for the imported annotations of type `AnnotationType.BOTH`.  

    The following attributes are initialized when [`set_dataset_config_and_initialize`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.set_dataset_config_and_initialize] is called.

    Attributes:
        dataset_config: Configuration of the dataset.
        train_dataset: Training set as a `SeriesBasedDataset` instance wrapping the PyTables database.
        val_dataset: Validation set as a `SeriesBasedDataset` instance wrapping the PyTables database.
        test_dataset: Test set as a `SeriesBasedDataset` instance wrapping the PyTables database.
        all_dataset: All set as a `SeriesBasedDataset` instance wrapping the PyTables database.        
        train_dataloader: Iterable PyTorch [`DataLoader`](https://pytorch.org/docs/stable/data.html#torch.utils.data.DataLoader) for training set.
        val_dataloader: Iterable PyTorch [`DataLoader`](https://pytorch.org/docs/stable/data.html#torch.utils.data.DataLoader) for validation set.
        test_dataloader: Iterable PyTorch [`DataLoader`](https://pytorch.org/docs/stable/data.html#torch.utils.data.DataLoader) for test set.
        all_dataloader: Iterable PyTorch [`DataLoader`](https://pytorch.org/docs/stable/data.html#torch.utils.data.DataLoader) for all set.          
    """

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

    train_dataset: Optional[SeriesBasedDataset] = field(default=None, init=False)
    val_dataset: Optional[SeriesBasedDataset] = field(default=None, init=False)
    test_dataset: Optional[SeriesBasedDataset] = field(default=None, init=False)
    all_dataset: Optional[SeriesBasedDataset] = field(default=None, init=False)

    is_series_based: bool = field(default=True, init=False)

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

    def set_dataset_config_and_initialize(self, dataset_config: SeriesBasedConfig, display_config_details: bool = True, workers: int | Literal["config"] = "config") -> None:
        """
        Initialize training set, validation est, test set etc.. This method must be called before any data can be accessed. It is required for the final initialization of [`dataset_config`][cesnet_tszoo.configs.series_based_config.SeriesBasedConfig].

        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_scalers` | Determines whether initialized scalers 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 to display the configuration values after initialization. `Default: True`  
            workers: The number of workers to use during initialization. `Default: "config"`  
        """

        assert isinstance(dataset_config, SeriesBasedConfig), "SeriesBasedCesnetDataset can only use SeriesBasedConfig."

        super(SeriesBasedCesnetDataset, self).set_dataset_config_and_initialize(dataset_config, display_config_details, 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",
                                             train_batch_size: int | Literal["config"] = "config",
                                             val_batch_size: int | Literal["config"] = "config",
                                             test_batch_size: int | Literal["config"] = "config",
                                             all_batch_size: int | Literal["config"] = "config",
                                             fill_missing_with: type | FillerType | Literal["mean_filler", "forward_filler", "linear_interpolation_filler"] | None | Literal["config"] = "config",
                                             scale_with: type | list[Scaler] | np.ndarray[Scaler] | ScalerType | Scaler | Literal["min_max_scaler", "standard_scaler", "max_abs_scaler", "log_scaler", "robust_scaler", "power_transformer", "quantile_transformer", "l2_normalizer"] | None | Literal["config"] = "config",
                                             partial_fit_initialized_scalers: bool | Literal["config"] = "config",
                                             train_workers: int | Literal["config"] = "config",
                                             val_workers: int | Literal["config"] = "config",
                                             test_workers: int | Literal["config"] = "config",
                                             all_workers: int | Literal["config"] = "config",
                                             init_workers: int | Literal["config"] = "config",
                                             workers: int | Literal["config"] = "config",
                                             display_config_details: bool = False):
        """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.                        |           
        | `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.  |
        | `all_batch_size`                   | Number of samples per batch for all set. Affected by whether the dataset is series-based or time-based. Refer to relevant config for details.   |                   
        | `fill_missing_with`                | Defines how to fill missing values in the dataset.                                                                                              |     
        | `scale_with`                       | Defines the scaler to transform the dataset.                                                                                                    |      
        | `partial_fit_initialized_scalers`  | If `True`, partial fitting on train set is performed when using initiliazed scalers.                                                            |   
        | `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.                                                                                                        |
        | `all_workers`                      | Number of workers for loading all data.                                                                                                         |     
        | `init_workers`                     | Number of workers for dataset configuration.                                                                                                    |                        

        Parameters:
            default_values: Default values for missing data, applied before fillers. `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`.
            all_batch_size: Number of samples per batch for all set. `Defaults: config`.                    
            fill_missing_with: Defines how to fill missing values in the dataset. `Defaults: config`. 
            scale_with: Defines the scaler to transform the dataset. `Defaults: config`.  
            partial_fit_initialized_scalers: If `True`, partial fitting on train set is performed when using initiliazed scalers. `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`.
            all_workers: Number of workers for loading all 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`. 
        """

        return super(SeriesBasedCesnetDataset, self).update_dataset_config_and_initialize(default_values, "config", "config", "config", "config", train_batch_size, val_batch_size, test_batch_size, all_batch_size, fill_missing_with, scale_with, "config", partial_fit_initialized_scalers, train_workers, val_workers, test_workers, all_workers, init_workers, workers, display_config_details)

    def get_data_about_set(self, about: SplitType | Literal["train", "val", "test", "all"]) -> dict:
        """
        Retrieves 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 = self.dataset_config.time_period

        result = {}

        if about == SplitType.TRAIN:
            if not self.dataset_config.has_train:
                raise ValueError("Train set is not used.")
            ts_ids = self.dataset_config.train_ts
        elif about == SplitType.VAL:
            if not self.dataset_config.has_val:
                raise ValueError("Val set is not used.")
            ts_ids = self.dataset_config.val_ts
        elif about == SplitType.TEST:
            if not self.dataset_config.has_test:
                raise ValueError("Test set is not used.")
            ts_ids = self.dataset_config.test_ts
        elif about == SplitType.ALL:
            if not self.dataset_config.has_all:
                raise ValueError("All set is not used.")
            ts_ids = self.dataset_config.all_ts
        else:
            raise NotImplementedError("Should not happen")

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

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

        return result

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

        if self.dataset_config.has_train:
            self.train_dataset = SeriesBasedDataset(self.dataset_path,
                                                    self.dataset_config._get_table_data_path(),
                                                    self.dataset_config.ts_id_name,
                                                    self.dataset_config.train_ts_row_ranges,
                                                    self.dataset_config.time_period,
                                                    self.dataset_config.features_to_take,
                                                    self.dataset_config.indices_of_features_to_take_no_ids,
                                                    self.dataset_config.default_values,
                                                    self.dataset_config.train_fillers,
                                                    self.dataset_config.include_time,
                                                    self.dataset_config.include_ts_id,
                                                    self.dataset_config.time_format,
                                                    self.dataset_config.scalers)
            self.logger.debug("train_dataset initiliazed.")

        if self.dataset_config.has_val:
            self.val_dataset = SeriesBasedDataset(self.dataset_path,
                                                  self.dataset_config._get_table_data_path(),
                                                  self.dataset_config.ts_id_name,
                                                  self.dataset_config.val_ts_row_ranges,
                                                  self.dataset_config.time_period,
                                                  self.dataset_config.features_to_take,
                                                  self.dataset_config.indices_of_features_to_take_no_ids,
                                                  self.dataset_config.default_values,
                                                  self.dataset_config.val_fillers,
                                                  self.dataset_config.include_time,
                                                  self.dataset_config.include_ts_id,
                                                  self.dataset_config.time_format,
                                                  self.dataset_config.scalers)
            self.logger.debug("val_dataset initiliazed.")

        if self.dataset_config.has_test:
            self.test_dataset = SeriesBasedDataset(self.dataset_path,
                                                   self.dataset_config._get_table_data_path(),
                                                   self.dataset_config.ts_id_name,
                                                   self.dataset_config.test_ts_row_ranges,
                                                   self.dataset_config.time_period,
                                                   self.dataset_config.features_to_take,
                                                   self.dataset_config.indices_of_features_to_take_no_ids,
                                                   self.dataset_config.default_values,
                                                   self.dataset_config.test_fillers,
                                                   self.dataset_config.include_time,
                                                   self.dataset_config.include_ts_id,
                                                   self.dataset_config.time_format,
                                                   self.dataset_config.scalers)
            self.logger.debug("test_dataset initiliazed.")

        if self.dataset_config.has_all:
            self.all_dataset = SeriesBasedDataset(self.dataset_path,
                                                  self.dataset_config._get_table_data_path(),
                                                  self.dataset_config.ts_id_name,
                                                  self.dataset_config.all_ts_row_ranges,
                                                  self.dataset_config.time_period,
                                                  self.dataset_config.features_to_take,
                                                  self.dataset_config.indices_of_features_to_take_no_ids,
                                                  self.dataset_config.default_values,
                                                  self.dataset_config.all_fillers,
                                                  self.dataset_config.include_time,
                                                  self.dataset_config.include_ts_id,
                                                  self.dataset_config.time_format,
                                                  self.dataset_config.scalers)
            self.logger.debug("all_dataset initiliazed.")

    def _initialize_scalers_and_details(self, workers: int) -> None:
        """
        Called in [`set_dataset_config_and_initialize`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.set_dataset_config_and_initialize]. 

        Goes through data to validate time series against `nan_threshold`, partial fit `scalers` and prepare `fillers`.
        """

        init_dataset = SeriesBasedInitializerDataset(self.dataset_path,
                                                     self.dataset_config._get_table_data_path(),
                                                     self.dataset_config.ts_id_name,
                                                     self.dataset_config.train_ts_row_ranges,
                                                     self.dataset_config.val_ts_row_ranges,
                                                     self.dataset_config.test_ts_row_ranges,
                                                     self.dataset_config.all_ts_row_ranges,
                                                     self.dataset_config.time_period,
                                                     self.dataset_config.features_to_take,
                                                     self.dataset_config.indices_of_features_to_take_no_ids,
                                                     self.dataset_config.default_values,
                                                     self.dataset_config.all_fillers)

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

        if workers == 0:
            init_dataset.pytables_worker_init()

        train_ts_ids_to_take = []
        val_ts_ids_to_take = []
        test_ts_ids_to_take = []
        all_ts_ids_to_take = []

        self.logger.info("Updating config on train/val/test/all and selected time period.")
        for i, data in enumerate(tqdm(dataloader)):
            train_data, count_values, is_train, is_val, is_test, offsetted_idx = data[0]

            missing_percentage = count_values[1] / (count_values[0] + count_values[1])

            # Filter time series based on missing data threshold
            if missing_percentage <= self.dataset_config.nan_threshold:
                if is_train:
                    train_ts_ids_to_take.append(offsetted_idx)
                elif is_val:
                    val_ts_ids_to_take.append(offsetted_idx)
                elif is_test:
                    test_ts_ids_to_take.append(offsetted_idx)

                all_ts_ids_to_take.append(i)

                # Partial fit scaler on train data if applicable
                if self.dataset_config.scale_with is not None and is_train and (not self.dataset_config.are_scalers_premade or self.dataset_config.partial_fit_initialized_scalers):
                    self.dataset_config.scalers.partial_fit(train_data)

        if workers == 0:
            init_dataset.cleanup()

        # Update sets based on filtered time series
        if self.dataset_config.has_train:
            if len(train_ts_ids_to_take) == 0:
                raise ValueError("No time series left in training set after applying nan_threshold.")
            self.dataset_config.train_ts_row_ranges = self.dataset_config.train_ts_row_ranges[train_ts_ids_to_take]
            self.dataset_config.train_ts = self.dataset_config.train_ts[train_ts_ids_to_take]

            if self.dataset_config.fill_missing_with is not None:
                self.dataset_config.train_fillers = self.dataset_config.train_fillers[train_ts_ids_to_take]

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

        if self.dataset_config.has_val:
            if len(val_ts_ids_to_take) == 0:
                raise ValueError("No time series left in validation set after applying nan_threshold.")
            self.dataset_config.val_ts_row_ranges = self.dataset_config.val_ts_row_ranges[val_ts_ids_to_take]
            self.dataset_config.val_ts = self.dataset_config.val_ts[val_ts_ids_to_take]

            if self.dataset_config.fill_missing_with is not None:
                self.dataset_config.val_fillers = self.dataset_config.val_fillers[val_ts_ids_to_take]

            self.logger.debug("Validation set updated: %s time series selected.", len(val_ts_ids_to_take))

        if self.dataset_config.has_test:
            if len(test_ts_ids_to_take) == 0:
                raise ValueError("No time series left in test set after applying nan_threshold.")
            self.dataset_config.test_ts_row_ranges = self.dataset_config.test_ts_row_ranges[test_ts_ids_to_take]
            self.dataset_config.test_ts = self.dataset_config.test_ts[test_ts_ids_to_take]

            if self.dataset_config.fill_missing_with is not None:
                self.dataset_config.test_fillers = self.dataset_config.test_fillers[test_ts_ids_to_take]

            self.logger.debug("Test set updated: %s time series selected.", len(test_ts_ids_to_take))

        if self.dataset_config.has_all:
            if len(all_ts_ids_to_take) == 0:
                raise ValueError("No series left in all set after applying nan_threshold.")
            self.dataset_config.all_ts = self.dataset_config.all_ts[all_ts_ids_to_take]
            self.dataset_config.all_ts_row_ranges = self.dataset_config.all_ts_row_ranges[all_ts_ids_to_take]

            if self.dataset_config.fill_missing_with is not None:
                self.dataset_config.all_fillers = self.dataset_config.all_fillers[all_ts_ids_to_take]

            self.logger.debug("All set updated: %s time series selected.", len(all_ts_ids_to_take))

        self.dataset_config.used_ts_ids = self.dataset_config.all_ts
        self.dataset_config.used_ts_row_ranges = self.dataset_config.all_ts_row_ranges
        self.dataset_config.used_fillers = self.dataset_config.all_fillers
        self.dataset_config.used_times = self.time_indices

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

    def _update_export_config_copy(self) -> None:
        """
        Called at the end of [`set_dataset_config_and_initialize`][cesnet_tszoo.datasets.series_based_cesnet_dataset.SeriesBasedCesnetDataset.set_dataset_config_and_initialize]. 

        Updates values of config used for saving config.
        """

        self._export_config_copy.database_name = self.database_name

        if self.dataset_config.has_train:
            self._export_config_copy.train_ts = self.dataset_config.train_ts.copy()
            self.logger.debug("Updated train_ts of _export_config_copy.")

        if self.dataset_config.has_val:
            self._export_config_copy.val_ts = self.dataset_config.val_ts.copy()
            self.logger.debug("Updated val_ts of _export_config_copy.")

        if self.dataset_config.has_test:
            self._export_config_copy.test_ts = self.dataset_config.test_ts.copy()
            self.logger.debug("Updated test_ts of _export_config_copy.")

        super(SeriesBasedCesnetDataset, self)._update_export_config_copy()

    def apply_scaler(self, scale_with: type | list[Scaler] | np.ndarray[Scaler] | ScalerType | Scaler | Literal["min_max_scaler", "standard_scaler", "max_abs_scaler", "log_scaler", "l2_normalizer"] | None | Literal["config"] = "config",
                     partial_fit_initialized_scalers: bool | Literal["config"] = "config", workers: int | Literal["config"] = "config") -> None:
        """Used for updating scaler and relevenat 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                                                                                                    |
        | ---------------------------------- | -------------------------------------------------------------------------------------------------------------- |
        | `scale_with`                       | Defines the scaler to transform the dataset.                                                                   |     
        | `partial_fit_initialized_scalers`  | If `True`, partial fitting on train set is performed when using initiliazed scalers.                           |    

        Parameters:
            scale_with: Defines the scaler to transform the dataset. `Defaults: config`.  
            partial_fit_initialized_scalers: If `True`, partial fitting on train set is performed when using initiliazed scalers. `Defaults: config`.  
            workers: How many workers to use when setting new scaler. `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 scaler values.")

        self.update_dataset_config_and_initialize(scale_with=scale_with, partial_fit_initialized_scalers=partial_fit_initialized_scalers, workers=workers)

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

        temp = np.where(np.isin(parent_dataset.ts_row_ranges[self.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.ts_row_ranges[self.ts_id_name]}")

        time_series_position = temp[0]

        filler = None if parent_dataset.fillers is None else parent_dataset.fillers[time_series_position:time_series_position + 1]
        scaler = None if parent_dataset.scalers is None else parent_dataset.scalers

        dataset = SeriesBasedDataset(self.dataset_path,
                                     self.dataset_config._get_table_data_path(),
                                     self.dataset_config.ts_id_name,
                                     parent_dataset.ts_row_ranges[time_series_position: time_series_position + 1],
                                     parent_dataset.time_period,
                                     self.dataset_config.features_to_take,
                                     self.dataset_config.indices_of_features_to_take_no_ids,
                                     self.dataset_config.default_values,
                                     filler,
                                     self.dataset_config.include_time,
                                     self.dataset_config.include_ts_id,
                                     self.dataset_config.time_format,
                                     scaler)
        self.logger.debug("Singular time series dataset initiliazed.")

        return dataset

    def _get_dataloader(self, dataset: SeriesBasedDataset, workers: int | Literal["config"], take_all: bool, batch_size: int, **kwargs) -> DataLoader:
        """Set series based dataloader for this dataset. """

        defaultKwargs = {'order': DataloaderOrder.SEQUENTIAL}
        kwargs = {**defaultKwargs, **kwargs}

        return self._get_series_based_dataloader(dataset, workers, take_all, batch_size, kwargs["order"])

set_dataset_config_and_initialize 

set_dataset_config_and_initialize(dataset_config: SeriesBasedConfig, display_config_details: bool = True, workers: int | Literal['config'] = 'config') -> None

Initialize training set, validation est, 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_scalers Determines whether initialized scalers 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 SeriesBasedConfig

Desired configuration of the dataset.

 required
display_config_details bool

Flag indicating whether to display the configuration values after initialization. Default: True

True
workers int | Literal['config']

The number of workers to use during initialization. Default: "config"

 'config'
Source code in cesnet_tszoo\datasets\series_based_cesnet_dataset.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def set_dataset_config_and_initialize(self, dataset_config: SeriesBasedConfig, display_config_details: bool = True, workers: int | Literal["config"] = "config") -> None:
    """
    Initialize training set, validation est, test set etc.. This method must be called before any data can be accessed. It is required for the final initialization of [`dataset_config`][cesnet_tszoo.configs.series_based_config.SeriesBasedConfig].

    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_scalers` | Determines whether initialized scalers 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 to display the configuration values after initialization. `Default: True`  
        workers: The number of workers to use during initialization. `Default: "config"`  
    """

    assert isinstance(dataset_config, SeriesBasedConfig), "SeriesBasedCesnetDataset can only use SeriesBasedConfig."

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

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. See cesnet_tszoo.utils.enums.DataloaderOrder.
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 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
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`][cesnet_tszoo.datasets.cesnet_dataset.CesnetDataset.get_train_df] or [`get_train_numpy`][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. See [cesnet_tszoo.utils.enums.DataloaderOrder][].  |
    | `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."

    defaultKwargs = {'take_all': False, "cache_loader": True}
    kwargs = {**defaultKwargs, **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._get_dataloader(dataset, 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._get_dataloader(self.train_dataset, 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._get_dataloader(self.train_dataset, 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 
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
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`][cesnet_tszoo.datasets.cesnet_dataset.CesnetDataset.get_val_df] or [`get_val_numpy`][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."

    defaultKwargs = {'take_all': False, "cache_loader": True}
    kwargs = {**defaultKwargs, **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._get_dataloader(dataset, 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._get_dataloader(self.val_dataset, 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._get_dataloader(self.val_dataset, 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 
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
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`][cesnet_tszoo.datasets.cesnet_dataset.CesnetDataset.get_test_df] or [`get_test_numpy`][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_all:
        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."

    defaultKwargs = {'take_all': False, "cache_loader": True}
    kwargs = {**defaultKwargs, **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._get_dataloader(dataset, 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._get_dataloader(self.test_dataset, 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._get_dataloader(self.test_dataset, 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 
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
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`][cesnet_tszoo.datasets.cesnet_dataset.CesnetDataset.get_all_df] or [`get_all_numpy`][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."

    defaultKwargs = {'take_all': False, "cache_loader": True}
    kwargs = {**defaultKwargs, **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._get_dataloader(dataset, 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._get_dataloader(self.all_dataset, 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._get_dataloader(self.all_dataset, 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 
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
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()

    dataloader = self.get_train_dataloader(workers=workers, take_all=not self.dataset_config.is_series_based, 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 
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
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()

    dataloader = self.get_val_dataloader(workers=workers, take_all=not self.dataset_config.is_series_based, 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 
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
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_all:
        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()

    dataloader = self.get_test_dataloader(workers=workers, take_all=not self.dataset_config.is_series_based, 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 
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
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()

    dataloader = self.get_all_dataloader(workers=workers, take_all=not self.dataset_config.is_series_based, 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 
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
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()

    dataloader = self.get_train_dataloader(workers=workers, take_all=not self.dataset_config.is_series_based, 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 
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
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()

    dataloader = self.get_val_dataloader(workers=workers, take_all=not self.dataset_config.is_series_based, 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 
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
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_all:
        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()

    dataloader = self.get_test_dataloader(workers=workers, take_all=not self.dataset_config.is_series_based, 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 
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
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()

    dataloader = self.get_all_dataloader(workers=workers, take_all=not self.dataset_config.is_series_based, cache_loader=False)

    return self._get_numpy(dataloader, ts_ids, time_period)

display_dataset_details 

display_dataset_details() -> None

Display information about the contents of the dataset.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py 
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
    def display_dataset_details(self) -> None:
        """Display information about the contents of the dataset.  """

        to_display = f'''
Dataset details:

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

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

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

        print(to_display)

display_config 

display_config() -> None

Displays the values of the initialized configuration.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py 
1220
1221
1222
1223
1224
1225
def display_config(self) -> None:
    """Displays the values of the initialized configuration. """
    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 displaying config.")

    print(self.dataset_config)

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 
1227
1228
1229
1230
def get_feature_names(self) -> list[str]:
    """Returns a list of all available feature names in the dataset. """

    return get_column_names(self.dataset_path, self.source_type, self.aggregation)

get_data_about_set 

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

Retrieves data related to the specified set.

Parameters:

Name Type Description Default
about SplitType | Literal['train', 'val', 'test', 'all']

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\series_based_cesnet_dataset.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def get_data_about_set(self, about: SplitType | Literal["train", "val", "test", "all"]) -> dict:
    """
    Retrieves 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 = self.dataset_config.time_period

    result = {}

    if about == SplitType.TRAIN:
        if not self.dataset_config.has_train:
            raise ValueError("Train set is not used.")
        ts_ids = self.dataset_config.train_ts
    elif about == SplitType.VAL:
        if not self.dataset_config.has_val:
            raise ValueError("Val set is not used.")
        ts_ids = self.dataset_config.val_ts
    elif about == SplitType.TEST:
        if not self.dataset_config.has_test:
            raise ValueError("Test set is not used.")
        ts_ids = self.dataset_config.test_ts
    elif about == SplitType.ALL:
        if not self.dataset_config.has_all:
            raise ValueError("All set is not used.")
        ts_ids = self.dataset_config.all_ts
    else:
        raise NotImplementedError("Should not happen")

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

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

    return result

get_available_ts_indices 

get_available_ts_indices()

Returns the available time series indices in this dataset.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py 
1245
1246
1247
def get_available_ts_indices(self):
    """Returns the available time series indices in this dataset. """
    return self.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 
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
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.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.additional_data}")

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

    for column, column_type in self.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', use_scalers: bool = True, 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'
use_scalers bool

Whether the data should be scaled. If True, scaling will be applied using the available scaler for the selected ts_id. Defaults: True.

 True
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 
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
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", use_scalers: bool = True, 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"`.
        use_scalers: Whether the data should be scaled. If `True`, scaling will be applied using the available scaler for the selected `ts_id`. `Defaults: True`.
        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, use_scalers)
    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 
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
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 
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
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 
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
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 
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
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 
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
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.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 
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
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.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.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.ts_id_name)
        on = AnnotationType.BOTH
        self.logger.info("Annotations detected as %s (both %s and id_time)", AnnotationType.BOTH, self.ts_id_name)

    elif self.ts_id_name in temp_df.columns:
        self.annotations.clear_time_series()
        ts_id_index = temp_df.columns.tolist().index(self.ts_id_name)
        on = AnnotationType.TS_ID
        self.logger.info("Annotations detected as %s (%s only)", AnnotationType.TS_ID, self.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.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: bool = True, 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_scalers Determines whether initialized scalers 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 bool

Flag indicating whether to display the configuration values after initialization. Default: True

True
workers int | Literal['config']

The number of workers to use during initialization. Default: "config"

 'config'
Source code in cesnet_tszoo\datasets\cesnet_dataset.py 
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
def import_config(self, identifier: str, display_config_details: bool = True, 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_scalers` | Determines whether initialized scalers 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"`  
    """

    # Load config
    config_file_path, is_built_in = get_config_path_and_whether_it_is_built_in(identifier, self.configs_root, self.database_name, self.source_type, self.aggregation, self.logger)

    if is_built_in:
        self.logger.info("Built-in config found: %s. Loading it.", identifier)
        config = pickle_load(config_file_path)
    else:
        self.logger.info("Custom config found: %s. Loading it.", identifier)
        config = pickle_load(config_file_path)

    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 imported config from %s", config_file_path)

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 
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
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.annotations_root):
        os.makedirs(self.annotations_root)
        self.logger.info("Created annotations directory at %s", self.annotations_root)

    path = os.path.join(self.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) -> 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 
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
def save_config(self, identifier: str, create_with_details_file: bool = True, force_write: bool = False) -> 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`            
    """

    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 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.configs_root):
        os.makedirs(self.configs_root)
        self.logger.info("Created config directory at %s", self.configs_root)

    path_pickle = os.path.join(self.configs_root, f"{identifier}.pickle")
    path_details = os.path.join(self.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 self.dataset_config.is_filler_custom:
        self.logger.warning("You are using a custom filler. Ensure the config is distributed with the source code of the filler.")

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

    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.logger.info("Config successfully saved")

save_benchmark 

save_benchmark(identifier: str, force_write: bool = False) -> 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 
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
def save_benchmark(self, identifier: str, force_write: bool = False) -> 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`            
    """

    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 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, otherwise default to the current identifier
    config_name = self.dataset_config.import_identifier if self.dataset_config.import_identifier is not None else identifier

    export_benchmark = ExportBenchmark(self.database_name,
                                       self.is_series_based,
                                       self.source_type.value,
                                       self.aggregation.value,
                                       config_name,
                                       annotations_ts_name,
                                       annotations_time_name,
                                       annotations_both_name)

    # If the config was not imported, save it
    if self.dataset_config.import_identifier is None:
        self.save_config(export_benchmark.config_identifier, force_write=force_write)
    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.benchmarks_root):
        os.makedirs(self.benchmarks_root)
        self.logger.info("Created benchmarks directory at %s", self.benchmarks_root)

    benchmark_path = os.path.join(self.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_scalers 

get_scalers() -> np.ndarray[Scaler] | Scaler | None

Return used scalers from config.

Source code in cesnet_tszoo\datasets\cesnet_dataset.py 
1731
1732
1733
1734
1735
1736
def get_scalers(self) -> np.ndarray[Scaler] | Scaler | None:
    """Return used scalers 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 scalers.")

    return self.dataset_config.scalers

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 
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
def check_errors(self) -> None:
    """
    Validates whether the dataset is corrupted. 

    Raises an exception if corrupted.
    """

    dataset, _ = load_database(self.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.")