

"""simple docstring"""


import collections
from typing import List, Optional, Union

from ...tokenization_utils_base import BatchEncoding
from ...utils import TensorType, add_end_docstrings, add_start_docstrings, logging
from ..bert.tokenization_bert_fast import BertTokenizerFast
from .tokenization_dpr import DPRContextEncoderTokenizer, DPRQuestionEncoderTokenizer, DPRReaderTokenizer


a					:str    = logging.get_logger(__name__)

a					:Optional[int]    = {"vocab_file": "vocab.txt", "tokenizer_file": "tokenizer.json"}

a					:Optional[int]    = {
    "vocab_file": {
        "facebook/dpr-ctx_encoder-single-nq-base": (
            "https://huggingface.co/facebook/dpr-ctx_encoder-single-nq-base/resolve/main/vocab.txt"
        ),
        "facebook/dpr-ctx_encoder-multiset-base": (
            "https://huggingface.co/facebook/dpr-ctx_encoder-multiset-base/resolve/main/vocab.txt"
        ),
    },
    "tokenizer_file": {
        "facebook/dpr-ctx_encoder-single-nq-base": (
            "https://huggingface.co/facebook/dpr-ctx_encoder-single-nq-base/resolve/main/tokenizer.json"
        ),
        "facebook/dpr-ctx_encoder-multiset-base": (
            "https://huggingface.co/facebook/dpr-ctx_encoder-multiset-base/resolve/main/tokenizer.json"
        ),
    },
}
a					:Union[str, Any]    = {
    "vocab_file": {
        "facebook/dpr-question_encoder-single-nq-base": (
            "https://huggingface.co/facebook/dpr-question_encoder-single-nq-base/resolve/main/vocab.txt"
        ),
        "facebook/dpr-question_encoder-multiset-base": (
            "https://huggingface.co/facebook/dpr-question_encoder-multiset-base/resolve/main/vocab.txt"
        ),
    },
    "tokenizer_file": {
        "facebook/dpr-question_encoder-single-nq-base": (
            "https://huggingface.co/facebook/dpr-question_encoder-single-nq-base/resolve/main/tokenizer.json"
        ),
        "facebook/dpr-question_encoder-multiset-base": (
            "https://huggingface.co/facebook/dpr-question_encoder-multiset-base/resolve/main/tokenizer.json"
        ),
    },
}
a					:str    = {
    "vocab_file": {
        "facebook/dpr-reader-single-nq-base": (
            "https://huggingface.co/facebook/dpr-reader-single-nq-base/resolve/main/vocab.txt"
        ),
        "facebook/dpr-reader-multiset-base": (
            "https://huggingface.co/facebook/dpr-reader-multiset-base/resolve/main/vocab.txt"
        ),
    },
    "tokenizer_file": {
        "facebook/dpr-reader-single-nq-base": (
            "https://huggingface.co/facebook/dpr-reader-single-nq-base/resolve/main/tokenizer.json"
        ),
        "facebook/dpr-reader-multiset-base": (
            "https://huggingface.co/facebook/dpr-reader-multiset-base/resolve/main/tokenizer.json"
        ),
    },
}

a					:str    = {
    "facebook/dpr-ctx_encoder-single-nq-base": 512,
    "facebook/dpr-ctx_encoder-multiset-base": 512,
}
a					:Tuple    = {
    "facebook/dpr-question_encoder-single-nq-base": 512,
    "facebook/dpr-question_encoder-multiset-base": 512,
}
a					:Optional[int]    = {
    "facebook/dpr-reader-single-nq-base": 512,
    "facebook/dpr-reader-multiset-base": 512,
}


a					:int    = {
    "facebook/dpr-ctx_encoder-single-nq-base": {"do_lower_case": True},
    "facebook/dpr-ctx_encoder-multiset-base": {"do_lower_case": True},
}
a					:Dict    = {
    "facebook/dpr-question_encoder-single-nq-base": {"do_lower_case": True},
    "facebook/dpr-question_encoder-multiset-base": {"do_lower_case": True},
}
a					:Union[str, Any]    = {
    "facebook/dpr-reader-single-nq-base": {"do_lower_case": True},
    "facebook/dpr-reader-multiset-base": {"do_lower_case": True},
}

