






"""simple docstring"""




import logging
import re

import pytorch_quantization
import pytorch_quantization.nn as quant_nn
import torch
from pytorch_quantization import calib
from pytorch_quantization.tensor_quant import QuantDescriptor


_snake_case          =   logging.getLogger(__name__)

_snake_case          =   50  # max width of layer names
_snake_case          =   70  # max width of quantizer names

def 			lowerCAmelCase__     (     UpperCamelCase__				):



  '''simple docstring'''
  _a  :       Optional[int]             =	parser.add_argument_group("""quant_trainer arguments"""				)
  group.add_argument("""--wprec"""     ,   type=UpperCamelCase__     ,   default=8     ,   help="""weight precision"""				)
  group.add_argument("""--aprec"""     ,   type=UpperCamelCase__     ,   default=8     ,   help="""activation precision"""				)
  group.add_argument("""--quant-per-tensor"""     ,   action="""store_true"""     ,   help="""per tensor weight scaling"""				)
  group.add_argument("""--quant-disable"""     ,   action="""store_true"""     ,   help="""disable all quantizers"""				)
  group.add_argument("""--quant-disable-embeddings"""     ,   action="""store_true"""     ,   help="""disable all embeddings quantizers"""				)
  group.add_argument("""--quant-disable-keyword"""     ,   type=UpperCamelCase__     ,   nargs="""+"""     ,   help="""disable quantizers by keyword"""				)
  group.add_argument("""--quant-disable-layer-module"""     ,   type=UpperCamelCase__     ,   help="""disable quantizers by keyword under layer."""				)
  group.add_argument("""--quant-enable-layer-module"""     ,   type=UpperCamelCase__     ,   help="""enable quantizers by keyword under layer"""				)
  group.add_argument("""--calibrator"""     ,   default="""max"""     ,   help="""which quantization range calibrator to use"""				)
  group.add_argument("""--percentile"""     ,   default=UpperCamelCase__     ,   type=UpperCamelCase__     ,   help="""percentile for PercentileCalibrator"""				)
  group.add_argument("""--fuse-qkv"""     ,   action="""store_true"""     ,   help="""use the same scale factor for qkv"""				)
  group.add_argument("""--clip-gelu"""     ,   metavar="""N"""     ,   type=UpperCamelCase__     ,   help="""clip gelu output maximum value to N"""				)
  group.add_argument(
      """--recalibrate-weights"""     ,   action="""store_true"""     ,   help=(
          """recalibrate weight amaxes by taking the max of the weights."""
          """ amaxes will be computed with the current quantization granularity (axis)."""
      )     ,   )

def 			lowerCAmelCase__     (     UpperCamelCase__				):



  '''simple docstring'''

  if args.calibrator == "max":
    _a  :       List[str]             =	"""max"""
  elif args.calibrator == "percentile":
    if args.percentile is None:
      raise ValueError("""Specify --percentile when using percentile calibrator"""				)
    _a  :       Dict             =	"""histogram"""
  elif args.calibrator == "mse":
    _a  :       Optional[Any]             =	"""histogram"""
  else:
    raise ValueError(F"""Invalid calibrator {args.calibrator}"""				)

  _a  :       str             =	QuantDescriptor(num_bits=args.aprec     ,   calib_method=UpperCamelCase__				)
  _a  :       Dict             =	QuantDescriptor(num_bits=args.wprec     ,   axis=(None if args.quant_per_tensor else (0,))				)
  quant_nn.QuantLinear.set_default_quant_desc_input(UpperCamelCase__				)
  quant_nn.QuantLinear.set_default_quant_desc_weight(UpperCamelCase__				)

