






"""simple docstring"""
import tempfile
import unittest

import numpy as np

from diffusers import (
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionPipeline,
    PNDMScheduler,
)
from diffusers.utils.testing_utils import is_onnx_available, nightly, require_onnxruntime, require_torch_gpu

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
			import onnxruntime as ort







class      UpperCamelCase_			(__A							,		unittest.TestCase	):
	__magic_name__       =							'''hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'''





	def 	_SCREAMING_SNAKE_CASE ( self   :					str      ,      lowerCAmelCase_   :					List[str]=0		)		->     Union[str, Any]:
				UpperCAmelCase_	:					Union[str, Any]     	=   np.random.RandomState(lowerCAmelCase_		)
				UpperCAmelCase_	:					Tuple     	=   {
				    "prompt": "A painting of a squirrel eating a burger",
				    "generator": generator,
				    "num_inference_steps": 2,
				    "guidance_scale": 7.5,
				    "output_type": "numpy",
				}
				return inputs





	def 	_SCREAMING_SNAKE_CASE ( self   :					List[str]		)		->     int:
				UpperCAmelCase_	:					List[Any]     	=   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint      ,      provider="CPUExecutionProvider"		)
				pipe.set_progress_bar_config(disable=lowerCAmelCase_		)

				UpperCAmelCase_	:					Optional[int]     	=   self.get_dummy_inputs()
				UpperCAmelCase_	:					List[Any]     	=   pipe(**lowerCAmelCase_		).images
				UpperCAmelCase_	:					Tuple     	=   image[0, -3:, -3:, -1]

				assert image.shape == (1, 128, 128, 3)
				UpperCAmelCase_	:					Union[str, Any]     	=   np.array([0.6_5_0_7_2, 0.5_8_4_9_2, 0.4_8_2_1_9, 0.5_5_5_2_1, 0.5_3_1_8_0, 0.5_5_9_3_9, 0.5_0_6_9_7, 0.3_9_8_0_0, 0.4_6_4_5_5]		)

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





	def 	_SCREAMING_SNAKE_CASE ( self   :					Union[str, Any]		)		->     int:
				UpperCAmelCase_	:					List[Any]     	=   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint      ,      provider="CPUExecutionProvider"		)
				UpperCAmelCase_	:					Tuple     	=   PNDMScheduler.from_config(pipe.scheduler.config      ,      skip_prk_steps=lowerCAmelCase_		)
				pipe.set_progress_bar_config(disable=lowerCAmelCase_		)

				UpperCAmelCase_	:					Any     	=   self.get_dummy_inputs()
				UpperCAmelCase_	:					Optional[Any]     	=   pipe(**lowerCAmelCase_		).images
				UpperCAmelCase_	:					Dict     	=   image[0, -3:, -3:, -1]

				assert image.shape == (1, 128, 128, 3)
				UpperCAmelCase_	:					Optional[int]     	=   np.array([0.6_5_8_6_3, 0.5_9_4_2_5, 0.4_9_3_2_6, 0.5_6_3_1_3, 0.5_3_8_7_5, 0.5_6_6_2_7, 0.5_1_0_6_5, 0.3_9_7_7_7, 0.4_6_3_3_0]		)

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





	def 	_SCREAMING_SNAKE_CASE ( self   :					int		)		->     Union[str, Any]:
				UpperCAmelCase_	:					Tuple     	=   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint      ,      provider="CPUExecutionProvider"		)
				UpperCAmelCase_	:					Union[str, Any]     	=   LMSDiscreteScheduler.from_config(pipe.scheduler.config		)
				pipe.set_progress_bar_config(disable=lowerCAmelCase_		)

				UpperCAmelCase_	:					Dict     	=   self.get_dummy_inputs()
				UpperCAmelCase_	:					int     	=   pipe(**lowerCAmelCase_		).images
				UpperCAmelCase_	:					Tuple     	=   image[0, -3:, -3:, -1]

				assert image.shape == (1, 128, 128, 3)
				UpperCAmelCase_	:					List[Any]     	=   np.array([0.5_3_7_5_5, 0.6_0_7_8_6, 0.4_7_4_0_2, 0.4_9_4_8_8, 0.5_1_8_6_9, 0.4_9_8_1_9, 0.4_7_9_8_5, 0.3_8_9_5_7, 0.4_4_2_7_9]		)

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





	def 	_SCREAMING_SNAKE_CASE ( self   :					Union[str, Any]		)		->     List[str]:
				UpperCAmelCase_	:					Union[str, Any]     	=   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint      ,      provider="CPUExecutionProvider"		)
				UpperCAmelCase_	:					int     	=   EulerDiscreteScheduler.from_config(pipe.scheduler.config		)
				pipe.set_progress_bar_config(disable=lowerCAmelCase_		)

				UpperCAmelCase_	:					int     	=   self.get_dummy_inputs()
				UpperCAmelCase_	:					Any     	=   pipe(**lowerCAmelCase_		).images
				UpperCAmelCase_	:					Optional[Any]     	=   image[0, -3:, -3:, -1]

				assert image.shape == (1, 128, 128, 3)
				UpperCAmelCase_	:					Any     	=   np.array([0.5_3_7_5_5, 0.6_0_7_8_6, 0.4_7_4_0_2, 0.4_9_4_8_8, 0.5_1_8_6_9, 0.4_9_8_1_9, 0.4_7_9_8_5, 0.3_8_9_5_7, 0.4_4_2_7_9]		)

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





	def 	_SCREAMING_SNAKE_CASE ( self   :					int		)		->     Tuple:
				UpperCAmelCase_	:					List[Any]     	=   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint      ,      provider="CPUExecutionProvider"		)
				UpperCAmelCase_	:					Union[str, Any]     	=   EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config		)
				pipe.set_progress_bar_config(disable=lowerCAmelCase_		)

				UpperCAmelCase_	:					str     	=   self.get_dummy_inputs()
				UpperCAmelCase_	:					List[str]     	=   pipe(**lowerCAmelCase_		).images
				UpperCAmelCase_	:					Optional[Any]     	=   image[0, -3:, -3:, -1]

				assert image.shape == (1, 128, 128, 3)
				UpperCAmelCase_	:					Optional[int]     	=   np.array([0.5_3_8_1_7, 0.6_0_8_1_2, 0.4_7_3_8_4, 0.4_9_5_3_0, 0.5_1_8_9_4, 0.4_9_8_1_4, 0.4_7_9_8_4, 0.3_8_9_5_8, 0.4_4_2_7_1]		)

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





	def 	_SCREAMING_SNAKE_CASE ( self   :					List[Any]		)		->     List[str]:
				UpperCAmelCase_	:					Optional[Any]     	=   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint      ,      provider="CPUExecutionProvider"		)
				UpperCAmelCase_	:					Union[str, Any]     	=   DPMSolverMultistepScheduler.from_config(pipe.scheduler.config		)
				pipe.set_progress_bar_config(disable=lowerCAmelCase_		)

				UpperCAmelCase_	:					Optional[Any]     	=   self.get_dummy_inputs()
				UpperCAmelCase_	:					str     	=   pipe(**lowerCAmelCase_		).images
				UpperCAmelCase_	:					Dict     	=   image[0, -3:, -3:, -1]

				assert image.shape == (1, 128, 128, 3)
				UpperCAmelCase_	:					Optional[Any]     	=   np.array([0.5_3_8_9_5, 0.6_0_8_0_8, 0.4_7_9_3_3, 0.4_9_6_0_8, 0.5_1_8_8_6, 0.4_9_9_5_0, 0.4_8_0_5_3, 0.3_8_9_5_7, 0.4_4_2_0_0]		)

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





	def 	_SCREAMING_SNAKE_CASE ( self   :					Tuple		)		->     List[Any]:
				UpperCAmelCase_	:					Optional[int]     	=   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint      ,      provider="CPUExecutionProvider"		)
				pipe.set_progress_bar_config(disable=lowerCAmelCase_		)

				UpperCAmelCase_	:					str     	=   self.get_dummy_inputs()
				UpperCAmelCase_	:					List[str]     	=   3 * [inputs["prompt"]]

				# forward
				UpperCAmelCase_	:					str     	=   pipe(**lowerCAmelCase_		)
				UpperCAmelCase_	:					Dict     	=   output.images[0, -3:, -3:, -1]

				UpperCAmelCase_	:					int     	=   self.get_dummy_inputs()
				UpperCAmelCase_	:					Any     	=   3 * [inputs.pop("prompt"		)]

				UpperCAmelCase_	:					Union[str, Any]     	=   pipe.tokenizer(
				    lowerCAmelCase_      ,      padding="max_length"      ,      max_length=pipe.tokenizer.model_max_length      ,      truncation=lowerCAmelCase_      ,      return_tensors="np"      ,      )
				UpperCAmelCase_	:					Tuple     	=   text_inputs["input_ids"]

				UpperCAmelCase_	:					Optional[int]     	=   pipe.text_encoder(input_ids=text_inputs.astype(np.intaa		)		)[0]

				UpperCAmelCase_	:					Dict     	=   prompt_embeds

				# forward
				UpperCAmelCase_	:					str     	=   pipe(**lowerCAmelCase_		)
				UpperCAmelCase_	:					str     	=   output.images[0, -3:, -3:, -1]

				assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()		).max() < 1e-4





	def 	_SCREAMING_SNAKE_CASE ( self   :					Dict		)		->     List[Any]:
				UpperCAmelCase_	:					List[Any]     	=   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint      ,      provider="CPUExecutionProvider"		)
				pipe.set_progress_bar_config(disable=lowerCAmelCase_		)

				UpperCAmelCase_	:					List[Any]     	=   self.get_dummy_inputs()
				UpperCAmelCase_	:					Union[str, Any]     	=   3 * ["this is a negative prompt"]
				UpperCAmelCase_	:					Optional[int]     	=   negative_prompt
				UpperCAmelCase_	:					List[str]     	=   3 * [inputs["prompt"]]

				# forward
				UpperCAmelCase_	:					List[Any]     	=   pipe(**lowerCAmelCase_		)
				UpperCAmelCase_	:					Tuple     	=   output.images[0, -3:, -3:, -1]

				UpperCAmelCase_	:					Dict     	=   self.get_dummy_inputs()
				UpperCAmelCase_	:					Union[str, Any]     	=   3 * [inputs.pop("prompt"		)]

				UpperCAmelCase_	:					Optional[int]     	=   []
				for p in [prompt, negative_prompt]:
							UpperCAmelCase_	:					Optional[int]     	=   pipe.tokenizer(
							    lowerCAmelCase_      ,      padding="max_length"      ,      max_length=pipe.tokenizer.model_max_length      ,      truncation=lowerCAmelCase_      ,      return_tensors="np"      ,      )
							UpperCAmelCase_	:					Union[str, Any]     	=   text_inputs["input_ids"]

							embeds.append(pipe.text_encoder(input_ids=text_inputs.astype(np.intaa		)		)[0]		)

				UpperCAmelCase_	,      UpperCAmelCase_	:					Optional[Any]     	=   embeds

				# forward
				UpperCAmelCase_	:					Optional[int]     	=   pipe(**lowerCAmelCase_		)
				UpperCAmelCase_	:					List[Any]     	=   output.images[0, -3:, -3:, -1]

				assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()		).max() < 1e-4







