


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:						List[str]   				=       Mapping[str, np.ndarray]
_lowerCAmelCase:						Optional[Any]   				=       Mapping[str, Any]  # Is a nested dict.
_lowerCAmelCase:						Any   				=       0.01




@dataclasses.dataclass(frozen=_SCREAMING_SNAKE_CASE   )
class       lowercase_					:
				snake_case	    =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'.
				snake_case	    =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.
				snake_case	    =42  # [num_res, num_atom_type]

				# Residue index as used in PDB. It is not necessarily continuous or 0-indexed.
				snake_case	    =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.
				snake_case	    =42  # [num_res, num_atom_type]

				# Chain indices for multi-chain predictions
				snake_case	    =None

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

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

				# Chain corresponding to each parent
				snake_case	    =None







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

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

							assert aatype is not None

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






def   _lowercase(					__a				:						Protein    ,     __a				:						int = 0     ):
							a__      =[]

							a__      =prot.remark
							if remark is not None:
														pdb_headers.append(f"""REMARK {remark}"""     )

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

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

							pdb_headers.append(f"""PARENT {' '.join(_UpperCAmelCase     )}"""     )

							return pdb_headers






def   _lowercase(					__a				:						Protein    ,     __a				:						str     ):
							a__      =[]
							a__      =pdb_str.split('\n'     )

							a__      =prot.remark
							if remark is not None:
														out_pdb_lines.append(f"""REMARK {remark}"""     )

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

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

							def make_parent_line(__a				:						Sequence[str]     ) -> str:
														return f"""PARENT {' '.join(_UpperCAmelCase     )}"""

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

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

																					out_pdb_lines.append(make_parent_line(_UpperCAmelCase     )     )

							return "\n".join(_UpperCAmelCase     )






def   _lowercase(					__a				:						Protein     ):
							a__      =residue_constants.restypes + ['X']

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

							a__      =residue_constants.atom_types

							a__      =[]

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

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

							a__      =get_pdb_headers(_UpperCAmelCase     )
							if len(_UpperCAmelCase     ) > 0:
														pdb_lines.extend(_UpperCAmelCase     )

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

																					a__      ='ATOM'
																					a__      =atom_name if len(_UpperCAmelCase     ) == 4 else f""" {atom_name}"""
																					a__      =''
																					a__      =''
																					a__      =1.00
																					a__      =atom_name[0]  # Protein supports only C, N, O, S, this works.
																					a__      =''

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

																					# PDB is a columnar format, every space matters here!
																					a__      =(
																					    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(_UpperCAmelCase     )
																					atom_index += 1

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

														if should_terminate:
																					# Close the chain.
																					a__      ='TER'
																					a__      =(
																					    f"""{chain_end:<6}{atom_index:>5}      {res_atoa(aatype[i]     ):>3} {chain_tag:>1}{residue_index[i]:>4}"""
																					)
																					pdb_lines.append(_UpperCAmelCase     )
																					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(_UpperCAmelCase    ,     _UpperCAmelCase     )     )

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






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







def   _lowercase(					__a				:						FeatureDict    ,     __a				:						ModelOutput    ,     __a				:						Optional[np.ndarray] = None    ,     __a				:						Optional[np.ndarray] = None    ,     __a				:						Optional[str] = None    ,     __a				:						Optional[Sequence[str]] = None    ,     __a				:						Optional[Sequence[int]] = 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=_UpperCAmelCase    ,     remark=_UpperCAmelCase    ,     parents=_UpperCAmelCase    ,     parents_chain_index=_UpperCAmelCase    ,     )








import logging
import os
import threading
import time


try:
      import warnings
except ImportError:
      a__      :     Optional[Any]         =		None

try:
      import msvcrt
except ImportError:
      a__      :     List[str]         =		None

try:
      import fcntl
except ImportError:
      a__      :     Optional[int]         =		None


# Backward compatibility
# ------------------------------------------------
try:
      TimeoutError
except NameError:
      a__      :     List[str]         =		OSError