def 			lowerCAmelCase__     (     UpperCamelCase__     ,   UpperCamelCase__     ,   UpperCamelCase__=False     ,   UpperCamelCase__=False				):



  '''simple docstring'''
  logger.info("""Configuring Model for Quantization"""				)
  logger.info(F"""using quantization package {pytorch_quantization.__file__}"""				)

  if not calib:
    if args.quant_disable_embeddings:
      set_quantizer_by_name(UpperCamelCase__     ,   ["""embeddings"""]     ,   which="""weight"""     ,   _disabled=UpperCamelCase__				)

    if args.quant_disable:
      set_quantizer_by_name(UpperCamelCase__     ,   [""""""]     ,   _disabled=UpperCamelCase__				)

    if args.quant_disable_keyword:
      set_quantizer_by_name(UpperCamelCase__     ,   args.quant_disable_keyword     ,   _disabled=UpperCamelCase__				)

    if args.quant_disable_layer_module:
      set_quantizer_by_name(UpperCamelCase__     ,   [R"""layer.\d+.""" + args.quant_disable_layer_module]     ,   _disabled=UpperCamelCase__				)

    if args.quant_enable_layer_module:
      set_quantizer_by_name(UpperCamelCase__     ,   [R"""layer.\d+.""" + args.quant_enable_layer_module]     ,   _disabled=UpperCamelCase__				)

    if args.recalibrate_weights:
      recalibrate_weights(UpperCamelCase__				)

    if args.fuse_qkv:
      fuse_qkv(UpperCamelCase__     ,   UpperCamelCase__				)

  if args.clip_gelu:
    clip_gelu(UpperCamelCase__     ,   args.clip_gelu				)

  # if args.local_rank in [-1, 0] and not calib:
  print_quant_summary(UpperCamelCase__				)

def 			lowerCAmelCase__     (     UpperCamelCase__				):



  '''simple docstring'''
  logger.info("""Enabling Calibration"""				)
  for name, module in model.named_modules():
    if name.endswith("""_quantizer"""				):
      if module._calibrator is not None:
        module.disable_quant()
        module.enable_calib()
      else:
        module.disable()
      logger.info(F"""{name:80}: {module}"""				)

def 			lowerCAmelCase__     (     UpperCamelCase__     ,   UpperCamelCase__				):



  '''simple docstring'''
  logger.info("""Loading calibrated amax"""				)
  for name, module in model.named_modules():
    if name.endswith("""_quantizer"""				):
      if module._calibrator is not None:
        if isinstance(module._calibrator     ,   calib.MaxCalibrator				):
          module.load_calib_amax()
        else:
          module.load_calib_amax("""percentile"""     ,   percentile=args.percentile				)
        module.enable_quant()
        module.disable_calib()
      else:
        module.enable()
  model.cuda()
  print_quant_summary(UpperCamelCase__				)

def 			lowerCAmelCase__     (     UpperCamelCase__     ,   UpperCamelCase__				):



  '''simple docstring'''

  def fusea(UpperCamelCase__     ,   UpperCamelCase__     ,   UpperCamelCase__				):
    for mod in [qq, qk, qv]:
      if not hasattr(UpperCamelCase__     ,   """_amax"""				):
        print("""          WARNING: NO AMAX BUFFER"""				)
        return
    _a  :       List[Any]             =	qq._amax.detach().item()
    _a  :       Union[str, Any]             =	qk._amax.detach().item()
    _a  :       Tuple             =	qv._amax.detach().item()

    _a  :       List[Any]             =	max(UpperCamelCase__     ,   UpperCamelCase__     ,   UpperCamelCase__				)
    qq._amax.fill_(UpperCamelCase__				)
    qk._amax.fill_(UpperCamelCase__				)
    qv._amax.fill_(UpperCamelCase__				)
    logger.info(F"""          q={q:5.2f} k={k:5.2f} v={v:5.2f} -> {amax:5.2f}"""				)

  for name, mod in model.named_modules():
    if name.endswith(""".attention.self"""				):
      logger.info(F"""FUSE_QKV: {name:{name_width}}"""				)
      fusea(mod.matmul_q_input_quantizer     ,   mod.matmul_k_input_quantizer     ,   mod.matmul_v_input_quantizer				)
      if args.quant_per_tensor:
        fusea(mod.query._weight_quantizer     ,   mod.key._weight_quantizer     ,   mod.value._weight_quantizer				)

