

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


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


@dataclasses.dataclass(frozen=_UpperCAmelCase )
class  A						:







    """simple docstring"""





    lowerCamelCase			=							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'.
    lowerCamelCase			=							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.
    lowerCamelCase			=							42  # [num_res, num_atom_type]

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

    # Chain indices for multi-chain predictions
    lowerCamelCase			=							None

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

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

    # Chain corresponding to each parent
    lowerCamelCase			=							None




def 						_snake_case(    SCREAMING_SNAKE_CASE__			:	str      )							->      Protein:





    '''simple docstring'''






    A__          =   R'(\[[A-Z]+\]\n)'
    A__          =   [tag.strip() for tag in re.split(SCREAMING_SNAKE_CASE__    ,			SCREAMING_SNAKE_CASE__      ) if len(SCREAMING_SNAKE_CASE__      ) > 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(SCREAMING_SNAKE_CASE__      )      ):
                if seq[i] not in residue_constants.restypes:
                    A__          =   'X'  # FIXME: strings are immutable
            A__          =   np.array(
                [residue_constants.restype_order.get(SCREAMING_SNAKE_CASE__    ,			residue_constants.restype_num      ) for res_symbol in seq]      )
        elif "[TERTIARY]" == g[0]:
            A__          =   []
            for axis in range(3      ):
                tertiary.append(list(map(SCREAMING_SNAKE_CASE__    ,			g[1][axis].split()      )      )      )
            A__          =   np.array(SCREAMING_SNAKE_CASE__      )
            A__          =   np.zeros((len(tertiary[0]      ) // 3, residue_constants.atom_type_num, 3)      ).astype(np.floataa      )
            for i, atom in enumerate(SCREAMING_SNAKE_CASE__      ):
                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(SCREAMING_SNAKE_CASE__      ),
                    residue_constants.atom_type_num,
                )      ).astype(np.floataa      )
            for i, atom in enumerate(SCREAMING_SNAKE_CASE__      ):
                A__          =   1
            atom_mask *= mask[..., None]

    assert aatype is not None

    return Protein(
        atom_positions=SCREAMING_SNAKE_CASE__    ,			atom_mask=SCREAMING_SNAKE_CASE__    ,			aatype=SCREAMING_SNAKE_CASE__    ,			residue_index=np.arange(len(SCREAMING_SNAKE_CASE__      )      )    ,			b_factors=SCREAMING_SNAKE_CASE__    ,			)
def 						_snake_case(    SCREAMING_SNAKE_CASE__			:	Protein    ,			SCREAMING_SNAKE_CASE__			:	int = 0      )							->      List[str]:





    '''simple docstring'''






    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(SCREAMING_SNAKE_CASE__    ,			SCREAMING_SNAKE_CASE__      ) if i == chain_id]

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

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

    return pdb_headers
def 						_snake_case(    SCREAMING_SNAKE_CASE__			:	Protein    ,			SCREAMING_SNAKE_CASE__			:	str      )							->      str:





    '''simple docstring'''






    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(SCREAMING_SNAKE_CASE__      )    ,			[]      )
                parent_dict[str(SCREAMING_SNAKE_CASE__      )].append(SCREAMING_SNAKE_CASE__      )

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

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

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

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

            out_pdb_lines.append(make_parent_line(SCREAMING_SNAKE_CASE__      )      )

    return "\n".join(SCREAMING_SNAKE_CASE__      )
def 						_snake_case(    SCREAMING_SNAKE_CASE__			:	Protein      )							->      str:





    '''simple docstring'''






    A__          =   residue_constants.restypes + ['X']

    def res_atoa(SCREAMING_SNAKE_CASE__			:	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(SCREAMING_SNAKE_CASE__      )
    if len(SCREAMING_SNAKE_CASE__      ) > 0:
        pdb_lines.extend(SCREAMING_SNAKE_CASE__      )

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

            A__          =   'ATOM'
            A__          =   atom_name if len(SCREAMING_SNAKE_CASE__      ) == 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(SCREAMING_SNAKE_CASE__      )
            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(SCREAMING_SNAKE_CASE__      )
            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(SCREAMING_SNAKE_CASE__    ,			SCREAMING_SNAKE_CASE__      )      )

    pdb_lines.append('END'      )
    pdb_lines.append(''      )
    return "\n".join(SCREAMING_SNAKE_CASE__      )
def 						_snake_case(    SCREAMING_SNAKE_CASE__			:	Protein      )							->      np.ndarray:





    '''simple docstring'''






    return residue_constants.STANDARD_ATOM_MASK[prot.aatype]

