

import os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


lowerCAmelCase__       = "\\n    Text data.\n    Second line of data."

lowerCAmelCase__       = "file"
@pytest.fixture(scope="session"			)
def      _lowerCAmelCase(	__A			):
  UpperCAmelCase					      =  tmp_path_factory.mktemp("data"			) / (FILE_PATH + ".zstd")
  UpperCAmelCase					      =  bytes(UpperCAmelCase__   ,       "utf-8"			)
  with zstd.open(UpperCAmelCase__   ,       "wb"			) as f:
    f.write(UpperCAmelCase__			)
  return path
@pytest.fixture
def      _lowerCAmelCase(	__A			):
  with open(os.path.join(tmpfs.local_root_dir   ,       UpperCAmelCase__			)   ,       "w"			) as f:
    f.write(UpperCAmelCase__			)
  return FILE_PATH
@pytest.mark.parametrize("compression_format"   ,       ["gzip", "xz", "zstd"]			)
def      _lowerCAmelCase(	__A   ,       __A   ,       __A   ,       __A   ,       __A   ,       __A			):
  UpperCAmelCase					      =  {"gzip": gz_file, "xz": xz_file, "zstd": zstd_path}
  UpperCAmelCase					      =  input_paths[compression_format]
  UpperCAmelCase					      =  tmp_path / "cache"
  UpperCAmelCase					      =  DownloadConfig(cache_dir=UpperCAmelCase__   ,       extract_compressed_file=UpperCAmelCase__			)
  UpperCAmelCase					      =  cached_path(UpperCAmelCase__   ,       download_config=UpperCAmelCase__			)
  with open(UpperCAmelCase__			) as f:
    UpperCAmelCase					      =  f.read()
  with open(UpperCAmelCase__			) as f:
    UpperCAmelCase					      =  f.read()
  assert extracted_file_content == expected_file_content
@pytest.mark.parametrize("default_extracted"   ,       [True, False]			)
@pytest.mark.parametrize("default_cache_dir"   ,       [True, False]			)
def      _lowerCAmelCase(	__A   ,       __A   ,       __A   ,       __A   ,       __A			):
  UpperCAmelCase					      =  "custom_cache"
  UpperCAmelCase					      =  "custom_extracted_dir"
  UpperCAmelCase					      =  tmp_path / "custom_extracted_path"
  if default_extracted:
    UpperCAmelCase					      =  ("downloads" if default_cache_dir else custom_cache_dir, "extracted")
  else:
    monkeypatch.setattr("datasets.config.EXTRACTED_DATASETS_DIR"   ,       UpperCAmelCase__			)
    monkeypatch.setattr("datasets.config.EXTRACTED_DATASETS_PATH"   ,       str(UpperCAmelCase__			)			)
    UpperCAmelCase					      =  custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

  UpperCAmelCase					      =  xz_file
  UpperCAmelCase					      =  (
      DownloadConfig(extract_compressed_file=UpperCAmelCase__			)
      if default_cache_dir
      else DownloadConfig(cache_dir=tmp_path / custom_cache_dir   ,       extract_compressed_file=UpperCAmelCase__			)
  )
  UpperCAmelCase					      =  cached_path(UpperCAmelCase__   ,       download_config=UpperCAmelCase__			)
  assert Path(UpperCAmelCase__			).parent.parts[-2:] == expected
def      _lowerCAmelCase(	__A			):
  # absolute path
  UpperCAmelCase					      =  str(Path(UpperCAmelCase__			).resolve()			)
  assert cached_path(UpperCAmelCase__			) == text_file
  # relative path
  UpperCAmelCase					      =  str(Path(UpperCAmelCase__			).resolve().relative_to(Path(os.getcwd()			)			)			)
  assert cached_path(UpperCAmelCase__			) == text_file
def      _lowerCAmelCase(	__A			):
  # absolute path
  UpperCAmelCase					      =  str(tmp_path.resolve() / "__missing_file__.txt"			)
  with pytest.raises(UpperCAmelCase__			):
    cached_path(UpperCAmelCase__			)
  # relative path
  UpperCAmelCase					      =  "./__missing_file__.txt"
  with pytest.raises(UpperCAmelCase__			):
    cached_path(UpperCAmelCase__			)
