"""simple docstring"""

import unittest

from transformers.testing_utils import is_torch_available, require_torch, tooslow

from ...generation.test_utils import torch_device
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
							import torch

							from transformers import (
							    CpmAntConfig,
							    CpmAntForCausalLM,
							    CpmAntModel,
							    CpmAntTokenizer,
							)
@require_torch
class      lowerCamelCase__      :


			def __init__(       self			,	SCREAMING_SNAKE_CASE			,	SCREAMING_SNAKE_CASE=2			,	SCREAMING_SNAKE_CASE=8			,	SCREAMING_SNAKE_CASE=True			,	SCREAMING_SNAKE_CASE=False			,	SCREAMING_SNAKE_CASE=False			,	SCREAMING_SNAKE_CASE=False			,	SCREAMING_SNAKE_CASE=False			,	SCREAMING_SNAKE_CASE=99			,	SCREAMING_SNAKE_CASE=32			,	SCREAMING_SNAKE_CASE=3			,	SCREAMING_SNAKE_CASE=4			,	SCREAMING_SNAKE_CASE=37			,	SCREAMING_SNAKE_CASE=32			,	SCREAMING_SNAKE_CASE=128			,	SCREAMING_SNAKE_CASE=8			,	SCREAMING_SNAKE_CASE=8			,	SCREAMING_SNAKE_CASE=8			,	SCREAMING_SNAKE_CASE=1.0			,	SCREAMING_SNAKE_CASE=True			,	):



					"""simple docstring"""







					snake_case						:  Optional[int]        =			parent
					snake_case						:  List[str]        =			batch_size
					snake_case						:  List[str]        =			seq_length
					snake_case						:  Optional[int]        =			is_training
					snake_case						:  List[str]        =			use_token_type_ids
					snake_case						:  Tuple        =			use_input_mask
					snake_case						:  List[Any]        =			use_labels
					snake_case						:  Tuple        =			use_mc_token_ids
					snake_case						:  List[Any]        =			vocab_size
					snake_case						:  Optional[int]        =			hidden_size
					snake_case						:  Tuple        =			num_hidden_layers
					snake_case						:  Optional[Any]        =			num_attention_heads
					snake_case						:  Tuple        =			intermediate_size
					snake_case						:  Tuple        =			num_buckets
					snake_case						:  List[str]        =			max_distance
					snake_case						:  List[Any]        =			prompt_length
					snake_case						:  Dict        =			prompt_types
					snake_case						:  Tuple        =			segment_types
					snake_case						:  Optional[Any]        =			init_std
					snake_case						:  str        =			return_dict


			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					snake_case						:  List[str]        =			{}
					snake_case						:  str        =			ids_tensor([self.batch_size, self.seq_length]			,	self.vocab_size							).type(torch.intaa							)
					snake_case						:  Any        =			False

					snake_case						:  Any        =			self.get_config()

					return (config, input_ids)


			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					return CpmAntConfig(
					    vocab_size=self.vocab_size			,	hidden_size=self.hidden_size			,	num_hidden_layers=self.num_hidden_layers			,	num_attention_heads=self.num_attention_heads			,	dim_ff=self.intermediate_size			,	position_bias_num_buckets=self.num_buckets			,	position_bias_max_distance=self.max_distance			,	prompt_types=self.prompt_types			,	prompt_length=self.prompt_length			,	segment_types=self.segment_types			,	use_cache=SCREAMING_SNAKE_CASE			,	init_std=self.init_std			,	return_dict=self.return_dict			,	)


			def 							lowerCamelCase_			(       self			,	SCREAMING_SNAKE_CASE			,	SCREAMING_SNAKE_CASE			,	*SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					snake_case						:  List[str]        =			CpmAntModel(config=SCREAMING_SNAKE_CASE							)
					model.to(SCREAMING_SNAKE_CASE							)
					model.eval()

					snake_case						:  int        =			model(**SCREAMING_SNAKE_CASE							).last_hidden_state

					self.parent.assertEqual(hidden_states.shape			,	(self.batch_size, self.seq_length, config.hidden_size)							)


			def 							lowerCamelCase_			(       self			,	SCREAMING_SNAKE_CASE			,	SCREAMING_SNAKE_CASE			,	*SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					snake_case						:  List[str]        =			CpmAntForCausalLM(SCREAMING_SNAKE_CASE							)
					model.to(SCREAMING_SNAKE_CASE							)
					snake_case						:  str        =			input_ids["input_ids"].to(SCREAMING_SNAKE_CASE							)
					model.eval()

					snake_case						:  str        =			model(**SCREAMING_SNAKE_CASE							)
					self.parent.assertEqual(
					    model_output.logits.shape			,	(self.batch_size, self.seq_length, config.vocab_size + config.prompt_types * config.prompt_length)			,	)



			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					snake_case						:  Dict        =			self.prepare_config_and_inputs()
					return config, inputs_dict







@require_torch
class      lowerCamelCase__      (					lowerCamelCase_			,					lowerCamelCase_			,					unittest.TestCase     ):
			a__	: Optional[int]        =       (CpmAntModel, CpmAntForCausalLM) if is_torch_available() else ()
			a__	: List[str]        =       (
			    {"""feature-extraction""": CpmAntModel, """text-generation""": CpmAntForCausalLM} if is_torch_available() else {}
			)

			a__	: List[Any]        =       False
			a__	: Any        =       False
			a__	: Any        =       False
			a__	: Optional[Any]        =       False
			a__	: Optional[int]        =       False


			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					snake_case						:  Tuple        =			CpmAntModelTester(self							)
					snake_case						:  Dict        =			ConfigTester(self			,	config_class=SCREAMING_SNAKE_CASE							)


			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					self.config_tester.create_and_test_config_common_properties()
					self.config_tester.create_and_test_config_to_json_string()
					self.config_tester.create_and_test_config_to_json_file()
					self.config_tester.create_and_test_config_from_and_save_pretrained()
					self.config_tester.check_config_can_be_init_without_params()
					self.config_tester.check_config_arguments_init()


			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					unittest.skip("CPMAnt doesn't support input_embeds."							)(self.test_inputs_embeds							)


			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					unittest.skip(
					    "CPMAnt doesn't support retain grad in hidden_states or attentions, because prompt management will peel off the output.hidden_states from graph.\
                 So is attentions. We strongly recommand you use loss to tune model."							)(self.test_retain_grad_hidden_states_attentions							)


			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					snake_case						:  Any        =			self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_cpmant_model(SCREAMING_SNAKE_CASE			,	SCREAMING_SNAKE_CASE							)



			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					snake_case						:  List[str]        =			self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_lm_head_model(SCREAMING_SNAKE_CASE			,	SCREAMING_SNAKE_CASE							)







@require_torch
class      lowerCamelCase__      (					unittest.TestCase     ):


			@tooslow
			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					snake_case						:  Optional[int]        =			"今天天气真好！"
					snake_case						:  Optional[int]        =			"openbmb/cpm-ant-10b"
					snake_case						:  Union[str, Any]        =			CpmAntModel.from_pretrained(SCREAMING_SNAKE_CASE							)
					snake_case						:  str        =			CpmAntTokenizer.from_pretrained(SCREAMING_SNAKE_CASE							)
					snake_case						:  Any        =			tokenizer(SCREAMING_SNAKE_CASE			,	return_tensors="pt"							)
					snake_case						:  Optional[int]        =			model(**SCREAMING_SNAKE_CASE							).last_hidden_state

					snake_case						:  Dict        =			torch.tensor(
					    [[[6.17_08, 5.92_44, 1.08_35], [6.52_07, 6.28_93, -11.33_24], [-1.01_07, -0.05_76, -5.95_77]]]			,	)
					self.assertTrue(torch.allclose(hidden_states[:, :3, :3]			,	SCREAMING_SNAKE_CASE			,	atol=1E-2							)							)







@require_torch
class      lowerCamelCase__      (					unittest.TestCase     ):


			@tooslow
			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					snake_case						:  int        =			"今天天气真好！"
					snake_case						:  Optional[int]        =			"openbmb/cpm-ant-10b"
					snake_case						:  Tuple        =			CpmAntForCausalLM.from_pretrained(SCREAMING_SNAKE_CASE							)
					snake_case						:  Any        =			CpmAntTokenizer.from_pretrained(SCREAMING_SNAKE_CASE							)
					snake_case						:  int        =			tokenizer(SCREAMING_SNAKE_CASE			,	return_tensors="pt"							)
					snake_case						:  List[Any]        =			model(**SCREAMING_SNAKE_CASE							).logits

					snake_case						:  Any        =			torch.tensor(
					    [[[-6.42_67, -6.40_83, -6.39_58], [-5.88_02, -5.94_47, -5.78_11], [-5.38_96, -5.48_20, -5.42_95]]]			,	)
					self.assertTrue(torch.allclose(hidden_states[:, :3, :3]			,	SCREAMING_SNAKE_CASE			,	atol=1E-2							)							)


			@tooslow
			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					snake_case						:  Optional[Any]        =			"openbmb/cpm-ant-10b"
					snake_case						:  str        =			CpmAntForCausalLM.from_pretrained(SCREAMING_SNAKE_CASE							)
					snake_case						:  Tuple        =			CpmAntTokenizer.from_pretrained(SCREAMING_SNAKE_CASE							)
					snake_case						:  List[Any]        =			"今天天气不错，"
					snake_case						:  Union[str, Any]        =			"今天天气不错，阳光明媚，我和妈妈一起去超市买东西。\n在超市里，我看到了一个很好玩的玩具，它的名字叫“机器人”。它有一个圆圆的脑袋，两只圆圆的眼睛，还有一个圆圆的"
					snake_case						:  Dict        =			tokenizer(SCREAMING_SNAKE_CASE			,	return_tensors="pt"							)
					snake_case						:  Tuple        =			model.generate(**SCREAMING_SNAKE_CASE							)
					snake_case						:  Union[str, Any]        =			tokenizer.batch_decode(SCREAMING_SNAKE_CASE							)
					self.assertEqual(SCREAMING_SNAKE_CASE			,	SCREAMING_SNAKE_CASE							)



			@tooslow
			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					snake_case						:  List[str]        =			"openbmb/cpm-ant-10b"
					snake_case						:  Dict        =			CpmAntForCausalLM.from_pretrained(SCREAMING_SNAKE_CASE							)
					snake_case						:  List[Any]        =			CpmAntTokenizer.from_pretrained(SCREAMING_SNAKE_CASE							)
					snake_case						:  Any        =			["今天天气不错，", "新年快乐，万事如意！"]
					snake_case						:  Union[str, Any]        =			[
					    "今天天气不错，阳光明媚，我和妈妈一起去超市买东西。\n在超市里，我看到了一个很好玩的玩具，它的名字叫“机器人”。它有一个圆圆的脑袋，两只圆圆的眼睛，还有一个圆圆的",
					    "新年快乐，万事如意！在这辞旧迎新的美好时刻，我谨代表《农村新技术》杂志社全体同仁，向一直以来关心、支持《农村新技术》杂志发展的各级领导、各界朋友和广大读者致以最诚挚的",
					]
					snake_case						:  Optional[int]        =			tokenizer(SCREAMING_SNAKE_CASE			,	return_tensors="pt"			,	padding=SCREAMING_SNAKE_CASE							)
					snake_case						:  Optional[int]        =			model.generate(**SCREAMING_SNAKE_CASE							)
					snake_case						:  Any        =			tokenizer.batch_decode(SCREAMING_SNAKE_CASE							)
					self.assertEqual(SCREAMING_SNAKE_CASE			,	SCREAMING_SNAKE_CASE							)







"""simple docstring"""

import json
import os
import subprocess
import sys
import warnings
from argparse import ArgumentParser
from contextlib import AbstractContextManager
from typing import Dict, List, Optional

import requests

from ..utils import logging
from . import BaseTransformersCLICommand


__A							=     logging.get_logger(__name__)  # pylint: disable=invalid-name


__A							=     "lfs-multipart-upload"
class      lowerCamelCase__      (					lowerCamelCase_     ):


			@staticmethod
			def 							lowerCamelCase_			(       SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					snake_case						:  Any        =			parser.add_parser(
					    "lfs-enable-largefiles"			,	help=(
					        "Deprecated: use `huggingface-cli` instead. Configure your repository to enable upload of files > 5GB."
					    )			,	)
					enable_parser.add_argument("path"			,	type=SCREAMING_SNAKE_CASE			,	help="Local path to repository you want to configure."							)
					enable_parser.set_defaults(func=lambda SCREAMING_SNAKE_CASE							: LfsEnableCommand(SCREAMING_SNAKE_CASE							)							)

					snake_case						:  str        =			parser.add_parser(
					    SCREAMING_SNAKE_CASE			,	help=(
					        "Deprecated: use `huggingface-cli` instead. "
					        "Command will get called by git-lfs, do not call it directly."
					    )			,	)
					upload_parser.set_defaults(func=lambda SCREAMING_SNAKE_CASE							: LfsUploadCommand(SCREAMING_SNAKE_CASE							)							)







class      lowerCamelCase__      :


			def __init__(       self			,	SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					snake_case						:  Any        =			args



			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					warnings.warn(
					    "Managing repositories through transformers-cli is deprecated. Please use `huggingface-cli` instead."							)
					snake_case						:  Any        =			os.path.abspath(self.args.path							)
					if not os.path.isdir(SCREAMING_SNAKE_CASE							):
							print("This does not look like a valid git repo."							)
							exit(1							)
					subprocess.run(
					    "git config lfs.customtransfer.multipart.path transformers-cli".split()			,	check=SCREAMING_SNAKE_CASE			,	cwd=SCREAMING_SNAKE_CASE							)
					subprocess.run(
					    F'''git config lfs.customtransfer.multipart.args {LFS_MULTIPART_UPLOAD_COMMAND}'''.split()			,	check=SCREAMING_SNAKE_CASE			,	cwd=SCREAMING_SNAKE_CASE			,	)
					print("Local repo set up for largefiles"							)







def 					UpperCamelCase__       (						lowercase__					:       Dict				):
		snake_case						:  Tuple        =			json.dumps(lowercase__				) + "\n"
		sys.stdout.write(lowercase__				)
		sys.stdout.flush()



def 					UpperCamelCase__       (						):
		snake_case						:  Union[str, Any]        =			json.loads(sys.stdin.readline().strip()				)

		if "terminate" in (msg.get("type"				), msg.get("event"				)):
				# terminate message received
				return None

		if msg.get("event"				) not in ("download", "upload"):
				logger.critical("Received unexpected message"				)
				sys.exit(1				)

		return msg
class      lowerCamelCase__      (					lowerCamelCase_     ):


			def __init__(       self			,	SCREAMING_SNAKE_CASE			,	SCREAMING_SNAKE_CASE			,	SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					snake_case						:  int        =			filepath
					snake_case						:  Any        =			seek_from
					snake_case						:  Any        =			read_limit
					snake_case						:  Any        =			0


			def __enter__(       self							):



					"""simple docstring"""







					snake_case						:  Optional[int]        =			open(self.filepath			,	"rb"							)
					self.f.seek(self.seek_from							)
					return self


			def __len__(       self							):



					"""simple docstring"""







					snake_case						:  int        =			os.fstat(self.f.fileno()							).st_size
					return min(self.read_limit			,	total_length - self.seek_from							)


			def 							lowerCamelCase_			(       self			,	SCREAMING_SNAKE_CASE=-1							):



					"""simple docstring"""







					if self.n_seen >= self.read_limit:
							return b""
					snake_case						:  Any        =			self.read_limit - self.n_seen
					snake_case						:  int        =			self.f.read(remaining_amount if n < 0 else min(SCREAMING_SNAKE_CASE			,	SCREAMING_SNAKE_CASE							)							)
					self.n_seen += len(SCREAMING_SNAKE_CASE							)
					return data


			def __iter__(       self							):



					"""simple docstring"""







					yield self.read(n=4 * 1_024 * 1_024							)



			def __exit__(       self			,	*SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					self.f.close()







class      lowerCamelCase__      :


			def __init__(       self			,	SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					snake_case						:  List[str]        =			args



			def 							lowerCamelCase_			(       self							):



					"""simple docstring"""







					snake_case						:  Tuple        =			json.loads(sys.stdin.readline().strip()							)
					if not (init_msg.get("event"							) == "init" and init_msg.get("operation"							) == "upload"):
							write_msg({"error": {"code": 32, "message": "Wrong lfs init operation"}}							)
							sys.exit(1							)

					# The transfer process should use the information it needs from the
					# initiation structure, and also perform any one-off setup tasks it
					# needs to do. It should then respond on stdout with a simple empty
					# confirmation structure, as follows:
					write_msg({}							)

					# After the initiation exchange, git-lfs will send any number of
					# transfer requests to the stdin of the transfer process, in a serial sequence.
					while True:
							snake_case						:  str        =			read_msg()
							if msg is None:
									# When all transfers have been processed, git-lfs will send
									# a terminate event to the stdin of the transfer process.
									# On receiving this message the transfer process should
									# clean up and terminate. No response is expected.
									sys.exit(0							)

							snake_case						:  List[Any]        =			msg["oid"]
							snake_case						:  Any        =			msg["path"]
							snake_case						:  List[Any]        =			msg["action"]["href"]
							snake_case						:  Tuple        =			msg["action"]["header"]
							snake_case						:  Optional[int]        =			int(header.pop("chunk_size"							)							)
							snake_case						:  List[str]        =			list(header.values()							)

							snake_case						:  Union[str, Any]        =			[]
							for i, presigned_url in enumerate(SCREAMING_SNAKE_CASE							):
									with FileSlice(SCREAMING_SNAKE_CASE			,	seek_from=i * chunk_size			,	read_limit=SCREAMING_SNAKE_CASE							) as data:
											snake_case						:  List[Any]        =			requests.put(SCREAMING_SNAKE_CASE			,	data=SCREAMING_SNAKE_CASE							)
											r.raise_for_status()
											parts.append(
											    {
											        "etag": r.headers.get("etag"							),
											        "partNumber": i + 1,
											    }							)
											# In order to support progress reporting while data is uploading / downloading,
											# the transfer process should post messages to stdout
											write_msg(
											    {
											        "event": "progress",
											        "oid": oid,
											        "bytesSoFar": (i + 1) * chunk_size,
											        "bytesSinceLast": chunk_size,
											    }							)
											# Not precise but that's ok.

							snake_case						:  str        =			requests.post(
							    SCREAMING_SNAKE_CASE			,	json={
							        "oid": oid,
							        "parts": parts,
							    }			,	)
							r.raise_for_status()

							write_msg({"event": "complete", "oid": oid}							)

