




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					(       _SCREAMING_SNAKE_CASE      ):

     '''simple docstring'''





     lowerCAmelCase    :	str       		=  42



class       a					(       nn.Module      ):

     '''simple docstring'''





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

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

         UpperCAmelCase_										=					None
         UpperCAmelCase_										=					nn.ModuleList([]    )

         # down
         UpperCAmelCase_										=					block_out_channels[0]
         for i, down_block_type in enumerate(__snake_case    ):
             UpperCAmelCase_										=					output_channel
             UpperCAmelCase_										=					block_out_channels[i]
             UpperCAmelCase_										=					i == len(__snake_case    ) - 1

             UpperCAmelCase_										=					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
         UpperCAmelCase_										=					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
         UpperCAmelCase_										=					nn.GroupNorm(num_channels=block_out_channels[-1] ,   num_groups=__snake_case ,   eps=1E-6    )
         UpperCAmelCase_										=					nn.SiLU()

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

         UpperCAmelCase_										=					False




     def   lowerCamelCase_       (     self       :      List[str] ,   __snake_case       :      int    ):
         UpperCAmelCase_										=					x
         UpperCAmelCase_										=					self.conv_in(__snake_case    )

         if self.training and self.gradient_checkpointing:

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

                 return custom_forward

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

         else:
             # down
             for down_block in self.down_blocks:
                 UpperCAmelCase_										=					down_block(__snake_case    )

             # middle
             UpperCAmelCase_										=					self.mid_block(__snake_case    )

         # post-process
         UpperCAmelCase_										=					self.conv_norm_out(__snake_case    )
         UpperCAmelCase_										=					self.conv_act(__snake_case    )
         UpperCAmelCase_										=					self.conv_out(__snake_case    )

         return sample



class       a					(       nn.Module      ):

     '''simple docstring'''





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

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

         UpperCAmelCase_										=					None
         UpperCAmelCase_										=					nn.ModuleList([]    )

         UpperCAmelCase_										=					in_channels if norm_type == '''spatial''' else None

         # mid
         UpperCAmelCase_										=					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
         UpperCAmelCase_										=					list(reversed(__snake_case    )    )
         UpperCAmelCase_										=					reversed_block_out_channels[0]
         for i, up_block_type in enumerate(__snake_case    ):
             UpperCAmelCase_										=					output_channel
             UpperCAmelCase_										=					reversed_block_out_channels[i]

             UpperCAmelCase_										=					i == len(__snake_case    ) - 1

             UpperCAmelCase_										=					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    )
             UpperCAmelCase_										=					output_channel

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

         UpperCAmelCase_										=					False




     def   lowerCamelCase_       (     self       :      str ,   __snake_case       :      Any ,   __snake_case       :      Optional[Any]=None    ):
         UpperCAmelCase_										=					z
         UpperCAmelCase_										=					self.conv_in(__snake_case    )

         UpperCAmelCase_										=					next(iter(self.up_blocks.parameters()    )    ).dtype
         if self.training and self.gradient_checkpointing:

             def create_custom_forward(__snake_case       :      Dict    ):
                 def custom_forward(*__snake_case       :      List[Any]    ):
                     return module(*__snake_case    )

                 return custom_forward

             if is_torch_version('''>=''' ,   '''1.11.0'''    ):
                 # middle
                 UpperCAmelCase_										=					torch.utils.checkpoint.checkpoint(
                     create_custom_forward(self.mid_block    ) ,   __snake_case ,   __snake_case ,   use_reentrant=__snake_case    )
                 UpperCAmelCase_										=					sample.to(__snake_case    )

                 # up
                 for up_block in self.up_blocks:
                     UpperCAmelCase_										=					torch.utils.checkpoint.checkpoint(
                         create_custom_forward(__snake_case    ) ,   __snake_case ,   __snake_case ,   use_reentrant=__snake_case    )
             else:
                 # middle
                 UpperCAmelCase_										=					torch.utils.checkpoint.checkpoint(
                     create_custom_forward(self.mid_block    ) ,   __snake_case ,   __snake_case    )
                 UpperCAmelCase_										=					sample.to(__snake_case    )

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

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

        # post-process
         if latent_embeds is None:
             UpperCAmelCase_										=					self.conv_norm_out(__snake_case    )
         else:
             UpperCAmelCase_										=					self.conv_norm_out(__snake_case ,   __snake_case    )
         UpperCAmelCase_										=					self.conv_act(__snake_case    )
         UpperCAmelCase_										=					self.conv_out(__snake_case    )

         return sample



