"""simple docstring"""




import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, CycleDiffusionPipeline, DDIMScheduler, UNetaDConditionModel
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()
class 						__A							( SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_    ,	unittest.TestCase		):
     _UpperCamelCase							:			Tuple							=		CycleDiffusionPipeline
     _UpperCamelCase							:			Optional[int]							=		TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
         "negative_prompt",
         "height",
         "width",
         "negative_prompt_embeds",
     }
     _UpperCamelCase							:			str							=		PipelineTesterMixin.required_optional_params - {"latents"}
     _UpperCamelCase							:			Optional[Any]							=		TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"source_prompt"}		)
     _UpperCamelCase							:			Tuple							=		IMAGE_TO_IMAGE_IMAGE_PARAMS
     _UpperCamelCase							:			List[str]							=		IMAGE_TO_IMAGE_IMAGE_PARAMS






     def     __A  (		self  ):
       torch.manual_seed(0  )
       _lowerCAmelCase						:       int  			=     UNetaDConditionModel(
           block_out_channels=(32, 64)				,     layers_per_block=2				,     sample_size=32				,     in_channels=4				,     out_channels=4				,     down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")				,     up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")				,     cross_attention_dim=32				,     )
       _lowerCAmelCase						:       str  			=     DDIMScheduler(
           beta_start=0.0_0_0_8_5				,     beta_end=0.0_1_2				,     beta_schedule="""scaled_linear"""				,     num_train_timesteps=1000				,     clip_sample=a__				,     set_alpha_to_one=a__				,     )
       torch.manual_seed(0  )
       _lowerCAmelCase						:       Optional[int]  			=     AutoencoderKL(
           block_out_channels=[32, 64]				,     in_channels=3				,     out_channels=3				,     down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]				,     up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]				,     latent_channels=4				,     )
       torch.manual_seed(0  )
       _lowerCAmelCase						:       Any  			=     CLIPTextConfig(
           bos_token_id=0				,     eos_token_id=2				,     hidden_size=32				,     intermediate_size=37				,     layer_norm_eps=1e-05				,     num_attention_heads=4				,     num_hidden_layers=5				,     pad_token_id=1				,     vocab_size=1000				,     )
       _lowerCAmelCase						:       Tuple  			=     CLIPTextModel(a__  )
       _lowerCAmelCase						:       Dict  			=     CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""  )

       _lowerCAmelCase						:       List[str]  			=     {
           """unet""": unet,
           """scheduler""": scheduler,
           """vae""": vae,
           """text_encoder""": text_encoder,
           """tokenizer""": tokenizer,
           """safety_checker""": None,
           """feature_extractor""": None,
       }
       return components






     def     __A  (		self				,     a__				,     a__=0  ):
       _lowerCAmelCase						:       str  			=     floats_tensor((1, 3, 32, 32)				,     rng=random.Random(a__  )  ).to(a__  )
       _lowerCAmelCase						:       str  			=     image / 2 + 0.5
       if str(a__  ).startswith("""mps"""  ):
         _lowerCAmelCase						:       Union[str, Any]  			=     torch.manual_seed(a__  )
       else:
         _lowerCAmelCase						:       Any  			=     torch.Generator(device=a__  ).manual_seed(a__  )
       _lowerCAmelCase						:       Optional[int]  			=     {
           """prompt""": """An astronaut riding an elephant""",
           """source_prompt""": """An astronaut riding a horse""",
           """image""": image,
           """generator""": generator,
           """num_inference_steps""": 2,
           """eta""": 0.1,
           """strength""": 0.8,
           """guidance_scale""": 3,
           """source_guidance_scale""": 1,
           """output_type""": """numpy""",
       }
       return inputs






     def     __A  (		self  ):
       _lowerCAmelCase						:       Any  			=     """cpu"""  # ensure determinism for the device-dependent torch.Generator

       _lowerCAmelCase						:       List[Any]  			=     self.get_dummy_components()
       _lowerCAmelCase						:       List[str]  			=     CycleDiffusionPipeline(**a__  )
       _lowerCAmelCase						:       List[str]  			=     pipe.to(a__  )
       pipe.set_progress_bar_config(disable=a__  )

       _lowerCAmelCase						:       int  			=     self.get_dummy_inputs(a__  )
       _lowerCAmelCase						:       Optional[Any]  			=     pipe(**a__  )
       _lowerCAmelCase						:       Optional[Any]  			=     output.images

       _lowerCAmelCase						:       Optional[int]  			=     images[0, -3:, -3:, -1]

       assert images.shape == (1, 32, 32, 3)
       _lowerCAmelCase						:       Dict  			=     np.array([0.4_4_5_9, 0.4_9_4_3, 0.4_5_4_4, 0.6_6_4_3, 0.5_4_7_4, 0.4_3_2_7, 0.5_7_0_1, 0.5_9_5_9, 0.5_1_7_9]  )

       assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1e-2






     @unittest.skipIf(torch_device != """cuda"""				,     """This test requires a GPU"""  )
     def     __A  (		self  ):
       _lowerCAmelCase						:       str  			=     self.get_dummy_components()
       for name, module in components.items():
         if hasattr(a__				,     """half"""  ):
           _lowerCAmelCase						:       Any  			=     module.half()
       _lowerCAmelCase						:       int  			=     CycleDiffusionPipeline(**a__  )
       _lowerCAmelCase						:       Dict  			=     pipe.to(a__  )
       pipe.set_progress_bar_config(disable=a__  )

       _lowerCAmelCase						:       Optional[int]  			=     self.get_dummy_inputs(a__  )
       _lowerCAmelCase						:       Optional[int]  			=     pipe(**a__  )
       _lowerCAmelCase						:       List[str]  			=     output.images

       _lowerCAmelCase						:       str  			=     images[0, -3:, -3:, -1]

       assert images.shape == (1, 32, 32, 3)
       _lowerCAmelCase						:       int  			=     np.array([0.3_5_0_6, 0.4_5_4_3, 0.4_4_6, 0.4_5_7_5, 0.5_1_9_5, 0.4_1_5_5, 0.5_2_7_3, 0.5_1_8, 0.4_1_1_6]  )

       assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1e-2






     @skip_mps
     def     __A  (		self  ):
       return super().test_save_load_local()






     @unittest.skip("""non-deterministic pipeline"""  )
     def     __A  (		self  ):
       return super().test_inference_batch_single_identical()






     @skip_mps
     def     __A  (		self  ):
       return super().test_dict_tuple_outputs_equivalent()






     @skip_mps
     def     __A  (		self  ):
       return super().test_save_load_optional_components()






     @skip_mps
     def     __A  (		self  ):
       return super().test_attention_slicing_forward_pass()







