






import os
import shutil
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
from datasets import Dataset

from transformers.models.realm.configuration_realm import RealmConfig
from transformers.models.realm.retrieval_realm import _REALM_BLOCK_RECORDS_FILENAME, RealmRetriever
from transformers.models.realm.tokenization_realm import VOCAB_FILES_NAMES, RealmTokenizer
class     a    ( _SCREAMING_SNAKE_CASE   ):



      """simple docstring"""


      def 	UpperCAmelCase      (      self					:			Any					)							->      int:
        __UpperCAmelCase							:	Any           =  tempfile.mkdtemp()
        __UpperCAmelCase							:	str           =  5

        # Realm tok
        __UpperCAmelCase							:	Union[str, Any]           =  [
            """[UNK]""",
            """[CLS]""",
            """[SEP]""",
            """[PAD]""",
            """[MASK]""",
            """test""",
            """question""",
            """this""",
            """is""",
            """the""",
            """first""",
            """second""",
            """third""",
            """fourth""",
            """fifth""",
            """record""",
            """want""",
            """##want""",
            """##ed""",
            """wa""",
            """un""",
            """runn""",
            """##ing""",
            """,""",
            """low""",
            """lowest""",
        ]
        __UpperCAmelCase							:	int           =  os.path.join(self.tmpdirname		,      """realm_tokenizer"""					)
        os.makedirs(A_		,      exist_ok=A_					)
        __UpperCAmelCase							:	Tuple           =  os.path.join(A_		,      VOCAB_FILES_NAMES["""vocab_file"""]					)
        with open(self.vocab_file		,      """w"""		,      encoding="""utf-8"""					) as vocab_writer:
          vocab_writer.write("""""".join([x + """\n""" for x in vocab_tokens]					)					)

        __UpperCAmelCase							:	Dict           =  os.path.join(self.tmpdirname		,      """realm_block_records"""					)
        os.makedirs(A_		,      exist_ok=A_					)
      def 	UpperCAmelCase      (      self					:			Dict					)							->      RealmTokenizer:
        return RealmTokenizer.from_pretrained(os.path.join(self.tmpdirname		,      """realm_tokenizer"""					)					)
      def 	UpperCAmelCase      (      self					:			List[Any]					)							->      str:
        shutil.rmtree(self.tmpdirname					)
      def 	UpperCAmelCase      (      self					:			List[Any]					)							->      List[Any]:
        __UpperCAmelCase							:	List[str]           =  RealmConfig(num_block_records=self.num_block_records					)
        return config
      def 	UpperCAmelCase      (      self					:			Any					)							->      Dict:
        __UpperCAmelCase							:	Tuple           =  Dataset.from_dict(
            {
                """id""": ["""0""", """1"""],
                """question""": ["""foo""", """bar"""],
                """answers""": [["""Foo""", """Bar"""], ["""Bar"""]],
            }					)
        return dataset
      def 	UpperCAmelCase      (      self					:			int					)							->      Tuple:
        __UpperCAmelCase							:	Any           =  np.array(
            [
                B"""This is the first record""",
                B"""This is the second record""",
                B"""This is the third record""",
                B"""This is the fourth record""",
                B"""This is the fifth record""",
                B"""This is a longer longer longer record""",
            ]		,      dtype=A_		,      )
        return block_records
      def 	UpperCAmelCase      (      self					:			Tuple					)							->      int:
        __UpperCAmelCase							:	Any           =  RealmRetriever(
            block_records=self.get_dummy_block_records()		,      tokenizer=self.get_tokenizer()		,      )
        return retriever
      def 	UpperCAmelCase      (      self					:			Tuple					)							->      List[str]:
        __UpperCAmelCase							:	Optional[int]           =  self.get_config()
        __UpperCAmelCase							:	Union[str, Any]           =  self.get_dummy_retriever()
        __UpperCAmelCase							:	Dict           =  retriever.tokenizer

        __UpperCAmelCase							:	Optional[int]           =  np.array([0, 3]		,      dtype="""long"""					)
        __UpperCAmelCase							:	int           =  tokenizer(["""Test question"""]					).input_ids
        __UpperCAmelCase							:	Optional[Any]           =  tokenizer(
            ["""the fourth"""]		,      add_special_tokens=A_		,      return_token_type_ids=A_		,      return_attention_mask=A_		,      ).input_ids
        __UpperCAmelCase							:	str           =  config.reader_seq_len

        __UpperCAmelCase      , __UpperCAmelCase      , __UpperCAmelCase      , __UpperCAmelCase							:	List[str]           =  retriever(
            A_		,      A_		,      answer_ids=A_		,      max_length=A_		,      return_tensors="""np"""					)

        self.assertEqual(len(A_					)		,      2					)
        self.assertEqual(len(A_					)		,      2					)
        self.assertEqual(len(A_					)		,      2					)
        self.assertEqual(concat_inputs.input_ids.shape		,      (2, 10)					)
        self.assertEqual(concat_inputs.attention_mask.shape		,      (2, 10)					)
        self.assertEqual(concat_inputs.token_type_ids.shape		,      (2, 10)					)
        self.assertEqual(concat_inputs.special_tokens_mask.shape		,      (2, 10)					)
        self.assertEqual(
            tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[0]					)		,      ["""[CLS]""", """test""", """question""", """[SEP]""", """this""", """is""", """the""", """first""", """record""", """[SEP]"""]		,      )
        self.assertEqual(
            tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[1]					)		,      ["""[CLS]""", """test""", """question""", """[SEP]""", """this""", """is""", """the""", """fourth""", """record""", """[SEP]"""]		,      )
      def 	UpperCAmelCase      (      self					:			Tuple					)							->      str:
        __UpperCAmelCase							:	List[Any]           =  self.get_config()
        __UpperCAmelCase							:	List[str]           =  self.get_dummy_retriever()
        __UpperCAmelCase							:	Dict           =  retriever.tokenizer

        __UpperCAmelCase							:	List[str]           =  np.array([0, 3, 5]		,      dtype="""long"""					)
        __UpperCAmelCase							:	int           =  tokenizer(["""Test question"""]					).input_ids
        __UpperCAmelCase							:	str           =  tokenizer(
            ["""the fourth""", """longer longer"""]		,      add_special_tokens=A_		,      return_token_type_ids=A_		,      return_attention_mask=A_		,      ).input_ids
        __UpperCAmelCase							:	Optional[int]           =  config.reader_seq_len

        __UpperCAmelCase      , __UpperCAmelCase      , __UpperCAmelCase      , __UpperCAmelCase							:	str           =  retriever(
            A_		,      A_		,      answer_ids=A_		,      max_length=A_		,      return_tensors="""np"""					)

        self.assertEqual([False, True, True]		,      A_					)
        self.assertEqual([[-1, -1, -1], [6, -1, -1], [6, 7, 8]]		,      A_					)
        self.assertEqual([[-1, -1, -1], [7, -1, -1], [7, 8, 9]]		,      A_					)




      def 	UpperCAmelCase      (      self					:			int					)							->      Tuple:
        __UpperCAmelCase							:	Tuple           =  self.get_dummy_retriever()
        retriever.save_pretrained(os.path.join(self.tmpdirname		,      """realm_block_records"""					)					)

        # Test local path
        __UpperCAmelCase							:	Optional[int]           =  retriever.from_pretrained(os.path.join(self.tmpdirname		,      """realm_block_records"""					)					)
        self.assertEqual(retriever.block_records[0]		,      B"""This is the first record"""					)

        # Test mocked remote path
        with patch("""transformers.models.realm.retrieval_realm.hf_hub_download"""					) as mock_hf_hub_download:
          __UpperCAmelCase							:	str           =  os.path.join(
              os.path.join(self.tmpdirname		,      """realm_block_records"""					)		,      _REALM_BLOCK_RECORDS_FILENAME					)
          __UpperCAmelCase							:	List[str]           =  RealmRetriever.from_pretrained("""google/realm-cc-news-pretrained-openqa"""					)

        self.assertEqual(retriever.block_records[0]		,      B"""This is the first record"""					)




import requests


def       A       (					lowercase    ,       lowercase )   ->  None:
		'''simple docstring'''

		UpperCamelCase		=     {'Content-Type': 'application/json'}
		UpperCamelCase		=     requests.post(lowercase    ,       json={'text': message_body}    ,       headers=lowercase )
		if response.status_code != 200:
				UpperCamelCase		=     (
				    'Request to slack returned an error '
				    f'''{response.status_code}, the response is:\n{response.text}'''
				)
				raise ValueError(lowercase )


if __name__ == "__main__":
					# Set the slack url to the one provided by Slack when you create the webhook at
					# https://my.slack.com/services/new/incoming-webhook/
					send_slack_message("<YOUR MESSAGE BODY>", "<SLACK CHANNEL URL>")
