



'''simple docstring'''






from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block

@dataclass
class 				a__(   lowerCamelCase__      ):
    lowercase__					=       42

class 				a__(   nn.Module      ):







    def __init__( self					:    Union[str, Any]		,     __snake_case					:    List[Any]=3		,     __snake_case					:    List[str]=3		,     __snake_case					:    List[Any]=("DownEncoderBlock2D",)		,     __snake_case					:    List[str]=(64,)		,     __snake_case					:    Tuple=2		,     __snake_case					:    List[Any]=32		,     __snake_case					:    List[str]="silu"		,     __snake_case					:    List[str]=True		,     ):
      super().__init__()
      a		:   Optional[Any]	  =	layers_per_block

      a		:   Tuple	  =	torch.nn.Convad(
          __snake_case		,     block_out_channels[0]		,     kernel_size=3		,     stride=1		,     padding=1		,     )

      a		:   str	  =	None
      a		:   Optional[int]	  =	nn.ModuleList([]      )

      # down
      a		:   int	  =	block_out_channels[0]
      for i, down_block_type in enumerate(__snake_case      ):
        a		:   Tuple	  =	output_channel
        a		:   List[str]	  =	block_out_channels[i]
        a		:   Optional[Any]	  =	i == len(__snake_case      ) - 1

        a		:   Tuple	  =	get_down_block(
            __snake_case		,     num_layers=self.layers_per_block		,     in_channels=__snake_case		,     out_channels=__snake_case		,     add_downsample=not is_final_block		,     resnet_eps=1e-6		,     downsample_padding=0		,     resnet_act_fn=__snake_case		,     resnet_groups=__snake_case		,     attention_head_dim=__snake_case		,     temb_channels=__snake_case		,     )
        self.down_blocks.append(__snake_case      )

      # mid
      a		:   int	  =	UNetMidBlockaD(
          in_channels=block_out_channels[-1]		,     resnet_eps=1e-6		,     resnet_act_fn=__snake_case		,     output_scale_factor=1		,     resnet_time_scale_shift='default'		,     attention_head_dim=block_out_channels[-1]		,     resnet_groups=__snake_case		,     temb_channels=__snake_case		,     )

      # out
      a		:   Union[str, Any]	  =	nn.GroupNorm(num_channels=block_out_channels[-1]		,     num_groups=__snake_case		,     eps=1e-6      )
      a		:   str	  =	nn.SiLU()

      a		:   int	  =	2 * out_channels if double_z else out_channels
      a		:   Tuple	  =	nn.Convad(block_out_channels[-1]		,     __snake_case		,     3		,     padding=1      )

      a		:   Optional[int]	  =	False







    def      lowercase_			( self					:    Tuple		,     __snake_case					:    Optional[Any]      ):
      a		:   Any	  =	x
      a		:   Dict	  =	self.conv_in(__snake_case      )

      if self.training and self.gradient_checkpointing:

        def create_custom_forward(__snake_case					:    Any      ):
          def custom_forward(*__snake_case					:    Optional[int]      ):
            return module(*__snake_case      )

          return custom_forward

        # down
        if is_torch_version('>='		,     '1.11.0'      ):
          for down_block in self.down_blocks:
            a		:   Tuple	  =	torch.utils.checkpoint.checkpoint(
                create_custom_forward(__snake_case      )		,     __snake_case		,     use_reentrant=__snake_case      )
          # middle
          a		:   Dict	  =	torch.utils.checkpoint.checkpoint(
              create_custom_forward(self.mid_block      )		,     __snake_case		,     use_reentrant=__snake_case      )
        else:
          for down_block in self.down_blocks:
            a		:   Tuple	  =	torch.utils.checkpoint.checkpoint(create_custom_forward(__snake_case      )		,     __snake_case      )
          # middle
          a		:   str	  =	torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block      )		,     __snake_case      )

      else:
        # down
        for down_block in self.down_blocks:
          a		:   List[str]	  =	down_block(__snake_case      )

        # middle
        a		:   Tuple	  =	self.mid_block(__snake_case      )

      # post-process
      a		:   Any	  =	self.conv_norm_out(__snake_case      )
      a		:   Optional[int]	  =	self.conv_act(__snake_case      )
      a		:   Optional[Any]	  =	self.conv_out(__snake_case      )

      return sample

