
"""simple docstring"""




import dataclasses
import re
import string
from typing import Any, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple

import numpy as np

from . import residue_constants


lowerCAmelCase__							=				Mapping[str, np.ndarray]
lowerCAmelCase__							=				Mapping[str, Any]  # Is a nested dict.
lowerCAmelCase__							=				0.01





@dataclasses.dataclass(frozen=__lowercase   )
class  snake_case     :
							UpperCAmelCase__						     =      42  # [num_res, num_atom_type, 3]

							# Amino-acid type for each residue represented as an integer between 0 and
							# 20, where 20 is 'X'.
							UpperCAmelCase__						     =      42  # [num_res]

							# Binary float mask to indicate presence of a particular atom. 1.0 if an atom
							# is present and 0.0 if not. This should be used for loss masking.
							UpperCAmelCase__						     =      42  # [num_res, num_atom_type]

							# Residue index as used in PDB. It is not necessarily continuous or 0-indexed.
							UpperCAmelCase__						     =      42  # [num_res]

							# B-factors, or temperature factors, of each residue (in sq. angstroms units),
							# representing the displacement of the residue from its ground truth mean
							# value.
							UpperCAmelCase__						     =      42  # [num_res, num_atom_type]

							# Chain indices for multi-chain predictions
							UpperCAmelCase__						     =      None

							# Optional remark about the protein. Included as a comment in output PDB
							# files
							UpperCAmelCase__						     =      None

							# Templates used to generate this protein (prediction-only)
							UpperCAmelCase__						     =      None

							# Chain corresponding to each parent
							UpperCAmelCase__						     =      None







