class_docstr: A dataloader used to load :class:`AvalancheDataset`s for which
  the FFCV support was previously enabled by using :func:`enable_ffcv`.
  
  This is not a pure wrapper of a FFCV loader: this hybrid dataloader
  is in charge of both creating the FFCV loader and merging
  the Avalanche-specific info contained in the :class:`DataAttribute`
  fields of the datasets (such as task labels).
class_docstr_tok: ["A","dataloader","used","to","load",":","class",":","`","AvalancheDataset","`","s","for","which","the","FFCV","support","was","previously","enabled","by","using",":","func",":","`","enable_ffcv","`",".","This","is","not","a","pure","wrapper","of","a","FFCV","loader",":","this","hybrid","dataloader","is","in","charge","of","both","creating","the","FFCV","loader","and","merging","the","Avalanche-specific","info","contained","in","the",":","class",":","`","DataAttribute","`","fields","of","the","datasets","(","such","as","task","labels",")","."]
code: def _extract_ffcv_info(
      dataset: AvalancheDataset,
      device: Optional[Union[str, torch.device]] = None,
      print_summary: bool = True,
  ):
      from avalanche.benchmarks.utils.ffcv_support.ffcv_transform_utils import (
          adapt_transforms,
          check_transforms_consistency,
      )
  
      # Obtain the leaf dataset, the indices,
      # and the transformations to apply
      flat_set_def = single_flat_dataset(dataset)
      if flat_set_def is None:
          raise RuntimeError(
              "The dataset cannot be traversed to the leaf dataset."
          )
  
      leaf_dataset, indices, transforms = flat_set_def
      if print_summary:
          print(
              "The input AvalancheDataset is a subset of the leaf dataset",
              leaf_dataset,
          )
          print("The input dataset contains", len(indices), "elements")
          print("The original chain of transformations is:")
          for t in transforms:
              print("\t", t)
          print("Will try to translate those transformations to FFCV")
  
      ffcv_info: FFCVInfo = leaf_dataset.ffcv_info
  
      ffcv_dataset_path = ffcv_info.path
      ffcv_decoder_dictionary = ffcv_info.decoder_dictionary
      decoder_includes_transformations = (
          ffcv_info.decoder_includes_transformations
      )
  
      if device is None:
          device = ffcv_info.device
      device = torch.device(device)
  
      # Map the indices so that we know how leaf
      # dataset indices are mapped in the AvalancheDataset
      reversed_indices = dict()
      for avl_idx, leaf_idx in enumerate(indices):
          reversed_indices[leaf_idx] = avl_idx
  
      # We will use the GetItemDataset to get those Avalanche-specific
      # dynamic fields that are not loaded by FFCV, such as the task label
      get_item_dataset = _GetItemDataset(
          dataset, reversed_indices=reversed_indices
      )
  
      if print_summary:
          if len(get_item_dataset.get_item_data_attributes) > 0:
              print(
                  "The following data attributes are returned in "
                  "the example tuple:"
              )
              for da in get_item_dataset.get_item_data_attributes:
                  print("\t", da.name)
          else:
              print(
                  "No data attributes are returned in the example tuple."
              )
  
      # Defensive copy
      # Alas, FFCV Loader internally modifies it, so this is also
      # needed when decoder_includes_transformations is True
      ffcv_decoder_dictionary = OrderedDict(ffcv_decoder_dictionary)
  
      if not decoder_includes_transformations:
          # Adapt the transformations (usually from torchvision) to FFCV.
          # Most torchvision transformations cannot be mapped to FFCV ones,
          # but they still work.
          ffcv_decoder_dictionary_lst = list(
              ffcv_decoder_dictionary.values()
          )
  
          adapted_transforms = adapt_transforms(
              transforms, ffcv_decoder_dictionary_lst, device=device
          )
  
          for i, field_name in enumerate(
              ffcv_decoder_dictionary.keys()
          ):
              ffcv_decoder_dictionary[field_name] = adapted_transforms[
                  i
              ]
  
      for field_name, field_decoder in ffcv_decoder_dictionary.items():
          if print_summary:
              print(f'Checking pipeline for field "{field_name}"')
          no_issues = check_transforms_consistency(field_decoder)
  
          if print_summary and no_issues:
              print(f"No issues for this field")
  
      if print_summary:
          print("### The final chain of transformations is: ###")
          for (
              field_name,
              field_transforms,
          ) in ffcv_decoder_dictionary.items():
              print(f'Field "{field_name}":')
              for t in field_transforms:
                  print("\t", t)
  
      return (
          ffcv_dataset_path,
          ffcv_decoder_dictionary,
          indices,
          get_item_dataset,
          device,
      )
  
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 2 for calling the HybridFfcvLoader obj's underlying member method code with example usage: obj._extract_ffcv_info(dataset, device, print_summary) and returns: ffcv_dataset_path,  ffcv_decoder_dictionary,  indices,  get_item_dataset,  device