class 				a__(   nn.Module      ):







    def __init__( self					:    Union[str, Any]		,     __snake_case					:    Union[str, Any]=3		,     __snake_case					:    str=3		,     __snake_case					:    List[str]=("UpDecoderBlock2D",)		,     __snake_case					:    Any=(64,)		,     __snake_case					:    Optional[int]=2		,     __snake_case					:    Any=32		,     __snake_case					:    Tuple="silu"		,     __snake_case					:    List[str]="group"		,     ):
      super().__init__()
      a		:   List[Any]	  =	layers_per_block

      a		:   Tuple	  =	nn.Convad(
          __snake_case		,     block_out_channels[-1]		,     kernel_size=3		,     stride=1		,     padding=1		,     )

      a		:   List[str]	  =	None
      a		:   Tuple	  =	nn.ModuleList([]      )

      a		:   List[Any]	  =	in_channels if norm_type == 'spatial' else None

      # mid
      a		:   Union[str, Any]	  =	UNetMidBlockaD(
          in_channels=block_out_channels[-1]		,     resnet_eps=1e-6		,     resnet_act_fn=__snake_case		,     output_scale_factor=1		,     resnet_time_scale_shift='default' if norm_type == 'group' else norm_type		,     attention_head_dim=block_out_channels[-1]		,     resnet_groups=__snake_case		,     temb_channels=__snake_case		,     )

      # up
      a		:   Any	  =	list(reversed(__snake_case      )      )
      a		:   Optional[int]	  =	reversed_block_out_channels[0]
      for i, up_block_type in enumerate(__snake_case      ):
        a		:   Optional[Any]	  =	output_channel
        a		:   List[Any]	  =	reversed_block_out_channels[i]

        a		:   str	  =	i == len(__snake_case      ) - 1

        a		:   Union[str, Any]	  =	get_up_block(
            __snake_case		,     num_layers=self.layers_per_block + 1		,     in_channels=__snake_case		,     out_channels=__snake_case		,     prev_output_channel=__snake_case		,     add_upsample=not is_final_block		,     resnet_eps=1e-6		,     resnet_act_fn=__snake_case		,     resnet_groups=__snake_case		,     attention_head_dim=__snake_case		,     temb_channels=__snake_case		,     resnet_time_scale_shift=__snake_case		,     )
        self.up_blocks.append(__snake_case      )
        a		:   List[str]	  =	output_channel

      # out
      if norm_type == "spatial":
        a		:   str	  =	SpatialNorm(block_out_channels[0]		,     __snake_case      )
      else:
        a		:   int	  =	nn.GroupNorm(num_channels=block_out_channels[0]		,     num_groups=__snake_case		,     eps=1e-6      )
      a		:   Optional[int]	  =	nn.SiLU()
      a		:   Optional[int]	  =	nn.Convad(block_out_channels[0]		,     __snake_case		,     3		,     padding=1      )

      a		:   Dict	  =	False







    def      lowercase_			( self					:    Optional[int]		,     __snake_case					:    Union[str, Any]		,     __snake_case					:    List[Any]=None      ):
      a		:   Optional[Any]	  =	z
      a		:   Optional[Any]	  =	self.conv_in(__snake_case      )

      a		:   Union[str, Any]	  =	next(iter(self.up_blocks.parameters()      )      ).dtype
      if self.training and self.gradient_checkpointing:

        def create_custom_forward(__snake_case					:    Union[str, Any]      ):
          def custom_forward(*__snake_case					:    Optional[int]      ):
            return module(*__snake_case      )

          return custom_forward

        if is_torch_version('>='		,     '1.11.0'      ):
          # middle
          a		:   str	  =	torch.utils.checkpoint.checkpoint(
              create_custom_forward(self.mid_block      )		,     __snake_case		,     __snake_case		,     use_reentrant=__snake_case      )
          a		:   Union[str, Any]	  =	sample.to(__snake_case      )

          # up
          for up_block in self.up_blocks:
            a		:   Any	  =	torch.utils.checkpoint.checkpoint(
                create_custom_forward(__snake_case      )		,     __snake_case		,     __snake_case		,     use_reentrant=__snake_case      )
        else:
          # middle
          a		:   List[Any]	  =	torch.utils.checkpoint.checkpoint(
              create_custom_forward(self.mid_block      )		,     __snake_case		,     __snake_case      )
          a		:   List[str]	  =	sample.to(__snake_case      )

          # up
          for up_block in self.up_blocks:
            a		:   Tuple	  =	torch.utils.checkpoint.checkpoint(create_custom_forward(__snake_case      )		,     __snake_case		,     __snake_case      )
      else:
        # middle
        a		:   Any	  =	self.mid_block(__snake_case		,     __snake_case      )
        a		:   str	  =	sample.to(__snake_case      )

        # up
        for up_block in self.up_blocks:
          a		:   Dict	  =	up_block(__snake_case		,     __snake_case      )

        # post-process
      if latent_embeds is None:
        a		:   Dict	  =	self.conv_norm_out(__snake_case      )
      else:
        a		:   int	  =	self.conv_norm_out(__snake_case		,     __snake_case      )
      a		:   List[str]	  =	self.conv_act(__snake_case      )
      a		:   str	  =	self.conv_out(__snake_case      )

      return sample

