
"""simple docstring"""




import json
import os

import torch

from diffusers import UNetaDModel


os.makedirs('hub/hopper-medium-v2/unet/hor32', exist_ok=True)
os.makedirs('hub/hopper-medium-v2/unet/hor128', exist_ok=True)

os.makedirs('hub/hopper-medium-v2/value_function', exist_ok=True)



def     __magic_name__		(					UpperCamelCase     :		Optional[int]     )     ->						List[str]:
    if hor == 128:
        a__		=			('DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D')
        a__		=			(32, 128, 256)
        a__		=			('UpResnetBlock1D', 'UpResnetBlock1D')

    elif hor == 32:
        a__		=			('DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D')
        a__		=			(32, 64, 128, 256)
        a__		=			('UpResnetBlock1D', 'UpResnetBlock1D', 'UpResnetBlock1D')
    a__		=			torch.load(f'/Users/bglickenhaus/Documents/diffuser/temporal_unet-hopper-mediumv2-hor{hor}.torch'     )
    a__		=			model.state_dict()
    a__		=			{
        'down_block_types': down_block_types,
        'block_out_channels': block_out_channels,
        'up_block_types': up_block_types,
        'layers_per_block': 1,
        'use_timestep_embedding': True,
        'out_block_type': 'OutConv1DBlock',
        'norm_num_groups': 8,
        'downsample_each_block': False,
        'in_channels': 14,
        'out_channels': 14,
        'extra_in_channels': 0,
        'time_embedding_type': 'positional',
        'flip_sin_to_cos': False,
        'freq_shift': 1,
        'sample_size': 65536,
        'mid_block_type': 'MidResTemporalBlock1D',
        'act_fn': 'mish',
    }
    a__		=			UNetaDModel(**UpperCamelCase     )
    print(f'length of state dict: {len(state_dict.keys()     )}'     )
    print(f'length of value function dict: {len(hf_value_function.state_dict().keys()     )}'     )
    a__		=			dict(zip(model.state_dict().keys()							,		hf_value_function.state_dict().keys()     )     )
    for k, v in mapping.items():
        a__		=			state_dict.pop(UpperCamelCase     )
    hf_value_function.load_state_dict(UpperCamelCase     )

    torch.save(hf_value_function.state_dict()							,		f'hub/hopper-medium-v2/unet/hor{hor}/diffusion_pytorch_model.bin'     )
    with open(f'hub/hopper-medium-v2/unet/hor{hor}/config.json'							,		'w'     ) as f:
        json.dump(UpperCamelCase							,		UpperCamelCase     )



def     __magic_name__		(					)     ->						Dict:
    a__		=			{
        'in_channels': 14,
        'down_block_types': ('DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D'),
        'up_block_types': (),
        'out_block_type': 'ValueFunction',
        'mid_block_type': 'ValueFunctionMidBlock1D',
        'block_out_channels': (32, 64, 128, 256),
        'layers_per_block': 1,
        'downsample_each_block': True,
        'sample_size': 65536,
        'out_channels': 14,
        'extra_in_channels': 0,
        'time_embedding_type': 'positional',
        'use_timestep_embedding': True,
        'flip_sin_to_cos': False,
        'freq_shift': 1,
        'norm_num_groups': 8,
        'act_fn': 'mish',
    }

    a__		=			torch.load('/Users/bglickenhaus/Documents/diffuser/value_function-hopper-mediumv2-hor32.torch'     )
    a__		=			model
    a__		=			UNetaDModel(**UpperCamelCase     )
    print(f'length of state dict: {len(state_dict.keys()     )}'     )
    print(f'length of value function dict: {len(hf_value_function.state_dict().keys()     )}'     )

    a__		=			dict(zip(state_dict.keys()							,		hf_value_function.state_dict().keys()     )     )
    for k, v in mapping.items():
        a__		=			state_dict.pop(UpperCamelCase     )

    hf_value_function.load_state_dict(UpperCamelCase     )

    torch.save(hf_value_function.state_dict()							,		'hub/hopper-medium-v2/value_function/diffusion_pytorch_model.bin'     )
    with open('hub/hopper-medium-v2/value_function/config.json'							,		'w'     ) as f:
        json.dump(UpperCamelCase							,		UpperCamelCase     )


if __name__ == "__main__":
    unet(32)
    # unet(128)
    value_function()




"""simple docstring"""




import random
import unittest

import torch

from diffusers import IFImgaImgSuperResolutionPipeline
from diffusers.utils import floats_tensor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import skip_mps, torch_device

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin
from . import IFPipelineTesterMixin







@skip_mps
class     lowercase(_lowercase    ,						_lowercase    ,						unittest.TestCase       ):
     __snake_case:	Optional[int]						=     IFImgaImgSuperResolutionPipeline
     __snake_case:	Dict						=     TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'width', 'height'}
     __snake_case:	Any						=     TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'original_image'}       )
     __snake_case:	List[Any]						=     PipelineTesterMixin.required_optional_params - {'latents'}



     def 			lowercase__  (							self   )		->     Tuple:







         """simple docstring"""



         return self._get_superresolution_dummy_components()



     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE=0   )		->     Union[str, Any]:







         """simple docstring"""



         if str(__SCREAMING_SNAKE_CASE   ).startswith('mps'   ):
             a__		=			torch.manual_seed(__SCREAMING_SNAKE_CASE   )
         else:
             a__		=			torch.Generator(device=__SCREAMING_SNAKE_CASE   ).manual_seed(__SCREAMING_SNAKE_CASE   )

         a__		=			floats_tensor((1, 3, 3_2, 3_2)     ,				rng=random.Random(__SCREAMING_SNAKE_CASE   )   ).to(__SCREAMING_SNAKE_CASE   )
         a__		=			floats_tensor((1, 3, 1_6, 1_6)     ,				rng=random.Random(__SCREAMING_SNAKE_CASE   )   ).to(__SCREAMING_SNAKE_CASE   )

         a__		=			{
             'prompt': 'A painting of a squirrel eating a burger',
             'image': image,
             'original_image': original_image,
             'generator': generator,
             'num_inference_steps': 2,
             'output_type': 'numpy',
         }

         return inputs



     @unittest.skipIf(
         torch_device != 'cuda' or not is_xformers_available()     ,				reason='XFormers attention is only available with CUDA and `xformers` installed'     ,				)
     def 			lowercase__  (							self   )		->     Optional[int]:







         """simple docstring"""



         self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1e-3   )



     def 			lowercase__  (							self   )		->     Any:







         """simple docstring"""



         self._test_save_load_optional_components()



     @unittest.skipIf(torch_device != 'cuda'     ,				reason='float16 requires CUDA'   )
     def 			lowercase__  (							self   )		->     List[Any]:







         """simple docstring"""



         super().test_save_load_floataa(expected_max_diff=1e-1   )



     def 			lowercase__  (							self   )		->     List[str]:







         """simple docstring"""



         self._test_attention_slicing_forward_pass(expected_max_diff=1e-2   )



     def 			lowercase__  (							self   )		->     Optional[Any]:







         """simple docstring"""



         self._test_save_load_local()




     def 			lowercase__  (							self   )		->     Any:







         """simple docstring"""



         self._test_inference_batch_single_identical(
             expected_max_diff=1e-2     ,				)


