


import argparse
import glob
import logging
import os
from argparse import Namespace
from importlib import import_module

import numpy as np
import torch
from lightning_base import BaseTransformer, add_generic_args, generic_train
from seqeval.metrics import accuracy_score, fa_score, precision_score, recall_score
from torch.nn import CrossEntropyLoss
from torch.utils.data import DataLoader, TensorDataset
from utils_ner import TokenClassificationTask


A	:		Tuple    =       logging.getLogger(__name__)



class 							A   (	UpperCAmelCase__							):
					'''simple docstring'''






					A__      =			'''token-classification'''

					def __init__(self   :						List[str]				, _UpperCAmelCase   :						Any		)						->       Tuple:




							"""simple docstring"""


							if type(_UpperCAmelCase		) == dict:
									lowercase__					      =		Namespace(**_UpperCAmelCase		)
							lowercase__					      =		import_module("""tasks"""		)
							try:
									lowercase__					      =		getattr(_UpperCAmelCase				, hparams.task_type		)
									lowercase__					      =		token_classification_task_clazz()
							except AttributeError:
									raise ValueError(
									    f'''Task {hparams.task_type} needs to be defined as a TokenClassificationTask subclass in {module}. '''
									    f'''Available tasks classes are: {TokenClassificationTask.__subclasses__()}'''		)
							lowercase__					      =		self.token_classification_task.get_labels(hparams.labels		)
							lowercase__					      =		CrossEntropyLoss().ignore_index
							super().__init__(_UpperCAmelCase				, len(self.labels		)				, self.mode		)

					def   lowerCamelCase__					(self   :						Dict				, **_UpperCAmelCase   :						int		)						->       Optional[Any]:




							"""simple docstring"""


							return self.model(**_UpperCAmelCase		)

					def   lowerCamelCase__					(self   :						Any				, _UpperCAmelCase   :						str				, _UpperCAmelCase   :						Dict		)						->       List[Any]:




							"""simple docstring"""


							lowercase__					      =		{"""input_ids""": batch[0], """attention_mask""": batch[1], """labels""": batch[3]}
							if self.config.model_type != "distilbert":
									lowercase__					      =		(
									    batch[2] if self.config.model_type in ["""bert""", """xlnet"""] else None
									)  # XLM and RoBERTa don"t use token_type_ids

							lowercase__					      =		self(**_UpperCAmelCase		)
							lowercase__					      =		outputs[0]
							# tensorboard_logs = {"loss": loss, "rate": self.lr_scheduler.get_last_lr()[-1]}
							return {"loss": loss}

					def   lowerCamelCase__					(self   :						Tuple		)						->       Dict:




							"""simple docstring"""


							lowercase__					      =		self.hparams
							for mode in ["train", "dev", "test"]:
									lowercase__					      =		self._feature_file(_UpperCAmelCase		)
									if os.path.exists(_UpperCAmelCase		) and not args.overwrite_cache:
											logger.info("""Loading features from cached file %s"""				, _UpperCAmelCase		)
											lowercase__					      =		torch.load(_UpperCAmelCase		)
									else:
											logger.info("""Creating features from dataset file at %s"""				, args.data_dir		)
											lowercase__					      =		self.token_classification_task.read_examples_from_file(args.data_dir				, _UpperCAmelCase		)
											lowercase__					      =		self.token_classification_task.convert_examples_to_features(
											    _UpperCAmelCase				, self.labels				, args.max_seq_length				, self.tokenizer				, cls_token_at_end=bool(self.config.model_type in ["""xlnet"""]		)				, cls_token=self.tokenizer.cls_token				, cls_token_segment_id=2 if self.config.model_type in ["""xlnet"""] else 0				, sep_token=self.tokenizer.sep_token				, sep_token_extra=_UpperCAmelCase				, pad_on_left=bool(self.config.model_type in ["""xlnet"""]		)				, pad_token=self.tokenizer.pad_token_id				, pad_token_segment_id=self.tokenizer.pad_token_type_id				, pad_token_label_id=self.pad_token_label_id				, )
											logger.info("""Saving features into cached file %s"""				, _UpperCAmelCase		)
											torch.save(_UpperCAmelCase				, _UpperCAmelCase		)

					def   lowerCamelCase__					(self   :						List[str]				, _UpperCAmelCase   :						int				, _UpperCAmelCase   :						int				, _UpperCAmelCase   :						bool = False		)						->       DataLoader:




							"""simple docstring"""


							lowercase__					      =		self._feature_file(_UpperCAmelCase		)
							logger.info("""Loading features from cached file %s"""				, _UpperCAmelCase		)
							lowercase__					      =		torch.load(_UpperCAmelCase		)
							lowercase__					      =		torch.tensor([f.input_ids for f in features]				, dtype=torch.long		)
							lowercase__					      =		torch.tensor([f.attention_mask for f in features]				, dtype=torch.long		)
							if features[0].token_type_ids is not None:
									lowercase__					      =		torch.tensor([f.token_type_ids for f in features]				, dtype=torch.long		)
							else:
									lowercase__					      =		torch.tensor([0 for f in features]				, dtype=torch.long		)
									# HACK(we will not use this anymore soon)
							lowercase__					      =		torch.tensor([f.label_ids for f in features]				, dtype=torch.long		)
							return DataLoader(
							    TensorDataset(_UpperCAmelCase				, _UpperCAmelCase				, _UpperCAmelCase				, _UpperCAmelCase		)				, batch_size=_UpperCAmelCase		)

					def   lowerCamelCase__					(self   :						Tuple				, _UpperCAmelCase   :						str				, _UpperCAmelCase   :						Any		)						->       Any:




							"""simple docstring"""


							"""Compute validation""" ""
							lowercase__					      =		{"""input_ids""": batch[0], """attention_mask""": batch[1], """labels""": batch[3]}
							if self.config.model_type != "distilbert":
									lowercase__					      =		(
									    batch[2] if self.config.model_type in ["""bert""", """xlnet"""] else None
									)  # XLM and RoBERTa don"t use token_type_ids
							lowercase__					      =		self(**_UpperCAmelCase		)
							lowercase__      ,     lowercase__					      =		outputs[:2]
							lowercase__					      =		logits.detach().cpu().numpy()
							lowercase__					      =		inputs["""labels"""].detach().cpu().numpy()
							return {"val_loss": tmp_eval_loss.detach().cpu(), "pred": preds, "target": out_label_ids}

					def   lowerCamelCase__					(self   :						Any				, _UpperCAmelCase   :						Any		)						->       int:




							"""simple docstring"""


							lowercase__					      =		torch.stack([x["""val_loss"""] for x in outputs]		).mean()
							lowercase__					      =		np.concatenate([x["""pred"""] for x in outputs]				, axis=0		)
							lowercase__					      =		np.argmax(_UpperCAmelCase				, axis=2		)
							lowercase__					      =		np.concatenate([x["""target"""] for x in outputs]				, axis=0		)

							lowercase__					      =		dict(enumerate(self.labels		)		)
							lowercase__					      =		[[] for _ in range(out_label_ids.shape[0]		)]
							lowercase__					      =		[[] for _ in range(out_label_ids.shape[0]		)]

							for i in range(out_label_ids.shape[0]		):
									for j in range(out_label_ids.shape[1]		):
											if out_label_ids[i, j] != self.pad_token_label_id:
													out_label_list[i].append(label_map[out_label_ids[i][j]]		)
													preds_list[i].append(label_map[preds[i][j]]		)

							lowercase__					      =		{
							    """val_loss""": val_loss_mean,
							    """accuracy_score""": accuracy_score(_UpperCAmelCase				, _UpperCAmelCase		),
							    """precision""": precision_score(_UpperCAmelCase				, _UpperCAmelCase		),
							    """recall""": recall_score(_UpperCAmelCase				, _UpperCAmelCase		),
							    """f1""": fa_score(_UpperCAmelCase				, _UpperCAmelCase		),
							}

							lowercase__					      =		dict(results.items()		)
							lowercase__					      =		results
							return ret, preds_list, out_label_list

					def   lowerCamelCase__					(self   :						str				, _UpperCAmelCase   :						List[str]		)						->       Any:




							"""simple docstring"""


							lowercase__      ,     lowercase__      ,     lowercase__					      =		self._eval_end(_UpperCAmelCase		)
							lowercase__					      =		ret["""log"""]
							return {"val_loss": logs["val_loss"], "log": logs, "progress_bar": logs}

					def   lowerCamelCase__					(self   :						Optional[Any]				, _UpperCAmelCase   :						Optional[Any]		)						->       Any:




							"""simple docstring"""


							lowercase__      ,     lowercase__      ,     lowercase__					      =		self._eval_end(_UpperCAmelCase		)

							# Converting to the dict required by pl
							# https://github.com/PyTorchLightning/pytorch-lightning/blob/master/\
							# pytorch_lightning/trainer/logging.py#L139
							lowercase__					      =		ret["""log"""]
							# `val_loss` is the key returned by `self._eval_end()` but actually refers to `test_loss`
							return {"avg_test_loss": logs["val_loss"], "log": logs, "progress_bar": logs}


					@staticmethod
					def   lowerCamelCase__					(_UpperCAmelCase   :						Any				, _UpperCAmelCase   :						Tuple		)						->       Tuple:




							"""simple docstring"""


							BaseTransformer.add_model_specific_args(_UpperCAmelCase				, _UpperCAmelCase		)
							parser.add_argument(
							    """--task_type"""				, default="""NER"""				, type=_UpperCAmelCase				, help="""Task type to fine tune in training (e.g. NER, POS, etc)"""		)
							parser.add_argument(
							    """--max_seq_length"""				, default=128				, type=_UpperCAmelCase				, help=(
							        """The maximum total input sequence length after tokenization. Sequences longer """
							        """than this will be truncated, sequences shorter will be padded."""
							    )				, )

							parser.add_argument(
							    """--labels"""				, default=""""""				, type=_UpperCAmelCase				, help="""Path to a file containing all labels. If not specified, CoNLL-2003 labels are used."""				, )
							parser.add_argument(
							    """--gpus"""				, default=0				, type=_UpperCAmelCase				, help="""The number of GPUs allocated for this, it is by default 0 meaning none"""				, )

							parser.add_argument(
							    """--overwrite_cache"""				, action="""store_true"""				, help="""Overwrite the cached training and evaluation sets"""		)

							return parser