class 				a__(   nn.Module      ):







    def __init__( self					:    Any		,     __snake_case					:    Optional[int]		,     __snake_case					:    Tuple		,     __snake_case					:    str		,     __snake_case					:    Any=None		,     __snake_case					:    Dict="random"		,     __snake_case					:    Tuple=False		,     __snake_case					:    int=True      ):
      super().__init__()
      a		:   List[Any]	  =	n_e
      a		:   int	  =	vq_embed_dim
      a		:   Optional[int]	  =	beta
      a		:   Dict	  =	legacy

      a		:   List[Any]	  =	nn.Embedding(self.n_e		,     self.vq_embed_dim      )
      self.embedding.weight.data.uniform_(-1.0 / self.n_e		,     1.0 / self.n_e      )

      a		:   int	  =	remap
      if self.remap is not None:
        self.register_buffer('used'		,     torch.tensor(np.load(self.remap      )      )      )
        a		:   List[str]	  =	self.used.shape[0]
        a		:   Optional[Any]	  =	unknown_index  # "random" or "extra" or integer
        if self.unknown_index == "extra":
          a		:   Any	  =	self.re_embed
          a		:   Optional[int]	  =	self.re_embed + 1
        print(
            F"""Remapping {self.n_e} indices to {self.re_embed} indices. """
            F"""Using {self.unknown_index} for unknown indices."""      )
      else:
        a		:   Optional[Any]	  =	n_e

      a		:   Union[str, Any]	  =	sane_index_shape







    def      lowercase_			( self					:    Optional[int]		,     __snake_case					:    Optional[int]      ):
      a		:   Optional[int]	  =	inds.shape
      assert len(__snake_case      ) > 1
      a		:   Optional[int]	  =	inds.reshape(ishape[0]		,     -1      )
      a		:   str	  =	self.used.to(__snake_case      )
      a		:   Union[str, Any]	  =	(inds[:, :, None] == used[None, None, ...]).long()
      a		:   List[Any]	  =	match.argmax(-1      )
      a		:   List[str]	  =	match.sum(2      ) < 1
      if self.unknown_index == "random":
        a		:   Dict	  =	torch.randint(0		,     self.re_embed		,     size=new[unknown].shape      ).to(device=new.device      )
      else:
        a		:   Any	  =	self.unknown_index
      return new.reshape(__snake_case      )







    def      lowercase_			( self					:    List[Any]		,     __snake_case					:    List[str]      ):
      a		:   str	  =	inds.shape
      assert len(__snake_case      ) > 1
      a		:   List[Any]	  =	inds.reshape(ishape[0]		,     -1      )
      a		:   Dict	  =	self.used.to(__snake_case      )
      if self.re_embed > self.used.shape[0]:  # extra token
        a		:   Optional[Any]	  =	0  # simply set to zero
      a		:   Optional[int]	  =	torch.gather(used[None, :][inds.shape[0] * [0], :]		,     1		,     __snake_case      )
      return back.reshape(__snake_case      )







    def      lowercase_			( self					:    Tuple		,     __snake_case					:    Dict      ):
      # reshape z -> (batch, height, width, channel) and flatten
      a		:   Union[str, Any]	  =	z.permute(0		,     2		,     3		,     1      ).contiguous()
      a		:   List[Any]	  =	z.view(-1		,     self.vq_embed_dim      )

      # distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
      a		:   Union[str, Any]	  =	torch.argmin(torch.cdist(__snake_case		,     self.embedding.weight      )		,     dim=1      )

      a		:   int	  =	self.embedding(__snake_case      ).view(z.shape      )
      a		:   str	  =	None
      a		:   Optional[int]	  =	None

      # compute loss for embedding
      if not self.legacy:
        a		:   List[str]	  =	self.beta * torch.mean((z_q.detach() - z) ** 2      ) + torch.mean((z_q - z.detach()) ** 2      )
      else:
        a		:   Dict	  =	torch.mean((z_q.detach() - z) ** 2      ) + self.beta * torch.mean((z_q - z.detach()) ** 2      )

      # preserve gradients
      a		:   Dict	  =	z + (z_q - z).detach()

      # reshape back to match original input shape
      a		:   str	  =	z_q.permute(0		,     3		,     1		,     2      ).contiguous()

      if self.remap is not None:
        a		:   List[str]	  =	min_encoding_indices.reshape(z.shape[0]		,     -1      )  # add batch axis
        a		:   List[Any]	  =	self.remap_to_used(__snake_case      )
        a		:   int	  =	min_encoding_indices.reshape(-1		,     1      )  # flatten

      if self.sane_index_shape:
        a		:   Tuple	  =	min_encoding_indices.reshape(z_q.shape[0]		,     z_q.shape[2]		,     z_q.shape[3]      )

      return z_q, loss, (perplexity, min_encodings, min_encoding_indices)







    def      lowercase_			( self					:    str		,     __snake_case					:    Optional[int]		,     __snake_case					:    Optional[int]      ):
      # shape specifying (batch, height, width, channel)
      if self.remap is not None:
        a		:   Optional[Any]	  =	indices.reshape(shape[0]		,     -1      )  # add batch axis
        a		:   Tuple	  =	self.unmap_to_all(__snake_case      )
        a		:   Optional[int]	  =	indices.reshape(-1      )  # flatten again

      # get quantized latent vectors
      a		:   Union[str, Any]	  =	self.embedding(__snake_case      )

      if shape is not None:
        a		:   int	  =	z_q.view(__snake_case      )
        # reshape back to match original input shape
        a		:   List[str]	  =	z_q.permute(0		,     3		,     1		,     2      ).contiguous()

      return z_q

