code: def enable_ffcv(
      benchmark: CLScenario,
      write_dir: Union[str, Path],
      device: torch.device,
      ffcv_parameters: Dict[str, Any],
      force_overwrite: bool = False,
      encoder_def: "EncoderDef" = None,
      decoder_def: "DecoderDef" = None,
      decoder_includes_transformations: Optional[bool] = None,
      print_summary: bool = True,
  ) -> None:
      """
      Enables the support for FFCV data loading for the given benchmark.
  
      Once the support is added, the strategies will create FFCV-based dataloaders
      instead of the usual PyTorch-based ones.
  
      Please note that FFCV is an optional dependency whose installation process
      is described in the official FFCV website.
  
      This function supposes that the benchmark is based on a few base datasets
      (usually one for train and one for test). This is the case for Split-* benchmarks
      and is also the usual case for the vast majority of benchmarks. The support for
      "sparse" datasets such as CORe50 will be added in the near future.
  
      When this function is first called on a benchmark, the underlying datasets are
      serialized on disk. If a `encoder_def` is given, that definition is used. Otherwise,
      a definition is searched in the leaf dataset (`_ffcv_encoder` field, if available).
      If such a definition is not found, it is created automatically.
      Refer to the FFCV documentation for more details on the encoder pipeline.
  
      Please note that the serialized datasets are independent of the benchmark seed,
      number of experiences, presence of task labels, etcetera. This means that the
      same folder can be reused for the same benchmark type.
  
      The definition of the decoder pipeline is created later, if not
      given using `decoder_def`. However, creating the decoder pipeline is a complex
      task and not all field types and transformations are fully supported. Consider
      passing an explicit `decoder_def` in case of unexpected outputs. If a decoder
      definition is not passed explicitly, Avalanche will try to use the dataset
      `_ffcv_decoder` field if available before attempting to create one automatically.
  
      See the `ffcv` examples for more info on how to tune the decoder definitions and for
      examples of advanced use of the FFCV support.
  
      :param benchmark: The benchmark for which the support for FFCV loader should be enabled.
      :param write_dir: Where the datasets should be serialized in FFCV format.
      :param device: The device used for training.
      :param ffcv_parameters: Parameters to be passed to FFCV writer and RGB fields.
      :param force_overwrite: If True, serialized datasets already found in `write_dir` will be
          overwritten.
      :param encoder_def: The definition of the dataset fields. See the FFCV guide for more details.
      :param decoder_def: The definition of the decoder pipeline. If not None, then
          `decoder_includes_transformations` must be passed.
      :param decoder_includes_transformations: If True, then Avalanche will treat `decoder_def` as
          the complete pipeline, transformations included. If False, Avalanche will suppose that only
          the decoder is passed for each field and transformations will be translated by Avalanche
          from the torchvision ones.
      :param print_summary: If True (default), will print some verbose info to stdout regaring the
          datasets and pipelines. Once you have a complete working FFCV pipeline, you can consider
          setting this to False.
      """
      global FFCV_EXPERIMENTAL_WARNED
  
      if not FFCV_EXPERIMENTAL_WARNED:
          warnings.warn(
              "The support for FFCV is experimental. Use at your own risk!"
          )
          FFCV_EXPERIMENTAL_WARNED = True
  
      from ffcv.writer import DatasetWriter
      from avalanche.benchmarks.utils.ffcv_support.ffcv_support_internals import (
          _make_ffcv_decoder,
          _make_ffcv_encoder,
      )
  
      if decoder_def is not None:
          if decoder_includes_transformations is None:
              raise ValueError(
                  "When defining the decoder pipeline, "
                  "please specify `decoder_includes_transformations`"
              )
          assert isinstance(decoder_includes_transformations, bool)
  
      if decoder_includes_transformations is None:
          decoder_includes_transformations = False
  
      write_dir = Path(write_dir)
      write_dir.mkdir(exist_ok=True, parents=True)
  
      flattened_datasets = flat_datasets_from_benchmark(benchmark)
  
      if print_summary:
          print(
              "FFCV will serialize", len(flattened_datasets), "datasets"
          )
  
      for idx, (dataset, _, _) in enumerate(flattened_datasets):
          if print_summary:
              print("-" * 25, "Dataset", idx, "-" * 25)
  
          # Note: it is appropriate to serialize the dataset in its raw
          # version (without transformations). Transformations will be
          # applied at loading time.
          with _SuppressTransformations(dataset):
              dataset_ffcv_path = write_dir / f"dataset{idx}.beton"
  
              # Obtain the encoder dictionary
              # The FFCV encoder is a ordered dictionary mapping each
              # field (by name) to the field encoder.
              #
              # Example:
              # {
              #   'image': RGBImageField(),
              #   'label: IntField()
              # }
              #
              # Some fields (especcially the RGBImageField) accept
              # some parameters that are here contained in ffcv_parameters.
              encoder_dict = _make_ffcv_encoder(
                  dataset, encoder_def, ffcv_parameters
              )
  
              if encoder_dict is None:
                  raise RuntimeError(
                      "Could not create the encoder pipeline for "
                      "the given dataset"
                  )
  
              if print_summary:
                  print("### Encoder ###")
                  for (
                      field_name,
                      encoder_pipeline,
                  ) in encoder_dict.items():
                      print(f'Field "{field_name}"')
                      print("\t", encoder_pipeline)
  
              # Obtain the decoder dictionary
              # The FFCV decoder is a ordered dictionary mapping each
              # field (by name) to the field pipeline.
              # A field pipeline is made of a decoder followed by
              # transformations.
              #
              # Example:
              # {
              #   'image': [
              #       SimpleRGBImageDecoder(),
              #       RandomHorizontalFlip(),
              #       ToTensor(),
              #       ...
              #   ],
              #   'label: [IntDecoder(), ToTensor(), Squeeze(), ...]
              # }
              #
              # However, unless the user specified a full custom decoder
              # pipeline, Avalanche will obtain only the decoder for each
              # field. The transformations, which may vary, will be added by the
              # data loader.
              decoder_dict = _make_ffcv_decoder(
                  dataset,
                  decoder_def,
                  ffcv_parameters,
                  encoder_dictionary=encoder_dict,
              )
  
              if decoder_dict is None:
                  raise RuntimeError(
                      "Could not create the decoder pipeline "
                      "for the given dataset"
                  )
  
              if print_summary:
                  print("### Decoder ###")
                  for (
                      field_name,
                      decoder_pipeline,
                  ) in decoder_dict.items():
                      print(f'Field "{field_name}"')
                      for pipeline_element in decoder_pipeline:
                          print("\t", pipeline_element)
  
                  if decoder_includes_transformations:
                      print(
                          "This pipeline already includes transformations"
                      )
                  else:
                      print(
                          "This pipeline does not include transformations"
                      )
  
              if force_overwrite or not dataset_ffcv_path.exists():
                  if print_summary:
                      print(
                          "Serializing dataset to:",
                          str(dataset_ffcv_path),
                      )
  
                  writer_kwarg_parameters = dict()
                  if "page_size" in ffcv_parameters:
                      writer_kwarg_parameters[
                          "page_size"
                      ] = ffcv_parameters["page_size"]
  
                  if "num_workers" in ffcv_parameters:
                      writer_kwarg_parameters[
                          "num_workers"
                      ] = ffcv_parameters["num_workers"]
  
                  writer = DatasetWriter(
                      str(dataset_ffcv_path),
                      OrderedDict(encoder_dict),
                      **writer_kwarg_parameters,
                  )
                  writer.from_indexed_dataset(dataset)
  
                  if print_summary:
                      print("Dataset serialized successfully")
  
              # Set the FFCV file path and encoder/decoder dictionaries
              # Those will be used later in the data loading process and may
              # also be useful for debugging purposes
              dataset.ffcv_info = FFCVInfo(
                  dataset_ffcv_path,
                  encoder_dict,
                  decoder_dict,
                  decoder_includes_transformations,
                  torch.device(device),
              )
  
      if print_summary:
          print("-" * 61)
  
imports: from dataclasses import dataclass
  from pathlib import Path
  from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Union
  from collections import OrderedDict
  import warnings
  import numpy
  import torch
  from torch.utils.data.sampler import Sampler
  from avalanche.benchmarks.scenarios.generic_scenario import CLScenario
  from avalanche.benchmarks.utils.data import AvalancheDataset
  from avalanche.benchmarks.utils.dataset_traversal_utils import flat_datasets_from_benchmark, single_flat_dataset
  from avalanche.benchmarks.utils.utils import concat_datasets
label_desc: Use image node_id 1 for calling a global function with example usage: enable_ffcv(benchmark, write_dir, device, ffcv_parameters, force_overwrite, encoder_def, decoder_def, decoder_includes_transformations, print_summary) without return types