





'''simple docstring'''






import json
import os
import shutil
import tempfile
import unittest
from multiprocessing import get_context
from pathlib import Path

import datasets
import numpy as np
from datasets import load_dataset
from parameterized import parameterized

from transformers import AutoProcessor
from transformers.models.wavaveca import WavaVecaCTCTokenizer, WavaVecaFeatureExtractor
from transformers.models.wavaveca.tokenization_wavaveca import VOCAB_FILES_NAMES
from transformers.testing_utils import require_pyctcdecode, require_torch, require_torchaudio, slow
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_pyctcdecode_available, is_torch_available

from ..wavaveca.test_feature_extraction_wavaveca import floats_list


if is_pyctcdecode_available():
      from huggingface_hub import snapshot_download
      from pyctcdecode import BeamSearchDecoderCTC

      from transformers.models.wavaveca_with_lm import WavaVecaProcessorWithLM
      from transformers.models.wavaveca_with_lm.processing_wavaveca_with_lm import WavaVecaDecoderWithLMOutput

if is_torch_available():
      from transformers import WavaVecaForCTC




@require_pyctcdecode
class  UpperCAmelCase     (			unittest.TestCase   ):




      def 				lowercase__    (			self			: str  )				->      Union[str, Any]:
            _lowerCAmelCase							     =							'''| <pad> <unk> <s> </s> a b c d e f g h i j k'''.split()
            _lowerCAmelCase							     =							dict(zip(lowercase_			,			range(len(lowercase_  )  )  )  )

            _lowerCAmelCase							     =							{
                '''unk_token''': '''<unk>''',
                '''bos_token''': '''<s>''',
                '''eos_token''': '''</s>''',
            }
            _lowerCAmelCase							     =							{
                '''feature_size''': 1,
                '''padding_value''': 0.0,
                '''sampling_rate''': 1_60_00,
                '''return_attention_mask''': False,
                '''do_normalize''': True,
            }

            _lowerCAmelCase							     =							tempfile.mkdtemp()
            _lowerCAmelCase							     =							os.path.join(self.tmpdirname			,			VOCAB_FILES_NAMES["""vocab_file"""]  )
            _lowerCAmelCase							     =							os.path.join(self.tmpdirname			,			lowercase_  )
            with open(self.vocab_file			,			"""w"""			,			encoding="""utf-8"""  ) as fp:
                  fp.write(json.dumps(lowercase_  ) + """\n"""  )

            with open(self.feature_extraction_file			,			"""w"""			,			encoding="""utf-8"""  ) as fp:
                  fp.write(json.dumps(lowercase_  ) + """\n"""  )

            # load decoder from hub
            _lowerCAmelCase							     =							'''hf-internal-testing/ngram-beam-search-decoder'''




      def 				lowercase__    (			self			: Any			,			**__snake_case			: Optional[Any]  )				->      Tuple:
            _lowerCAmelCase							     =							self.add_kwargs_tokens_map.copy()
            kwargs.update(lowercase_  )
            return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname			,			**lowercase_  )




      def 				lowercase__    (			self			: str			,			**__snake_case			: Dict  )				->      Union[str, Any]:
            return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname			,			**lowercase_  )




      def 				lowercase__    (			self			: Union[str, Any]			,			**__snake_case			: int  )				->      int:
            return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name			,			**lowercase_  )




      def 				lowercase__    (			self			: Tuple  )				->      Dict:
            shutil.rmtree(self.tmpdirname  )




      def 				lowercase__    (			self			: Optional[int]  )				->      List[str]:
            _lowerCAmelCase							     =							self.get_tokenizer()
            _lowerCAmelCase							     =							self.get_feature_extractor()
            _lowerCAmelCase							     =							self.get_decoder()

            _lowerCAmelCase							     =							WavaVecaProcessorWithLM(tokenizer=lowercase_			,			feature_extractor=lowercase_			,			decoder=lowercase_  )

            processor.save_pretrained(self.tmpdirname  )
            _lowerCAmelCase							     =							WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname  )

            # tokenizer
            self.assertEqual(processor.tokenizer.get_vocab()			,			tokenizer.get_vocab()  )
            self.assertIsInstance(processor.tokenizer			,			lowercase_  )

            # feature extractor
            self.assertEqual(processor.feature_extractor.to_json_string()			,			feature_extractor.to_json_string()  )
            self.assertIsInstance(processor.feature_extractor			,			lowercase_  )

            # decoder
            self.assertEqual(processor.decoder._alphabet.labels			,			decoder._alphabet.labels  )
            self.assertEqual(
                processor.decoder.model_container[decoder._model_key]._unigram_set			,			decoder.model_container[decoder._model_key]._unigram_set			,			)
            self.assertIsInstance(processor.decoder			,			lowercase_  )




      def 				lowercase__    (			self			: int  )				->      Dict:
            _lowerCAmelCase							     =							WavaVecaProcessorWithLM(
                tokenizer=self.get_tokenizer()			,			feature_extractor=self.get_feature_extractor()			,			decoder=self.get_decoder()  )
            processor.save_pretrained(self.tmpdirname  )

            # make sure that error is thrown when decoder alphabet doesn't match
            _lowerCAmelCase							     =							WavaVecaProcessorWithLM.from_pretrained(
                self.tmpdirname			,			alpha=5.0			,			beta=3.0			,			score_boundary=-7.0			,			unk_score_offset=3  )

            # decoder
            self.assertEqual(processor.language_model.alpha			,			5.0  )
            self.assertEqual(processor.language_model.beta			,			3.0  )
            self.assertEqual(processor.language_model.score_boundary			,			-7.0  )
            self.assertEqual(processor.language_model.unk_score_offset			,			3  )




      def 				lowercase__    (			self			: Dict  )				->      Any:
            _lowerCAmelCase							     =							self.get_tokenizer()
            # add token to trigger raise
            tokenizer.add_tokens(["""xx"""]  )
            with self.assertRaisesRegex(lowercase_			,			"""include"""  ):
                  WavaVecaProcessorWithLM(
                      tokenizer=lowercase_			,			feature_extractor=self.get_feature_extractor()			,			decoder=self.get_decoder()  )




      def 				lowercase__    (			self			: int  )				->      Union[str, Any]:
            _lowerCAmelCase							     =							self.get_feature_extractor()
            _lowerCAmelCase							     =							self.get_tokenizer()
            _lowerCAmelCase							     =							self.get_decoder()

            _lowerCAmelCase							     =							WavaVecaProcessorWithLM(tokenizer=lowercase_			,			feature_extractor=lowercase_			,			decoder=lowercase_  )

            _lowerCAmelCase							     =							floats_list((3, 10_00)  )

            _lowerCAmelCase							     =							feature_extractor(lowercase_			,			return_tensors="""np"""  )
            _lowerCAmelCase							     =							processor(lowercase_			,			return_tensors="""np"""  )

            for key in input_feat_extract.keys():
                  self.assertAlmostEqual(input_feat_extract[key].sum()			,			input_processor[key].sum()			,			delta=1E-2  )




      def 				lowercase__    (			self			: Optional[int]  )				->      str:
            _lowerCAmelCase							     =							self.get_feature_extractor()
            _lowerCAmelCase							     =							self.get_tokenizer()
            _lowerCAmelCase							     =							self.get_decoder()

            _lowerCAmelCase							     =							WavaVecaProcessorWithLM(tokenizer=lowercase_			,			feature_extractor=lowercase_			,			decoder=lowercase_  )

            _lowerCAmelCase							     =							'''This is a test string'''

            _lowerCAmelCase							     =							processor(text=lowercase_  )

            _lowerCAmelCase							     =							tokenizer(lowercase_  )

            for key in encoded_tok.keys():
                  self.assertListEqual(encoded_tok[key]			,			encoded_processor[key]  )




      def 				lowercase__    (			self			: List[str]			,			__snake_case			: Any=(2, 10, 16)			,			__snake_case			: str=77  )				->      List[str]:
            np.random.seed(lowercase_  )
            return np.random.rand(*lowercase_  )




      def 				lowercase__    (			self			: Optional[Any]  )				->      Dict:
            _lowerCAmelCase							     =							self.get_feature_extractor()
            _lowerCAmelCase							     =							self.get_tokenizer()
            _lowerCAmelCase							     =							self.get_decoder()

            _lowerCAmelCase							     =							WavaVecaProcessorWithLM(tokenizer=lowercase_			,			feature_extractor=lowercase_			,			decoder=lowercase_  )

            _lowerCAmelCase							     =							self._get_dummy_logits(shape=(10, 16)			,			seed=13  )

            _lowerCAmelCase							     =							processor.decode(lowercase_  )

            _lowerCAmelCase							     =							decoder.decode_beams(lowercase_  )[0]

            self.assertEqual(decoded_decoder[0]			,			decoded_processor.text  )
            self.assertEqual("""</s> <s> </s>"""			,			decoded_processor.text  )
            self.assertEqual(decoded_decoder[-2]			,			decoded_processor.logit_score  )
            self.assertEqual(decoded_decoder[-1]			,			decoded_processor.lm_score  )




      @parameterized.expand([[None], ["""fork"""], ["""spawn"""]]  )
      def 				lowercase__    (			self			: Optional[Any]			,			__snake_case			: Union[str, Any]  )				->      Any:
            _lowerCAmelCase							     =							self.get_feature_extractor()
            _lowerCAmelCase							     =							self.get_tokenizer()
            _lowerCAmelCase							     =							self.get_decoder()

            _lowerCAmelCase							     =							WavaVecaProcessorWithLM(tokenizer=lowercase_			,			feature_extractor=lowercase_			,			decoder=lowercase_  )

            _lowerCAmelCase							     =							self._get_dummy_logits()

            # note: pool should be instantiated *after* Wav2Vec2ProcessorWithLM.
            #       otherwise, the LM won't be available to the pool's sub-processes.
            # manual logic used to allow parameterized test for both pool=None and pool=Pool(...)
            if pool_context is None:
                  _lowerCAmelCase							     =							processor.batch_decode(lowercase_  )
            else:
                  with get_context(lowercase_  ).Pool() as pool:
                        _lowerCAmelCase							     =							processor.batch_decode(lowercase_			,			lowercase_  )

            _lowerCAmelCase							     =							list(lowercase_  )

            with get_context("""fork"""  ).Pool() as p:
                  _lowerCAmelCase							     =							decoder.decode_beams_batch(lowercase_			,			lowercase_  )

            _lowerCAmelCase							     =							[], [], []
            for beams in decoded_beams:
                  texts_decoder.append(beams[0][0]  )
                  logit_scores_decoder.append(beams[0][-2]  )
                  lm_scores_decoder.append(beams[0][-1]  )

            self.assertListEqual(lowercase_			,			decoded_processor.text  )
            self.assertListEqual(["""<s> <s> </s>""", """<s> <s> <s>"""]			,			decoded_processor.text  )
            self.assertListEqual(lowercase_			,			decoded_processor.logit_score  )
            self.assertListEqual(lowercase_			,			decoded_processor.lm_score  )




      def 				lowercase__    (			self			: List[str]  )				->      List[str]:
            _lowerCAmelCase							     =							self.get_feature_extractor()
            _lowerCAmelCase							     =							self.get_tokenizer()
            _lowerCAmelCase							     =							self.get_decoder()

            _lowerCAmelCase							     =							WavaVecaProcessorWithLM(tokenizer=lowercase_			,			feature_extractor=lowercase_			,			decoder=lowercase_  )

            _lowerCAmelCase							     =							self._get_dummy_logits()

            _lowerCAmelCase							     =							15
            _lowerCAmelCase							     =							-20.0
            _lowerCAmelCase							     =							-4.0

            _lowerCAmelCase							     =							processor.batch_decode(
                lowercase_			,			beam_width=lowercase_			,			beam_prune_logp=lowercase_			,			token_min_logp=lowercase_			,			)
            _lowerCAmelCase							     =							decoded_processor_out.text

            _lowerCAmelCase							     =							list(lowercase_  )

            with get_context("""fork"""  ).Pool() as pool:
                  _lowerCAmelCase							     =							decoder.decode_beams_batch(
                      lowercase_			,			lowercase_			,			beam_width=lowercase_			,			beam_prune_logp=lowercase_			,			token_min_logp=lowercase_			,			)

            _lowerCAmelCase							     =							[d[0][0] for d in decoded_decoder_out]
            _lowerCAmelCase							     =							[d[0][2] for d in decoded_decoder_out]
            _lowerCAmelCase							     =							[d[0][3] for d in decoded_decoder_out]

            self.assertListEqual(lowercase_			,			lowercase_  )
            self.assertListEqual(["""</s> <s> <s>""", """<s> <s> <s>"""]			,			lowercase_  )

            self.assertTrue(np.array_equal(lowercase_			,			decoded_processor_out.logit_score  )  )
            self.assertTrue(np.allclose([-20.0_54, -18.4_47]			,			lowercase_			,			atol=1E-3  )  )

            self.assertTrue(np.array_equal(lowercase_			,			decoded_processor_out.lm_score  )  )
            self.assertTrue(np.allclose([-15.5_54, -13.94_74]			,			lowercase_			,			atol=1E-3  )  )




      def 				lowercase__    (			self			: List[Any]  )				->      List[str]:
            _lowerCAmelCase							     =							self.get_feature_extractor()
            _lowerCAmelCase							     =							self.get_tokenizer()
            _lowerCAmelCase							     =							self.get_decoder()

            _lowerCAmelCase							     =							WavaVecaProcessorWithLM(tokenizer=lowercase_			,			feature_extractor=lowercase_			,			decoder=lowercase_  )

            _lowerCAmelCase							     =							self._get_dummy_logits()

            _lowerCAmelCase							     =							2.0
            _lowerCAmelCase							     =							5.0
            _lowerCAmelCase							     =							-20.0
            _lowerCAmelCase							     =							True

            _lowerCAmelCase							     =							processor.batch_decode(
                lowercase_			,			alpha=lowercase_			,			beta=lowercase_			,			unk_score_offset=lowercase_			,			lm_score_boundary=lowercase_			,			)
            _lowerCAmelCase							     =							decoded_processor_out.text

            _lowerCAmelCase							     =							list(lowercase_  )
            decoder.reset_params(
                alpha=lowercase_			,			beta=lowercase_			,			unk_score_offset=lowercase_			,			lm_score_boundary=lowercase_			,			)

            with get_context("""fork"""  ).Pool() as pool:
                  _lowerCAmelCase							     =							decoder.decode_beams_batch(
                      lowercase_			,			lowercase_			,			)

            _lowerCAmelCase							     =							[d[0][0] for d in decoded_decoder_out]

            self.assertListEqual(lowercase_			,			lowercase_  )
            self.assertListEqual(["""<s> </s> <s> </s> </s>""", """</s> </s> <s> </s> </s>"""]			,			lowercase_  )
            _lowerCAmelCase							     =							processor.decoder.model_container[processor.decoder._model_key]
            self.assertEqual(lm_model.alpha			,			2.0  )
            self.assertEqual(lm_model.beta			,			5.0  )
            self.assertEqual(lm_model.unk_score_offset			,			-20.0  )
            self.assertEqual(lm_model.score_boundary			,			lowercase_  )




      def 				lowercase__    (			self			: Dict  )				->      int:
            _lowerCAmelCase							     =							WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""  )

            _lowerCAmelCase							     =							processor.decoder.model_container[processor.decoder._model_key]
            _lowerCAmelCase							     =							Path(language_model._kenlm_model.path.decode("""utf-8"""  )  ).parent.parent.absolute()

            _lowerCAmelCase							     =							os.listdir(lowercase_  )
            _lowerCAmelCase							     =							['''alphabet.json''', '''language_model''']

            downloaded_decoder_files.sort()
            expected_decoder_files.sort()

            # test that only decoder relevant files from
            # https://huggingface.co/hf-internal-testing/processor_with_lm/tree/main
            # are downloaded and none of the rest (e.g. README.md, ...)
            self.assertListEqual(lowercase_			,			lowercase_  )




      def 				lowercase__    (			self			: Union[str, Any]  )				->      Dict:
            _lowerCAmelCase							     =							snapshot_download("""hf-internal-testing/processor_with_lm"""  )

            _lowerCAmelCase							     =							WavaVecaProcessorWithLM.from_pretrained(lowercase_  )

            _lowerCAmelCase							     =							processor.decoder.model_container[processor.decoder._model_key]
            _lowerCAmelCase							     =							Path(language_model._kenlm_model.path.decode("""utf-8"""  )  ).parent.parent.absolute()

            _lowerCAmelCase							     =							os.listdir(lowercase_  )
            _lowerCAmelCase							     =							os.listdir(lowercase_  )

            local_decoder_files.sort()
            expected_decoder_files.sort()

            # test that both decoder form hub and local files in cache are the same
            self.assertListEqual(lowercase_			,			lowercase_  )




      def 				lowercase__    (			self			: str  )				->      Any:
            _lowerCAmelCase							     =							WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""  )
            _lowerCAmelCase							     =							AutoProcessor.from_pretrained("""hf-internal-testing/processor_with_lm"""  )

            _lowerCAmelCase							     =							floats_list((3, 10_00)  )

            _lowerCAmelCase							     =							processor_wavaveca(lowercase_			,			return_tensors="""np"""  )
            _lowerCAmelCase							     =							processor_auto(lowercase_			,			return_tensors="""np"""  )

            for key in input_wavaveca.keys():
                  self.assertAlmostEqual(input_wavaveca[key].sum()			,			input_auto[key].sum()			,			delta=1E-2  )

            _lowerCAmelCase							     =							self._get_dummy_logits()

            _lowerCAmelCase							     =							processor_wavaveca.batch_decode(lowercase_  )
            _lowerCAmelCase							     =							processor_auto.batch_decode(lowercase_  )

            self.assertListEqual(decoded_wavaveca.text			,			decoded_auto.text  )




      def 				lowercase__    (			self			: Union[str, Any]  )				->      List[str]:
            _lowerCAmelCase							     =							self.get_feature_extractor()
            _lowerCAmelCase							     =							self.get_tokenizer()
            _lowerCAmelCase							     =							self.get_decoder()

            _lowerCAmelCase							     =							WavaVecaProcessorWithLM(tokenizer=lowercase_			,			feature_extractor=lowercase_			,			decoder=lowercase_  )

            self.assertListEqual(
                processor.model_input_names			,			feature_extractor.model_input_names			,			msg="""`processor` and `feature_extractor` model input names do not match"""			,			)




      @staticmethod
      def 				lowercase__    (			__snake_case			: Tuple			,			__snake_case			: str  )				->      int:
            _lowerCAmelCase							     =							[d[key] for d in offsets]
            return retrieved_list




      def 				lowercase__    (			self			: str  )				->      List[Any]:
            _lowerCAmelCase							     =							WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""  )
            _lowerCAmelCase							     =							self._get_dummy_logits()[0]

            _lowerCAmelCase							     =							processor.decode(lowercase_			,			output_word_offsets=lowercase_  )
            # check Wav2Vec2CTCTokenizerOutput keys for word
            self.assertEqual(len(outputs.keys()  )			,			4  )
            self.assertTrue("""text""" in outputs  )
            self.assertTrue("""word_offsets""" in outputs  )
            self.assertTrue(isinstance(lowercase_			,			lowercase_  )  )

            self.assertEqual(""" """.join(self.get_from_offsets(outputs["""word_offsets"""]			,			"""word"""  )  )			,			outputs.text  )
            self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]			,			"""word"""  )			,			["""<s>""", """<s>""", """</s>"""]  )
            self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]			,			"""start_offset"""  )			,			[0, 2, 4]  )
            self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]			,			"""end_offset"""  )			,			[1, 3, 5]  )




      def 				lowercase__    (			self			: Optional[int]  )				->      Any:
            _lowerCAmelCase							     =							WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""  )
            _lowerCAmelCase							     =							self._get_dummy_logits()

            _lowerCAmelCase							     =							processor.batch_decode(lowercase_			,			output_word_offsets=lowercase_  )

            # check Wav2Vec2CTCTokenizerOutput keys for word
            self.assertEqual(len(outputs.keys()  )			,			4  )
            self.assertTrue("""text""" in outputs  )
            self.assertTrue("""word_offsets""" in outputs  )
            self.assertTrue(isinstance(lowercase_			,			lowercase_  )  )

            self.assertListEqual(
                [""" """.join(self.get_from_offsets(lowercase_			,			"""word"""  )  ) for o in outputs["""word_offsets"""]]			,			outputs.text  )
            self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""][0]			,			"""word"""  )			,			["""<s>""", """<s>""", """</s>"""]  )
            self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""][0]			,			"""start_offset"""  )			,			[0, 2, 4]  )
            self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""][0]			,			"""end_offset"""  )			,			[1, 3, 5]  )




      @slow
      @require_torch
      @require_torchaudio
      def 				lowercase__    (			self			: int  )				->      Union[str, Any]:
            import torch

            _lowerCAmelCase							     =							load_dataset("""common_voice"""			,			"""en"""			,			split="""train"""			,			streaming=lowercase_  )
            _lowerCAmelCase							     =							ds.cast_column("""audio"""			,			datasets.Audio(sampling_rate=1_60_00  )  )
            _lowerCAmelCase							     =							iter(lowercase_  )
            _lowerCAmelCase							     =							next(lowercase_  )

            _lowerCAmelCase							     =							AutoProcessor.from_pretrained("""patrickvonplaten/wav2vec2-base-100h-with-lm"""  )
            _lowerCAmelCase							     =							WavaVecaForCTC.from_pretrained("""patrickvonplaten/wav2vec2-base-100h-with-lm"""  )

            # compare to filename `common_voice_en_100038.mp3` of dataset viewer on https://huggingface.co/datasets/common_voice/viewer/en/train
            _lowerCAmelCase							     =							processor(sample["""audio"""]["""array"""]			,			return_tensors="""pt"""  ).input_values

            with torch.no_grad():
                  _lowerCAmelCase							     =							model(lowercase_  ).logits.cpu().numpy()

            _lowerCAmelCase							     =							processor.decode(logits[0]			,			output_word_offsets=lowercase_  )

            _lowerCAmelCase							     =							model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
            _lowerCAmelCase							     =							[
                {
                    '''start_time''': d['''start_offset'''] * time_offset,
                    '''end_time''': d['''end_offset'''] * time_offset,
                    '''word''': d['''word'''],
                }
                for d in output['''word_offsets''']
            ]

            _lowerCAmelCase							     =							'''WHY DOES MILISANDRA LOOK LIKE SHE WANTS TO CONSUME JOHN SNOW ON THE RIVER AT THE WALL'''

            # output words
            self.assertEqual(""" """.join(self.get_from_offsets(lowercase_			,			"""word"""  )  )			,			lowercase_  )
            self.assertEqual(""" """.join(self.get_from_offsets(lowercase_			,			"""word"""  )  )			,			output.text  )

            # output times
            _lowerCAmelCase							     =							torch.tensor(self.get_from_offsets(lowercase_			,			"""start_time"""  )  )
            _lowerCAmelCase							     =							torch.tensor(self.get_from_offsets(lowercase_			,			"""end_time"""  )  )

            # fmt: off
            _lowerCAmelCase							     =							torch.tensor([1.41_99, 1.65_99, 2.25_99, 3.0, 3.24, 3.59_99, 3.79_99, 4.09_99, 4.26, 4.94, 5.28, 5.65_99, 5.78, 5.94, 6.32, 6.53_99, 6.65_99]  )
            _lowerCAmelCase							     =							torch.tensor([1.53_99, 1.89_99, 2.9, 3.16, 3.53_99, 3.72, 4.01_99, 4.17_99, 4.76, 5.15_99, 5.55_99, 5.69_99, 5.86, 6.19_99, 6.38, 6.61_99, 6.94]  )
            # fmt: on

            self.assertTrue(torch.allclose(lowercase_			,			lowercase_			,			atol=0.01  )  )
            self.assertTrue(torch.allclose(lowercase_			,			lowercase_			,			atol=0.01  )  )