class 				a__(   lowerCamelCase__      ):







    def __init__( self					:    Optional[Any]		,     __snake_case					:    Any		,     __snake_case					:    Any=False      ):
      a		:   Any	  =	parameters
      a  ,      a		:   int	  =	torch.chunk(__snake_case		,     2		,     dim=1      )
      a		:   str	  =	torch.clamp(self.logvar		,     -30.0		,     20.0      )
      a		:   Dict	  =	deterministic
      a		:   Union[str, Any]	  =	torch.exp(0.5 * self.logvar      )
      a		:   Optional[int]	  =	torch.exp(self.logvar      )
      if self.deterministic:
        a		:   Union[str, Any]	  =	torch.zeros_like(
            self.mean		,     device=self.parameters.device		,     dtype=self.parameters.dtype      )







    def      lowercase_			( self					:    Union[str, Any]		,     __snake_case					:    Optional[torch.Generator] = None      ):
      # make sure sample is on the same device as the parameters and has same dtype
      a		:   List[Any]	  =	randn_tensor(
          self.mean.shape		,     generator=__snake_case		,     device=self.parameters.device		,     dtype=self.parameters.dtype      )
      a		:   Optional[Any]	  =	self.mean + self.std * sample
      return x







    def      lowercase_			( self					:    Dict		,     __snake_case					:    Tuple=None      ):
      if self.deterministic:
        return torch.Tensor([0.0]      )
      else:
        if other is None:
          return 0.5 * torch.sum(torch.pow(self.mean		,     2      ) + self.var - 1.0 - self.logvar		,     dim=[1, 2, 3]      )
        else:
          return 0.5 * torch.sum(
              torch.pow(self.mean - other.mean		,     2      ) / other.var
              + self.var / other.var
              - 1.0
              - self.logvar
              + other.logvar		,     dim=[1, 2, 3]		,     )







    def      lowercase_			( self					:    Union[str, Any]		,     __snake_case					:    int		,     __snake_case					:    Dict=[1, 2, 3]      ):
      if self.deterministic:
        return torch.Tensor([0.0]      )
      a		:   Optional[Any]	  =	np.log(2.0 * np.pi      )
      return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean		,     2      ) / self.var		,     dim=__snake_case      )







    def      lowercase_			( self					:    str      ):
      return self.mean