# Data
# ------------------------------------------------
a__      :     Union[str, Any]         =		[
    'Timeout',
    'BaseFileLock',
    'WindowsFileLock',
    'UnixFileLock',
    'SoftFileLock',
    'FileLock',
]

a__      :     str         =		'3.0.12'


a__      :     Optional[int]         =		None

def      UpperCAmelCase_    (    )	->				Tuple:
     '''simple docstring'''


     global _logger
     A_          =       _logger or logging.getLogger(__name__				)
     return _logger





class UpperCAmelCase_				( _SCREAMING_SNAKE_CASE ):


  def __init__(							self ,__snake_case ):



       """simple docstring"""





       A_          =       lock_file
       return None




  def __str__(							self ):



       """simple docstring"""





       A_          =       f'The file lock \'{self.lock_file}\' could not be acquired.'
       return temp





class UpperCAmelCase_				:


  def __init__(							self ,__snake_case ):



       """simple docstring"""





       A_          =       lock
       return None


  def __enter__(							self ):



       """simple docstring"""





       return self.lock




  def __exit__(							self ,__snake_case ,__snake_case ,__snake_case ):



       """simple docstring"""





       self.lock.release()
       return None





class UpperCAmelCase_				:


  def __init__(							self ,__snake_case ,__snake_case=-1 ,__snake_case=None ):



       """simple docstring"""





       A_          =       max_filename_length if max_filename_length is not None else 2_5_5
       # Hash the filename if it's too long
       A_          =       self.hash_filename_if_too_long(__snake_case ,__snake_case )
       # The path to the lock file.
       A_          =       lock_file

       # The file descriptor for the *_lock_file* as it is returned by the
       # os.open() function.
       # This file lock is only NOT None, if the object currently holds the
       # lock.
       A_          =       None

       # The default timeout value.
       A_          =       timeout

       # We use this lock primarily for the lock counter.
       A_          =       threading.Lock()

       # The lock counter is used for implementing the nested locking
       # mechanism. Whenever the lock is acquired, the counter is increased and
       # the lock is only released, when this value is 0 again.
       A_          =       0
       return None


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return self._lock_file


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return self._timeout


  @timeout.setter
  def 						__UpperCAmelCase     (							self ,__snake_case ):



       """simple docstring"""





       A_          =       float(__snake_case )
       return None


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       raise NotImplementedError()


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       raise NotImplementedError()


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return self._lock_file_fd is not None


  def 						__UpperCAmelCase     (							self ,__snake_case=None ,__snake_case=0.05 ):



       """simple docstring"""





       if timeout is None:
            A_          =       self.timeout

       # Increment the number right at the beginning.
       # We can still undo it, if something fails.
       with self._thread_lock:
            self._lock_counter += 1

       A_          =       id(self )
       A_          =       self._lock_file
       A_          =       time.time()
       try:
            while True:
                 with self._thread_lock:
                      if not self.is_locked:
                           logger().debug(f'Attempting to acquire lock {lock_id} on {lock_filename}' )
                           self._acquire()

                 if self.is_locked:
                      logger().debug(f'Lock {lock_id} acquired on {lock_filename}' )
                      break
                 elif timeout >= 0 and time.time() - start_time > timeout:
                      logger().debug(f'Timeout on acquiring lock {lock_id} on {lock_filename}' )
                      raise Timeout(self._lock_file )
                 else:
                      logger().debug(
                          f'Lock {lock_id} not acquired on {lock_filename}, waiting {poll_intervall} seconds ...' )
                      time.sleep(__snake_case )
       except:  # noqa
            # Something did go wrong, so decrement the counter.
            with self._thread_lock:
                 A_          =       max(0 ,self._lock_counter - 1 )

            raise
       return _Acquire_ReturnProxy(lock=self )


  def 						__UpperCAmelCase     (							self ,__snake_case=False ):



       """simple docstring"""





       with self._thread_lock:
            if self.is_locked:
                 self._lock_counter -= 1

                 if self._lock_counter == 0 or force:
                      A_          =       id(self )
                      A_          =       self._lock_file

                      logger().debug(f'Attempting to release lock {lock_id} on {lock_filename}' )
                      self._release()
                      A_          =       0
                      logger().debug(f'Lock {lock_id} released on {lock_filename}' )

       return None


  def __enter__(							self ):



       """simple docstring"""





       self.acquire()
       return self


  def __exit__(							self ,__snake_case ,__snake_case ,__snake_case ):



       """simple docstring"""





       self.release()
       return None


  def __del__(							self ):



       """simple docstring"""





       self.release(force=__snake_case )
       return None




  def 						__UpperCAmelCase     (							self ,__snake_case ,__snake_case ):



       """simple docstring"""





       A_          =       os.path.basename(__snake_case )
       if len(__snake_case ) > max_length and max_length > 0:
            A_          =       os.path.dirname(__snake_case )
            A_          =       str(hash(__snake_case ) )
            A_          =       filename[: max_length - len(__snake_case ) - 8] + '''...''' + hashed_filename + '''.lock'''
            return os.path.join(__snake_case ,__snake_case )
       else:
            return path