class       a					(       nn.Module      ):

     '''simple docstring'''





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

         UpperCAmelCase_										=					nn.Embedding(self.n_e ,   self.vq_embed_dim    )
         self.embedding.weight.data.uniform_(-1.0 / self.n_e ,   1.0 / self.n_e    )

         UpperCAmelCase_										=					remap
         if self.remap is not None:
             self.register_buffer('''used''' ,   torch.tensor(np.load(self.remap    )    )    )
             UpperCAmelCase_										=					self.used.shape[0]
             UpperCAmelCase_										=					unknown_index  # "random" or "extra" or integer
             if self.unknown_index == "extra":
                 UpperCAmelCase_										=					self.re_embed
                 UpperCAmelCase_										=					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:
             UpperCAmelCase_										=					n_e

         UpperCAmelCase_										=					sane_index_shape
     def   lowerCamelCase_       (     self       :      List[str] ,   __snake_case       :      List[Any]    ):
         UpperCAmelCase_										=					inds.shape
         assert len(__snake_case    ) > 1
         UpperCAmelCase_										=					inds.reshape(ishape[0] ,   -1    )
         UpperCAmelCase_										=					self.used.to(__snake_case    )
         UpperCAmelCase_										=					(inds[:, :, None] == used[None, None, ...]).long()
         UpperCAmelCase_										=					match.argmax(-1    )
         UpperCAmelCase_										=					match.sum(2    ) < 1
         if self.unknown_index == "random":
             UpperCAmelCase_										=					torch.randint(0 ,   self.re_embed ,   size=new[unknown].shape    ).to(device=new.device    )
         else:
             UpperCAmelCase_										=					self.unknown_index
         return new.reshape(__snake_case    )
     def   lowerCamelCase_       (     self       :      Tuple ,   __snake_case       :      Any    ):
         UpperCAmelCase_										=					inds.shape
         assert len(__snake_case    ) > 1
         UpperCAmelCase_										=					inds.reshape(ishape[0] ,   -1    )
         UpperCAmelCase_										=					self.used.to(__snake_case    )
         if self.re_embed > self.used.shape[0]:  # extra token
             UpperCAmelCase_										=					0  # simply set to zero
         UpperCAmelCase_										=					torch.gather(used[None, :][inds.shape[0] * [0], :] ,   1 ,   __snake_case    )
         return back.reshape(__snake_case    )
     def   lowerCamelCase_       (     self       :      Dict ,   __snake_case       :      Union[str, Any]    ):
         # reshape z -> (batch, height, width, channel) and flatten
         UpperCAmelCase_										=					z.permute(0 ,   2 ,   3 ,   1    ).contiguous()
         UpperCAmelCase_										=					z.view(-1 ,   self.vq_embed_dim    )

         # distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
         UpperCAmelCase_										=					torch.argmin(torch.cdist(__snake_case ,   self.embedding.weight    ) ,   dim=1    )

         UpperCAmelCase_										=					self.embedding(__snake_case    ).view(z.shape    )
         UpperCAmelCase_										=					None
         UpperCAmelCase_										=					None

         # compute loss for embedding
         if not self.legacy:
             UpperCAmelCase_										=					self.beta * torch.mean((z_q.detach() - z) ** 2    ) + torch.mean((z_q - z.detach()) ** 2    )
         else:
             UpperCAmelCase_										=					torch.mean((z_q.detach() - z) ** 2    ) + self.beta * torch.mean((z_q - z.detach()) ** 2    )

         # preserve gradients
         UpperCAmelCase_										=					z + (z_q - z).detach()

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

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

         if self.sane_index_shape:
             UpperCAmelCase_										=					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   lowerCamelCase_       (     self       :      int ,   __snake_case       :      List[str] ,   __snake_case       :      Any    ):
         # shape specifying (batch, height, width, channel)
         if self.remap is not None:
             UpperCAmelCase_										=					indices.reshape(shape[0] ,   -1    )  # add batch axis
             UpperCAmelCase_										=					self.unmap_to_all(__snake_case    )
             UpperCAmelCase_										=					indices.reshape(-1    )  # flatten again

         # get quantized latent vectors
         UpperCAmelCase_										=					self.embedding(__snake_case    )

         if shape is not None:
             UpperCAmelCase_										=					z_q.view(__snake_case    )
             # reshape back to match original input shape
             UpperCAmelCase_										=					z_q.permute(0 ,   3 ,   1 ,   2    ).contiguous()

         return z_q



