



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

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

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

    # Chain indices for multi-chain predictions
    lowercase__				=					None

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

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

    # Chain corresponding to each parent
    lowercase__				=					None
def    lowerCamelCase__ (		_A       ):
  a		:      Optional[int]       =  r'(\[[A-Z]+\]\n)'
  a		:      int       =  [tag.strip() for tag in re.split(lowerCamelCase__				,    lowerCamelCase__       ) if len(lowerCamelCase__       ) > 0]
  a		:      int       =  zip(tags[0::2]				,    [l.split('\n'       ) for l in tags[1::2]]       )

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

  assert aatype is not None

  return Protein(
      atom_positions=lowerCamelCase__				,    atom_mask=lowerCamelCase__				,    aatype=lowerCamelCase__				,    residue_index=np.arange(len(lowerCamelCase__       )       )				,    b_factors=lowerCamelCase__				,    )
def    lowerCamelCase__ (		_A				,    _A = 0       ):
  a		:      str       =  []

  a		:      Union[str, Any]       =  prot.remark
  if remark is not None:
    pdb_headers.append(f"""REMARK {remark}"""       )

  a		:      Any       =  prot.parents
  a		:      List[str]       =  prot.parents_chain_index
  if parents is not None and parents_chain_index is not None:
    a		:      Union[str, Any]       =  [p for i, p in zip(lowerCamelCase__				,    lowerCamelCase__       ) if i == chain_id]

  if parents is None or len(lowerCamelCase__       ) == 0:
    a		:      Dict       =  ['N/A']

  pdb_headers.append(f"""PARENT {" ".join(lowerCamelCase__       )}"""       )

  return pdb_headers
def    lowerCamelCase__ (		_A				,    _A       ):
  a		:      Any       =  []
  a		:      str       =  pdb_str.split('\n'       )

  a		:      List[str]       =  prot.remark
  if remark is not None:
    out_pdb_lines.append(f"""REMARK {remark}"""       )

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

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

  def make_parent_line(_A       ) -> str:
    return f"""PARENT {" ".join(lowerCamelCase__       )}"""

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

  a		:      Optional[int]       =  0
  for i, l in enumerate(lowerCamelCase__       ):
    if "PARENT" not in l and "REMARK" not in l:
      out_pdb_lines.append(lowerCamelCase__       )
    if "TER" in l and "END" not in lines[i + 1]:
      chain_counter += 1
      if not chain_counter >= len(lowerCamelCase__       ):
        a		:      List[Any]       =  parents_per_chain[chain_counter]
      else:
        a		:      Union[str, Any]       =  ['N/A']

      out_pdb_lines.append(make_parent_line(lowerCamelCase__       )       )

  return "\n".join(lowerCamelCase__       )
def    lowerCamelCase__ (		_A       ):
  a		:      Tuple       =  residue_constants.restypes + ['X']

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

  a		:      Tuple       =  residue_constants.atom_types

  a		:      Optional[Any]       =  []

  a		:      Optional[Any]       =  prot.atom_mask
  a		:      str       =  prot.aatype
  a		:      List[str]       =  prot.atom_positions
  a		:      str       =  prot.residue_index.astype(np.intaa       )
  a		:      Union[str, Any]       =  prot.b_factors
  a		:      Tuple       =  prot.chain_index

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

  a		:      List[str]       =  get_pdb_headers(lowerCamelCase__       )
  if len(lowerCamelCase__       ) > 0:
    pdb_lines.extend(lowerCamelCase__       )

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

      a		:      int       =  'ATOM'
      a		:      Dict       =  atom_name if len(lowerCamelCase__       ) == 4 else f""" {atom_name}"""
      a		:      Union[str, Any]       =  ''
      a		:      Optional[int]       =  ''
      a		:      Optional[int]       =  1.00
      a		:      Dict       =  atom_name[0]  # Protein supports only C, N, O, S, this works.
      a		:      Tuple       =  ''

      a		:      int       =  'A'
      if chain_index is not None:
        a		:      Optional[int]       =  chain_tags[chain_index[i]]

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

    a		:      Tuple       =  i == n - 1
    if chain_index is not None:
      if i != n - 1 and chain_index[i + 1] != prev_chain_index:
        a		:      Tuple       =  True
        a		:      List[str]       =  chain_index[i + 1]

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

  pdb_lines.append('END'       )
  pdb_lines.append(''       )
  return "\n".join(lowerCamelCase__       )
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=lowerCamelCase__				,    remark=lowerCamelCase__				,    parents=lowerCamelCase__				,    parents_chain_index=lowerCamelCase__				,    )