class UpperCAmelCase_				( _SCREAMING_SNAKE_CASE ):


  def __init__(							self ,__snake_case ,__snake_case=-1 ,__snake_case=None ):



       """simple docstring"""





       from .file_utils import relative_to_absolute_path

       super().__init__(__snake_case ,timeout=__snake_case ,max_filename_length=__snake_case )
       A_          =       '''\\\\?\\''' + relative_to_absolute_path(self.lock_file )


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       os.O_RDWR | os.O_CREAT | os.O_TRUNC

       try:
            A_          =       os.open(self._lock_file ,__snake_case )
       except OSError:
            pass
       else:
            try:
                 msvcrt.locking(__snake_case ,msvcrt.LK_NBLCK ,1 )
            except OSError:
                 os.close(__snake_case )
            else:
                 A_          =       fd
       return None




  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       self._lock_file_fd
       A_          =       None
       msvcrt.locking(__snake_case ,msvcrt.LK_UNLCK ,1 )
       os.close(__snake_case )

       try:
            os.remove(self._lock_file )
       # Probably another instance of the application
       # that acquired the file lock.
       except OSError:
            pass
       return None





class UpperCAmelCase_				( _SCREAMING_SNAKE_CASE ):


  def __init__(							self ,__snake_case ,__snake_case=-1 ,__snake_case=None ):



       """simple docstring"""





       A_          =       os.statvfs(os.path.dirname(__snake_case ) ).f_namemax
       super().__init__(__snake_case ,timeout=__snake_case ,max_filename_length=__snake_case )


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       os.O_RDWR | os.O_CREAT | os.O_TRUNC
       A_          =       os.open(self._lock_file ,__snake_case )

       try:
            fcntl.flock(__snake_case ,fcntl.LOCK_EX | fcntl.LOCK_NB )
       except OSError:
            os.close(__snake_case )
       else:
            A_          =       fd
       return None




  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       self._lock_file_fd
       A_          =       None
       fcntl.flock(__snake_case ,fcntl.LOCK_UN )
       os.close(__snake_case )
       return None





class UpperCAmelCase_				( _SCREAMING_SNAKE_CASE ):


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
       try:
            A_          =       os.open(self._lock_file ,__snake_case )
       except OSError:
            pass
       else:
            A_          =       fd
       return None




  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       os.close(self._lock_file_fd )
       A_          =       None

       try:
            os.remove(self._lock_file )
       # The file is already deleted and that's what we want.
       except OSError:
            pass
       return None




a__      :     Union[str, Any]         =		None

if msvcrt:
      a__      :     Dict         =		WindowsFileLock
elif fcntl:
      a__      :     Dict         =		UnixFileLock
else:
      a__      :     Optional[int]         =		SoftFileLock

      if warnings is not None:
            warnings.warn('only soft file lock is available')



