code: def data_incremental_benchmark(
      benchmark_instance: DatasetScenario[
          TDatasetStream, TDatasetExperience, TCLDataset
      ],
      experience_size: int,
      shuffle: bool = False,
      split_streams: Sequence[str] = ("train",),
      custom_split_strategy: Optional[
          Callable[
              [DatasetExperience[TCLDataset]], Sequence[TCLDataset]
          ]
      ] = None,
      *,
      benchmark_factory: Optional[
          Callable[
              [
                  DatasetScenario[
                      TDatasetStream, TDatasetExperience, TCLDataset
                  ],
                  TStreamsUserDict,
                  bool,
              ],
              DatasetScenario[
                  DatasetStream[DatasetExperience[TCLDataset]],
                  DatasetExperience[TCLDataset],
                  TCLDataset,
              ],
          ]
      ] = _smart_benchmark_factory,
      experience_factory: Optional[
          Callable[
              [DatasetStream[DatasetExperience[TCLDataset]], int],
              DatasetExperience[TCLDataset],
          ]
      ] = _make_plain_experience,
  ) -> DatasetScenario[
      DatasetStream[DatasetExperience[TCLDataset]],
      DatasetExperience[TCLDataset],
      TCLDataset,
  ]:
      """
      High-level benchmark generator for a Data Incremental setup.
  
      This generator accepts an existing benchmark instance and returns a version
      of it in which experiences have been split in order to produce a
      Data Incremental stream.
  
      In its base form this generator will split train experiences in experiences
      of a fixed, configurable, size. The split can be also performed on other
      streams (like the test one) if needed.
  
      The `custom_split_strategy` parameter can be used if a more specific
      splitting is required.
  
      Beware that experience splitting is NOT executed in a lazy way. This
      means that the splitting process takes place immediately. Consider
      optimizing the split process for speed when using a custom splitting
      strategy.
  
      Please note that each mini-experience will have a task labels field
      equal to the one of the originating experience.
  
      The `complete_test_set_only` field of the resulting benchmark instance
      will be `True` only if the same field of original benchmark instance is
      `True` and if the resulting test stream contains exactly one experience.
  
      :param benchmark_instance: The benchmark to split.
      :param experience_size: The size of the experience, as an int. Ignored
          if `custom_split_strategy` is used.
      :param shuffle: If True, experiences will be split by first shuffling
          instances in each experience. This will use the default PyTorch
          random number generator at its current state. Defaults to False.
          Ignored if `custom_split_strategy` is used.
      :param split_streams: The list of streams to split. By default only the
          "train" stream will be split.
      :param custom_split_strategy: A function that implements a custom splitting
          strategy. The function must accept an experience and return a list
          of datasets each describing an experience. Defaults to None, which means
          that the standard splitting strategy will be used (which creates
          experiences of size `experience_size`).
          A good starting to understand the mechanism is to look at the
          implementation of the standard splitting function
          :func:`split_dataset`.
      :param benchmark_factory: The scenario factory. Defaults to
          `_smart_experience_factory`, which will try to create a benchmark of the
          same class of the originating one. Can be None, in which case a generic
          :class:`DatasetScenario` will be used coupled with the factory defined
          by the `experience_factory` parameter.
      :param experience_factory: The experience factory. Ignored if
          `scenario_factory` is not None. Otherwise, defaults to
          :class:`DatasetExperience`.
      :return: The Data Incremental benchmark instance.
      """
  
      split_strategy: Callable[
          [DatasetExperience[TCLDataset]], Sequence[TCLDataset]
      ]
      if custom_split_strategy is None:
          # functools.partial is a more compact option
          # However, MyPy does not understand what a partial is -_-
          def split_strategy_wrapper(exp):
              ss = FixedSizeExperienceSplitter(
                  experience=exp,
                  experience_size=experience_size,
                  shuffle=shuffle,
              )
              return [e.dataset for e in ss]
  
          split_strategy = split_strategy_wrapper
      else:
          split_strategy = custom_split_strategy
  
      stream_definitions: Dict[str, StreamDef[TCLDataset]] = dict(
          benchmark_instance.stream_definitions
      )
  
      for stream_name in split_streams:
          if stream_name not in stream_definitions:
              raise ValueError(
                  f"Stream {stream_name} could not be found in the "
                  f"benchmark instance"
              )
  
          stream: TDatasetStream = getattr(
              benchmark_instance, f"{stream_name}_stream"
          )
  
          split_datasets: List[TCLDataset] = []
          split_task_labels: List[Set[int]] = []
  
          exp: DatasetExperience[TCLDataset]
          for exp in stream:
              experiences = split_strategy(exp)
              split_datasets += experiences
              for _ in range(len(experiences)):
                  split_task_labels.append(set(exp.task_labels))
  
          stream_def = StreamDef(
              LazyDatasetSequence(split_datasets, len(split_datasets)),
              split_task_labels,
              stream_definitions[stream_name].origin_dataset,
              False,
          )
          stream_def.exps_data.load_all_experiences()
  
          stream_definitions[stream_name] = stream_def
  
      complete_test_set_only = (
          benchmark_instance.complete_test_set_only
          and len(stream_definitions["test"].exps_data) == 1
      )
  
      if benchmark_factory is not None:
          # Try to create a benchmark of the same class of the
          # initial benchmark.
          return benchmark_factory(
              benchmark_instance,
              stream_definitions,
              complete_test_set_only,
          )
  
      # Generic benchmark class
      if experience_factory is None:
          experience_factory = _make_plain_experience
  
      return DatasetScenario(
          stream_definitions=stream_definitions,
          complete_test_set_only=complete_test_set_only,
          stream_factory=FactoryBasedStream,
          experience_factory=experience_factory,
      )
  
file_path: avalanche/avalanche/benchmarks/scenarios/deprecated/generators.py
imports: from typing import TYPE_CHECKING, Sequence, Optional, Dict, TypeVar, Union, Any, List, Callable, Set, Tuple
  import torch
  from avalanche.benchmarks.scenarios.dataset_scenario import split_validation_random
  from avalanche.benchmarks.scenarios.deprecated.classification_scenario import ClassificationScenario
  from avalanche.benchmarks.scenarios.deprecated.dataset_scenario import DatasetScenario, DatasetStream, FactoryBasedStream, StreamDef, TStreamsUserDict
  from avalanche.benchmarks.scenarios.detection_scenario import DetectionScenario
  from avalanche.benchmarks.scenarios.deprecated.generic_benchmark_creation import *
  from avalanche.benchmarks.scenarios.deprecated import DatasetExperience
  from avalanche.benchmarks.scenarios.deprecated.lazy_dataset_sequence import LazyDatasetSequence
  from avalanche.benchmarks.scenarios.deprecated.new_classes.nc_scenario import NCScenario
  from avalanche.benchmarks.scenarios.deprecated.new_instances.ni_scenario import NIScenario
  from avalanche.benchmarks.scenarios.online import FixedSizeExperienceSplitter
  from avalanche.benchmarks.utils.classification_dataset import SupportedDataset, _as_taskaware_supervised_classification_dataset, _make_taskaware_classification_dataset, _concat_taskaware_classification_datasets_sequentially
label_desc: Use image node_id 6 for calling a global function with example usage: data_incremental_benchmark(benchmark_instance, experience_size, shuffle, split_streams, custom_split_strategy) and returns: DatasetScenario, benchmark_factory, unknown