@nightly
@require_onnxruntime
@require_torch_gpu
class      UpperCamelCase_			(unittest.TestCase	):





	@property
	def 	_SCREAMING_SNAKE_CASE ( self   :					Any		)		->     int:
				return (
				    "CUDAExecutionProvider",
				    {
				        "gpu_mem_limit": "15000000000",  # 15GB
				        "arena_extend_strategy": "kSameAsRequested",
				    },
				)





	@property
	def 	_SCREAMING_SNAKE_CASE ( self   :					List[str]		)		->     Union[str, Any]:
				UpperCAmelCase_	:					Union[str, Any]     	=   ort.SessionOptions()
				UpperCAmelCase_	:					Optional[int]     	=   False
				return options





	def 	_SCREAMING_SNAKE_CASE ( self   :					Dict		)		->     Tuple:
				# using the PNDM scheduler by default
				UpperCAmelCase_	:					str     	=   OnnxStableDiffusionPipeline.from_pretrained(
				    "CompVis/stable-diffusion-v1-4"      ,      revision="onnx"      ,      safety_checker=lowerCAmelCase_      ,      feature_extractor=lowerCAmelCase_      ,      provider=self.gpu_provider      ,      sess_options=self.gpu_options      ,      )
				sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_		)

				UpperCAmelCase_	:					List[Any]     	=   "A painting of a squirrel eating a burger"
				np.random.seed(0		)
				UpperCAmelCase_	:					Any     	=   sd_pipe([prompt]      ,      guidance_scale=6.0      ,      num_inference_steps=10      ,      output_type="np"		)
				UpperCAmelCase_	:					Tuple     	=   output.images

				UpperCAmelCase_	:					Optional[int]     	=   image[0, -3:, -3:, -1]

				assert image.shape == (1, 512, 512, 3)
				UpperCAmelCase_	:					Optional[int]     	=   np.array([0.0_4_5_2, 0.0_3_9_0, 0.0_0_8_7, 0.0_3_5_0, 0.0_6_1_7, 0.0_3_6_4, 0.0_5_4_4, 0.0_5_2_3, 0.0_7_2_0]		)

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





	def 	_SCREAMING_SNAKE_CASE ( self   :					Tuple		)		->     Union[str, Any]:
				UpperCAmelCase_	:					Dict     	=   DDIMScheduler.from_pretrained(
				    "runwayml/stable-diffusion-v1-5"      ,      subfolder="scheduler"      ,      revision="onnx"		)
				UpperCAmelCase_	:					str     	=   OnnxStableDiffusionPipeline.from_pretrained(
				    "runwayml/stable-diffusion-v1-5"      ,      revision="onnx"      ,      scheduler=lowerCAmelCase_      ,      safety_checker=lowerCAmelCase_      ,      feature_extractor=lowerCAmelCase_      ,      provider=self.gpu_provider      ,      sess_options=self.gpu_options      ,      )
				sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_		)

				UpperCAmelCase_	:					Optional[int]     	=   "open neural network exchange"
				UpperCAmelCase_	:					List[Any]     	=   np.random.RandomState(0		)
				UpperCAmelCase_	:					int     	=   sd_pipe([prompt]      ,      guidance_scale=7.5      ,      num_inference_steps=10      ,      generator=lowerCAmelCase_      ,      output_type="np"		)
				UpperCAmelCase_	:					Union[str, Any]     	=   output.images
				UpperCAmelCase_	:					List[Any]     	=   image[0, -3:, -3:, -1]

				assert image.shape == (1, 512, 512, 3)
				UpperCAmelCase_	:					List[Any]     	=   np.array([0.2_8_6_7, 0.1_9_7_4, 0.1_4_8_1, 0.7_2_9_4, 0.7_2_5_1, 0.6_6_6_7, 0.4_1_9_4, 0.5_6_4_2, 0.6_4_8_6]		)

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





	def 	_SCREAMING_SNAKE_CASE ( self   :					Optional[int]		)		->     Optional[Any]:
				UpperCAmelCase_	:					Dict     	=   LMSDiscreteScheduler.from_pretrained(
				    "runwayml/stable-diffusion-v1-5"      ,      subfolder="scheduler"      ,      revision="onnx"		)
				UpperCAmelCase_	:					Optional[int]     	=   OnnxStableDiffusionPipeline.from_pretrained(
				    "runwayml/stable-diffusion-v1-5"      ,      revision="onnx"      ,      scheduler=lowerCAmelCase_      ,      safety_checker=lowerCAmelCase_      ,      feature_extractor=lowerCAmelCase_      ,      provider=self.gpu_provider      ,      sess_options=self.gpu_options      ,      )
				sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_		)

				UpperCAmelCase_	:					List[Any]     	=   "open neural network exchange"
				UpperCAmelCase_	:					Optional[Any]     	=   np.random.RandomState(0		)
				UpperCAmelCase_	:					Dict     	=   sd_pipe([prompt]      ,      guidance_scale=7.5      ,      num_inference_steps=10      ,      generator=lowerCAmelCase_      ,      output_type="np"		)
				UpperCAmelCase_	:					Tuple     	=   output.images
				UpperCAmelCase_	:					Optional[Any]     	=   image[0, -3:, -3:, -1]

				assert image.shape == (1, 512, 512, 3)
				UpperCAmelCase_	:					Dict     	=   np.array([0.2_3_0_6, 0.1_9_5_9, 0.1_5_9_3, 0.6_5_4_9, 0.6_3_9_4, 0.5_4_0_8, 0.5_0_6_5, 0.6_0_1_0, 0.6_1_6_1]		)

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





	def 	_SCREAMING_SNAKE_CASE ( self   :					int		)		->     Any:
				UpperCAmelCase_	:					List[Any]     	=   0

				def test_callback_fn(lowerCAmelCase_   :					int      ,      lowerCAmelCase_   :					int      ,      lowerCAmelCase_   :					np.ndarray		) -> None:
							UpperCAmelCase_	:					Optional[Any]     	=   True
							nonlocal number_of_steps
							number_of_steps += 1
							if step == 0:
										assert latents.shape == (1, 4, 64, 64)
										UpperCAmelCase_	:					Optional[Any]     	=   latents[0, -3:, -3:, -1]
										UpperCAmelCase_	:					str     	=   np.array(
										    [-0.6_7_7_2, -0.3_8_3_5, -1.2_4_5_6, 0.1_9_0_5, -1.0_9_7_4, 0.6_9_6_7, -1.9_3_5_3, 0.0_1_7_8, 1.0_1_6_7]		)

										assert np.abs(latents_slice.flatten() - expected_slice		).max() < 1e-3
							elif step == 5:
										assert latents.shape == (1, 4, 64, 64)
										UpperCAmelCase_	:					Optional[int]     	=   latents[0, -3:, -3:, -1]
										UpperCAmelCase_	:					Optional[int]     	=   np.array(
										    [-0.3_3_5_1, 0.2_2_4_1, -0.1_8_3_7, -0.2_3_2_5, -0.6_5_7_7, 0.3_3_9_3, -0.0_2_4_1, 0.5_8_9_9, 1.3_8_7_5]		)

										assert np.abs(latents_slice.flatten() - expected_slice		).max() < 1e-3

				UpperCAmelCase_	:					List[Any]     	=   False

				UpperCAmelCase_	:					List[Any]     	=   OnnxStableDiffusionPipeline.from_pretrained(
				    "runwayml/stable-diffusion-v1-5"      ,      revision="onnx"      ,      safety_checker=lowerCAmelCase_      ,      feature_extractor=lowerCAmelCase_      ,      provider=self.gpu_provider      ,      sess_options=self.gpu_options      ,      )
				pipe.set_progress_bar_config(disable=lowerCAmelCase_		)

				UpperCAmelCase_	:					List[str]     	=   "Andromeda galaxy in a bottle"

				UpperCAmelCase_	:					int     	=   np.random.RandomState(0		)
				pipe(
				    prompt=lowerCAmelCase_      ,      num_inference_steps=5      ,      guidance_scale=7.5      ,      generator=lowerCAmelCase_      ,      callback=lowerCAmelCase_      ,      callback_steps=1      ,      )
				assert test_callback_fn.has_been_called
				assert number_of_steps == 6





	def 	_SCREAMING_SNAKE_CASE ( self   :					Any		)		->     Optional[Any]:
				UpperCAmelCase_	:					str     	=   OnnxStableDiffusionPipeline.from_pretrained(
				    "runwayml/stable-diffusion-v1-5"      ,      revision="onnx"      ,      safety_checker=lowerCAmelCase_      ,      feature_extractor=lowerCAmelCase_      ,      provider=self.gpu_provider      ,      sess_options=self.gpu_options      ,      )
				assert isinstance(lowerCAmelCase_      ,      lowerCAmelCase_		)
				assert pipe.safety_checker is None

				UpperCAmelCase_	:					Dict     	=   pipe("example prompt"      ,      num_inference_steps=2		).images[0]
				assert image is not None

				# check that there's no error when saving a pipeline with one of the models being None
				with tempfile.TemporaryDirectory() as tmpdirname:
							pipe.save_pretrained(lowerCAmelCase_		)
							UpperCAmelCase_	:					Any     	=   OnnxStableDiffusionPipeline.from_pretrained(lowerCAmelCase_		)

				# sanity check that the pipeline still works
				assert pipe.safety_checker is None
				UpperCAmelCase_	:					Tuple     	=   pipe("example prompt"      ,      num_inference_steps=2		).images[0]
				assert image is not None