@slow
@require_torch_gpu
class 						__A							( unittest.TestCase		):






     def     __A  (		self  ):
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()






     def     __A  (		self  ):
       _lowerCAmelCase						:       Tuple  			=     load_image(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
           """/cycle-diffusion/black_colored_car.png"""  )
       _lowerCAmelCase						:       Optional[Any]  			=     load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy"""  )
       _lowerCAmelCase						:       Optional[int]  			=     init_image.resize((512, 512)  )

       _lowerCAmelCase						:       Optional[Any]  			=     """CompVis/stable-diffusion-v1-4"""
       _lowerCAmelCase						:       List[str]  			=     DDIMScheduler.from_pretrained(a__				,     subfolder="""scheduler"""  )
       _lowerCAmelCase						:       Tuple  			=     CycleDiffusionPipeline.from_pretrained(
           a__				,     scheduler=a__				,     safety_checker=a__				,     torch_dtype=torch.floataa				,     revision="""fp16"""  )

       pipe.to(a__  )
       pipe.set_progress_bar_config(disable=a__  )
       pipe.enable_attention_slicing()

       _lowerCAmelCase						:       Dict  			=     """A black colored car"""
       _lowerCAmelCase						:       str  			=     """A blue colored car"""

       _lowerCAmelCase						:       int  			=     torch.manual_seed(0  )
       _lowerCAmelCase						:       Optional[int]  			=     pipe(
           prompt=a__				,     source_prompt=a__				,     image=a__				,     num_inference_steps=100				,     eta=0.1				,     strength=0.8_5				,     guidance_scale=3				,     source_guidance_scale=1				,     generator=a__				,     output_type="""np"""				,     )
       _lowerCAmelCase						:       int  			=     output.images

       # the values aren't exactly equal, but the images look the same visually
       assert np.abs(image - expected_image  ).max() < 5e-1






     def     __A  (		self  ):
       _lowerCAmelCase						:       List[str]  			=     load_image(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
           """/cycle-diffusion/black_colored_car.png"""  )
       _lowerCAmelCase						:       str  			=     load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"""  )
       _lowerCAmelCase						:       List[Any]  			=     init_image.resize((512, 512)  )

       _lowerCAmelCase						:       Tuple  			=     """CompVis/stable-diffusion-v1-4"""
       _lowerCAmelCase						:       Dict  			=     DDIMScheduler.from_pretrained(a__				,     subfolder="""scheduler"""  )
       _lowerCAmelCase						:       Optional[int]  			=     CycleDiffusionPipeline.from_pretrained(a__				,     scheduler=a__				,     safety_checker=a__  )

       pipe.to(a__  )
       pipe.set_progress_bar_config(disable=a__  )
       pipe.enable_attention_slicing()

       _lowerCAmelCase						:       Union[str, Any]  			=     """A black colored car"""
       _lowerCAmelCase						:       Any  			=     """A blue colored car"""

       _lowerCAmelCase						:       Optional[int]  			=     torch.manual_seed(0  )
       _lowerCAmelCase						:       Dict  			=     pipe(
           prompt=a__				,     source_prompt=a__				,     image=a__				,     num_inference_steps=100				,     eta=0.1				,     strength=0.8_5				,     guidance_scale=3				,     source_guidance_scale=1				,     generator=a__				,     output_type="""np"""				,     )
       _lowerCAmelCase						:       Optional[int]  			=     output.images

       assert np.abs(image - expected_image  ).max() < 2e-2





