






"""simple docstring"""



from typing import Dict, Tuple, overload

import torch
import torch.types
from torch import nn

from . import residue_constants as rc
from .rigid_utils import Rigid, Rotation
from .tensor_utils import batched_gather


@overload
def   SCREAMING_SNAKE_CASE__					(     snake_case					:   Dict ,				snake_case					:   Tuple ,				snake_case					:   List[str]							)->		torch.Tensor:






     '''simple docstring'''



     ...


@overload
def   SCREAMING_SNAKE_CASE__					(     snake_case					:   int ,				snake_case					:   Union[str, Any] ,				snake_case					:   Union[str, Any]							)->		Tuple[torch.Tensor, torch.Tensor]:






     '''simple docstring'''



     ...


def   SCREAMING_SNAKE_CASE__					(     snake_case					:   Dict ,				snake_case					:   Union[str, Any] ,				snake_case					:   str							)->		Any:






     '''simple docstring'''



     UpperCAmelCase__	:						Any       =       aatype == rc.restype_order["G"]
     UpperCAmelCase__	:						Union[str, Any]       =       rc.atom_order["CA"]
     UpperCAmelCase__	:						Any       =       rc.atom_order["CB"]
     UpperCAmelCase__	:						List[str]       =       torch.where(
         is_gly[..., None].expand(*((-1,) * len(is_gly.shape							)) ,				3							) ,				all_atom_positions[..., ca_idx, :] ,				all_atom_positions[..., cb_idx, :] ,				)

     if all_atom_masks is not None:
          UpperCAmelCase__	:						Dict       =       torch.where(
              snake_case ,				all_atom_masks[..., ca_idx] ,				all_atom_masks[..., cb_idx] ,				)
          return pseudo_beta, pseudo_beta_mask
     else:
          return pseudo_beta


def   SCREAMING_SNAKE_CASE__					(     snake_case					:   Any ,				snake_case					:   Dict							)->		torch.Tensor:






     '''simple docstring'''



     UpperCAmelCase__	:						Tuple       =       batched_gather(
         snake_case ,				batch["residx_atom37_to_atom14"] ,				dim=-2 ,				no_batch_dims=len(atomaa.shape[:-2]							) ,				)

     UpperCAmelCase__	:						Tuple       =       atomaa_data * batch["atom37_atom_exists"][..., None]

     return atomaa_data


def   SCREAMING_SNAKE_CASE__					(     snake_case					:   Any							)->		torch.Tensor:






     '''simple docstring'''



     UpperCAmelCase__	:						Dict       =       template_feats["template_aatype"]
     UpperCAmelCase__	:						Any       =       template_feats["template_torsion_angles_sin_cos"]
     UpperCAmelCase__	:						Optional[Any]       =       template_feats["template_alt_torsion_angles_sin_cos"]
     UpperCAmelCase__	:						List[str]       =       template_feats["template_torsion_angles_mask"]
     UpperCAmelCase__	:						Optional[Any]       =       torch.cat(
         [
             nn.functional.one_hot(snake_case ,				22							),
             torsion_angles_sin_cos.reshape(*torsion_angles_sin_cos.shape[:-2] ,				14							),
             alt_torsion_angles_sin_cos.reshape(*alt_torsion_angles_sin_cos.shape[:-2] ,				14							),
             torsion_angles_mask,
         ] ,				dim=-1 ,				)

     return template_angle_feat


