


# 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					:		Tuple      =    get_logger()

_snake_case					:		Optional[dict]      =    None

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




							"""simple docstring"""


							def __init__(							self	:  Any   ,     lowerCamelCase	:  int=None   ,     lowerCamelCase	:  List[Any]=None   ,     **lowerCamelCase	:  int    )     ->   int:
														super().__init__(features=lowerCamelCase    )
														import jax
														from jaxlib.xla_client import Device

														if isinstance(lowerCamelCase   ,     lowerCamelCase    ):
																					raise ValueError(
																					    F'Expected {device} to be a `str` not {type(lowerCamelCase    )}, 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`."    )
														__snake_case		:      List[Any]										=       device if isinstance(lowerCamelCase   ,     lowerCamelCase    ) 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:
																					__snake_case		:      Optional[int]										=       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]    )}.'    )
																					__snake_case		:      Any										=       str(jax.devices()[0]    )
														__snake_case		:      int										=       jnp_array_kwargs

							@staticmethod
							def    __snake_case (							)     ->   Dict[str, "jaxlib.xla_extension.Device"]:
														import jax

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

							def    __snake_case (							self	:  Any   ,     lowerCamelCase	:  Union[str, Any]    )     ->   int:
														import jax
														import jax.numpy as jnp

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

							def    __snake_case (							self	:  str   ,     lowerCamelCase	:  List[Any]    )     ->   Union[str, Any]:
														import jax
														import jax.numpy as jnp

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

														__snake_case		:      Dict										=       {}

														if isinstance(lowerCamelCase   ,     (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:
																												__snake_case		:      Optional[Any]										=       {"dtype": jnp.intaa}
																					else:
																												__snake_case		:      Optional[Any]										=       {"dtype": jnp.intaa}
														elif isinstance(lowerCamelCase   ,     (np.number, np.ndarray)    ) and np.issubdtype(value.dtype   ,     np.floating    ):
																					__snake_case		:      List[Any]										=       {"dtype": jnp.floataa}
														elif config.PIL_AVAILABLE and "PIL" in sys.modules:
																					import PIL.Image

																					if isinstance(lowerCamelCase   ,     PIL.Image.Image    ):
																												__snake_case		:      Tuple										=       np.asarray(lowerCamelCase    )

        # 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:
																					__snake_case		:      Optional[int]										=       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(lowerCamelCase   ,     **{**default_dtype, **self.jnp_array_kwargs}    )

							def    __snake_case (							self	:  Dict   ,     lowerCamelCase	:  Optional[Any]    )     ->   str:
														import jax

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

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

							def    __snake_case (							self	:  Union[str, Any]   ,     lowerCamelCase	:  dict    )     ->   Optional[Any]:
														return map_nested(self._recursive_tensorize   ,     lowerCamelCase   ,     map_list=lowerCamelCase    )

							def    __snake_case (							self	:  Optional[Any]   ,     lowerCamelCase	:  pa.Table    )     ->   Mapping:
														__snake_case		:      List[str]										=       self.numpy_arrow_extractor().extract_row(lowerCamelCase    )
														__snake_case		:      str										=       self.python_features_decoder.decode_row(lowerCamelCase    )
														return self.recursive_tensorize(lowerCamelCase    )

							def    __snake_case (							self	:  Tuple   ,     lowerCamelCase	:  pa.Table    )     ->   "jax.Array":
														__snake_case		:      Dict										=       self.numpy_arrow_extractor().extract_column(lowerCamelCase    )
														__snake_case		:      Tuple										=       self.python_features_decoder.decode_column(lowerCamelCase   ,     pa_table.column_names[0]    )
														__snake_case		:      Optional[int]										=       self.recursive_tensorize(lowerCamelCase    )
														__snake_case		:      Tuple										=       self._consolidate(lowerCamelCase    )
														return column






							def    __snake_case (							self	:  Union[str, Any]   ,     lowerCamelCase	:  pa.Table    )     ->   Mapping:
														__snake_case		:      Dict										=       self.numpy_arrow_extractor().extract_batch(lowerCamelCase    )
														__snake_case		:      Optional[int]										=       self.python_features_decoder.decode_batch(lowerCamelCase    )
														__snake_case		:      Dict										=       self.recursive_tensorize(lowerCamelCase    )
														for column_name in batch:
																					__snake_case		:      Dict										=       self._consolidate(batch[column_name]    )
														return batch





from typing import Dict

import numpy as np
import torch

from . import residue_constants as rc
from .tensor_utils import tensor_tree_map, tree_map







def     lowerCAmelCase_					( __lowerCamelCase     ):
							__snake_case		:      Any										=       []
							__snake_case		:      Optional[Any]										=       []
							__snake_case		:      List[Any]										=       []

							for rt in rc.restypes:
														__snake_case		:      Any										=       rc.restype_name_to_atomaa_names[rc.restype_atoa[rt]]
														restype_atomaa_to_atomaa_list.append([(rc.atom_order[name] if name else 0) for name in atom_names]     )
														__snake_case		:      Tuple										=       {name: i for i, name in enumerate(__lowerCamelCase     )}
														restype_atomaa_to_atomaa_list.append(
														    [(atom_name_to_idxaa[name] if name in atom_name_to_idxaa else 0) for name in rc.atom_types]     )

														restype_atomaa_mask_list.append([(1.0 if name else 0.0) for name in atom_names]     )

							# Add dummy mapping for restype 'UNK'
							restype_atomaa_to_atomaa_list.append([0] * 1_4     )
							restype_atomaa_to_atomaa_list.append([0] * 3_7     )
							restype_atomaa_mask_list.append([0.0] * 1_4     )

							__snake_case		:      int										=       torch.tensor(
							    __lowerCamelCase							,     dtype=torch.intaa							,     device=protein["aatype"].device							,     )
							__snake_case		:      List[str]										=       torch.tensor(
							    __lowerCamelCase							,     dtype=torch.intaa							,     device=protein["aatype"].device							,     )
							__snake_case		:      Optional[Any]										=       torch.tensor(
							    __lowerCamelCase							,     dtype=torch.floataa							,     device=protein["aatype"].device							,     )
							__snake_case		:      Optional[int]										=       protein["aatype"].to(torch.long     )

							# create the mapping for (residx, atom14) --> atom37, i.e. an array
							# with shape (num_res, 14) containing the atom37 indices for this protein
							__snake_case		:      Optional[Any]										=       restype_atomaa_to_atomaa[protein_aatype]
							__snake_case		:      Tuple										=       restype_atomaa_mask[protein_aatype]

							__snake_case		:      Optional[Any]										=       residx_atomaa_mask
							__snake_case		:      Union[str, Any]										=       residx_atomaa_to_atomaa.long()

							# create the gather indices for mapping back
							__snake_case		:      Dict										=       restype_atomaa_to_atomaa[protein_aatype]
							__snake_case		:      Dict										=       residx_atomaa_to_atomaa.long()

							# create the corresponding mask
							__snake_case		:      List[str]										=       torch.zeros([2_1, 3_7]							,     dtype=torch.floataa							,     device=protein["aatype"].device     )
							for restype, restype_letter in enumerate(rc.restypes     ):
														__snake_case		:      List[str]										=       rc.restype_atoa[restype_letter]
														__snake_case		:      List[Any]										=       rc.residue_atoms[restype_name]
														for atom_name in atom_names:
																					__snake_case		:      Union[str, Any]										=       rc.atom_order[atom_name]
																					__snake_case		:      str										=       1

							__snake_case		:      List[str]										=       restype_atomaa_mask[protein_aatype]
							__snake_case		:      List[str]										=       residx_atomaa_mask

							return protein







def     lowerCAmelCase_					( __lowerCamelCase     ):
							__snake_case		:      str										=       tree_map(lambda __lowerCamelCase     : torch.tensor(__lowerCamelCase							,     device=batch["aatype"].device     )							,     __lowerCamelCase							,     np.ndarray     )
							__snake_case		:      str										=       tensor_tree_map(lambda __lowerCamelCase     : np.array(__lowerCamelCase     )							,     make_atomaa_masks(__lowerCamelCase     )     )
							return out

