






"""simple docstring"""



import inspect
import warnings

import torch

from .state import AcceleratorState, GradientState
from .utils import DistributedType, honor_type, is_tpu_available


if is_tpu_available(check_device=False):
						import torch_xla.core.xla_model as xm



def 						lowercase   (    __snake_case    :			List[str]		, __snake_case    :			List[Any]     ):

		if isinstance(__lowercase		, (list, tuple)     ):
				return honor_type(__lowercase		, (move_to_device(__lowercase		, __lowercase     ) for t in state)     )
		elif isinstance(__lowercase		, __lowercase     ):
				return type(__lowercase     )({k: move_to_device(__lowercase		, __lowercase     ) for k, v in state.items()}     )
		elif isinstance(__lowercase		, torch.Tensor     ):
				return state.to(__lowercase     )
		return state






class  _UpperCAmelCase	(			torch.optim.Optimizer							):




					def __init__(    self					:						Union[str, Any]   ,						A					:						List[str]   ,						A					:						List[Any]=True   ,						A					:						Tuple=None       )  ->  Optional[Any]:
							lowercase_			:			Any								=       optimizer
							lowercase_			:			Union[str, Any]								=       scaler
							lowercase_			:			List[str]								=       AcceleratorState()
							lowercase_			:			int								=       GradientState()
							lowercase_			:			Optional[Any]								=       device_placement
							lowercase_			:			Optional[int]								=       False
							lowercase_			:			Optional[Any]								=       None

							# Handle device placement
							if device_placement:
									lowercase_			:			Optional[Any]								=       self.optimizer.state_dict()
									if self.accelerator_state.distributed_type == DistributedType.TPU:
											xm.send_cpu_data_to_device(UpperCamelCase_   ,						self.accelerator_state.device       )
									else:
											lowercase_			:			Optional[int]								=       move_to_device(UpperCamelCase_   ,						self.accelerator_state.device       )
									self.optimizer.load_state_dict(UpperCamelCase_       )




					@property
					def   A  (    self					:						Union[str, Any]       )  ->  str:
							return self.optimizer.state




					@state.setter
					def   A  (    self					:						Any   ,						A					:						Union[str, Any]       )  ->  Dict:
							lowercase_			:			str								=       state




					@property
					def   A  (    self					:						List[Any]       )  ->  Tuple:
							return self.optimizer.param_groups




					@param_groups.setter
					def   A  (    self					:						List[Any]   ,						A					:						Any       )  ->  Optional[int]:
							lowercase_			:			Optional[int]								=       param_groups




					@property
					def   A  (    self					:						int       )  ->  Dict:
							return self.optimizer.defaults




					@defaults.setter
					def   A  (    self					:						List[Any]   ,						A					:						Union[str, Any]       )  ->  Union[str, Any]:
							lowercase_			:			Dict								=       defaults




					def   A  (    self					:						str   ,						A					:						Tuple       )  ->  Optional[int]:
							self.optimizer.add_param_group(UpperCamelCase_       )




					def   A  (    self					:						Dict   ,						A					:						Dict       )  ->  int:



							if self.accelerator_state.distributed_type == DistributedType.TPU and self.device_placement:
									xm.send_cpu_data_to_device(UpperCamelCase_   ,						self.accelerator_state.device       )
							self.optimizer.load_state_dict(UpperCamelCase_       )




					def   A  (    self					:						Optional[int]       )  ->  int:
							return self.optimizer.state_dict()




					def   A  (    self					:						Dict   ,						A					:						List[Any]=None       )  ->  Optional[Any]:



							if self.gradient_state.sync_gradients:
									lowercase_			:			int								=       '''set_to_none''' in inspect.signature(self.optimizer.zero_grad       ).parameters
									if accept_arg:
											if set_to_none is None:
													lowercase_			:			Optional[int]								=       False
											self.optimizer.zero_grad(set_to_none=UpperCamelCase_       )
									else:
											if set_to_none is not None:
													raise ValueError('''`set_to_none` for Optimizer.zero_grad` is not supported by this optimizer.'''       )
											self.optimizer.zero_grad()




					def   A  (    self					:						int   ,						A					:						List[Any]=None       )  ->  List[str]:



							if self.gradient_state.sync_gradients:
									if self.accelerator_state.distributed_type == DistributedType.TPU:
											lowercase_			:			Optional[int]								=       {'''closure''': closure} if closure is not None else {}
											xm.optimizer_step(self.optimizer   ,						optimizer_args=UpperCamelCase_       )
									elif self.scaler is not None:
											lowercase_			:			Optional[Any]								=       False
											if self._last_scale is None:
													# `get_scale` is an async operation requiring full synchronization
													# on CPU and GPUs before finishing. As a result, we store away
													# the prior one to reduce the call overhead
													lowercase_			:			Dict								=       self.scaler.get_scale()
													lowercase_			:			Optional[Any]								=       True
											self.scaler.step(self.optimizer   ,						UpperCamelCase_       )
											self.scaler.update()
											lowercase_			:			List[Any]								=       self.scaler.get_scale()
											if not new_scale:
													# If we reduced the loss scale, it means the optimizer step was skipped because of gradient overflow.
													lowercase_			:			int								=       scale_after < self._last_scale
											lowercase_			:			List[Any]								=       scale_after

									else:
											self.optimizer.step(UpperCamelCase_       )




					def   A  (    self					:						int   ,						A					:						Any       )  ->  Union[str, Any]:



							for param_group in self.optimizer.param_groups:
									lowercase_			:			List[str]								=       [parameters_map.get(UpperCamelCase_   ,						UpperCamelCase_       ) for p in param_group['''params''']]




					@property
					def   A  (    self					:						Any       )  ->  Optional[Any]:
							warnings.warn(
							    '''The `is_overflow` property is deprecated and will be removed in version 1.0 of Accelerate use '''
							    '''`optimizer.step_was_skipped` instead.'''   ,						UpperCamelCase_   ,						)
							return self._is_overflow




					@property
					def   A  (    self					:						int       )  ->  Optional[int]:
							return self._is_overflow




					def __getstate__(    self					:						Dict       )  ->  Optional[int]:
							return self.__dict__.copy()






					def __setstate__(    self					:						str   ,						A					:						str       )  ->  str:
							self.__dict__.update(UpperCamelCase_       )