class      __a							(UpperCamelCase_):






     '''simple docstring'''
     _SCREAMING_SNAKE_CASE						:Dict       = VOCAB_FILES_NAMES
     _SCREAMING_SNAKE_CASE						:str       = CONTEXT_ENCODER_PRETRAINED_VOCAB_FILES_MAP
     _SCREAMING_SNAKE_CASE						:Dict       = CONTEXT_ENCODER_PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     _SCREAMING_SNAKE_CASE						:int       = CONTEXT_ENCODER_PRETRAINED_INIT_CONFIGURATION
     _SCREAMING_SNAKE_CASE						:int       = DPRContextEncoderTokenizer





class      __a							(UpperCamelCase_):






     '''simple docstring'''
     _SCREAMING_SNAKE_CASE						:Union[str, Any]       = VOCAB_FILES_NAMES
     _SCREAMING_SNAKE_CASE						:str       = QUESTION_ENCODER_PRETRAINED_VOCAB_FILES_MAP
     _SCREAMING_SNAKE_CASE						:Dict       = QUESTION_ENCODER_PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     _SCREAMING_SNAKE_CASE						:List[Any]       = QUESTION_ENCODER_PRETRAINED_INIT_CONFIGURATION
     _SCREAMING_SNAKE_CASE						:Tuple       = DPRQuestionEncoderTokenizer





a					:int    = collections.namedtuple(
    "DPRSpanPrediction", ["span_score", "relevance_score", "doc_id", "start_index", "end_index", "text"]
)

a					:Optional[Any]    = collections.namedtuple("DPRReaderOutput", ["start_logits", "end_logits", "relevance_logits"])


a					:List[str]    = r"\n    Return a dictionary with the token ids of the input strings and other information to give to `.decode_best_spans`.\n    It converts the strings of a question and different passages (title and text) in a sequence of IDs (integers),\n    using the tokenizer and vocabulary. The resulting `input_ids` is a matrix of size `(n_passages, sequence_length)`\n    with the format:\n\n    [CLS] <question token ids> [SEP] <titles ids> [SEP] <texts ids>\n\n    Args:\n        questions (`str` or `List[str]`):\n            The questions to be encoded. You can specify one question for many passages. In this case, the question\n            will be duplicated like `[questions] * n_passages`. Otherwise you have to specify as many questions as in\n            `titles` or `texts`.\n        titles (`str` or `List[str]`):\n            The passages titles to be encoded. This can be a string or a list of strings if there are several passages.\n        texts (`str` or `List[str]`):\n            The passages texts to be encoded. This can be a string or a list of strings if there are several passages.\n        padding (`bool`, `str` or [`~utils.PaddingStrategy`], *optional*, defaults to `False`):\n            Activates and controls padding. Accepts the following values:\n\n            - `True` or `'longest'`: Pad to the longest sequence in the batch (or no padding if only a single sequence\n              if provided).\n            - `'max_length'`: Pad to a maximum length specified with the argument `max_length` or to the maximum\n              acceptable input length for the model if that argument is not provided.\n            - `False` or `'do_not_pad'` (default): No padding (i.e., can output a batch with sequences of different\n              lengths).\n        truncation (`bool`, `str` or [`~tokenization_utils_base.TruncationStrategy`], *optional*, defaults to `False`):\n            Activates and controls truncation. Accepts the following values:\n\n            - `True` or `'longest_first'`: Truncate to a maximum length specified with the argument `max_length` or to\n              the maximum acceptable input length for the model if that argument is not provided. This will truncate\n              token by token, removing a token from the longest sequence in the pair if a pair of sequences (or a batch\n              of pairs) is provided.\n            - `'only_first'`: Truncate to a maximum length specified with the argument `max_length` or to the maximum\n              acceptable input length for the model if that argument is not provided. This will only truncate the first\n              sequence of a pair if a pair of sequences (or a batch of pairs) is provided.\n            - `'only_second'`: Truncate to a maximum length specified with the argument `max_length` or to the maximum\n              acceptable input length for the model if that argument is not provided. This will only truncate the\n              second sequence of a pair if a pair of sequences (or a batch of pairs) is provided.\n            - `False` or `'do_not_truncate'` (default): No truncation (i.e., can output batch with sequence lengths\n              greater than the model maximum admissible input size).\n        max_length (`int`, *optional*):\n                Controls the maximum length to use by one of the truncation/padding parameters.\n\n                If left unset or set to `None`, this will use the predefined model maximum length if a maximum length\n                is required by one of the truncation/padding parameters. If the model has no specific maximum input\n                length (like XLNet) truncation/padding to a maximum length will be deactivated.\n        return_tensors (`str` or [`~utils.TensorType`], *optional*):\n                If set, will return tensors instead of list of python integers. Acceptable values are:\n\n                - `'tf'`: Return TensorFlow `tf.constant` objects.\n                - `'pt'`: Return PyTorch `torch.Tensor` objects.\n                - `'np'`: Return Numpy `np.ndarray` objects.\n        return_attention_mask (`bool`, *optional*):\n            Whether or not to return the attention mask. If not set, will return the attention mask according to the\n            specific tokenizer's default, defined by the `return_outputs` attribute.\n\n            [What are attention masks?](../glossary#attention-mask)\n\n    Return:\n        `Dict[str, List[List[int]]]`: A dictionary with the following keys:\n\n        - `input_ids`: List of token ids to be fed to a model.\n        - `attention_mask`: List of indices specifying which tokens should be attended to by the model.\n    "