class       a					(       _SCREAMING_SNAKE_CASE      ):

     '''simple docstring'''





     def __init__(     self       :      Tuple ,   __snake_case       :      str ,   __snake_case       :      Any=False    ):
         UpperCAmelCase_										=					parameters
         UpperCAmelCase_			,							UpperCAmelCase_										=					torch.chunk(__snake_case ,   2 ,   dim=1    )
         UpperCAmelCase_										=					torch.clamp(self.logvar ,   -30.0 ,   20.0    )
         UpperCAmelCase_										=					deterministic
         UpperCAmelCase_										=					torch.exp(0.5 * self.logvar    )
         UpperCAmelCase_										=					torch.exp(self.logvar    )
         if self.deterministic:
             UpperCAmelCase_										=					UpperCAmelCase_										=					torch.zeros_like(
                 self.mean ,   device=self.parameters.device ,   dtype=self.parameters.dtype    )
     def   lowerCamelCase_       (     self       :      Optional[int] ,   __snake_case       :      str = None    ):
         # make sure sample is on the same device as the parameters and has same dtype
         UpperCAmelCase_										=					randn_tensor(
             self.mean.shape ,   generator=__snake_case ,   device=self.parameters.device ,   dtype=self.parameters.dtype    )
         UpperCAmelCase_										=					self.mean + self.std * sample
         return x
     def   lowerCamelCase_       (     self       :      Optional[Any] ,   __snake_case       :      str=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   lowerCamelCase_       (     self       :      Union[str, Any] ,   __snake_case       :      Union[str, Any] ,   __snake_case       :      Any=[1, 2, 3]    ):
         if self.deterministic:
             return torch.Tensor([0.0]    )
         UpperCAmelCase_										=					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   lowerCamelCase_       (     self       :      List[str]    ):
         return self.mean











import logging
import math
from functools import partial
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union

import torch

from .tensor_utils import tensor_tree_map, tree_map




def  SCREAMING_SNAKE_CASE     (	__UpperCamelCase      :						Union[dict, list, tuple, torch.Tensor]				)  ->							List[Tuple[int, ...]]:
    UpperCAmelCase_										=					[]
    if isinstance(__UpperCamelCase	,					__UpperCamelCase				):
        for v in tree.values():
            shapes.extend(_fetch_dims(__UpperCamelCase				)				)
    elif isinstance(__UpperCamelCase	,					(list, tuple)				):
        for t in tree:
            shapes.extend(_fetch_dims(__UpperCamelCase				)				)
    elif isinstance(__UpperCamelCase	,					torch.Tensor				):
        shapes.append(tree.shape				)
    else:
        raise ValueError('''Not supported'''				)

    return shapes




@torch.jit.ignore
def  SCREAMING_SNAKE_CASE     (	__UpperCamelCase      :						int	,					__UpperCamelCase      :						Tuple[int, ...]				)  ->							Tuple[int, ...]:
    UpperCAmelCase_										=					[]
    for d in reversed(__UpperCamelCase				):
        idx.append(flat_idx % d				)
        UpperCAmelCase_										=					flat_idx // d

    return tuple(reversed(__UpperCamelCase				)				)




@torch.jit.ignore
def  SCREAMING_SNAKE_CASE     (	__UpperCamelCase      :						Sequence[int]	,					__UpperCamelCase      :						Sequence[int]	,					__UpperCamelCase      :						Sequence[int]	,					__UpperCamelCase      :						Optional[Sequence[bool]] = None	,					__UpperCamelCase      :						Optional[Sequence[bool]] = None	,					)  ->							List[Tuple[slice, ...]]:

    # start_edges and end_edges both indicate whether, starting from any given
    # dimension, the start/end index is at the top/bottom edge of the
    # corresponding tensor, modeled as a tree
    def reduce_edge_list(__UpperCamelCase      :						List[bool]				) -> None:
        UpperCAmelCase_										=					True
        for i in range(len(__UpperCamelCase				)				):
            UpperCAmelCase_										=					-1 * (i + 1)
            l[reversed_idx] &= tally
            UpperCAmelCase_										=					l[reversed_idx]

    if start_edges is None:
        UpperCAmelCase_										=					[s == 0 for s in start]
        reduce_edge_list(__UpperCamelCase				)
    if end_edges is None:
        UpperCAmelCase_										=					[e == (d - 1) for e, d in zip(__UpperCamelCase	,					__UpperCamelCase				)]
        reduce_edge_list(__UpperCamelCase				)

    # Base cases. Either start/end are empty and we're done, or the final,
    # one-dimensional tensor can be simply sliced
    if len(__UpperCamelCase				) == 0:
        return [()]
    elif len(__UpperCamelCase				) == 1:
        return [(slice(start[0]	,					end[0] + 1				),)]

    UpperCAmelCase_										=					[]
    UpperCAmelCase_										=					[]

    # Dimensions common to start and end can be selected directly
    for s, e in zip(__UpperCamelCase	,					__UpperCamelCase				):
        if s == e:
            path_list.append(slice(__UpperCamelCase	,					s + 1				)				)
        else:
            break

    UpperCAmelCase_										=					tuple(__UpperCamelCase				)
    UpperCAmelCase_										=					len(__UpperCamelCase				)

    # start == end, and we're done
    if divergence_idx == len(__UpperCamelCase				):
        return [path]

    def upper() -> Tuple[Tuple[slice, ...], ...]:
        assert start_edges is not None
        assert end_edges is not None

        UpperCAmelCase_										=					start[divergence_idx]
        return tuple(
            path + (slice(__UpperCamelCase	,					sdi + 1				),) + s
            for s in _get_minimal_slice_set(
                start[divergence_idx + 1 :]	,					[d - 1 for d in dims[divergence_idx + 1 :]]	,					dims[divergence_idx + 1 :]	,					start_edges=start_edges[divergence_idx + 1 :]	,					end_edges=[True for _ in end_edges[divergence_idx + 1 :]]	,					)				)

    def lower() -> Tuple[Tuple[slice, ...], ...]:
        assert start_edges is not None
        assert end_edges is not None

        UpperCAmelCase_										=					end[divergence_idx]
        return tuple(
            path + (slice(__UpperCamelCase	,					edi + 1				),) + s
            for s in _get_minimal_slice_set(
                [0 for _ in start[divergence_idx + 1 :]]	,					end[divergence_idx + 1 :]	,					dims[divergence_idx + 1 :]	,					start_edges=[True for _ in start_edges[divergence_idx + 1 :]]	,					end_edges=end_edges[divergence_idx + 1 :]	,					)				)

    # If both start and end are at the edges of the subtree rooted at
    # divergence_idx, we can just select the whole subtree at once
    if start_edges[divergence_idx] and end_edges[divergence_idx]:
        slices.append(path + (slice(start[divergence_idx]	,					end[divergence_idx] + 1				),)				)
    # If just start is at the edge, we can grab almost all of the subtree,
    # treating only the ragged bottom edge as an edge case
    elif start_edges[divergence_idx]:
        slices.append(path + (slice(start[divergence_idx]	,					end[divergence_idx]				),)				)
        slices.extend(lower()				)
    # Analogous to the previous case, but the top is ragged this time
    elif end_edges[divergence_idx]:
        slices.extend(upper()				)
        slices.append(path + (slice(start[divergence_idx] + 1	,					end[divergence_idx] + 1				),)				)
    # If both sides of the range are ragged, we need to handle both sides
    # separately. If there's contiguous meat in between them, we can index it
    # in one big chunk
    else:
        slices.extend(upper()				)
        UpperCAmelCase_										=					end[divergence_idx] - start[divergence_idx]
        if middle_ground > 1:
            slices.append(path + (slice(start[divergence_idx] + 1	,					end[divergence_idx]				),)				)
        slices.extend(lower()				)

    return slices




@torch.jit.ignore
def  SCREAMING_SNAKE_CASE     (	__UpperCamelCase      :						torch.Tensor	,					__UpperCamelCase      :						int	,					__UpperCamelCase      :						int	,					__UpperCamelCase      :						int				)  ->							torch.Tensor:
    UpperCAmelCase_										=					t.shape[:no_batch_dims]
    UpperCAmelCase_										=					list(_flat_idx_to_idx(__UpperCamelCase	,					__UpperCamelCase				)				)
    # _get_minimal_slice_set is inclusive
    UpperCAmelCase_										=					list(_flat_idx_to_idx(flat_end - 1	,					__UpperCamelCase				)				)

    # Get an ordered list of slices to perform
    UpperCAmelCase_										=					_get_minimal_slice_set(
        __UpperCamelCase	,					__UpperCamelCase	,					__UpperCamelCase	,					)

    UpperCAmelCase_										=					[t[s] for s in slices]

    return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]				) for s in sliced_tensors]				)