def   SCREAMING_SNAKE_CASE__					(     snake_case					:   Tuple ,				snake_case					:   Optional[int] ,				snake_case					:   Tuple ,				snake_case					:   Any ,				snake_case					:   List[str] = False ,				snake_case					:   str = 1E-2_0 ,				snake_case					:   List[str] = 1E8 ,				)->		torch.Tensor:






     '''simple docstring'''



     UpperCAmelCase__	:						Any       =       batch["template_pseudo_beta_mask"]
     UpperCAmelCase__	:						Dict       =       template_mask[..., None] * template_mask[..., None, :]

     # Compute distogram (this seems to differ slightly from Alg. 5)
     UpperCAmelCase__	:						Optional[Any]       =       batch["template_pseudo_beta"]
     UpperCAmelCase__	:						int       =       torch.sum((tpb[..., None, :] - tpb[..., None, :, :]) ** 2 ,				dim=-1 ,				keepdim=snake_case							)
     UpperCAmelCase__	:						int       =       torch.linspace(snake_case ,				snake_case ,				snake_case ,				device=tpb.device							) ** 2
     UpperCAmelCase__	:						Tuple       =       torch.cat([lower[1:], lower.new_tensor([inf]							)] ,				dim=-1							)
     UpperCAmelCase__	:						List[Any]       =       ((dgram > lower) * (dgram < upper)).type(dgram.dtype							)

     UpperCAmelCase__	:						Optional[Any]       =       [dgram, template_mask_ad[..., None]]

     UpperCAmelCase__	:						Tuple       =       nn.functional.one_hot(
         batch["template_aatype"] ,				rc.restype_num + 2 ,				)

     UpperCAmelCase__	:						Optional[int]       =       batch["template_aatype"].shape[-1]
     to_concat.append(aatype_one_hot[..., None, :, :].expand(*aatype_one_hot.shape[:-2] ,				snake_case ,				-1 ,				-1							)							)
     to_concat.append(aatype_one_hot[..., None, :].expand(*aatype_one_hot.shape[:-2] ,				-1 ,				snake_case ,				-1							)							)

     UpperCAmelCase__       ,							UpperCAmelCase__       ,							UpperCAmelCase__	:						str       =       [rc.atom_order[a] for a in ["N", "CA", "C"]]
     UpperCAmelCase__	:						Dict       =       Rigid.make_transform_from_reference(
         n_xyz=batch["template_all_atom_positions"][..., n, :] ,				ca_xyz=batch["template_all_atom_positions"][..., ca, :] ,				c_xyz=batch["template_all_atom_positions"][..., c, :] ,				eps=snake_case ,				)
     UpperCAmelCase__	:						List[str]       =       rigids.get_trans()[..., None, :, :]
     UpperCAmelCase__	:						Tuple       =       rigids[..., None].invert_apply(snake_case							)

     UpperCAmelCase__	:						List[Any]       =       torch.rsqrt(eps + torch.sum(rigid_vec**2 ,				dim=-1							)							)

     UpperCAmelCase__	:						Dict       =       batch["template_all_atom_mask"]
     UpperCAmelCase__	:						Optional[int]       =       t_aa_masks[..., n] * t_aa_masks[..., ca] * t_aa_masks[..., c]
     UpperCAmelCase__	:						Optional[Any]       =       template_mask[..., None] * template_mask[..., None, :]

     UpperCAmelCase__	:						List[str]       =       inv_distance_scalar * template_mask_ad
     UpperCAmelCase__	:						str       =       rigid_vec * inv_distance_scalar[..., None]

     if not use_unit_vector:
          UpperCAmelCase__	:						List[str]       =       unit_vector * 0.0

     to_concat.extend(torch.unbind(unit_vector[..., None, :] ,				dim=-1							)							)
     to_concat.append(template_mask_ad[..., None]							)

     UpperCAmelCase__	:						List[str]       =       torch.cat(snake_case ,				dim=-1							)
     UpperCAmelCase__	:						int       =       act * template_mask_ad[..., None]

     return act


def   SCREAMING_SNAKE_CASE__					(     snake_case					:   int							)->		torch.Tensor:






     '''simple docstring'''



     UpperCAmelCase__	:						int       =       nn.functional.one_hot(batch["extra_msa"] ,				23							)
     UpperCAmelCase__	:						List[Any]       =       [
         msa_ahot,
         batch["extra_has_deletion"].unsqueeze(-1							),
         batch["extra_deletion_value"].unsqueeze(-1							),
     ]
     return torch.cat(snake_case ,				dim=-1							)