def 			lowerCAmelCase__     (     UpperCamelCase__     ,   UpperCamelCase__				):



  '''simple docstring'''

  for name, mod in model.named_modules():
    if name.endswith(""".output.dense"""				) and not name.endswith("""attention.output.dense"""				):
      _a  :       Optional[int]             =	mod._input_quantizer._amax.data.detach().item()
      mod._input_quantizer._amax.data.detach().clamp_(max=UpperCamelCase__				)
      _a  :       Dict             =	mod._input_quantizer._amax.data.detach().item()
      logger.info(F"""CLIP_GELU: {name:{name_width}} amax: {amax_init:5.2f} -> {amax:5.2f}"""				)

def 			lowerCAmelCase__     (     UpperCamelCase__				):



  '''simple docstring'''

  for name, mod in model.named_modules():
    if hasattr(UpperCamelCase__     ,   """_weight_quantizer"""				) and mod._weight_quantizer.axis is not None:
      _a  :       List[str]             =	mod.weight.shape[0]
      _a  :       Any             =	mod._weight_quantizer._amax.detach()
      _a  :       List[str]             =	torch.ones(UpperCamelCase__     ,   dtype=amax.dtype     ,   device=amax.device				) * amax
      print(F"""expanding {name} {amax} -> {mod._weight_quantizer._amax}"""				)

def 			lowerCAmelCase__     (     UpperCamelCase__				):



  '''simple docstring'''

  for name, mod in model.named_modules():
    if hasattr(UpperCamelCase__     ,   """_weight_quantizer"""				):
      if not hasattr(mod.weight_quantizer     ,   """_amax"""				):
        print("""RECALIB: {name:{name_width}} WARNING: NO AMAX BUFFER"""				)
        continue

      # determine which axes to reduce across
      # e.g. a 4D tensor quantized per axis 0 should reduce over (1,2,3)
      _a  :       List[str]             =	set() if mod._weight_quantizer.axis is None else set(mod._weight_quantizer.axis				)
      _a  :       List[str]             =	set(range(len(mod.weight.size()				)				)				) - axis_set
      _a  :       Union[str, Any]             =	pytorch_quantization.utils.reduce_amax(mod.weight     ,   axis=UpperCamelCase__     ,   keepdims=UpperCamelCase__				).detach()
      logger.info(F"""RECALIB: {name:{name_width}} {mod._weight_quantizer._amax.flatten()} -> {amax.flatten()}"""				)
      _a  :       List[Any]             =	amax

def 			lowerCAmelCase__     (     UpperCamelCase__     ,   UpperCamelCase__=2_5     ,   UpperCamelCase__=1_8_0     ,   UpperCamelCase__=None				):



  '''simple docstring'''

  if ignore is None:
    _a  :       int             =	[]
  elif not isinstance(UpperCamelCase__     ,   UpperCamelCase__				):
    _a  :       Optional[int]             =	[ignore]

  _a  :       int             =	0
  for name, mod in model.named_modules():
    if not hasattr(UpperCamelCase__     ,   """weight"""				):
      continue
    _a  :       str             =	max(UpperCamelCase__     ,   len(UpperCamelCase__				)				)

  for name, mod in model.named_modules():
    _a  :       List[Any]             =	getattr(UpperCamelCase__     ,   """_input_quantizer"""     ,   UpperCamelCase__				)
    _a  :       List[Any]             =	getattr(UpperCamelCase__     ,   """_weight_quantizer"""     ,   UpperCamelCase__				)
    if not hasattr(UpperCamelCase__     ,   """weight"""				):
      continue
    if type(UpperCamelCase__				) in ignore:
      continue
    if [True for s in ignore if type(UpperCamelCase__				) is str and s in name]:
      continue
    _a  :       Tuple             =	F"""Act:{input_q.extra_repr()}"""
    _a  :       Optional[Any]             =	F"""Wgt:{weight_q.extra_repr()}"""
    _a  :       Dict             =	F"""{name:{name_width}} {act_str} {wgt_str}"""
    if len(UpperCamelCase__				) <= line_width:
      logger.info(UpperCamelCase__				)
    else:
      logger.info(F"""{name:{name_width}} {act_str}"""				)
      logger.info(F"""{'  ':{name_width}} {wgt_str}"""				)