"""simple docstring"""




import numpy as np
import qiskit





def 							SCREAMING_SNAKE_CASE     (			_lowerCamelCase						:       int = 8			,_lowerCamelCase						:       int | None = None	)       ->      str:
  _lowerCAmelCase						:       int  			=     np.random.default_rng(seed=_lowerCamelCase	)

  # Roughly 25% of the qubits will contribute to the key.
  # So we take more than we need.
  _lowerCAmelCase						:       Tuple  			=     6 * key_len
  # Measurement basis for Alice's qubits.
  _lowerCAmelCase						:       Dict  			=     rng.integers(2			,size=_lowerCamelCase	)
  # The set of states Alice will prepare.
  _lowerCAmelCase						:       Tuple  			=     rng.integers(2			,size=_lowerCamelCase	)
  # Measurement basis for Bob's qubits.
  _lowerCAmelCase						:       Union[str, Any]  			=     rng.integers(2			,size=_lowerCamelCase	)

  # Quantum Circuit to simulate BB84
  _lowerCAmelCase						:       Dict  			=     qiskit.QuantumCircuit(_lowerCamelCase			,name="""BB84"""	)

  # Alice prepares her qubits according to rules above.
  for index, _ in enumerate(_lowerCamelCase	):
    if alice_state[index] == 1:
      bbaa_circ.x(_lowerCamelCase	)
    if alice_basis[index] == 1:
      bbaa_circ.h(_lowerCamelCase	)
  bbaa_circ.barrier()

  # Bob measures the received qubits according to rules above.
  for index, _ in enumerate(_lowerCamelCase	):
    if bob_basis[index] == 1:
      bbaa_circ.h(_lowerCamelCase	)

  bbaa_circ.barrier()
  bbaa_circ.measure_all()

  # Simulate the quantum circuit.
  _lowerCAmelCase						:       int  			=     qiskit.Aer.get_backend("""aer_simulator"""	)
  # We only need to run one shot because the key is unique.
  # Multiple shots will produce the same key.
  _lowerCAmelCase						:       List[str]  			=     qiskit.execute(_lowerCamelCase			,_lowerCamelCase			,shots=1			,seed_simulator=_lowerCamelCase	)
  # Returns the result of measurement.
  _lowerCAmelCase						:       List[Any]  			=     job.result().get_counts(_lowerCamelCase	).most_frequent()

  # Extracting the generated key from the simulation results.
  # Only keep measurement results where Alice and Bob chose the same basis.
  _lowerCAmelCase						:       str  			=     """""".join(
      [
          result_bit
          for alice_basis_bit, bob_basis_bit, result_bit in zip(
              _lowerCamelCase			,_lowerCamelCase			,_lowerCamelCase	)
          if alice_basis_bit == bob_basis_bit
      ]	)

  # Get final key. Pad with 0 if too short, otherwise truncate.
  _lowerCAmelCase						:       List[Any]  			=     gen_key[:key_len] if len(_lowerCamelCase	) >= key_len else gen_key.ljust(_lowerCamelCase			,"""0"""	)
  return key


if __name__ == "__main__":
     print(F"""The generated key is : {bbaa(8, seed=0)}""")
     from doctest import testmod

     testmod()