def        _lowerCamelCase ( __a    ):
		SCREAMING_SNAKE_CASE_						 = r'''(\[[A-Z]+\]\n)'''
		SCREAMING_SNAKE_CASE_						 = [tag.strip() for tag in re.split(__a,							__a    ) if len(__a    ) > 0]
		SCREAMING_SNAKE_CASE_						 = zip(tags[0::2],							[l.split('''\n'''    ) for l in tags[1::2]]    )

		SCREAMING_SNAKE_CASE_						 = ["N", "CA", "C"]
		SCREAMING_SNAKE_CASE_						 = None
		SCREAMING_SNAKE_CASE_						 = None
		SCREAMING_SNAKE_CASE_						 = None
		for g in groups:
				if "[PRIMARY]" == g[0]:
						SCREAMING_SNAKE_CASE_						 = g[1][0].strip()
						for i in range(len(__a    )    ):
								if seq[i] not in residue_constants.restypes:
										SCREAMING_SNAKE_CASE_						 = '''X'''  # FIXME: strings are immutable
						SCREAMING_SNAKE_CASE_						 = np.array(
						    [residue_constants.restype_order.get(__a,							residue_constants.restype_num    ) for res_symbol in seq]    )
				elif "[TERTIARY]" == g[0]:
						SCREAMING_SNAKE_CASE_						 = []
						for axis in range(3    ):
								tertiary.append(list(map(__a,							g[1][axis].split()    )    )    )
						SCREAMING_SNAKE_CASE_						 = np.array(__a    )
						SCREAMING_SNAKE_CASE_						 = np.zeros((len(tertiary[0]    ) // 3, residue_constants.atom_type_num, 3)    ).astype(np.floataa    )
						for i, atom in enumerate(__a    ):
								SCREAMING_SNAKE_CASE_						 = np.transpose(tertiary_np[:, i::3]    )
						atom_positions *= PICO_TO_ANGSTROM
				elif "[MASK]" == g[0]:
						SCREAMING_SNAKE_CASE_						 = np.array(list(map({'''-''': 0, '''+''': 1}.get,							g[1][0].strip()    )    )    )
						SCREAMING_SNAKE_CASE_						 = np.zeros(
						    (
						        len(__a    ),
						        residue_constants.atom_type_num,
						    )    ).astype(np.floataa    )
						for i, atom in enumerate(__a    ):
								SCREAMING_SNAKE_CASE_						 = 1
						atom_mask *= mask[..., None]

		assert aatype is not None

		return Protein(
		    atom_positions=__a,							atom_mask=__a,							aatype=__a,							residue_index=np.arange(len(__a    )    ),							b_factors=__a,							)



def        _lowerCamelCase ( __a,							__a = 0    ):
		SCREAMING_SNAKE_CASE_						 = []

		SCREAMING_SNAKE_CASE_						 = prot.remark
		if remark is not None:
				pdb_headers.append(F'REMARK {remark}'    )

		SCREAMING_SNAKE_CASE_						 = prot.parents
		SCREAMING_SNAKE_CASE_						 = prot.parents_chain_index
		if parents is not None and parents_chain_index is not None:
				SCREAMING_SNAKE_CASE_						 = [p for i, p in zip(__a,							__a    ) if i == chain_id]

		if parents is None or len(__a    ) == 0:
				SCREAMING_SNAKE_CASE_						 = ['''N/A''']

		pdb_headers.append(F'PARENT {" ".join(__a    )}'    )

		return pdb_headers



def        _lowerCamelCase ( __a,							__a    ):
		SCREAMING_SNAKE_CASE_						 = []
		SCREAMING_SNAKE_CASE_						 = pdb_str.split('''\n'''    )

		SCREAMING_SNAKE_CASE_						 = prot.remark
		if remark is not None:
				out_pdb_lines.append(F'REMARK {remark}'    )

		SCREAMING_SNAKE_CASE_						 = 42
		if prot.parents is not None and len(prot.parents    ) > 0:
				SCREAMING_SNAKE_CASE_						 = []
				if prot.parents_chain_index is not None:
						SCREAMING_SNAKE_CASE_						 = {}
						for p, i in zip(prot.parents,							prot.parents_chain_index    ):
								parent_dict.setdefault(str(__a    ),							[]    )
								parent_dict[str(__a    )].append(__a    )

						SCREAMING_SNAKE_CASE_						 = max([int(__a    ) for chain_idx in parent_dict]    )
						for i in range(max_idx + 1    ):
								SCREAMING_SNAKE_CASE_						 = parent_dict.get(str(__a    ),							['''N/A''']    )
								parents_per_chain.append(__a    )
				else:
						parents_per_chain.append(list(prot.parents    )    )
		else:
				SCREAMING_SNAKE_CASE_						 = [['''N/A''']]

		def make_parent_line(__a    ) -> str:
				return F'PARENT {" ".join(__a    )}'

		out_pdb_lines.append(make_parent_line(parents_per_chain[0]    )    )

		SCREAMING_SNAKE_CASE_						 = 0
		for i, l in enumerate(__a    ):
				if "PARENT" not in l and "REMARK" not in l:
						out_pdb_lines.append(__a    )
				if "TER" in l and "END" not in lines[i + 1]:
						chain_counter += 1
						if not chain_counter >= len(__a    ):
								SCREAMING_SNAKE_CASE_						 = parents_per_chain[chain_counter]
						else:
								SCREAMING_SNAKE_CASE_						 = ['''N/A''']

						out_pdb_lines.append(make_parent_line(__a    )    )

		return "\n".join(__a    )



def        _lowerCamelCase ( __a    ):
		SCREAMING_SNAKE_CASE_						 = residue_constants.restypes + ['''X''']

		def res_atoa(__a    ) -> str:
				return residue_constants.restype_atoa.get(restypes[r],							'''UNK'''    )

		SCREAMING_SNAKE_CASE_						 = residue_constants.atom_types

		SCREAMING_SNAKE_CASE_						 = []

		SCREAMING_SNAKE_CASE_						 = prot.atom_mask
		SCREAMING_SNAKE_CASE_						 = prot.aatype
		SCREAMING_SNAKE_CASE_						 = prot.atom_positions
		SCREAMING_SNAKE_CASE_						 = prot.residue_index.astype(np.intaa    )
		SCREAMING_SNAKE_CASE_						 = prot.b_factors
		SCREAMING_SNAKE_CASE_						 = prot.chain_index

		if np.any(aatype > residue_constants.restype_num    ):
				raise ValueError('''Invalid aatypes.'''    )

		SCREAMING_SNAKE_CASE_						 = get_pdb_headers(__a    )
		if len(__a    ) > 0:
				pdb_lines.extend(__a    )

		SCREAMING_SNAKE_CASE_						 = aatype.shape[0]
		SCREAMING_SNAKE_CASE_						 = 1
		SCREAMING_SNAKE_CASE_						 = 0
		SCREAMING_SNAKE_CASE_						 = string.ascii_uppercase
		SCREAMING_SNAKE_CASE_						 = None
		# Add all atom sites.
		for i in range(__a    ):
				SCREAMING_SNAKE_CASE_						 = res_atoa(aatype[i]    )
				for atom_name, pos, mask, b_factor in zip(__a,							atom_positions[i],							atom_mask[i],							b_factors[i]    ):
						if mask < 0.5:
								continue

						SCREAMING_SNAKE_CASE_						 = '''ATOM'''
						SCREAMING_SNAKE_CASE_						 = atom_name if len(__a    ) == 4 else F' {atom_name}'
						SCREAMING_SNAKE_CASE_						 = ''''''
						SCREAMING_SNAKE_CASE_						 = ''''''
						SCREAMING_SNAKE_CASE_						 = 1.0_0
						SCREAMING_SNAKE_CASE_						 = atom_name[0]  # Protein supports only C, N, O, S, this works.
						SCREAMING_SNAKE_CASE_						 = ''''''

						SCREAMING_SNAKE_CASE_						 = '''A'''
						if chain_index is not None:
								SCREAMING_SNAKE_CASE_						 = chain_tags[chain_index[i]]

						# PDB is a columnar format, every space matters here!
						SCREAMING_SNAKE_CASE_						 = (
						    F'{record_type:<6}{atom_index:>5} {name:<4}{alt_loc:>1}'
						    F'{res_name_a:>3} {chain_tag:>1}'
						    F'{residue_index[i]:>4}{insertion_code:>1}   '
						    F'{pos[0]:>8.3f}{pos[1]:>8.3f}{pos[2]:>8.3f}'
						    F'{occupancy:>6.2f}{b_factor:>6.2f}          '
						    F'{element:>2}{charge:>2}'
						)
						pdb_lines.append(__a    )
						atom_index += 1

				SCREAMING_SNAKE_CASE_						 = i == n - 1
				if chain_index is not None:
						if i != n - 1 and chain_index[i + 1] != prev_chain_index:
								SCREAMING_SNAKE_CASE_						 = True
								SCREAMING_SNAKE_CASE_						 = chain_index[i + 1]

				if should_terminate:
						# Close the chain.
						SCREAMING_SNAKE_CASE_						 = '''TER'''
						SCREAMING_SNAKE_CASE_						 = (
						    F'{chain_end:<6}{atom_index:>5}      {res_atoa(aatype[i]    ):>3} {chain_tag:>1}{residue_index[i]:>4}'
						)
						pdb_lines.append(__a    )
						atom_index += 1

						if i != n - 1:
								# "prev" is a misnomer here. This happens at the beginning of
								# each new chain.
								pdb_lines.extend(get_pdb_headers(__a,							__a    )    )

		pdb_lines.append('''END'''    )
		pdb_lines.append(''''''    )
		return "\n".join(__a    )



def        _lowerCamelCase ( __a    ):
		return residue_constants.STANDARD_ATOM_MASK[prot.aatype]



def        _lowerCamelCase ( __a,							__a,							__a = None,							__a = None,							__a = None,							__a = None,							__a = None,							):
		return Protein(
		    aatype=features['''aatype'''],							atom_positions=result['''final_atom_positions'''],							atom_mask=result['''final_atom_mask'''],							residue_index=features['''residue_index'''] + 1,							b_factors=b_factors if b_factors is not None else np.zeros_like(result['''final_atom_mask''']    ),							chain_index=__a,							remark=__a,							parents=__a,							parents_chain_index=__a,							)

"""simple docstring"""




import os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
						from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
						import torch_xla.core.xla_model as xm



def        _lowerCamelCase ( __a    ):
		if is_torch_version('''<''',							'''2.0.0'''    ) or not hasattr(__a,							'''_dynamo'''    ):
				return False
		return isinstance(__a,							torch._dynamo.eval_frame.OptimizedModule    )



def        _lowerCamelCase ( __a,							__a = True    ):
		SCREAMING_SNAKE_CASE_						 = (torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

		SCREAMING_SNAKE_CASE_						 = is_compiled_module(__a    )
		if is_compiled:
				SCREAMING_SNAKE_CASE_						 = model
				SCREAMING_SNAKE_CASE_						 = model._orig_mod

		if is_deepspeed_available():
				options += (DeepSpeedEngine,)

		while isinstance(__a,							__a    ):
				SCREAMING_SNAKE_CASE_						 = model.module

		if not keep_fpaa_wrapper:
				SCREAMING_SNAKE_CASE_						 = getattr(__a,							'''forward'''    )
				SCREAMING_SNAKE_CASE_						 = model.__dict__.pop('''_original_forward''',							__a    )
				if original_forward is not None:
						while hasattr(__a,							'''__wrapped__'''    ):
								SCREAMING_SNAKE_CASE_						 = forward.__wrapped__
								if forward == original_forward:
										break
						SCREAMING_SNAKE_CASE_						 = forward
				if getattr(__a,							'''_converted_to_transformer_engine''',							__a    ):
						convert_model(__a,							to_transformer_engine=__a    )

		if is_compiled:
				SCREAMING_SNAKE_CASE_						 = model
				SCREAMING_SNAKE_CASE_						 = compiled_model

		return model



def        _lowerCamelCase ( ):
		PartialState().wait_for_everyone()



def        _lowerCamelCase ( __a,							__a    ):
		if PartialState().distributed_type == DistributedType.TPU:
				xm.save(__a,							__a    )
		elif PartialState().local_process_index == 0:
				torch.save(__a,							__a    )



@contextmanager
def        _lowerCamelCase ( **__a    ):
		for key, value in kwargs.items():
				SCREAMING_SNAKE_CASE_						 = str(__a    )

		yield

		for key in kwargs:
				if key.upper() in os.environ:
						del os.environ[key.upper()]



def        _lowerCamelCase ( __a    ):
		if not hasattr(__a,							'''__qualname__'''    ) and not hasattr(__a,							'''__name__'''    ):
				SCREAMING_SNAKE_CASE_						 = getattr(__a,							'''__class__''',							__a    )
		if hasattr(__a,							'''__qualname__'''    ):
				return obj.__qualname__
		if hasattr(__a,							'''__name__'''    ):
				return obj.__name__
		return str(__a    )



def        _lowerCamelCase ( __a,							__a    ):
		for key, value in source.items():
				if isinstance(__a,							__a    ):
						SCREAMING_SNAKE_CASE_						 = destination.setdefault(__a,							{}    )
						merge_dicts(__a,							__a    )
				else:
						SCREAMING_SNAKE_CASE_						 = value

		return destination



def        _lowerCamelCase ( __a = None    ):
		if port is None:
				SCREAMING_SNAKE_CASE_						 = 29_500
		with socket.socket(socket.AF_INET,							socket.SOCK_STREAM    ) as s:
				return s.connect_ex(('''localhost''', port)    ) == 0