def 			lowerCAmelCase__     (     UpperCamelCase__				):



  '''simple docstring'''
  _a  :       Optional[int]             =	0
  for name, mod in model.named_modules():
    if isinstance(UpperCamelCase__     ,   pytorch_quantization.nn.TensorQuantizer				):
      print(F"""{name:80} {mod}"""				)
      count += 1
  print(F"""{count} TensorQuantizers found in model"""				)

def 			lowerCAmelCase__     (     UpperCamelCase__     ,   UpperCamelCase__     ,   UpperCamelCase__     ,   UpperCamelCase__     ,   UpperCamelCase__				):



  '''simple docstring'''
  _a  :       Dict             =	getattr(UpperCamelCase__     ,   UpperCamelCase__     ,   UpperCamelCase__				)
  if quantizer_mod is not None:
    assert hasattr(UpperCamelCase__     ,   UpperCamelCase__				)
    setattr(UpperCamelCase__     ,   UpperCamelCase__     ,   UpperCamelCase__				)
  else:
    logger.warning(F"""{name} has no {quantizer}"""				)

def 			lowerCAmelCase__     (     UpperCamelCase__     ,   UpperCamelCase__     ,   UpperCamelCase__="both"     ,   **UpperCamelCase__				):



  '''simple docstring'''
  _a  :       Any             =	F"""Warning: changing {which} quantizers of {name:{qname_width}}"""
  for k, v in kwargs.items():
    s += F""" {k}={v}"""
    if which in ["input", "both"]:
      set_quantizer(UpperCamelCase__     ,   UpperCamelCase__     ,   """_input_quantizer"""     ,   UpperCamelCase__     ,   UpperCamelCase__				)
    if which in ["weight", "both"]:
      set_quantizer(UpperCamelCase__     ,   UpperCamelCase__     ,   """_weight_quantizer"""     ,   UpperCamelCase__     ,   UpperCamelCase__				)
  logger.info(UpperCamelCase__				)



def 			lowerCAmelCase__     (     UpperCamelCase__     ,   UpperCamelCase__     ,   **UpperCamelCase__				):



  '''simple docstring'''

  for name, mod in model.named_modules():
    if hasattr(UpperCamelCase__     ,   """_input_quantizer"""				) or hasattr(UpperCamelCase__     ,   """_weight_quantizer"""				):
      for n in names:
        if re.search(UpperCamelCase__     ,   UpperCamelCase__				):
          set_quantizers(UpperCamelCase__     ,   UpperCamelCase__     ,   **UpperCamelCase__				)
    elif name.endswith("""_quantizer"""				):
      for n in names:
        if re.search(UpperCamelCase__     ,   UpperCamelCase__				):
          _a  :       Dict             =	F"""Warning: changing {name:{name_width}}"""
          for k, v in kwargs.items():
            s += F""" {k}={v}"""
            setattr(UpperCamelCase__     ,   UpperCamelCase__     ,   UpperCamelCase__				)
          logger.info(UpperCamelCase__				)









"""simple docstring"""




import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


_snake_case          =   False






class        UpperCamelCase			(	unittest.TestCase      ):
 pass