'''simple docstring'''






from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


lowerCAmelCase:      Optional[Any]		 =		logging.get_logger(__name__)

@add_end_docstrings(lowerCamelCase__			)
class 							a__(      lowerCamelCase__			):







    def __init__(   self				:						Union[str, Any]		,    **__snake_case				:						List[str]			):
      super().__init__(**__snake_case			)

      if self.framework != "pt":
        raise ValueError(F"""The {self.__class__} is only available in PyTorch."""			)
      # No specific FOR_XXX available yet







    def __call__(   self				:						Any		,    __snake_case				:						Union[np.ndarray, bytes, str]		,    **__snake_case				:						Optional[Any]			):
      return super().__call__(__snake_case		,    **__snake_case			)







    def     lowercase_ (   self				:						Dict		,    **__snake_case				:						Optional[int]			):
      a		:      str       =  {}
      if "candidate_labels" in kwargs:
        a		:      Optional[int]       =  kwargs['candidate_labels']
      if "hypothesis_template" in kwargs:
        a		:      Any       =  kwargs['hypothesis_template']

      return preprocess_params, {}, {}







    def     lowercase_ (   self				:						int		,    __snake_case				:						Optional[Any]		,    __snake_case				:						List[Any]=None		,    __snake_case				:						List[str]="This is a sound of {}."			):
      if isinstance(__snake_case		,    __snake_case			):
        if audio.startswith('http://'			) or audio.startswith('https://'			):
          # We need to actually check for a real protocol, otherwise it's impossible to use a local file
          # like http_huggingface_co.png
          a		:      List[Any]       =  requests.get(__snake_case			).content
        else:
          with open(__snake_case		,    'rb'			) as f:
            a		:      List[str]       =  f.read()

      if isinstance(__snake_case		,    __snake_case			):
        a		:      Optional[Any]       =  ffmpeg_read(__snake_case		,    self.feature_extractor.sampling_rate			)

      if not isinstance(__snake_case		,    np.ndarray			):
        raise ValueError('We expect a numpy ndarray as input'			)
      if len(audio.shape			) != 1:
        raise ValueError('We expect a single channel audio input for ZeroShotAudioClassificationPipeline'			)

      a		:      str       =  self.feature_extractor(
          [audio]		,    sampling_rate=self.feature_extractor.sampling_rate		,    return_tensors='pt'			)
      a		:      Tuple       =  candidate_labels
      a		:      Union[str, Any]       =  [hypothesis_template.format(__snake_case			) for x in candidate_labels]
      a		:      int       =  self.tokenizer(__snake_case		,    return_tensors=self.framework		,    padding=__snake_case			)
      a		:      Any       =  [text_inputs]
      return inputs







    def     lowercase_ (   self				:						Optional[Any]		,    __snake_case				:						str			):
      a		:      Any       =  model_inputs.pop('candidate_labels'			)
      a		:      List[Any]       =  model_inputs.pop('text_inputs'			)
      if isinstance(text_inputs[0]		,    __snake_case			):
        a		:      str       =  text_inputs[0]
      else:
        # Batching case.
        a		:      Dict       =  text_inputs[0][0]

      a		:      List[Any]       =  self.model(**__snake_case		,    **__snake_case			)

      a		:      Any       =  {
          'candidate_labels': candidate_labels,
          'logits': outputs.logits_per_audio,
      }
      return model_outputs







    def     lowercase_ (   self				:						Union[str, Any]		,    __snake_case				:						str			):
      a		:      Optional[int]       =  model_outputs.pop('candidate_labels'			)
      a		:      List[str]       =  model_outputs['logits'][0]

      if self.framework == "pt":
        a		:      str       =  logits.softmax(dim=0			)
        a		:      List[Any]       =  probs.tolist()
      else:
        raise ValueError('`tf` framework not supported.'			)

      a		:      Dict       =  [
          {'score': score, 'label': candidate_label}
          for score, candidate_label in sorted(zip(__snake_case		,    __snake_case			)		,    key=lambda __snake_case			: -x[0]			)
      ]
      return result