def  SCREAMING_SNAKE_CASE     (	__UpperCamelCase      :						Callable	,					__UpperCamelCase      :						Dict[str, Any]	,					__UpperCamelCase      :						int	,					__UpperCamelCase      :						int	,					__UpperCamelCase      :						bool = False	,					__UpperCamelCase      :						Any = None	,					__UpperCamelCase      :						bool = False	,					)  ->							Any:
    if not (len(__UpperCamelCase				) > 0):
        raise ValueError('''Must provide at least one input'''				)

    UpperCAmelCase_										=					[shape[:no_batch_dims] for shape in _fetch_dims(__UpperCamelCase				)]
    UpperCAmelCase_										=					tuple([max(__UpperCamelCase				) for s in zip(*__UpperCamelCase				)]				)

    def _prep_inputs(__UpperCamelCase      :						torch.Tensor				) -> torch.Tensor:
        if not low_mem:
            if not sum(t.shape[:no_batch_dims]				) == no_batch_dims:
                UpperCAmelCase_										=					t.expand(orig_batch_dims + t.shape[no_batch_dims:]				)
            UpperCAmelCase_										=					t.reshape(-1	,					*t.shape[no_batch_dims:]				)
        else:
            UpperCAmelCase_										=					t.expand(orig_batch_dims + t.shape[no_batch_dims:]				)
        return t

    UpperCAmelCase_										=					tensor_tree_map(_prep_inputs	,					__UpperCamelCase				)
    UpperCAmelCase_										=					None
    if _out is not None:
        UpperCAmelCase_										=					tensor_tree_map(lambda __UpperCamelCase				: t.view([-1] + list(t.shape[no_batch_dims:]				)				)	,					_out				)

    UpperCAmelCase_										=					1
    for d in orig_batch_dims:
        flat_batch_dim *= d

    UpperCAmelCase_										=					flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

    def _select_chunk(__UpperCamelCase      :						torch.Tensor				) -> torch.Tensor:
        return t[i : i + chunk_size] if t.shape[0] != 1 else t

    UpperCAmelCase_										=					0
    UpperCAmelCase_										=					prepped_outputs
    for _ in range(__UpperCamelCase				):
        # Chunk the input
        if not low_mem:
            UpperCAmelCase_										=					_select_chunk
        else:
            UpperCAmelCase_										=					partial(
                _chunk_slice	,					flat_start=__UpperCamelCase	,					flat_end=min(__UpperCamelCase	,					i + chunk_size				)	,					no_batch_dims=len(__UpperCamelCase				)	,					)

        UpperCAmelCase_										=					tensor_tree_map(__UpperCamelCase	,					__UpperCamelCase				)

        # Run the layer on the chunk
        UpperCAmelCase_										=					layer(**__UpperCamelCase				)

        # Allocate space for the output
        if out is None:
            UpperCAmelCase_										=					tensor_tree_map(lambda __UpperCamelCase				: t.new_zeros((flat_batch_dim,) + t.shape[1:]				)	,					__UpperCamelCase				)

        # Put the chunk in its pre-allocated space
        if isinstance(__UpperCamelCase	,					__UpperCamelCase				):

            def assign(__UpperCamelCase      :						dict	,					__UpperCamelCase      :						dict				) -> None:
                for k, v in da.items():
                    if isinstance(__UpperCamelCase	,					__UpperCamelCase				):
                        assign(__UpperCamelCase	,					da[k]				)
                    else:
                        if _add_into_out:
                            v[i : i + chunk_size] += da[k]
                        else:
                            UpperCAmelCase_										=					da[k]

            assign(__UpperCamelCase	,					__UpperCamelCase				)
        elif isinstance(__UpperCamelCase	,					__UpperCamelCase				):
            for xa, xa in zip(__UpperCamelCase	,					__UpperCamelCase				):
                if _add_into_out:
                    xa[i : i + chunk_size] += xa
                else:
                    UpperCAmelCase_										=					xa
        elif isinstance(__UpperCamelCase	,					torch.Tensor				):
            if _add_into_out:
                out[i : i + chunk_size] += output_chunk
            else:
                UpperCAmelCase_										=					output_chunk
        else:
            raise ValueError('''Not supported'''				)

        i += chunk_size

    UpperCAmelCase_										=					tensor_tree_map(lambda __UpperCamelCase				: t.view(orig_batch_dims + t.shape[1:]				)	,					__UpperCamelCase				)

    return out