'''simple docstring'''






from __future__ import annotations

lowerCAmelCase:   str					=		'Muhammad Umer Farooq'
lowerCAmelCase:   List[str]					=		'MIT'
lowerCAmelCase:   Tuple					=		'1.0.0'
lowerCAmelCase:   List[Any]					=		'Muhammad Umer Farooq'
lowerCAmelCase:   Optional[Any]					=		'contact@muhammadumerfarooq.me'
lowerCAmelCase:   Dict					=		'Alpha'

import re
from html.parser import HTMLParser
from urllib import parse

import requests

class 				a__(   lowerCamelCase__      ):







    def __init__( self					:    Dict		,     __snake_case					:    str      ):
      super().__init__()
      a		:   list[str]	  =	[]
      a		:   List[Any]	  =	domain







    def      lowercase_			( self					:    Dict		,     __snake_case					:    str		,     __snake_case					:    list[tuple[str, str | None]]      ):
      # Only parse the 'anchor' tag.
      if tag == "a":
        # Check the list of defined attributes.
        for name, value in attrs:
          # If href is defined, and not empty nor # print it.
          if name == "href" and value != "#" and value != "":
            # If not already in urls.
            if value not in self.urls:
              a		:   Tuple	  =	parse.urljoin(self.domain		,     __snake_case      )
              self.urls.append(__snake_case      )
def  lowerCamelCase__			(		_A							):
  return ".".join(get_sub_domain_name(_A							).split('.'							)[-2:]							)
def  lowerCamelCase__			(		_A							):
  return parse.urlparse(_A							).netloc
def  lowerCamelCase__			(		_A = "https://github.com"							):
  a		:   Any	  =	get_domain_name(_A							)

  # Initialize the parser
  a		:   Tuple	  =	Parser(_A							)

  try:
    # Open URL
    a		:   List[Any]	  =	requests.get(_A							)

    # pass the raw HTML to the parser to get links
    parser.feed(r.text							)

    # Get links and loop through
    a		:   Union[str, Any]	  =	set()
    for link in parser.urls:
      # open URL.
      # read = requests.get(link)
      try:
        a		:   int	  =	requests.get(_A							)
        # Get the valid email.
        a		:   Optional[Any]	  =	re.findall('[a-zA-Z0-9]+@' + domain					,     read.text							)
        # If not in list then append it.
        for email in emails:
          valid_emails.add(_A							)
      except ValueError:
        pass
  except ValueError:
    raise SystemExit(1							)

  # Finally return a sorted list of email addresses with no duplicates.
  return sorted(_A							)


if __name__ == "__main__":
     lowerCAmelCase:   Any					=		emails_from_url('https://github.com')
     print(F"{len(emails)} emails found:")
     print('\n'.join(sorted(emails)))