def      _lowerCAmelCase(	__A			):
  UpperCAmelCase					      =  get_from_cache(F"tmp://{tmpfs_file}"			)
  with open(UpperCAmelCase__			) as f:
    UpperCAmelCase					      =  f.read()
  assert output_file_content == FILE_CONTENT
@patch("datasets.config.HF_DATASETS_OFFLINE"   ,       UpperCAmelCase__			)
def      _lowerCAmelCase(	):
  with pytest.raises(UpperCAmelCase__			):
    cached_path("https://huggingface.co"			)
@patch("datasets.config.HF_DATASETS_OFFLINE"   ,       UpperCAmelCase__			)
def      _lowerCAmelCase(	__A			):
  UpperCAmelCase					      =  tmp_path_factory.mktemp("data"			) / "file.html"
  with pytest.raises(UpperCAmelCase__			):
    http_get("https://huggingface.co"   ,       temp_file=UpperCAmelCase__			)
  with pytest.raises(UpperCAmelCase__			):
    http_head("https://huggingface.co"			)
@patch("datasets.config.HF_DATASETS_OFFLINE"   ,       UpperCAmelCase__			)
def      _lowerCAmelCase(	__A			):
  UpperCAmelCase					      =  tmp_path_factory.mktemp("data"			) / "file.html"
  with pytest.raises(UpperCAmelCase__			):
    ftp_get("ftp://huggingface.co"   ,       temp_file=UpperCAmelCase__			)
  with pytest.raises(UpperCAmelCase__			):
    ftp_head("ftp://huggingface.co"			)





@patch("datasets.config.HF_DATASETS_OFFLINE"   ,       UpperCAmelCase__			)
def      _lowerCAmelCase(	__A			):
  UpperCAmelCase					      =  tmp_path_factory.mktemp("data"			) / "file.html"
  with pytest.raises(UpperCAmelCase__			):
    fsspec_get("s3://huggingface.co"   ,       temp_file=UpperCAmelCase__			)
  with pytest.raises(UpperCAmelCase__			):
    fsspec_head("s3://huggingface.co"			)






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


lowerCAmelCase__       = logging.getLogger(__name__)