@add_start_docstrings(UpperCamelCase_)
class      __a							:






     '''simple docstring'''

     def __call__(						self	,    _a	,    _a = None	,    _a = None	,    _a = False	,    _a = False	,    _a = None	,    _a = None	,    _a = None	,    **_a	,    )  ->      BatchEncoding:


           """simple docstring"""


           if titles is None and texts is None:
                 return super().__call__(
                     _a	,    padding=_a	,    truncation=_a	,    max_length=_a	,    return_tensors=_a	,    return_attention_mask=_a	,    **_a	,    )
           elif titles is None or texts is None:
                 SCREAMING_SNAKE_CASE__   :      str      		=	titles if texts is None else texts
                 return super().__call__(
                     _a	,    _a	,    padding=_a	,    truncation=_a	,    max_length=_a	,    return_tensors=_a	,    return_attention_mask=_a	,    **_a	,    )
           SCREAMING_SNAKE_CASE__   :      Union[str, Any]      		=	titles if not isinstance(_a	,    _a  ) else [titles]
           SCREAMING_SNAKE_CASE__   :      Any      		=	texts if not isinstance(_a	,    _a  ) else [texts]
           SCREAMING_SNAKE_CASE__   :      int      		=	len(_a  )
           SCREAMING_SNAKE_CASE__   :      int      		=	questions if not isinstance(_a	,    _a  ) else [questions] * n_passages
           assert len(_a  ) == len(
               _a  ), f'''There should be as many titles than texts but got {len(_a  )} titles and {len(_a  )} texts.'''
           SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	super().__call__(_a	,    _a	,    padding=_a	,    truncation=_a  )["""input_ids"""]
           SCREAMING_SNAKE_CASE__   :      Optional[Any]      		=	super().__call__(_a	,    add_special_tokens=_a	,    padding=_a	,    truncation=_a  )["""input_ids"""]
           SCREAMING_SNAKE_CASE__   :      List[str]      		=	{
               """input_ids""": [
                   (encoded_question_and_title + encoded_text)[:max_length]
                   if max_length is not None and truncation
                   else encoded_question_and_title + encoded_text
                   for encoded_question_and_title, encoded_text in zip(_a	,    _a  )
               ]
           }
           if return_attention_mask is not False:
                 SCREAMING_SNAKE_CASE__   :      str      		=	[]
                 for input_ids in encoded_inputs["input_ids"]:
                       attention_mask.append([int(input_id != self.pad_token_id  ) for input_id in input_ids]  )
                 SCREAMING_SNAKE_CASE__   :      str      		=	attention_mask
           return self.pad(_a	,    padding=_a	,    max_length=_a	,    return_tensors=_a  )

     def 					_a			(						self	,    _a	,    _a	,    _a = 16	,    _a = 64	,    _a = 4	,    )  ->      List[DPRSpanPrediction]:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      List[Any]      		=	reader_input["""input_ids"""]
           SCREAMING_SNAKE_CASE__		,		SCREAMING_SNAKE_CASE__		,		SCREAMING_SNAKE_CASE__   :      Any      		=	reader_output[:3]
           SCREAMING_SNAKE_CASE__   :      Dict      		=	len(_a  )
           SCREAMING_SNAKE_CASE__   :      str      		=	sorted(range(_a  )	,    reverse=_a	,    key=relevance_logits.__getitem__  )
           SCREAMING_SNAKE_CASE__   :      List[DPRReaderOutput]      		=	[]
           for doc_id in sorted_docs:
                 SCREAMING_SNAKE_CASE__   :      Any      		=	list(input_ids[doc_id]  )
                 # assuming question & title information is at the beginning of the sequence
                 SCREAMING_SNAKE_CASE__   :      List[Any]      		=	sequence_ids.index(self.sep_token_id	,    2  ) + 1  # second sep id
                 if sequence_ids[-1] == self.pad_token_id:
                       SCREAMING_SNAKE_CASE__   :      int      		=	sequence_ids.index(self.pad_token_id  )
                 else:
                       SCREAMING_SNAKE_CASE__   :      List[Any]      		=	len(_a  )

                 SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	self._get_best_spans(
                     start_logits=start_logits[doc_id][passage_offset:sequence_len]	,    end_logits=end_logits[doc_id][passage_offset:sequence_len]	,    max_answer_length=_a	,    top_spans=_a	,    )
                 for start_index, end_index in best_spans:
                       start_index += passage_offset
                       end_index += passage_offset
                       nbest_spans_predictions.append(
                           DPRSpanPrediction(
                               span_score=start_logits[doc_id][start_index] + end_logits[doc_id][end_index]	,    relevance_score=relevance_logits[doc_id]	,    doc_id=_a	,    start_index=_a	,    end_index=_a	,    text=self.decode(sequence_ids[start_index : end_index + 1]  )	,    )  )
                 if len(_a  ) >= num_spans:
                       break
           return nbest_spans_predictions[:num_spans]





     def 					_a			(						self	,    _a	,    _a	,    _a	,    _a	,    )  ->      List[DPRSpanPrediction]:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      List[Any]      		=	[]
           for start_index, start_score in enumerate(_a  ):
                 for answer_length, end_score in enumerate(end_logits[start_index : start_index + max_answer_length]  ):
                       scores.append(((start_index, start_index + answer_length), start_score + end_score)  )
           SCREAMING_SNAKE_CASE__   :      str      		=	sorted(_a	,    key=lambda _a  : x[1]	,    reverse=_a  )
           SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	[]
           for (start_index, end_index), score in scores:
                 assert start_index <= end_index, f'''Wrong span indices: [{start_index}:{end_index}]'''
                 SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	end_index - start_index + 1
                 assert length <= max_answer_length, f'''Span is too long: {length} > {max_answer_length}'''
                 if any(
                     start_index <= prev_start_index <= prev_end_index <= end_index
                     or prev_start_index <= start_index <= end_index <= prev_end_index
                     for (prev_start_index, prev_end_index) in chosen_span_intervals  ):
                       continue
                 chosen_span_intervals.append((start_index, end_index)  )

                 if len(_a  ) == top_spans:
                       break
           return chosen_span_intervals