"""simple docstring"""

import os
from datetime import datetime as dt

from github import Github


lowercase__    :  int							    =	[
    '''good first issue''',
    '''good second issue''',
    '''good difficult issue''',
    '''enhancement''',
    '''new pipeline/model''',
    '''new scheduler''',
    '''wip''',
]




def __lowercase			(							):
    snake_case_ :     Optional[Any]				=	Github(os.environ['''GITHUB_TOKEN''']    )
    snake_case_ :     Any				=	g.get_repo('''huggingface/diffusers'''    )
    snake_case_ :     Any				=	repo.get_issues(state='''open'''    )

    for issue in open_issues:
        snake_case_ :     str				=	sorted(issue.get_comments()		,			key=lambda _a    : i.created_at		,			reverse=_a    )
        snake_case_ :     Dict				=	comments[0] if len(_a    ) > 0 else None
        if (
            last_comment is not None
            and last_comment.user.login == "github-actions[bot]"
            and (dt.utcnow() - issue.updated_at).days > 7
            and (dt.utcnow() - issue.created_at).days >= 30
            and not any(label.name.lower() in LABELS_TO_EXEMPT for label in issue.get_labels()    )
        ):
            # Closes the issue after 7 days of inactivity since the Stalebot notification.
            issue.edit(state='''closed'''    )
        elif (
            "stale" in issue.get_labels()
            and last_comment is not None
            and last_comment.user.login != "github-actions[bot]"
        ):
            # Opens the issue if someone other than Stalebot commented.
            issue.edit(state='''open'''    )
            issue.remove_from_labels('''stale'''    )
        elif (
            (dt.utcnow() - issue.updated_at).days > 23
            and (dt.utcnow() - issue.created_at).days >= 30
            and not any(label.name.lower() in LABELS_TO_EXEMPT for label in issue.get_labels()    )
        ):
            # Post a Stalebot notification after 23 days of inactivity.
            issue.create_comment(
                '''This issue has been automatically marked as stale because it has not had '''
                '''recent activity. If you think this still needs to be addressed '''
                '''please comment on this thread.\n\nPlease note that issues that do not follow the '''
                '''[contributing guidelines](https://github.com/huggingface/diffusers/blob/main/CONTRIBUTING.md) '''
                '''are likely to be ignored.'''    )
            issue.add_to_labels('''stale'''    )


if __name__ == "__main__":
     main()