'''simple docstring'''





import pickle

import numpy as np
from matplotlib import pyplot as plt



class  snake_case__						:
 def __init__(   self						:							List[Any]      ,     __a						:							str      ,     __a						:							Dict      ,     __a						:							List[Any]      ,     __a						:							str      ,     __a						:							str      ,     __a						:							List[str]=0.2      ,     __a						:							Any=0.2  )      ->   Any:







       '''simple docstring'''
       __snake_case :   Any													=    bp_numa
       __snake_case :   str													=    bp_numa
       __snake_case :   Optional[Any]													=    bp_numa
       __snake_case :   Any													=    conva_get[:2]
       __snake_case :   Dict													=    conva_get[2]
       __snake_case :   Optional[int]													=    size_pa
       __snake_case :   str													=    rate_w
       __snake_case :   Optional[Any]													=    rate_t
       __snake_case :   Optional[Any]													=    [
           np.mat(-1 * np.random.rand(self.conva[0]      ,     self.conva[0]  ) + 0.5  )
           for i in range(self.conva[1]  )
       ]
       __snake_case :   List[Any]													=    np.mat(-1 * np.random.rand(self.num_bpa      ,     self.num_bpa  ) + 0.5  )
       __snake_case :   Any													=    np.mat(-1 * np.random.rand(self.num_bpa      ,     self.num_bpa  ) + 0.5  )
       __snake_case :   Optional[int]													=    -2 * np.random.rand(self.conva[1]  ) + 1
       __snake_case :   Optional[Any]													=    -2 * np.random.rand(self.num_bpa  ) + 1
       __snake_case :   str													=    -2 * np.random.rand(self.num_bpa  ) + 1
 def A_   (   self						:							int      ,     __a						:							Dict  )      ->   Optional[Any]:







       '''simple docstring'''
       # save model dict with pickle
       __snake_case :   int													=    {
           'num_bp1': self.num_bpa,
           'num_bp2': self.num_bpa,
           'num_bp3': self.num_bpa,
           'conv1': self.conva,
           'step_conv1': self.step_conva,
           'size_pooling1': self.size_poolinga,
           'rate_weight': self.rate_weight,
           'rate_thre': self.rate_thre,
           'w_conv1': self.w_conva,
           'wkj': self.wkj,
           'vji': self.vji,
           'thre_conv1': self.thre_conva,
           'thre_bp2': self.thre_bpa,
           'thre_bp3': self.thre_bpa,
       }
       with open(__a      ,     'wb'  ) as f:
             pickle.dump(__a      ,     __a  )

       print(f'''Model saved： {save_path}'''  )
 @classmethod
 def A_   (   cls						:							List[Any]      ,     __a						:							Union[str, Any]  )      ->   int:







       '''simple docstring'''
       # read saved model
       with open(__a      ,     'rb'  ) as f:
             __snake_case :   Dict													=    pickle.load(__a  )  # noqa: S301

       __snake_case :   Tuple													=    model_dic.get('conv1'  )
       conv_get.append(model_dic.get('step_conv1'  )  )
       __snake_case :   Optional[Any]													=    model_dic.get('size_pooling1'  )
       __snake_case :   int													=    model_dic.get('num_bp1'  )
       __snake_case :   Optional[Any]													=    model_dic.get('num_bp2'  )
       __snake_case :   Optional[Any]													=    model_dic.get('num_bp3'  )
       __snake_case :   Any													=    model_dic.get('rate_weight'  )
       __snake_case :   Optional[int]													=    model_dic.get('rate_thre'  )
       # create model instance
       __snake_case :   Any													=    CNN(__a      ,     __a      ,     __a      ,     __a      ,     __a      ,     __a      ,     __a  )
       # modify model parameter
       __snake_case :   Dict													=    model_dic.get('w_conv1'  )
       __snake_case :   Any													=    model_dic.get('wkj'  )
       __snake_case :   List[str]													=    model_dic.get('vji'  )
       __snake_case :   int													=    model_dic.get('thre_conv1'  )
       __snake_case :   Optional[int]													=    model_dic.get('thre_bp2'  )
       __snake_case :   List[Any]													=    model_dic.get('thre_bp3'  )
       return conv_ins
 def A_   (   self						:							List[Any]      ,     __a						:							Tuple  )      ->   Optional[int]:







       '''simple docstring'''
       return 1 / (1 + np.exp(-1 * x  ))
 def A_   (   self						:							Any      ,     __a						:							Union[str, Any]  )      ->   List[Any]:







       '''simple docstring'''
       return round(__a      ,     3  )
 def A_   (   self						:							Optional[Any]      ,     __a						:							Tuple      ,     __a						:							List[str]      ,     __a						:							Dict      ,     __a						:							Optional[int]      ,     __a						:							List[str]  )      ->   str:







       '''simple docstring'''
       # convolution process
       __snake_case :   int													=    convs[0]
       __snake_case :   List[str]													=    convs[1]
       __snake_case :   Optional[Any]													=    np.shape(__a  )[0]
       # get the data slice of original image data, data_focus
       __snake_case :   str													=    []
       for i_focus in range(0      ,     size_data - size_conv + 1      ,     __a  ):
             for j_focus in range(0      ,     size_data - size_conv + 1      ,     __a  ):
                   __snake_case :   Any													=    data[
                       i_focus : i_focus + size_conv, j_focus : j_focus + size_conv
                   ]
                   data_focus.append(__a  )
        # calculate the feature map of every single kernel, and saved as list of matrix
       __snake_case :   Optional[int]													=    []
       __snake_case :   Dict													=    int((size_data - size_conv) / conv_step + 1  )
       for i_map in range(__a  ):
             __snake_case :   Optional[int]													=    []
             for i_focus in range(len(__a  )  ):
                   __snake_case :   List[Any]													=    (
                       np.sum(np.multiply(data_focus[i_focus]      ,     w_convs[i_map]  )  )
                       - thre_convs[i_map]
                   )
                   featuremap.append(self.sig(__a  )  )
             __snake_case :   List[Any]													=    np.asmatrix(__a  ).reshape(
                 __a      ,     __a  )
             data_featuremap.append(__a  )

       # expanding the data slice to One dimenssion
       __snake_case :   Union[str, Any]													=    []
       for each_focus in data_focus:
             focusa_list.extend(self.Expand_Mat(__a  )  )
       __snake_case :   List[Any]													=    np.asarray(__a  )
       return focus_list, data_featuremap
 def A_   (   self						:							Any      ,     __a						:							int      ,     __a						:							Tuple      ,     __a						:							List[Any]="average_pool"  )      ->   Dict:







       '''simple docstring'''
       # pooling process
       __snake_case :   List[str]													=    len(featuremaps[0]  )
       __snake_case :   Tuple													=    int(size_map / size_pooling  )
       __snake_case :   int													=    []
       for i_map in range(len(__a  )  ):
             __snake_case :   str													=    featuremaps[i_map]
             __snake_case :   Optional[Any]													=    []
             for i_focus in range(0      ,     __a      ,     __a  ):
                   for j_focus in range(0      ,     __a      ,     __a  ):
                         __snake_case :   Dict													=    feature_map[
                             i_focus : i_focus + size_pooling,
                             j_focus : j_focus + size_pooling,
                         ]
                         if pooling_type == "average_pool":
                               # average pooling
                               map_pooled.append(np.average(__a  )  )
                         elif pooling_type == "max_pooling":
                               # max pooling
                               map_pooled.append(np.max(__a  )  )
             __snake_case :   List[str]													=    np.asmatrix(__a  ).reshape(__a      ,     __a  )
             featuremap_pooled.append(__a  )
       return featuremap_pooled
 def A_   (   self						:							List[str]      ,     __a						:							Union[str, Any]  )      ->   int:







       '''simple docstring'''
       # expanding three dimension data to one dimension list
       __snake_case :   Tuple													=    []
       for i in range(len(__a  )  ):
             __snake_case :   Optional[int]													=    np.shape(data[i]  )
             __snake_case :   List[str]													=    data[i].reshape(1      ,     shapes[0] * shapes[1]  )
             __snake_case :   List[Any]													=    data_listed.getA().tolist()[0]
             data_expanded.extend(__a  )
       __snake_case :   Optional[int]													=    np.asarray(__a  )
       return data_expanded
 def A_   (   self						:							Union[str, Any]      ,     __a						:							int  )      ->   Any:







       '''simple docstring'''
       # expanding matrix to one dimension list
       __snake_case :   int													=    np.asarray(__a  )
       __snake_case :   str													=    np.shape(__a  )
       __snake_case :   Any													=    data_mat.reshape(1      ,     shapes[0] * shapes[1]  )
       return data_expanded
 def A_   (   self						:							List[Any]      ,     __a						:							str      ,     __a						:							Optional[int]      ,     __a						:							List[str]      ,     __a						:							int      ,     __a						:							List[str]  )      ->   Dict:







       '''simple docstring'''
       __snake_case :   Union[str, Any]													=    []
       __snake_case :   Tuple													=    0
       for i_map in range(__a  ):
             __snake_case :   Union[str, Any]													=    np.ones((size_map, size_map)  )
             for i in range(0      ,     __a      ,     __a  ):
                   for j in range(0      ,     __a      ,     __a  ):
                         __snake_case :   Any													=    pd_pool[
                             i_pool
                         ]
                         __snake_case :   List[Any]													=    i_pool + 1
             __snake_case :   List[Any]													=    np.multiply(
                 __a      ,     np.multiply(out_map[i_map]      ,     (1 - out_map[i_map])  )  )
             pd_all.append(__a  )
       return pd_all
 def A_   (   self						:							Tuple      ,     __a						:							List[str]      ,     __a						:							Optional[int]      ,     __a						:							Union[str, Any]      ,     __a						:							Dict      ,     __a						:							Tuple      ,     __a						:							Optional[int]=bool  )      ->   List[Any]:







       '''simple docstring'''
       # model traning
       print('----------------------Start Training-------------------------'  )
       print((' - - Shape: Train_Data  ', np.shape(__a  ))  )
       print((' - - Shape: Teach_Data  ', np.shape(__a  ))  )
       __snake_case :   str													=    0
       __snake_case :   List[str]													=    []
       __snake_case :   List[Any]													=    10000
       while rp < n_repeat and mse >= error_accuracy:
             __snake_case :   int													=    0
             print(f'''-------------Learning Time {rp}--------------'''  )
             for p in range(len(__a  )  ):
                   # print('------------Learning Image: %d--------------'%p)
                   __snake_case :   List[Any]													=    np.asmatrix(datas_train[p]  )
                   __snake_case :   Optional[Any]													=    np.asarray(datas_teach[p]  )
                   __snake_case							,     __snake_case :   List[Any]													=    self.convolute(
                       __a      ,     self.conva      ,     self.w_conva      ,     self.thre_conva      ,     conv_step=self.step_conva      ,     )
                   __snake_case :   Tuple													=    self.pooling(__a      ,     self.size_poolinga  )
                   __snake_case :   Dict													=    np.shape(__a  )
                   __snake_case :   Tuple													=    self._expand(__a  )
                   __snake_case :   str													=    data_bp_input

                   __snake_case :   List[Any]													=    np.dot(__a      ,     self.vji.T  ) - self.thre_bpa
                   __snake_case :   Any													=    self.sig(__a  )
                   __snake_case :   Tuple													=    np.dot(__a      ,     self.wkj.T  ) - self.thre_bpa
                   __snake_case :   Optional[Any]													=    self.sig(__a  )

                   # --------------Model Leaning ------------------------
                   # calculate error and gradient---------------
                   __snake_case :   Tuple													=    np.multiply(
                       (data_teach - bp_outa)      ,     np.multiply(__a      ,     (1 - bp_outa)  )  )
                   __snake_case :   Tuple													=    np.multiply(
                       np.dot(__a      ,     self.wkj  )      ,     np.multiply(__a      ,     (1 - bp_outa)  )  )
                   __snake_case :   Union[str, Any]													=    np.dot(__a      ,     self.vji  )

                   __snake_case :   Optional[int]													=    pd_i_all / (self.size_poolinga * self.size_poolinga)
                   __snake_case :   Tuple													=    pd_conva_pooled.T.getA().tolist()
                   __snake_case :   Optional[Any]													=    self._calculate_gradient_from_pool(
                       __a      ,     __a      ,     shape_featuremapa[0]      ,     shape_featuremapa[1]      ,     self.size_poolinga      ,     )
                   # weight and threshold learning process---------
                   # convolution layer
                   for k_conv in range(self.conva[1]  ):
                         __snake_case :   int													=    self._expand_mat(pd_conva_all[k_conv]  )
                         __snake_case :   Optional[int]													=    self.rate_weight * np.dot(__a      ,     __a  )

                         __snake_case :   List[Any]													=    self.w_conva[k_conv] + delta_w.reshape(
                             (self.conva[0], self.conva[0])  )

                         __snake_case :   Optional[Any]													=    (
                             self.thre_conva[k_conv]
                             - np.sum(pd_conva_all[k_conv]  ) * self.rate_thre
                         )
                   # all connected layer
                   __snake_case :   Tuple													=    self.wkj + pd_k_all.T * bp_outa * self.rate_weight
                   __snake_case :   Optional[Any]													=    self.vji + pd_j_all.T * bp_outa * self.rate_weight
                   __snake_case :   str													=    self.thre_bpa - pd_k_all * self.rate_thre
                   __snake_case :   Optional[int]													=    self.thre_bpa - pd_j_all * self.rate_thre
                   # calculate the sum error of all single image
                   __snake_case :   Any													=    np.sum(abs(data_teach - bp_outa  )  )
                   error_count += errors
                   # print('   ----Teach      ',data_teach)
                   # print('   ----BP_output  ',bp_out3)
             __snake_case :   Tuple													=    rp + 1
             __snake_case :   Tuple													=    error_count / patterns
             all_mse.append(__a  )

       def draw_error():
             __snake_case :   Union[str, Any]													=    [error_accuracy for i in range(int(n_repeat * 1.2  )  )]
             plt.plot(__a      ,     '+-'  )
             plt.plot(__a      ,     'r--'  )
             plt.xlabel('Learning Times'  )
             plt.ylabel('All_mse'  )
             plt.grid(__a      ,     alpha=0.5  )
             plt.show()

       print('------------------Training Complished---------------------'  )
       print((' - - Training epoch: ', rp, f'''     - - Mse: {mse:.6f}''')  )
       if draw_e:
             draw_error()
       return mse
 def A_   (   self						:							Tuple      ,     __a						:							Union[str, Any]  )      ->   List[Any]:







       '''simple docstring'''
       # model predict
       __snake_case :   str													=    []
       print('-------------------Start Testing-------------------------'  )
       print((' - - Shape: Test_Data  ', np.shape(__a  ))  )
       for p in range(len(__a  )  ):
             __snake_case :   int													=    np.asmatrix(datas_test[p]  )
             __snake_case							,     __snake_case :   str													=    self.convolute(
                 __a      ,     self.conva      ,     self.w_conva      ,     self.thre_conva      ,     conv_step=self.step_conva      ,     )
             __snake_case :   List[str]													=    self.pooling(__a      ,     self.size_poolinga  )
             __snake_case :   List[Any]													=    self._expand(__a  )

             __snake_case :   Optional[Any]													=    data_bp_input
             __snake_case :   Optional[Any]													=    bp_outa * self.vji.T - self.thre_bpa
             __snake_case :   Any													=    self.sig(__a  )
             __snake_case :   Any													=    bp_outa * self.wkj.T - self.thre_bpa
             __snake_case :   str													=    self.sig(__a  )
             produce_out.extend(bp_outa.getA().tolist()  )
       __snake_case :   List[Any]													=    [list(map(self.do_round      ,     __a  )  ) for each in produce_out]
       return np.asarray(__a  )





 def A_   (   self						:							Optional[Any]      ,     __a						:							Optional[int]  )      ->   Tuple:







       '''simple docstring'''
       # return the data of image after convoluting process so we can check it out
       __snake_case :   int													=    np.asmatrix(__a  )
       __snake_case							,     __snake_case :   int													=    self.convolute(
           __a      ,     self.conva      ,     self.w_conva      ,     self.thre_conva      ,     conv_step=self.step_conva      ,     )
       __snake_case :   Dict													=    self.pooling(__a      ,     self.size_poolinga  )

       return data_conveda, data_pooleda


if __name__ == "__main__":
    pass