class       a					:

     '''simple docstring'''





     def __init__(     self       :      List[Any] ,   __snake_case       :      int = 5_12 ,   ):
         UpperCAmelCase_										=					max_chunk_size
         UpperCAmelCase_										=					None
         UpperCAmelCase_										=					None
     def   lowerCamelCase_       (     self       :      List[Any] ,   __snake_case       :      Callable ,   __snake_case       :      tuple ,   __snake_case       :      int    ):
         logging.info('''Tuning chunk size...'''    )

         if min_chunk_size >= self.max_chunk_size:
             return min_chunk_size

         UpperCAmelCase_										=					[2**l for l in range(int(math.log(self.max_chunk_size ,   2    )    ) + 1    )]
         UpperCAmelCase_										=					[c for c in candidates if c > min_chunk_size]
         UpperCAmelCase_										=					[min_chunk_size] + candidates
         candidates[-1] += 4

         def test_chunk_size(__snake_case       :      int    ) -> bool:
             try:
                 with torch.no_grad():
                     fn(*__snake_case ,   chunk_size=__snake_case    )
                 return True
             except RuntimeError:
                 return False

         UpperCAmelCase_										=					0
         UpperCAmelCase_										=					len(__snake_case    ) - 1
         while i > min_viable_chunk_size_index:
             UpperCAmelCase_										=					test_chunk_size(candidates[i]    )
             if not viable:
                 UpperCAmelCase_										=					(min_viable_chunk_size_index + i) // 2
             else:
                 UpperCAmelCase_										=					i
                 UpperCAmelCase_										=					(i + len(__snake_case    ) - 1) // 2

         return candidates[min_viable_chunk_size_index]
     def   lowerCamelCase_       (     self       :      int ,   __snake_case       :      Iterable ,   __snake_case       :      Iterable    ):
         UpperCAmelCase_										=					True
         for aa, aa in zip(__snake_case ,   __snake_case    ):
             assert type(__snake_case    ) == type(__snake_case    )
             if isinstance(__snake_case ,   (list, tuple)    ):
                 consistent &= self._compare_arg_caches(__snake_case ,   __snake_case    )
             elif isinstance(__snake_case ,   __snake_case    ):
                 UpperCAmelCase_										=					[v for _, v in sorted(aa.items() ,   key=lambda __snake_case    : x[0]    )]
                 UpperCAmelCase_										=					[v for _, v in sorted(aa.items() ,   key=lambda __snake_case    : x[0]    )]
                 consistent &= self._compare_arg_caches(__snake_case ,   __snake_case    )
             else:
                 consistent &= aa == aa

         return consistent




     def   lowerCamelCase_       (     self       :      str ,   __snake_case       :      Callable ,   __snake_case       :      tuple ,   __snake_case       :      int ,   ):
         UpperCAmelCase_										=					True
         UpperCAmelCase_										=					tree_map(lambda __snake_case    : a.shape if isinstance(__snake_case ,   torch.Tensor    ) else a ,   __snake_case ,   __snake_case    )
         if self.cached_arg_data is not None:
             # If args have changed shape/value, we need to re-tune
             assert len(self.cached_arg_data    ) == len(__snake_case    )
             UpperCAmelCase_										=					self._compare_arg_caches(self.cached_arg_data ,   __snake_case    )
         else:
             # Otherwise, we can reuse the precomputed value
             UpperCAmelCase_										=					False

         if not consistent:
             UpperCAmelCase_										=					self._determine_favorable_chunk_size(
                 __snake_case ,   __snake_case ,   __snake_case ,   )
             UpperCAmelCase_										=					arg_data

         assert self.cached_chunk_size is not None

         return self.cached_chunk_size