if __name__ == "__main__":
	A	:		List[str]    =       argparse.ArgumentParser()
	add_generic_args(parser, os.getcwd())
	A	:		str    =       NERTransformer.add_model_specific_args(parser, os.getcwd())
	A	:		str    =       parser.parse_args()
	A	:		Union[str, Any]    =       NERTransformer(args)
	A	:		Optional[Any]    =       generic_train(model, args)

	if args.do_predict:
		# See https://github.com/huggingface/transformers/issues/3159
		# pl use this default format to create a checkpoint:
		# https://github.com/PyTorchLightning/pytorch-lightning/blob/master\
		# /pytorch_lightning/callbacks/model_checkpoint.py#L322
		A	:		Dict    =       sorted(glob.glob(os.path.join(args.output_dir, 'checkpoint-epoch=*.ckpt'), recursive=True))
		A	:		Union[str, Any]    =       model.load_from_checkpoint(checkpoints[-1])
		trainer.test(model)





import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


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



class 							A   (	UpperCAmelCase__							):
					'''simple docstring'''






					def __init__(self   :						List[str]				, _UpperCAmelCase   :						AutoencoderKL				, _UpperCAmelCase   :						CLIPTextModel				, _UpperCAmelCase   :						CLIPTokenizer				, _UpperCAmelCase   :						UNetaDConditionModel				, _UpperCAmelCase   :						Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]				, _UpperCAmelCase   :						StableDiffusionSafetyChecker				, _UpperCAmelCase   :						CLIPImageProcessor				, )						->       Dict:




							"""simple docstring"""


							super().__init__()
							self.register_modules(
							    vae=_UpperCAmelCase				, text_encoder=_UpperCAmelCase				, tokenizer=_UpperCAmelCase				, unet=_UpperCAmelCase				, scheduler=_UpperCAmelCase				, safety_checker=_UpperCAmelCase				, feature_extractor=_UpperCAmelCase				, )

					def   lowerCamelCase__					(self   :						Optional[Any]				, _UpperCAmelCase   :						Optional[Union[str, int]] = "auto"		)						->       Tuple:




							"""simple docstring"""


							if slice_size == "auto":
									# half the attention head size is usually a good trade-off between
									# speed and memory
									lowercase__					      =		self.unet.config.attention_head_dim // 2
							self.unet.set_attention_slice(_UpperCAmelCase		)

					def   lowerCamelCase__					(self   :						Optional[Any]		)						->       Optional[Any]:




							"""simple docstring"""


							self.enable_attention_slicing(_UpperCAmelCase		)


					@torch.no_grad()
					def __call__(self   :						Any				, _UpperCAmelCase   :						Union[str, List[str]]				, _UpperCAmelCase   :						int = 512				, _UpperCAmelCase   :						int = 512				, _UpperCAmelCase   :						int = 50				, _UpperCAmelCase   :						float = 7.5				, _UpperCAmelCase   :						Optional[Union[str, List[str]]] = None				, _UpperCAmelCase   :						Optional[int] = 1				, _UpperCAmelCase   :						float = 0.0				, _UpperCAmelCase   :						Optional[torch.Generator] = None				, _UpperCAmelCase   :						Optional[torch.FloatTensor] = None				, _UpperCAmelCase   :						Optional[str] = "pil"				, _UpperCAmelCase   :						bool = True				, _UpperCAmelCase   :						Optional[Callable[[int, int, torch.FloatTensor], None]] = None				, _UpperCAmelCase   :						int = 1				, _UpperCAmelCase   :						Optional[torch.FloatTensor] = None				, **_UpperCAmelCase   :						Any				, )						->       Tuple:




							"""simple docstring"""


							if isinstance(_UpperCAmelCase				, _UpperCAmelCase		):
									lowercase__					      =		1
							elif isinstance(_UpperCAmelCase				, _UpperCAmelCase		):
									lowercase__					      =		len(_UpperCAmelCase		)
							else:
									raise ValueError(f'''`prompt` has to be of type `str` or `list` but is {type(_UpperCAmelCase		)}'''		)

							if height % 8 != 0 or width % 8 != 0:
									raise ValueError(f'''`height` and `width` have to be divisible by 8 but are {height} and {width}.'''		)

							if (callback_steps is None) or (
							    callback_steps is not None and (not isinstance(_UpperCAmelCase				, _UpperCAmelCase		) or callback_steps <= 0)
							):
									raise ValueError(
									    f'''`callback_steps` has to be a positive integer but is {callback_steps} of type'''
									    f''' {type(_UpperCAmelCase		)}.'''		)

							# get prompt text embeddings
							lowercase__					      =		self.tokenizer(
							    _UpperCAmelCase				, padding="""max_length"""				, max_length=self.tokenizer.model_max_length				, return_tensors="""pt"""				, )
							lowercase__					      =		text_inputs.input_ids

							if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
									lowercase__					      =		self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]		)
									logger.warning(
									    """The following part of your input was truncated because CLIP can only handle sequences up to"""
									    f''' {self.tokenizer.model_max_length} tokens: {removed_text}'''		)
									lowercase__					      =		text_input_ids[:, : self.tokenizer.model_max_length]

							if text_embeddings is None:
									lowercase__					      =		self.text_encoder(text_input_ids.to(self.device		)		)[0]

							# duplicate text embeddings for each generation per prompt, using mps friendly method
							lowercase__      ,     lowercase__      ,     lowercase__					      =		text_embeddings.shape
							lowercase__					      =		text_embeddings.repeat(1				, _UpperCAmelCase				, 1		)
							lowercase__					      =		text_embeddings.view(bs_embed * num_images_per_prompt				, _UpperCAmelCase				, -1		)

							# here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
							# of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
							# corresponds to doing no classifier free guidance.
							lowercase__					      =		guidance_scale > 1.0
							# get unconditional embeddings for classifier free guidance
							if do_classifier_free_guidance:
									lowercase__					      =		42
									if negative_prompt is None:
											lowercase__					      =		[""""""]
									elif type(_UpperCAmelCase		) is not type(_UpperCAmelCase		):
											raise TypeError(
											    f'''`negative_prompt` should be the same type to `prompt`, but got {type(_UpperCAmelCase		)} !='''
											    f''' {type(_UpperCAmelCase		)}.'''		)
									elif isinstance(_UpperCAmelCase				, _UpperCAmelCase		):
											lowercase__					      =		[negative_prompt]
									elif batch_size != len(_UpperCAmelCase		):
											raise ValueError(
											    f'''`negative_prompt`: {negative_prompt} has batch size {len(_UpperCAmelCase		)}, but `prompt`:'''
											    f''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
											    """ the batch size of `prompt`."""		)
									else:
											lowercase__					      =		negative_prompt

									lowercase__					      =		text_input_ids.shape[-1]
									lowercase__					      =		self.tokenizer(
									    _UpperCAmelCase				, padding="""max_length"""				, max_length=_UpperCAmelCase				, truncation=_UpperCAmelCase				, return_tensors="""pt"""				, )
									lowercase__					      =		self.text_encoder(uncond_input.input_ids.to(self.device		)		)[0]

									# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
									lowercase__					      =		uncond_embeddings.shape[1]
									lowercase__					      =		uncond_embeddings.repeat(_UpperCAmelCase				, _UpperCAmelCase				, 1		)
									lowercase__					      =		uncond_embeddings.view(batch_size * num_images_per_prompt				, _UpperCAmelCase				, -1		)

									# For classifier free guidance, we need to do two forward passes.
									# Here we concatenate the unconditional and text embeddings into a single batch
									# to avoid doing two forward passes
									lowercase__					      =		torch.cat([uncond_embeddings, text_embeddings]		)

							# get the initial random noise unless the user supplied it

							# Unlike in other pipelines, latents need to be generated in the target device
							# for 1-to-1 results reproducibility with the CompVis implementation.
							# However this currently doesn't work in `mps`.
							lowercase__					      =		(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
							lowercase__					      =		(batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)
							lowercase__					      =		text_embeddings.dtype
							if latents is None:
									if self.device.type == "mps":
											# randn does not exist on mps
											lowercase__					      =		torch.randn(
											    _UpperCAmelCase				, generator=_UpperCAmelCase				, device="""cpu"""				, dtype=_UpperCAmelCase		).to(self.device		)
											lowercase__					      =		torch.randn(_UpperCAmelCase				, generator=_UpperCAmelCase				, device="""cpu"""				, dtype=_UpperCAmelCase		).to(
											    self.device		)
									else:
											lowercase__					      =		torch.randn(
											    _UpperCAmelCase				, generator=_UpperCAmelCase				, device=self.device				, dtype=_UpperCAmelCase		)
											lowercase__					      =		torch.randn(_UpperCAmelCase				, generator=_UpperCAmelCase				, device=self.device				, dtype=_UpperCAmelCase		)
							else:
									if latents_reference.shape != latents_shape:
											raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}'''		)
									lowercase__					      =		latents_reference.to(self.device		)
									lowercase__					      =		latents.to(self.device		)

							# This is the key part of the pipeline where we
							# try to ensure that the generated images w/ the same seed
							# but different sizes actually result in similar images
							lowercase__					      =		(latents_shape[3] - latents_shape_reference[3]) // 2
							lowercase__					      =		(latents_shape[2] - latents_shape_reference[2]) // 2
							lowercase__					      =		latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
							lowercase__					      =		latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
							lowercase__					      =		0 if dx < 0 else dx
							lowercase__					      =		0 if dy < 0 else dy
							lowercase__					      =		max(-dx				, 0		)
							lowercase__					      =		max(-dy				, 0		)
							# import pdb
							# pdb.set_trace()
							lowercase__					      =		latents_reference[:, :, dy : dy + h, dx : dx + w]

							# set timesteps
							self.scheduler.set_timesteps(_UpperCAmelCase		)

							# Some schedulers like PNDM have timesteps as arrays
							# It's more optimized to move all timesteps to correct device beforehand
							lowercase__					      =		self.scheduler.timesteps.to(self.device		)

							# scale the initial noise by the standard deviation required by the scheduler
							lowercase__					      =		latents * self.scheduler.init_noise_sigma

							# prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
							# eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
							# eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
							# and should be between [0, 1]
							lowercase__					      =		"""eta""" in set(inspect.signature(self.scheduler.step		).parameters.keys()		)
							lowercase__					      =		{}
							if accepts_eta:
									lowercase__					      =		eta

							for i, t in enumerate(self.progress_bar(_UpperCAmelCase		)		):
									# expand the latents if we are doing classifier free guidance
									lowercase__					      =		torch.cat([latents] * 2		) if do_classifier_free_guidance else latents
									lowercase__					      =		self.scheduler.scale_model_input(_UpperCAmelCase				, _UpperCAmelCase		)

									# predict the noise residual
									lowercase__					      =		self.unet(_UpperCAmelCase				, _UpperCAmelCase				, encoder_hidden_states=_UpperCAmelCase		).sample

									# perform guidance
									if do_classifier_free_guidance:
											lowercase__      ,     lowercase__					      =		noise_pred.chunk(2		)
											lowercase__					      =		noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

									# compute the previous noisy sample x_t -> x_t-1
									lowercase__					      =		self.scheduler.step(_UpperCAmelCase				, _UpperCAmelCase				, _UpperCAmelCase				, **_UpperCAmelCase		).prev_sample

									# call the callback, if provided
									if callback is not None and i % callback_steps == 0:
											callback(_UpperCAmelCase				, _UpperCAmelCase				, _UpperCAmelCase		)

							lowercase__					      =		1 / 0.18_215 * latents
							lowercase__					      =		self.vae.decode(_UpperCAmelCase		).sample

							lowercase__					      =		(image / 2 + 0.5).clamp(0				, 1		)

							# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
							lowercase__					      =		image.cpu().permute(0				, 2				, 3				, 1		).float().numpy()

							if self.safety_checker is not None:
									lowercase__					      =		self.feature_extractor(self.numpy_to_pil(_UpperCAmelCase		)				, return_tensors="""pt"""		).to(
									    self.device		)
									lowercase__      ,     lowercase__					      =		self.safety_checker(
									    images=_UpperCAmelCase				, clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype		)		)
							else:
									lowercase__					      =		None

							if output_type == "pil":
									lowercase__					      =		self.numpy_to_pil(_UpperCAmelCase		)

							if not return_dict:
									return (image, has_nsfw_concept)

							return StableDiffusionPipelineOutput(images=_UpperCAmelCase				, nsfw_content_detected=_UpperCAmelCase		)

