code: def _cache_for_image_dataset(num_workers, tmpdir, fabric=None):
      from PIL import Image
      from torchvision.transforms import PILToTensor
  
      dataset_size = 85
  
      cache_dir = os.path.join(tmpdir, "cache")
      distributed_env = _DistributedEnv.detect()
  
      cache = Cache(cache_dir, chunk_size=10)
      dataset = ImageDataset(tmpdir, cache, dataset_size, 10)
      dataloader = CacheDataLoader(
          dataset, num_workers=num_workers, batch_size=4
      )
  
      for _ in dataloader:
          pass
  
      # Not strictly required but added to avoid race condition
      if distributed_env.world_size > 1:
          fabric.barrier()
  
      assert cache.filled
  
      for i in range(len(dataset)):
          cached_data = dataset[i]
          original_data = dataset.data[i]
          assert cached_data["class"] == original_data["class"]
          original_array = PILToTensor()(
              Image.open(original_data["image"])
          )
          assert torch.equal(original_array, cached_data["image"])
  
      if distributed_env.world_size == 1:
          indexes = []
          dataloader = CacheDataLoader(
              dataset, num_workers=num_workers, batch_size=4
          )
          for batch in dataloader:
              if batch:
                  indexes.extend(batch["index"].numpy().tolist())
          assert len(indexes) == dataset_size
  
      seed_everything(42)
  
      dataloader = CacheDataLoader(
          dataset, num_workers=num_workers, batch_size=4, shuffle=True
      )
      dataloader_iter = iter(dataloader)
  
      indexes = []
      for batch in dataloader_iter:
          indexes.extend(batch["index"].numpy().tolist())
  
      if distributed_env.world_size == 1:
          assert len(indexes) == dataset_size
  
      indexes2 = []
      for batch in dataloader_iter:
          indexes2.extend(batch["index"].numpy().tolist())
  
      assert indexes2 != indexes
  
      streaming_dataset = StreamingDataset(input_dir=cache_dir)
      for i in range(len(streaming_dataset)):
          cached_data = streaming_dataset[i]
          original_data = dataset.data[i]
          assert cached_data["class"] == original_data["class"]
          original_array = PILToTensor()(
              Image.open(original_data["image"])
          )
          assert torch.equal(original_array, cached_data["image"])
  
      streaming_dataset_iter = iter(streaming_dataset)
      for _ in streaming_dataset_iter:
          pass
  
imports: import json
  import os
  import sys
  from functools import partial
  import numpy
  import pytest
  import torch
  from lightning import seed_everything
  from lightning.data.streaming import Cache
  from lightning.data.streaming.dataloader import CacheDataLoader
  from lightning.data.streaming.dataset import StreamingDataset
  from lightning.data.streaming.item_loader import TokensLoader
  from lightning.data.streaming.serializers import Serializer
  from lightning.data.utilities.env import _DistributedEnv
  from lightning.fabric import Fabric
  from lightning.pytorch.demos.boring_classes import RandomDataset
  from lightning_utilities.core.imports import RequirementCache
  from lightning_utilities.test.warning import no_warning_call
  from torch.utils.data import Dataset
label_desc: Use image node_id 1 for calling a global function with example usage: _cache_for_image_dataset(num_workers, tmpdir, fabric) without return types