def   SCREAMING_SNAKE_CASE__					(     snake_case					:   Tuple ,				snake_case					:   List[Any] ,				snake_case					:   Dict ,				snake_case					:   Optional[int] ,				)->		Rigid:






     '''simple docstring'''



     UpperCAmelCase__	:						List[Any]       =       rrgdf[aatype, ...]

     # [*, N, 8] transformations, i.e.
     #   One [*, N, 8, 3, 3] rotation matrix and
     #   One [*, N, 8, 3]    translation matrix
     UpperCAmelCase__	:						List[str]       =       r.from_tensor_axa(snake_case							)

     UpperCAmelCase__	:						Dict       =       alpha.new_zeros((*((1,) * len(alpha.shape[:-1]							)), 2)							)
     UpperCAmelCase__	:						Tuple       =       1

     # [*, N, 8, 2]
     UpperCAmelCase__	:						int       =       torch.cat([bb_rot.expand(*alpha.shape[:-2] ,				-1 ,				-1							), alpha] ,				dim=-2							)

     # [*, N, 8, 3, 3]
     # Produces rotation matrices of the form:
     # [
     #   [1, 0  , 0  ],
     #   [0, a_2,-a_1],
     #   [0, a_1, a_2]
     # ]
     # This follows the original code rather than the supplement, which uses
     # different indices.

     UpperCAmelCase__	:						int       =       alpha.new_zeros(default_r.get_rots().get_rot_mats().shape							)
     UpperCAmelCase__	:						str       =       1
     UpperCAmelCase__	:						List[str]       =       alpha[..., 1]
     UpperCAmelCase__	:						List[Any]       =       -alpha[..., 0]
     UpperCAmelCase__	:						Optional[Any]       =       alpha

     UpperCAmelCase__	:						Union[str, Any]       =       default_r.compose(Rigid(Rotation(rot_mats=snake_case							) ,				snake_case							)							)

     UpperCAmelCase__	:						Any       =       all_frames[..., 5]
     UpperCAmelCase__	:						int       =       all_frames[..., 6]
     UpperCAmelCase__	:						Optional[int]       =       all_frames[..., 7]

     UpperCAmelCase__	:						Optional[Any]       =       all_frames[..., 4]
     UpperCAmelCase__	:						int       =       chia_frame_to_bb.compose(snake_case							)
     UpperCAmelCase__	:						Dict       =       chia_frame_to_bb.compose(snake_case							)
     UpperCAmelCase__	:						int       =       chia_frame_to_bb.compose(snake_case							)

     UpperCAmelCase__	:						Tuple       =       Rigid.cat(
         [
             all_frames[..., :5],
             chia_frame_to_bb.unsqueeze(-1							),
             chia_frame_to_bb.unsqueeze(-1							),
             chia_frame_to_bb.unsqueeze(-1							),
         ] ,				dim=-1 ,				)

     UpperCAmelCase__	:						Any       =       r[..., None].compose(snake_case							)

     return all_frames_to_global



def   SCREAMING_SNAKE_CASE__					(     snake_case					:   Union[str, Any] ,				snake_case					:   int ,				snake_case					:   Dict ,				snake_case					:   Optional[Any] ,				snake_case					:   Dict ,				snake_case					:   int ,				)->		torch.Tensor:






     '''simple docstring'''



     UpperCAmelCase__	:						int       =       group_idx[aatype, ...]

     # [*, N, 14, 8]
     UpperCAmelCase__	:						Optional[Any]       =       nn.functional.one_hot(
         snake_case ,				num_classes=default_frames.shape[-3] ,				)

     # [*, N, 14, 8]
     UpperCAmelCase__	:						List[Any]       =       r[..., None, :] * group_mask_one_hot

     # [*, N, 14]
     UpperCAmelCase__	:						Dict       =       t_atoms_to_global.map_tensor_fn(lambda snake_case							: torch.sum(snake_case ,				dim=-1							)							)

     # [*, N, 14, 1]
     UpperCAmelCase__	:						List[str]       =       atom_mask[aatype, ...].unsqueeze(-1							)

     # [*, N, 14, 3]
     UpperCAmelCase__	:						Tuple       =       lit_positions[aatype, ...]
     UpperCAmelCase__	:						List[str]       =       t_atoms_to_global.apply(snake_case							)
     UpperCAmelCase__	:						str       =       pred_positions * atom_mask

     return pred_positions