def 						_snake_case(    SCREAMING_SNAKE_CASE__			:	FeatureDict    ,			SCREAMING_SNAKE_CASE__			:	ModelOutput    ,			SCREAMING_SNAKE_CASE__			:	Optional[np.ndarray] = None    ,			SCREAMING_SNAKE_CASE__			:	Optional[np.ndarray] = None    ,			SCREAMING_SNAKE_CASE__			:	Optional[str] = None    ,			SCREAMING_SNAKE_CASE__			:	Optional[Sequence[str]] = None    ,			SCREAMING_SNAKE_CASE__			:	Optional[Sequence[int]] = None    ,			)							->      Protein:





    '''simple docstring'''






    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=SCREAMING_SNAKE_CASE__    ,			remark=SCREAMING_SNAKE_CASE__    ,			parents=SCREAMING_SNAKE_CASE__    ,			parents_chain_index=SCREAMING_SNAKE_CASE__    ,			)


'''simple docstring'''




import os
from typing import Any, Callable, Dict, List, Optional, Tuple, Union

import torch
from torch import nn

from ...models.controlnet import ControlNetModel, ControlNetOutput
from ...models.modeling_utils import ModelMixin
from ...utils import logging


__SCREAMING_SNAKE_CASE       :     str     =  logging.get_logger(__name__)

class 	lowerCamelCase_	(snake_case__    ):
      '''simple docstring'''

      def __init__(       self						:				Optional[int]						,					A						:				Union[List[ControlNetModel], Tuple[ControlNetModel]]       ):
             super().__init__()
             _UpperCAmelCase     :  Optional[int]							      =    nn.ModuleList(A       )




      def      _A	(       self						:				Dict						,					A						:				torch.FloatTensor						,					A						:				Union[torch.Tensor, float, int]						,					A						:				torch.Tensor						,					A						:				List[torch.tensor]						,					A						:				List[float]						,					A						:				Optional[torch.Tensor] = None						,					A						:				Optional[torch.Tensor] = None						,					A						:				Optional[torch.Tensor] = None						,					A						:				Optional[Dict[str, Any]] = None						,					A						:				bool = False						,					A						:				bool = True						,					):
             for i, (image, scale, controlnet) in enumerate(zip(A						,					A						,					self.nets       )       ):
                    _UpperCAmelCase      ,  _UpperCAmelCase     :  str							      =    controlnet(
                        A						,					A						,					A						,					A						,					A						,					A						,					A						,					A						,					A						,					A						,					A						,					)

                    # merge samples
                    if i == 0:
                           _UpperCAmelCase      ,  _UpperCAmelCase     :  List[Any]							      =    down_samples, mid_sample
                    else:
                           _UpperCAmelCase     :  Optional[int]							      =    [
                               samples_prev + samples_curr
                               for samples_prev, samples_curr in zip(A						,					A       )
                           ]
                           mid_block_res_sample += mid_sample

             return down_block_res_samples, mid_block_res_sample




      def      _A	(       self						:				List[str]						,					A						:				Union[str, os.PathLike]						,					A						:				bool = True						,					A						:				Callable = None						,					A						:				bool = False						,					A						:				Optional[str] = None						,					):
             _UpperCAmelCase     :  str							      =    0
             _UpperCAmelCase     :  str							      =    save_directory
             for controlnet in self.nets:
                    controlnet.save_pretrained(
                        A						,					is_main_process=A						,					save_function=A						,					safe_serialization=A						,					variant=A						,					)

                    idx += 1
                    _UpperCAmelCase     :  Tuple							      =    model_path_to_save + F"""_{idx}"""





      @classmethod
      def      _A	(       cls						:				int						,					A						:				Optional[Union[str, os.PathLike]]						,					**A						:				Tuple       ):
             _UpperCAmelCase     :  str							      =    0
             _UpperCAmelCase     :  int							      =    []

             # load controlnet and append to list until no controlnet directory exists anymore
             # first controlnet has to be saved under `./mydirectory/controlnet` to be compliant with `DiffusionPipeline.from_prertained`
             # second, third, ... controlnets have to be saved under `./mydirectory/controlnet_1`, `./mydirectory/controlnet_2`, ...
             _UpperCAmelCase     :  int							      =    pretrained_model_path
             while os.path.isdir(A       ):
                    _UpperCAmelCase     :  List[str]							      =    ControlNetModel.from_pretrained(A						,					**A       )
                    controlnets.append(A       )

                    idx += 1
                    _UpperCAmelCase     :  Dict							      =    pretrained_model_path + F"""_{idx}"""

             logger.info(F"""{len(A       )} controlnets loaded from {pretrained_model_path}."""       )

             if len(A       ) == 0:
                    raise ValueError(
                        F"""No ControlNets found under {os.path.dirname(A       )}. Expected at least {pretrained_model_path + '_0'}."""       )

             return cls(A       )

