docstr: The BinaryWriter enables to chunk dataset into an efficient streaming format for cloud training.
  
  Arguments:
      cache_dir: The path to where the chunks will be saved.
      chunk_bytes: The maximum number of bytes within a chunk.
      chunk_size: The maximum number of items within a chunk.
      compression: The compression algorithm to use.
      serializers: Provide your own serializers.
docstr_tok: ["The","BinaryWriter","enables","to","chunk","dataset","into","an","efficient","streaming","format","for","cloud","training",".","Arguments",":","cache_dir",":","The","path","to","where","the","chunks","will","be","saved",".","chunk_bytes",":","The","maximum","number","of","bytes","within","a","chunk",".","chunk_size",":","The","maximum","number","of","items","within","a","chunk",".","compression",":","The","compression","algorithm","to","use",".","serializers",":","Provide","your","own","serializers","."]
code: def __init__(
      self,
      cache_dir: str,
      chunk_size: Optional[int] = None,
      chunk_bytes: Optional[Union[int, str]] = None,
      compression: Optional[str] = None,
      follow_tensor_dimension: bool = True,
      serializers: Optional[Dict[str, Serializer]] = None,
  ):
      """The BinaryWriter enables to chunk dataset into an efficient streaming format for cloud training.
  
      Arguments:
          cache_dir: The path to where the chunks will be saved.
          chunk_bytes: The maximum number of bytes within a chunk.
          chunk_size: The maximum number of items within a chunk.
          compression: The compression algorithm to use.
          serializers: Provide your own serializers.
  
      """
      self._cache_dir = cache_dir
  
      if (
          isinstance(self._cache_dir, str)
          and not os.path.exists(self._cache_dir)
      ) or self._cache_dir is None:
          raise FileNotFoundError(
              f"The provided cache directory `{self._cache_dir}` doesn't exist."
          )
  
      if (chunk_size is None and chunk_bytes is None) or (
          chunk_size and chunk_bytes
      ):
          raise ValueError(
              "Either one of the `chunk_size` or the `chunk_bytes` need to be provided."
          )
  
      self._serializers: Dict[str, Serializer] = _get_serializers(
          serializers
      )
      self._chunk_size = chunk_size
      self._chunk_bytes = (
          _convert_bytes_to_int(chunk_bytes)
          if isinstance(chunk_bytes, str)
          else chunk_bytes
      )
      self._compression = compression
  
      self._data_format: Optional[List[str]] = None
      self._data_spec: Optional[PyTree] = None
  
      if self._compression:
          if len(_COMPRESSORS) == 0:
              raise ValueError(
                  "No compresion algorithms are installed."
              )
          if self._compression not in _COMPRESSORS:
              raise ValueError(
                  f"The provided compression {self._compression} isn't available in {sorted(_COMPRESSORS)}"
              )
          self._compressor: Compressor = _COMPRESSORS[self._compression]
  
      self._serialized_items: Dict[int, Item] = {}
      self._chunk_index = 0
      self._min_index: Optional[int] = None
      self._max_index: Optional[int] = None
      self._chunks_info: List[Dict[str, Any]] = []
      self._worker_env: Optional[_WorkerEnv] = None
      self._rank: Optional[int] = None
      self._is_done = False
      self._distributed_env = _DistributedEnv.detect()
      self._follow_tensor_dimension = follow_tensor_dimension
  
imports: import json
  import os
  import warnings
  from dataclasses import dataclass
  from time import sleep
  from typing import Any, Dict, List, Optional, Tuple, Union
  import numpy
  import torch
  from lightning.data.streaming.compression import _COMPRESSORS, Compressor
  from lightning.data.streaming.constants import _INDEX_FILENAME, _TORCH_GREATER_EQUAL_2_1_0
  from lightning.data.streaming.serializers import Serializer, _get_serializers
  from lightning.data.utilities.env import _DistributedEnv, _WorkerEnv
  from lightning.data.utilities.format import _convert_bytes_to_int, _human_readable_bytes
label_desc: Use image node_id 1 to create a new BinaryWriter object  with example: obj = BinaryWriter(cache_dir, chunk_size, chunk_bytes, compression, follow_tensor_dimension, serializers)