@nightly
@require_torch_gpu
class        UpperCamelCase			(	unittest.TestCase      ):


 def _lowercase	(     self		:						Optional[Any]							)					->       Optional[Any]:
   # clean up the VRAM after each test
   super().tearDown()
   gc.collect()
   torch.cuda.empty_cache()


 def _lowercase	(     self		:						Tuple							)					->       List[Any]:
   _a  :       Tuple             =	VersatileDiffusionPipeline.from_pretrained("""shi-labs/versatile-diffusion"""							,						torch_dtype=torch.floataa							)
   pipe.to(UpperCAmelCase__							)
   pipe.set_progress_bar_config(disable=UpperCAmelCase__							)

   _a  :       List[str]             =	load_image(
       """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg"""							)

   _a  :       Optional[Any]             =	torch.manual_seed(0							)
   _a  :       Union[str, Any]             =	pipe.dual_guided(
       prompt="""first prompt"""							,						image=UpperCAmelCase__							,						text_to_image_strength=0.7_5							,						generator=UpperCAmelCase__							,						guidance_scale=7.5							,						num_inference_steps=2							,						output_type="""numpy"""							,						).images

   with tempfile.TemporaryDirectory() as tmpdirname:
     pipe.save_pretrained(UpperCAmelCase__							)
     _a  :       Dict             =	VersatileDiffusionPipeline.from_pretrained(UpperCAmelCase__							,						torch_dtype=torch.floataa							)
   pipe.to(UpperCAmelCase__							)
   pipe.set_progress_bar_config(disable=UpperCAmelCase__							)

   _a  :       Optional[Any]             =	generator.manual_seed(0							)
   _a  :       str             =	pipe.dual_guided(
       prompt="""first prompt"""							,						image=UpperCAmelCase__							,						text_to_image_strength=0.7_5							,						generator=UpperCAmelCase__							,						guidance_scale=7.5							,						num_inference_steps=2							,						output_type="""numpy"""							,						).images

   assert np.abs(image - new_image							).sum() < 1E-5, "Models don't have the same forward pass"







 def _lowercase	(     self		:						Optional[int]							)					->       Optional[int]:
   _a  :       Optional[int]             =	VersatileDiffusionPipeline.from_pretrained("""shi-labs/versatile-diffusion"""							,						torch_dtype=torch.floataa							)
   pipe.to(UpperCAmelCase__							)
   pipe.set_progress_bar_config(disable=UpperCAmelCase__							)

   _a  :       int             =	"""cyberpunk 2077"""
   _a  :       Tuple             =	load_image(
       """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg"""							)
   _a  :       Tuple             =	torch.manual_seed(0							)
   _a  :       Any             =	pipe.dual_guided(
       prompt=UpperCAmelCase__							,						image=UpperCAmelCase__							,						text_to_image_strength=0.7_5							,						generator=UpperCAmelCase__							,						guidance_scale=7.5							,						num_inference_steps=50							,						output_type="""numpy"""							,						).images

   _a  :       List[str]             =	image[0, 253:256, 253:256, -1]

   assert image.shape == (1, 512, 512, 3)
   _a  :       Optional[int]             =	np.array([0.1_4_4_8, 0.1_6_1_9, 0.1_7_4_1, 0.1_0_8_6, 0.1_1_4_7, 0.1_1_2_8, 0.1_1_9_9, 0.1_1_6_5, 0.1_0_0_1]							)

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

   _a  :       int             =	"""A painting of a squirrel eating a burger """
   _a  :       Tuple             =	torch.manual_seed(0							)
   _a  :       Union[str, Any]             =	pipe.text_to_image(
       prompt=UpperCAmelCase__							,						generator=UpperCAmelCase__							,						guidance_scale=7.5							,						num_inference_steps=50							,						output_type="""numpy"""							).images

   _a  :       int             =	image[0, 253:256, 253:256, -1]

   assert image.shape == (1, 512, 512, 3)
   _a  :       int             =	np.array([0.3_3_6_7, 0.3_1_6_9, 0.2_6_5_6, 0.3_8_7_0, 0.4_7_9_0, 0.3_7_9_6, 0.4_0_0_9, 0.4_8_7_8, 0.4_7_7_8]							)

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

   _a  :       str             =	pipe.image_variation(UpperCAmelCase__							,						generator=UpperCAmelCase__							,						output_type="""numpy"""							).images

   _a  :       str             =	image[0, 253:256, 253:256, -1]

   assert image.shape == (1, 512, 512, 3)
   _a  :       Optional[Any]             =	np.array([0.3_0_7_6, 0.3_1_2_3, 0.3_2_8_4, 0.3_7_8_2, 0.3_7_7_0, 0.3_8_9_4, 0.4_2_9_7, 0.4_3_3_1, 0.4_4_5_6]							)

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