"""simple docstring"""



#
# Sort large text files in a minimum amount of memory
#
import argparse
import os




class  lowerCAmelCase__ :
   SCREAMING_SNAKE_CASE_         ='''block_{0}.dat'''
   def __init__(		self			:  Tuple       , snake_case__			:  List[str]       ):


        '''simple docstring'''
        UpperCAmelCase__	:						List[Any]       =       filename
        UpperCAmelCase__	:						List[Any]       =       []
   def 			__a       (		self			:  Optional[Any]       , snake_case__			:  Union[str, Any]       , snake_case__			:  Union[str, Any]       ):


        '''simple docstring'''
        UpperCAmelCase__	:						Dict       =       self.BLOCK_FILENAME_FORMAT.format(snake_case__       )
        with open(snake_case__       , "w"       ) as file:
             file.write(snake_case__       )
        self.block_filenames.append(snake_case__       )
   def 			__a       (		self			:  List[str]       ):


        '''simple docstring'''
        return self.block_filenames
   def 			__a       (		self			:  Optional[Any]       , snake_case__			:  List[str]       , snake_case__			:  str=None       ):


        '''simple docstring'''
        UpperCAmelCase__	:						int       =       0
        with open(self.filename       ) as file:
             while True:
                  UpperCAmelCase__	:						Union[str, Any]       =       file.readlines(snake_case__       )

                  if lines == []:
                       break

                  if sort_key is None:
                       lines.sort()
                  else:
                       lines.sort(key=snake_case__       )

                  self.write_block("".join(snake_case__       )       , snake_case__       )
                  i += 1




   def 			__a       (		self			:  Optional[Any]       ):


        '''simple docstring'''
        map(os.remove       , self.block_filenames       )




class  lowerCAmelCase__ :
   def 			__a       (		self			:  Union[str, Any]       , snake_case__			:  Optional[Any]       ):


        '''simple docstring'''
        UpperCAmelCase__	:						Any       =       -1
        UpperCAmelCase__	:						Tuple       =       None

        for i in range(len(snake_case__       )       ):
             if min_str is None or choices[i] < min_str:
                  UpperCAmelCase__	:						List[str]       =       i

        return min_index




class  lowerCAmelCase__ :
   def __init__(		self			:  Dict       , snake_case__			:  Optional[Any]       ):


        '''simple docstring'''
        UpperCAmelCase__	:						Optional[Any]       =       files
        UpperCAmelCase__	:						str       =       set()
        UpperCAmelCase__	:						Any       =       len(snake_case__       )
        UpperCAmelCase__	:						Tuple       =       {i: None for i in range(self.num_buffers       )}
   def 			__a       (		self			:  List[str]       ):


        '''simple docstring'''
        return {
            i: self.buffers[i] for i in range(self.num_buffers       ) if i not in self.empty
        }
   def 			__a       (		self			:  str       ):


        '''simple docstring'''
        for i in range(self.num_buffers       ):
             if self.buffers[i] is None and i not in self.empty:
                  UpperCAmelCase__	:						Tuple       =       self.files[i].readline()

                  if self.buffers[i] == "":
                       self.empty.add(snake_case__       )
                       self.files[i].close()

        if len(self.empty       ) == self.num_buffers:
             return False

        return True




   def 			__a       (		self			:  List[str]       , snake_case__			:  Tuple       ):


        '''simple docstring'''
        UpperCAmelCase__	:						Any       =       self.buffers[index]
        UpperCAmelCase__	:						Union[str, Any]       =       None

        return value




