




from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import get_down_block, get_mid_block, get_out_block, get_up_block


@dataclass
class    lowerCamelCase    (		__UpperCAmelCase						):






   '''simple docstring'''





   UpperCamelCase__								=4_2






class    lowerCamelCase    (		__UpperCAmelCase				,__UpperCAmelCase						):






   '''simple docstring'''





   @register_to_config
   def __init__( self			:   Optional[int]       ,						lowerCamelCase_			:   int = 65536       ,						lowerCamelCase_			:   Optional[int] = None       ,						lowerCamelCase_			:   int = 2       ,						lowerCamelCase_			:   int = 2       ,						lowerCamelCase_			:   int = 0       ,						lowerCamelCase_			:   str = "fourier"       ,						lowerCamelCase_			:   bool = True       ,						lowerCamelCase_			:   bool = False       ,						lowerCamelCase_			:   float = 0.0       ,						lowerCamelCase_			:   Tuple[str] = ("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")       ,						lowerCamelCase_			:   Tuple[str] = ("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")       ,						lowerCamelCase_			:   Tuple[str] = "UNetMidBlock1D"       ,						lowerCamelCase_			:   str = None       ,						lowerCamelCase_			:   Tuple[int] = (32, 32, 64)       ,						lowerCamelCase_			:   str = None       ,						lowerCamelCase_			:   int = 8       ,						lowerCamelCase_			:   int = 1       ,						lowerCamelCase_			:   bool = False       ,						)						->   Tuple:
        super().__init__()
        __magic_name__							:	int    							=							sample_size

        # time
        if time_embedding_type == "fourier":
             __magic_name__							:	Optional[int]    							=							GaussianFourierProjection(
                 embedding_size=8       ,						set_W_to_weight=lowerCAmelCase_       ,						log=lowerCAmelCase_       ,						flip_sin_to_cos=lowerCAmelCase_  )
             __magic_name__							:	Dict    							=							2 * block_out_channels[0]
        elif time_embedding_type == "positional":
             __magic_name__							:	List[str]    							=							Timesteps(
                 block_out_channels[0]       ,						flip_sin_to_cos=lowerCAmelCase_       ,						downscale_freq_shift=lowerCAmelCase_  )
             __magic_name__							:	Dict    							=							block_out_channels[0]

        if use_timestep_embedding:
             __magic_name__							:	str    							=							block_out_channels[0] * 4
             __magic_name__							:	List[Any]    							=							TimestepEmbedding(
                 in_channels=lowerCAmelCase_       ,						time_embed_dim=lowerCAmelCase_       ,						act_fn=lowerCAmelCase_       ,						out_dim=block_out_channels[0]       ,						)

        __magic_name__							:	Dict    							=							nn.ModuleList([]  )
        __magic_name__							:	int    							=							None
        __magic_name__							:	Any    							=							nn.ModuleList([]  )
        __magic_name__							:	Union[str, Any]    							=							None

        # down
        __magic_name__							:	Any    							=							in_channels
        for i, down_block_type in enumerate(lowerCAmelCase_  ):
             __magic_name__							:	Tuple    							=							output_channel
             __magic_name__							:	str    							=							block_out_channels[i]

             if i == 0:
                  input_channel += extra_in_channels

             __magic_name__							:	str    							=							i == len(lowerCAmelCase_  ) - 1

             __magic_name__							:	List[Any]    							=							get_down_block(
                 lowerCAmelCase_       ,						num_layers=lowerCAmelCase_       ,						in_channels=lowerCAmelCase_       ,						out_channels=lowerCAmelCase_       ,						temb_channels=block_out_channels[0]       ,						add_downsample=not is_final_block or downsample_each_block       ,						)
             self.down_blocks.append(lowerCAmelCase_  )

        # mid
        __magic_name__							:	Dict    							=							get_mid_block(
            lowerCAmelCase_       ,						in_channels=block_out_channels[-1]       ,						mid_channels=block_out_channels[-1]       ,						out_channels=block_out_channels[-1]       ,						embed_dim=block_out_channels[0]       ,						num_layers=lowerCAmelCase_       ,						add_downsample=lowerCAmelCase_       ,						)

        # up
        __magic_name__							:	List[str]    							=							list(reversed(lowerCAmelCase_  )  )
        __magic_name__							:	List[str]    							=							reversed_block_out_channels[0]
        if out_block_type is None:
             __magic_name__							:	Any    							=							out_channels
        else:
             __magic_name__							:	Union[str, Any]    							=							block_out_channels[0]

        for i, up_block_type in enumerate(lowerCAmelCase_  ):
             __magic_name__							:	List[str]    							=							output_channel
             __magic_name__							:	Optional[Any]    							=							(
                 reversed_block_out_channels[i + 1] if i < len(lowerCAmelCase_  ) - 1 else final_upsample_channels
             )

             __magic_name__							:	Optional[Any]    							=							i == len(lowerCAmelCase_  ) - 1

             __magic_name__							:	List[str]    							=							get_up_block(
                 lowerCAmelCase_       ,						num_layers=lowerCAmelCase_       ,						in_channels=lowerCAmelCase_       ,						out_channels=lowerCAmelCase_       ,						temb_channels=block_out_channels[0]       ,						add_upsample=not is_final_block       ,						)
             self.up_blocks.append(lowerCAmelCase_  )
             __magic_name__							:	Optional[int]    							=							output_channel

        # out
        __magic_name__							:	Any    							=							norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4       ,						32  )
        __magic_name__							:	Optional[Any]    							=							get_out_block(
            out_block_type=lowerCAmelCase_       ,						num_groups_out=lowerCAmelCase_       ,						embed_dim=block_out_channels[0]       ,						out_channels=lowerCAmelCase_       ,						act_fn=lowerCAmelCase_       ,						fc_dim=block_out_channels[-1] // 4       ,						)


   def 							UpperCAmelCase__       ( self			:   Union[str, Any]       ,						lowerCamelCase_			:   torch.FloatTensor       ,						lowerCamelCase_			:   Union[torch.Tensor, float, int]       ,						lowerCamelCase_			:   bool = True       ,						)						->   Union[str, Any]:
        __magic_name__							:	Optional[int]    							=							timestep
        if not torch.is_tensor(lowerCAmelCase_  ):
             __magic_name__							:	Tuple    							=							torch.tensor([timesteps]       ,						dtype=torch.long       ,						device=sample.device  )
        elif torch.is_tensor(lowerCAmelCase_  ) and len(timesteps.shape  ) == 0:
             __magic_name__							:	int    							=							timesteps[None].to(sample.device  )

        __magic_name__							:	Optional[int]    							=							self.time_proj(lowerCAmelCase_  )
        if self.config.use_timestep_embedding:
             __magic_name__							:	int    							=							self.time_mlp(lowerCAmelCase_  )
        else:
             __magic_name__							:	Optional[int]    							=							timestep_embed[..., None]
             __magic_name__							:	str    							=							timestep_embed.repeat([1, 1, sample.shape[2]]  ).to(sample.dtype  )
             __magic_name__							:	Union[str, Any]    							=							timestep_embed.broadcast_to((sample.shape[:1] + timestep_embed.shape[1:])  )

        # 2. down
        __magic_name__							:	Dict    							=							()
        for downsample_block in self.down_blocks:
             __magic_name__							,    __magic_name__							:	Dict    							=							downsample_block(hidden_states=lowerCAmelCase_       ,						temb=lowerCAmelCase_  )
             down_block_res_samples += res_samples

        # 3. mid
        if self.mid_block:
             __magic_name__							:	Optional[Any]    							=							self.mid_block(lowerCAmelCase_       ,						lowerCAmelCase_  )

        # 4. up
        for i, upsample_block in enumerate(self.up_blocks  ):
             __magic_name__							:	Dict    							=							down_block_res_samples[-1:]
             __magic_name__							:	Optional[Any]    							=							down_block_res_samples[:-1]
             __magic_name__							:	Dict    							=							upsample_block(lowerCAmelCase_       ,						res_hidden_states_tuple=lowerCAmelCase_       ,						temb=lowerCAmelCase_  )

        # 5. post-process
        if self.out_block:
             __magic_name__							:	List[Any]    							=							self.out_block(lowerCAmelCase_       ,						lowerCAmelCase_  )

        if not return_dict:
             return (sample,)

        return UNetaDOutput(sample=lowerCAmelCase_  )









import inspect
import os
import unittest
from pathlib import Path

import torch

import accelerate
from accelerate.test_utils import execute_subprocess_async
from accelerate.test_utils.testing import run_command


class    lowerCamelCase    (		unittest.TestCase						):






			'''simple docstring'''





			UpperCamelCase__								=inspect.getfile(accelerate.test_utils						)
			UpperCamelCase__								=os.path.sep.join(mod_file.split(os.path.sep						)[:-1] + ['''scripts''', '''test_cli.py''']						)

			UpperCamelCase__								=['''accelerate''', '''launch''']
			UpperCamelCase__								=Path.home() / '''.cache/huggingface/accelerate'''
			UpperCamelCase__								='''default_config.yaml'''
			UpperCamelCase__								=config_folder / config_file
			UpperCamelCase__								=config_folder / '''_default_config.yaml'''

			UpperCamelCase__								=Path('''tests/test_configs'''						)

			@classmethod
			def 							UpperCAmelCase__       ( cls			:   Optional[int]  )						->   int:
								if cls.config_path.is_file():
													cls.config_path.rename(cls.changed_path  )

			@classmethod
			def 							UpperCAmelCase__       ( cls			:   Dict  )						->   List[str]:
								if cls.changed_path.is_file():
													cls.changed_path.rename(cls.config_path  )

			def 							UpperCAmelCase__       ( self			:   str  )						->   Optional[Any]:
								__magic_name__							:	Dict    							=							self.base_cmd
								if torch.cuda.is_available() and (torch.cuda.device_count() > 1):
													cmd += ["--multi_gpu"]
								execute_subprocess_async(cmd + [self.test_file_path]       ,						env=os.environ.copy()  )

			def 							UpperCAmelCase__       ( self			:   int  )						->   Optional[int]:
								for config in sorted(self.test_config_path.glob('''**/*.yaml'''  )  ):
													with self.subTest(config_file=lowerCamelCase_  ):
																		execute_subprocess_async(
																		    self.base_cmd + ['''--config_file''', str(lowerCamelCase_  ), self.test_file_path]       ,						env=os.environ.copy()  )


			def 							UpperCAmelCase__       ( self			:   int  )						->   Union[str, Any]:
								execute_subprocess_async(['''accelerate''', '''test''']       ,						env=os.environ.copy()  )






class    lowerCamelCase    (		unittest.TestCase						):






			'''simple docstring'''





			UpperCamelCase__								='''test-tpu'''
			UpperCamelCase__								='''us-central1-a'''
			UpperCamelCase__								='''ls'''
			UpperCamelCase__								=['''accelerate''', '''tpu-config''']
			UpperCamelCase__								='''cd /usr/share'''
			UpperCamelCase__								='''tests/test_samples/test_command_file.sh'''
			UpperCamelCase__								='''Running gcloud compute tpus tpu-vm ssh'''

			def 							UpperCAmelCase__       ( self			:   Optional[Any]  )						->   Optional[Any]:
								__magic_name__							:	str    							=							run_command(
								    self.cmd
								    + ['''--command''', self.command, '''--tpu_zone''', self.tpu_zone, '''--tpu_name''', self.tpu_name, '''--debug''']       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   Union[str, Any]  )						->   int:
								__magic_name__							:	int    							=							run_command(
								    self.cmd
								    + [
								        '''--config_file''',
								        '''tests/test_configs/0_12_0.yaml''',
								        '''--command''',
								        self.command,
								        '''--tpu_zone''',
								        self.tpu_zone,
								        '''--tpu_name''',
								        self.tpu_name,
								        '''--debug''',
								    ]       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   List[Any]  )						->   Tuple:
								__magic_name__							:	Any    							=							run_command(
								    self.cmd + ['''--config_file''', '''tests/test_configs/latest.yaml''', '''--debug''']       ,						return_stdout=lowerCamelCase_  )
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   Optional[int]  )						->   Tuple:
								__magic_name__							:	Tuple    							=							run_command(
								    self.cmd + ['''--config_file''', '''tests/test_configs/latest.yaml''', '''--command''', self.command, '''--debug''']       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   List[Any]  )						->   Dict:
								__magic_name__							:	str    							=							run_command(
								    self.cmd
								    + [
								        '''--config_file''',
								        '''tests/test_configs/latest.yaml''',
								        '''--command''',
								        self.command,
								        '''--command''',
								        '''echo "Hello World"''',
								        '''--debug''',
								    ]       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls; echo "Hello World" --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   Optional[Any]  )						->   Optional[int]:
								__magic_name__							:	Union[str, Any]    							=							run_command(
								    self.cmd
								    + ['''--config_file''', '''tests/test_configs/latest.yaml''', '''--command_file''', self.command_file, '''--debug''']       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   int  )						->   List[Any]:
								__magic_name__							:	List[Any]    							=							run_command(
								    self.cmd
								    + [
								        '''--config_file''',
								        '''tests/test_configs/0_12_0.yaml''',
								        '''--command_file''',
								        self.command_file,
								        '''--tpu_zone''',
								        self.tpu_zone,
								        '''--tpu_name''',
								        self.tpu_name,
								        '''--debug''',
								    ]       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   str  )						->   Dict:
								__magic_name__							:	Tuple    							=							run_command(
								    self.cmd + ['''--config_file''', '''tests/test_configs/latest.yaml''', '''--install_accelerate''', '''--debug''']       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; pip install accelerate -U; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)


			def 							UpperCAmelCase__       ( self			:   Any  )						->   Tuple:
								__magic_name__							:	Tuple    							=							run_command(
								    self.cmd
								    + [
								        '''--config_file''',
								        '''tests/test_configs/latest.yaml''',
								        '''--install_accelerate''',
								        '''--accelerate_version''',
								        '''12.0.0''',
								        '''--debug''',
								    ]       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; pip install accelerate==12.0.0; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)