@add_end_docstrings(UpperCamelCase_)
class      __a							(UpperCamelCase_   ,	UpperCamelCase_):






     '''simple docstring'''
     _SCREAMING_SNAKE_CASE						:List[Any]       = VOCAB_FILES_NAMES
     _SCREAMING_SNAKE_CASE						:Union[str, Any]       = READER_PRETRAINED_VOCAB_FILES_MAP
     _SCREAMING_SNAKE_CASE						:Optional[int]       = READER_PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     _SCREAMING_SNAKE_CASE						:List[Any]       = READER_PRETRAINED_INIT_CONFIGURATION
     _SCREAMING_SNAKE_CASE						:Tuple       = ["""input_ids""", """attention_mask"""]
     _SCREAMING_SNAKE_CASE						:List[Any]       = DPRReaderTokenizer







"""simple docstring"""


from __future__ import annotations

import time

import numpy as np

a					:Optional[Any]    = [8, 5, 9, 7]
a					:List[Any]    = [
    [2, 0, 1, 1],
    [0, 1, 2, 1],
    [4, 0, 0, 3],
    [0, 2, 1, 0],
    [1, 0, 3, 0],
]
a					:int    = [
    [3, 2, 1, 4],
    [0, 2, 5, 2],
    [5, 1, 0, 5],
    [1, 5, 3, 0],
    [3, 0, 3, 3],
]