class  lowerCAmelCase__ :
   def __init__(		self			:  Optional[Any]       , snake_case__			:  int       ):


        '''simple docstring'''
        UpperCAmelCase__	:						str       =       merge_strategy
   def 			__a       (		self			:  Dict       , snake_case__			:  Dict       , snake_case__			:  Optional[int]       , snake_case__			:  Optional[int]       ):


        '''simple docstring'''
        UpperCAmelCase__	:						Dict       =       FilesArray(self.get_file_handles(snake_case__       , snake_case__       )       )
        with open(snake_case__       , "w"       , snake_case__       ) as outfile:
             while buffers.refresh():
                  UpperCAmelCase__	:						Any       =       self.merge_strategy.select(buffers.get_dict()       )
                  outfile.write(buffers.unshift(snake_case__       )       )




   def 			__a       (		self			:  Tuple       , snake_case__			:  Dict       , snake_case__			:  List[str]       ):


        '''simple docstring'''
        UpperCAmelCase__	:						str       =       {}

        for i in range(len(snake_case__       )       ):
             UpperCAmelCase__	:						List[Any]       =       open(filenames[i]       , "r"       , snake_case__       )  # noqa: UP015

        return files




class  lowerCAmelCase__ :
   def __init__(		self			:  str       , snake_case__			:  Optional[Any]       ):


        '''simple docstring'''
        UpperCAmelCase__	:						List[str]       =       block_size
   def 			__a       (		self			:  Optional[Any]       , snake_case__			:  Optional[int]       , snake_case__			:  Any=None       ):


        '''simple docstring'''
        UpperCAmelCase__	:						Union[str, Any]       =       self.get_number_blocks(snake_case__       , self.block_size       )
        UpperCAmelCase__	:						Union[str, Any]       =       FileSplitter(snake_case__       )
        splitter.split(self.block_size       , snake_case__       )

        UpperCAmelCase__	:						int       =       FileMerger(NWayMerge()       )
        UpperCAmelCase__	:						int       =       self.block_size / (num_blocks + 1)
        merger.merge(splitter.get_block_filenames()       , filename + ".out"       , snake_case__       )

        splitter.cleanup()




   def 			__a       (		self			:  Union[str, Any]       , snake_case__			:  Optional[int]       , snake_case__			:  Optional[Any]       ):


        '''simple docstring'''
        return (os.stat(snake_case__       ).st_size / block_size) + 1




def   SCREAMING_SNAKE_CASE__					(     snake_case					:   Tuple							)->		Union[str, Any]:






     '''simple docstring'''



     if string[-1].lower() == "k":
          return int(string[:-1]							) * 1024
     elif string[-1].lower() == "m":
          return int(string[:-1]							) * 1024 * 1024
     elif string[-1].lower() == "g":
          return int(string[:-1]							) * 1024 * 1024 * 1024
     else:
          return int(snake_case							)


def   SCREAMING_SNAKE_CASE__					(     )->		Any:






     '''simple docstring'''



     UpperCAmelCase__	:						Any       =       argparse.ArgumentParser()
     parser.add_argument(
         "-m" ,				"--mem" ,				help="amount of memory to use for sorting" ,				default="100M"							)
     parser.add_argument(
         "filename" ,				metavar="<filename>" ,				nargs=1 ,				help="name of file to sort"							)
     UpperCAmelCase__	:						int       =       parser.parse_args()

     UpperCAmelCase__	:						Optional[int]       =       ExternalSort(parse_memory(args.mem							)							)
     sorter.sort(args.filename[0]							)


if __name__ == "__main__":
 main()