"""simple docstring"""



import math
import os

import cva
import numpy as np

__A		:	int      =		255.0



def 						lowercase   (    __snake_case    :			float		, __snake_case    :			float     ):
		lowercase_			:			Any								=       np.mean((original - contrast) ** 2     )
		if mse == 0:
				return 1_0_0

		return 2_0 * math.logaa(PIXEL_MAX / math.sqrt(__snake_case     )     )



def 						lowercase   (    ):
		lowercase_			:			Any								=       os.path.dirname(os.path.realpath(__snake_case     )     )
		# Loading images (original image and compressed image)
		lowercase_			:			Union[str, Any]								=       cva.imread(os.path.join(__snake_case		, '''image_data/original_image.png'''     )     )
		lowercase_			:			List[Any]								=       cva.imread(os.path.join(__snake_case		, '''image_data/compressed_image.png'''     )		, 1     )

		lowercase_			:			Tuple								=       cva.imread(os.path.join(__snake_case		, '''image_data/PSNR-example-base.png'''     )     )
		lowercase_			:			List[str]								=       cva.imread(
		    os.path.join(__snake_case		, '''image_data/PSNR-example-comp-10.jpg'''     )		, 1     )

		# Value expected: 29.73dB
		print('''-- First Test --'''     )
		print(F'''PSNR value is {peak_signal_to_noise_ratio(__snake_case		, __snake_case     )} dB'''     )

		# # Value expected: 31.53dB (Wikipedia Example)
		print('''\n-- Second Test --'''     )
		print(F'''PSNR value is {peak_signal_to_noise_ratio(__snake_case		, __snake_case     )} dB'''     )


if __name__ == "__main__":
						main()



