

# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING, Dict, Optional

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.logging import get_logger
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter


if TYPE_CHECKING:
       import jax
       import jaxlib

snake_case_			  = get_logger()

snake_case_			  = None


class       SCREAMING_SNAKE_CASE__					(TensorFormatter[Mapping, """jax.Array""", Mapping]     ):





    def __init__(      self						,   a=None						,   a=None						,   **a):
         super().__init__(features=a)
         import jax
         from jaxlib.xla_client import Device

         if isinstance(a						,   a):
              raise ValueError(
                  f"""Expected {device} to be a `str` not {type(a)}, as `jaxlib.xla_extension.Device` """
                  'is not serializable neither with `pickle` nor with `dill`. Instead you can surround '
                  'the device with `str()` to get its string identifier that will be internally mapped '
                  'to the actual `jaxlib.xla_extension.Device`.')
         lowercase__    :   int      =       device if isinstance(a						,   a) else str(jax.devices()[0])
         # using global variable since `jaxlib.xla_extension.Device` is not serializable neither
         # with `pickle` nor with `dill`, so we need to use a global variable instead
         global DEVICE_MAPPING
         if DEVICE_MAPPING is None:
              lowercase__    :   Optional[Any]      =       self._map_devices_to_str()
         if self.device not in list(DEVICE_MAPPING.keys()):
              logger.warning(
                  f"""Device with string identifier {self.device} not listed among the available """
                  f"""devices: {list(DEVICE_MAPPING.keys())}, so falling back to the default """
                  f"""device: {str(jax.devices()[0])}.""")
              lowercase__    :   str      =       str(jax.devices()[0])
         lowercase__    :   Dict      =       jnp_array_kwargs





    @staticmethod
    def       snake_case_				(      ):
         import jax

         return {str(a): device for device in jax.devices()}





    def       snake_case_				(      self						,   a):
         import jax
         import jax.numpy as jnp

         if isinstance(a						,   a) and column:
              if all(
                  isinstance(a						,   jax.Array) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column):
                   return jnp.stack(a						,   axis=0)
         return column





    def       snake_case_				(      self						,   a):
         import jax
         import jax.numpy as jnp

         if isinstance(a						,   (str, bytes, type(a))):
              return value
         elif isinstance(a						,   (np.character, np.ndarray)) and np.issubdtype(value.dtype						,   np.character):
              return value.tolist()

         lowercase__    :   str      =       {}

         if isinstance(a						,   (np.number, np.ndarray)) and np.issubdtype(value.dtype						,   np.integer):
              # the default int precision depends on the jax config
              # see https://jax.readthedocs.io/en/latest/notebooks/Common_Gotchas_in_JAX.html#double-64bit-precision
              if jax.config.jax_enable_xaa:
                   lowercase__    :   Union[str, Any]      =       {'dtype': jnp.intaa}
              else:
                   lowercase__    :   Tuple      =       {'dtype': jnp.intaa}
         elif isinstance(a						,   (np.number, np.ndarray)) and np.issubdtype(value.dtype						,   np.floating):
              lowercase__    :   Optional[int]      =       {'dtype': jnp.floataa}
         elif config.PIL_AVAILABLE and "PIL" in sys.modules:
              import PIL.Image

              if isinstance(a						,   PIL.Image.Image):
                   lowercase__    :   Tuple      =       np.asarray(a)

        # using global variable since `jaxlib.xla_extension.Device` is not serializable neither
        # with `pickle` nor with `dill`, so we need to use a global variable instead
         global DEVICE_MAPPING
         if DEVICE_MAPPING is None:
              lowercase__    :   List[str]      =       self._map_devices_to_str()

         with jax.default_device(DEVICE_MAPPING[self.device]):
              # calling jnp.array on a np.ndarray does copy the data
              # see https://github.com/google/jax/issues/4486
              return jnp.array(a						,   **{**default_dtype, **self.jnp_array_kwargs})





    def       snake_case_				(      self						,   a):
         import jax

         # support for torch, tf, jax etc.
         if config.TORCH_AVAILABLE and "torch" in sys.modules:
              import torch

              if isinstance(a						,   torch.Tensor):
                   return self._tensorize(data_struct.detach().cpu().numpy()[()])
         if hasattr(a						,   '__array__') and not isinstance(a						,   jax.Array):
              lowercase__    :   int      =       data_struct.__array__()
         # support for nested types like struct of list of struct
         if isinstance(a						,   np.ndarray):
              if data_struct.dtype == object:  # jax arrays cannot be instantied from an array of objects
                   return self._consolidate([self.recursive_tensorize(a) for substruct in data_struct])
         elif isinstance(a						,   (list, tuple)):
              return self._consolidate([self.recursive_tensorize(a) for substruct in data_struct])
         return self._tensorize(a)





    def       snake_case_				(      self						,   a):
         return map_nested(self._recursive_tensorize						,   a						,   map_list=a)





    def       snake_case_				(      self						,   a):
         lowercase__    :   Any      =       self.numpy_arrow_extractor().extract_row(a)
         lowercase__    :   List[Any]      =       self.python_features_decoder.decode_row(a)
         return self.recursive_tensorize(a)





    def       snake_case_				(      self						,   a):
         lowercase__    :   Tuple      =       self.numpy_arrow_extractor().extract_column(a)
         lowercase__    :   Optional[Any]      =       self.python_features_decoder.decode_column(a						,   pa_table.column_names[0])
         lowercase__    :   Any      =       self.recursive_tensorize(a)
         lowercase__    :   Any      =       self._consolidate(a)
         return column





    def       snake_case_				(      self						,   a):
         lowercase__    :   str      =       self.numpy_arrow_extractor().extract_batch(a)
         lowercase__    :   str      =       self.python_features_decoder.decode_batch(a)
         lowercase__    :   str      =       self.recursive_tensorize(a)
         for column_name in batch:
              lowercase__    :   int      =       self._consolidate(batch[column_name])
         return batch




def 	snake_case__			(    SCREAMING_SNAKE_CASE_		:    str			):



     '''simple docstring'''





     if not all(char in '01' for char in bin_string			):
          raise ValueError('Non-binary value was passed to the function'			)
     if not bin_string:
          raise ValueError('Empty string was passed to the function'			)
     lowercase__    :   Union[str, Any]      =       ''
     while len(SCREAMING_SNAKE_CASE_			) % 3 != 0:
          lowercase__    :   List[str]      =       '0' + bin_string
     lowercase__    :   Any      =       [
         bin_string[index : index + 3]
         for index in range(len(SCREAMING_SNAKE_CASE_			)			)
         if index % 3 == 0
     ]
     for bin_group in bin_string_in_3_list:
          lowercase__    :   str      =       0
          for index, val in enumerate(SCREAMING_SNAKE_CASE_			):
               oct_val += int(2 ** (2 - index) * int(SCREAMING_SNAKE_CASE_			)			)
          oct_string += str(SCREAMING_SNAKE_CASE_			)
     return oct_string


if __name__ == "__main__":
       from doctest import testmod

       testmod()