lowerCAmelCase__       = 50  # max width of layer names
lowerCAmelCase__       = 70  # max width of quantizer names
def      _lowerCAmelCase(	__A			):
  UpperCAmelCase					      =  parser.add_argument_group("quant_trainer arguments"			)
  group.add_argument("--wprec"   ,       type=__A   ,       default=8   ,       help="weight precision"			)
  group.add_argument("--aprec"   ,       type=__A   ,       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=__A   ,       nargs="+"   ,       help="disable quantizers by keyword"			)
  group.add_argument("--quant-disable-layer-module"   ,       type=__A   ,       help="disable quantizers by keyword under layer."			)
  group.add_argument("--quant-enable-layer-module"   ,       type=__A   ,       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=__A   ,       type=__A   ,       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=__A   ,       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(	__A			):

  if args.calibrator == "max":
    UpperCAmelCase					      =  "max"
  elif args.calibrator == "percentile":
    if args.percentile is None:
      raise ValueError("Specify --percentile when using percentile calibrator"			)
    UpperCAmelCase					      =  "histogram"
  elif args.calibrator == "mse":
    UpperCAmelCase					      =  "histogram"
  else:
    raise ValueError(F"Invalid calibrator {args.calibrator}"			)

  UpperCAmelCase					      =  QuantDescriptor(num_bits=args.aprec   ,       calib_method=__A			)
  UpperCAmelCase					      =  QuantDescriptor(num_bits=args.wprec   ,       axis=(None if args.quant_per_tensor else (0,))			)
  quant_nn.QuantLinear.set_default_quant_desc_input(__A			)
  quant_nn.QuantLinear.set_default_quant_desc_weight(__A			)
def      _lowerCAmelCase(	__A   ,       __A   ,       __A=False   ,       __A=False			):
  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(__A   ,       ["embeddings"]   ,       which="weight"   ,       _disabled=__A			)

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

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

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

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

    if args.recalibrate_weights:
      recalibrate_weights(__A			)

    if args.fuse_qkv:
      fuse_qkv(__A   ,       __A			)

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

  # if args.local_rank in [-1, 0] and not calib:
  print_quant_summary(__A			)
def      _lowerCAmelCase(	__A			):
  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(	__A   ,       __A			):
  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(__A			)
def      _lowerCAmelCase(	__A   ,       __A			):

  def fusea(__A   ,       __A   ,       __A			):
    for mod in [qq, qk, qv]:
      if not hasattr(__A   ,       "_amax"			):
        print("          WARNING: NO AMAX BUFFER"			)
        return
    UpperCAmelCase					      =  qq._amax.detach().item()
    UpperCAmelCase					      =  qk._amax.detach().item()
    UpperCAmelCase					      =  qv._amax.detach().item()

    UpperCAmelCase					      =  max(__A   ,       __A   ,       __A			)
    qq._amax.fill_(__A			)
    qk._amax.fill_(__A			)
    qv._amax.fill_(__A			)
    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(	__A   ,       __A			):

  for name, mod in model.named_modules():
    if name.endswith(".output.dense"			) and not name.endswith("attention.output.dense"			):
      UpperCAmelCase					      =  mod._input_quantizer._amax.data.detach().item()
      mod._input_quantizer._amax.data.detach().clamp_(max=__A			)
      UpperCAmelCase					      =  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(	__A			):

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

  for name, mod in model.named_modules():
    if hasattr(__A   ,       "_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)
      UpperCAmelCase					      =  set() if mod._weight_quantizer.axis is None else set(mod._weight_quantizer.axis			)
      UpperCAmelCase					      =  set(range(len(mod.weight.size()			)			)			) - axis_set
      UpperCAmelCase					      =  pytorch_quantization.utils.reduce_amax(mod.weight   ,       axis=__A   ,       keepdims=__A			).detach()
      logger.info(F"RECALIB: {name:{name_width}} {mod._weight_quantizer._amax.flatten()} -> {amax.flatten()}"			)
      UpperCAmelCase					      =  amax
def      _lowerCAmelCase(	__A   ,       __A=25   ,       __A=180   ,       __A=None			):

  if ignore is None:
    UpperCAmelCase					      =  []
  elif not isinstance(__A   ,       __A			):
    UpperCAmelCase					      =  [ignore]

  UpperCAmelCase					      =  0
  for name, mod in model.named_modules():
    if not hasattr(__A   ,       "weight"			):
      continue
    UpperCAmelCase					      =  max(__A   ,       len(__A			)			)

  for name, mod in model.named_modules():
    UpperCAmelCase					      =  getattr(__A   ,       "_input_quantizer"   ,       __A			)
    UpperCAmelCase					      =  getattr(__A   ,       "_weight_quantizer"   ,       __A			)
    if not hasattr(__A   ,       "weight"			):
      continue
    if type(__A			) in ignore:
      continue
    if [True for s in ignore if type(__A			) is str and s in name]:
      continue
    UpperCAmelCase					      =  F"Act:{input_q.extra_repr()}"
    UpperCAmelCase					      =  F"Wgt:{weight_q.extra_repr()}"
    UpperCAmelCase					      =  F"{name:{name_width}} {act_str} {wgt_str}"
    if len(__A			) <= line_width:
      logger.info(__A			)
    else:
      logger.info(F"{name:{name_width}} {act_str}"			)
      logger.info(F"{'  ':{name_width}} {wgt_str}"			)
def      _lowerCAmelCase(	__A			):
  UpperCAmelCase					      =  0
  for name, mod in model.named_modules():
    if isinstance(__A   ,       pytorch_quantization.nn.TensorQuantizer			):
      print(F"{name:80} {mod}"			)
      count += 1
  print(F"{count} TensorQuantizers found in model"			)
def      _lowerCAmelCase(	__A   ,       __A   ,       __A   ,       __A   ,       __A			):
  UpperCAmelCase					      =  getattr(__A   ,       __A   ,       __A			)
  if quantizer_mod is not None:
    assert hasattr(__A   ,       __A			)
    setattr(__A   ,       __A   ,       __A			)
  else:
    logger.warning(F"{name} has no {quantizer}"			)
def      _lowerCAmelCase(	__A   ,       __A   ,       __A="both"   ,       **__A			):
  UpperCAmelCase					      =  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(__A   ,       __A   ,       "_input_quantizer"   ,       __A   ,       __A			)
    if which in ["weight", "both"]:
      set_quantizer(__A   ,       __A   ,       "_weight_quantizer"   ,       __A   ,       __A			)
  logger.info(__A			)





def      _lowerCAmelCase(	__A   ,       __A   ,       **__A			):

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