class      __a							:






     '''simple docstring'''

     def __init__(						self	,    _a	,    _a	,    _a	,    )  ->      None:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      str      		=	claim_vector
           SCREAMING_SNAKE_CASE__   :      Any      		=	allocated_resources_table
           SCREAMING_SNAKE_CASE__   :      Any      		=	maximum_claim_table

     def 					_a			(						self  )  ->      list[int]:


           """simple docstring"""


           return [
               sum(p_item[i] for p_item in self.__allocated_resources_table  )
               for i in range(len(self.__allocated_resources_table[0]  )  )
           ]

     def 					_a			(						self  )  ->      list[int]:


           """simple docstring"""


           return np.array(self.__claim_vector  ) - np.array(
               self.__processes_resource_summation()  )

     def 					_a			(						self  )  ->      list[list[int]]:


           """simple docstring"""


           return [
               list(np.array(self.__maximum_claim_table[i]  ) - np.array(_a  )  )
               for i, allocated_resource in enumerate(self.__allocated_resources_table  )
           ]

     def 					_a			(						self  )  ->      dict[int, list[int]]:


           """simple docstring"""


           return {self.__need().index(_a  ): i for i in self.__need()}

     def 					_a			(						self	,    **_a  )  ->      None:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	self.__need()
           SCREAMING_SNAKE_CASE__   :      Any      		=	self.__allocated_resources_table
           SCREAMING_SNAKE_CASE__   :      Dict      		=	self.__available_resources()
           SCREAMING_SNAKE_CASE__   :      Dict      		=	self.__need_index_manager()
           for kw, val in kwargs.items():
                 if kw and val is True:
                       self.__pretty_data()
           print("""_""" * 50 + """\n"""  )
           while need_list:
                 SCREAMING_SNAKE_CASE__   :      List[str]      		=	False
                 for each_need in need_list:
                       SCREAMING_SNAKE_CASE__   :      Dict      		=	True
                       for index, need in enumerate(_a  ):
                             if need > available_resources[index]:
                                   SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	False
                                   break
                       if execution:
                             SCREAMING_SNAKE_CASE__   :      Any      		=	True
                             # get the original index of the process from ind_ctrl db
                             for original_need_index, need_clone in need_index_manager.items():
                                   if each_need == need_clone:
                                         SCREAMING_SNAKE_CASE__   :      Tuple      		=	original_need_index
                             print(f'''Process {process_number + 1} is executing.'''  )
                             # remove the process run from stack
                             need_list.remove(_a  )
                             # update available/freed resources stack
                             SCREAMING_SNAKE_CASE__   :      Dict      		=	np.array(_a  ) + np.array(
                                 alloc_resources_table[process_number]  )
                             print(
                                 """Updated available resource stack for processes: """
                                 + """ """.join([str(_a  ) for x in available_resources]  )  )
                             break
                 if safe:
                       print("""The process is in a safe state.\n"""  )
                 else:
                       print("""System in unsafe state. Aborting...\n"""  )
                       break





     def 					_a			(						self  )  ->      Any:


           """simple docstring"""


           print(""" """ * 9 + """Allocated Resource Table"""  )
           for item in self.__allocated_resources_table:
                 print(
                     f'''P{self.__allocated_resources_table.index(_a  ) + 1}'''
                     + """ """.join(f'''{it:>8}''' for it in item  )
                     + """\n"""  )
           print(""" """ * 9 + """System Resource Table"""  )
           for item in self.__maximum_claim_table:
                 print(
                     f'''P{self.__maximum_claim_table.index(_a  ) + 1}'''
                     + """ """.join(f'''{it:>8}''' for it in item  )
                     + """\n"""  )
           print(
               """Current Usage by Active Processes: """
               + """ """.join(str(_a  ) for x in self.__claim_vector  )  )
           print(
               """Initial Available Resources:       """
               + """ """.join(str(_a  ) for x in self.__available_resources()  )  )
           time.sleep(1  )


if __name__ == "__main__":
  import doctest

  doctest.testmod()




