'''simple docstring'''
import json
import os
import pickle
import subprocess
from hashlib import mda
from pathlib import Path
from tempfile import gettempdir
from textwrap import dedent
from types import CodeType, FunctionType
from unittest import TestCase
from unittest.mock import patch

import pytest
from multiprocess import Pool
from packaging import version

import datasets
from datasets.fingerprint import Hasher, fingerprint_transform
from datasets.table import InMemoryTable

from .utils import (
    require_regex,
    require_spacy,
    require_spacy_model,
    require_tiktoken,
    require_torch,
    require_transformers,
)






class       __lowercase		:







     """simple docstring"""






     def __init__(		self  ,      _lowerCamelCase     ):
         __UpperCamelCase						:						Optional[Any]		     =	foo







     def __call__(		self     ):
         return self.foo






class       __lowercase		(    datasets.Dataset    ):







     """simple docstring"""






     @fingerprint_transform(inplace=lowerCAmelCase_     )
     def lowerCAmelCase  (		self  ,      _lowerCamelCase  ,      *_lowerCamelCase  ,      **_lowerCamelCase     ):
         return DatasetChild(self.data  ,      fingerprint=lowerCAmelCase_     )







     @fingerprint_transform(inplace=lowerCAmelCase_     )
     def lowerCAmelCase  (		self  ,      _lowerCamelCase  ,      *_lowerCamelCase  ,      **_lowerCamelCase     ):
         return DatasetChild(self.data  ,      fingerprint=lowerCAmelCase_     )






class       __lowercase		:







     """simple docstring"""






     def __init__(		self  ,      _lowerCamelCase     ):
         __UpperCamelCase						:						str		     =	callable







     def __call__(		self  ,      *_lowerCamelCase  ,      **_lowerCamelCase     ):
         if self.callable is not None:
             return self.callable(*lowerCAmelCase_  ,      **lowerCAmelCase_     )







     def __getstate__(		self     ):
         raise pickle.PicklingError()






class       __lowercase		(    _lowerCamelCase    ):







     """simple docstring"""






     @require_transformers
     @pytest.mark.integration
     def lowerCAmelCase  (		self     ):
         from transformers import AutoTokenizer

         def encode(_lowerCamelCase     ):
             return tokenizer(lowerCAmelCase_     )

         # TODO: add hash consistency tests across sessions
         __UpperCamelCase						:						Dict		     =	AutoTokenizer.from_pretrained('bert-base-uncased'     )
         __UpperCamelCase						:						Optional[Any]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						List[str]		     =	mda(datasets.utils.py_utils.dumps(lambda _lowerCamelCase     : tokenizer(lowerCAmelCase_     )     )     ).hexdigest()
         __UpperCamelCase						:						Any		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						Tuple		     =	AutoTokenizer.from_pretrained('bert-base-cased'     )
         __UpperCamelCase						:						Optional[Any]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						Union[str, Any]		     =	mda(datasets.utils.py_utils.dumps(lambda _lowerCamelCase     : tokenizer(lowerCAmelCase_     )     )     ).hexdigest()
         __UpperCamelCase						:						Tuple		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						str		     =	AutoTokenizer.from_pretrained('bert-base-uncased'     )
         __UpperCamelCase						:						Optional[Any]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						List[str]		     =	mda(datasets.utils.py_utils.dumps(lambda _lowerCamelCase     : tokenizer(lowerCAmelCase_     )     )     ).hexdigest()
         __UpperCamelCase						:						Any		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







     @require_transformers
     @pytest.mark.integration
     def lowerCAmelCase  (		self     ):
         from transformers import AutoTokenizer

         __UpperCamelCase						:						Union[str, Any]		     =	AutoTokenizer.from_pretrained('gpt2'     )
         __UpperCamelCase						:						Union[str, Any]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         tokenizer('Hello world !'     )  # call once to change the tokenizer's cache
         __UpperCamelCase						:						Any		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







     @require_regex
     def lowerCAmelCase  (		self     ):
         import regex

         __UpperCamelCase						:						Dict		     =	regex.Regex('foo'     )
         __UpperCamelCase						:						Union[str, Any]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						Union[str, Any]		     =	regex.Regex('bar'     )
         __UpperCamelCase						:						Dict		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						Dict		     =	regex.Regex('foo'     )
         __UpperCamelCase						:						str		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )






class       __lowercase		(    _lowerCamelCase    ):







     """simple docstring"""






     def lowerCAmelCase  (		self     ):
         def func():
             return foo

         __UpperCamelCase						:						int		     =	[0]
         __UpperCamelCase						:						Optional[Any]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						int		     =	[1]
         __UpperCamelCase						:						Union[str, Any]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						Optional[Any]		     =	[0]
         __UpperCamelCase						:						Any		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







     def lowerCAmelCase  (		self     ):
         def func():
             pass

         __UpperCamelCase						:						int		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()

         def func():
             pass

         __UpperCamelCase						:						List[str]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







     def lowerCAmelCase  (		self     ):
         __UpperCamelCase						:						Optional[int]		     =	mda(datasets.utils.py_utils.dumps(Foo([0]     )     )     ).hexdigest()
         __UpperCamelCase						:						List[Any]		     =	mda(datasets.utils.py_utils.dumps(Foo([1]     )     )     ).hexdigest()
         __UpperCamelCase						:						str		     =	mda(datasets.utils.py_utils.dumps(Foo([0]     )     )     ).hexdigest()
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







     def lowerCAmelCase  (		self     ):
         __UpperCamelCase						:						int		     =	mda(datasets.utils.py_utils.dumps(Foo([0]     ).__call__     )     ).hexdigest()
         __UpperCamelCase						:						Tuple		     =	mda(datasets.utils.py_utils.dumps(Foo([1]     ).__call__     )     ).hexdigest()
         __UpperCamelCase						:						Dict		     =	mda(datasets.utils.py_utils.dumps(Foo([0]     ).__call__     )     ).hexdigest()
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







     def lowerCAmelCase  (		self     ):
         __UpperCamelCase						:						List[Any]		     =	(
             'co_argcount',
             'co_kwonlyargcount',
             'co_nlocals',
             'co_stacksize',
             'co_flags',
             'co_code',
             'co_consts',
             'co_names',
             'co_varnames',
             'co_filename',
             'co_name',
             'co_firstlineno',
             'co_lnotab',
             'co_freevars',
             'co_cellvars',
         )

         def _create_code_pyaa(*_lowerCamelCase     ):
             if hasattr(args[-3]  ,      'encode'     ):
                 __UpperCamelCase						:						Any		     =	list(lowerCAmelCase_     )
                 __UpperCamelCase						:						Tuple		     =	args[-3].encode()  # co_lnotab
                 __UpperCamelCase						:						List[Any]		     =	args[-1_0].encode()  # co_code
             if hasattr(lowerCAmelCase_  ,      'co_posonlyargcount'     ):
                 if len(lowerCAmelCase_     ) == 1_6:
                     return CodeType(*lowerCAmelCase_     )
                 elif len(lowerCAmelCase_     ) == 1_5:
                     return CodeType(args[0]  ,      0  ,      *args[1:]     )
                 return CodeType(args[0]  ,      0  ,      0  ,      *args[1:]     )
             elif hasattr(lowerCAmelCase_  ,      'co_kwonlyargcount'     ):
                 if len(lowerCAmelCase_     ) == 1_6:
                     return CodeType(args[0]  ,      *args[2:]     )
                 elif len(lowerCAmelCase_     ) == 1_5:
                     return CodeType(*lowerCAmelCase_     )
                 return CodeType(args[0]  ,      0  ,      *args[1:]     )
             if len(lowerCAmelCase_     ) == 1_6:
                 return CodeType(args[0]  ,      *args[3:]     )
             elif len(lowerCAmelCase_     ) == 1_5:
                 return CodeType(args[0]  ,      *args[2:]     )
             return CodeType(*lowerCAmelCase_     )

         def create_ipython_func(_lowerCamelCase  ,      _lowerCamelCase     ):
             def func():
                 return returned_obj

             __UpperCamelCase						:						Dict		     =	func.__code__
             # Use _create_code from dill in order to make it work for different python versions
             if datasets.config.PY_VERSION < version.parse('3.8'     ):
                 __UpperCamelCase						:						Any		     =	_create_code_pyaa(
                     *[getattr(lowerCAmelCase_  ,      lowerCAmelCase_     ) if k != 'co_filename' else co_filename for k in code_args_pyaa]     )
             else:
                 __UpperCamelCase						:						Optional[int]		     =	code.replace(co_filename=lowerCAmelCase_     )
             return FunctionType(lowerCAmelCase_  ,      func.__globals__  ,      func.__name__  ,      func.__defaults__  ,      func.__closure__     )

         __UpperCamelCase     ,       __UpperCamelCase						:						List[str]		     =	'<ipython-input-2-e0383a102aae>', [0]
         __UpperCamelCase						:						Optional[int]		     =	mda(datasets.utils.py_utils.dumps(create_ipython_func(lowerCAmelCase_  ,      lowerCAmelCase_     )     )     ).hexdigest()
         __UpperCamelCase     ,       __UpperCamelCase						:						Optional[int]		     =	'<ipython-input-2-e0383a102aae>', [1]
         __UpperCamelCase						:						Dict		     =	mda(datasets.utils.py_utils.dumps(create_ipython_func(lowerCAmelCase_  ,      lowerCAmelCase_     )     )     ).hexdigest()
         __UpperCamelCase     ,       __UpperCamelCase						:						Union[str, Any]		     =	'<ipython-input-5-713f6613acf3>', [0]
         __UpperCamelCase						:						str		     =	mda(datasets.utils.py_utils.dumps(create_ipython_func(lowerCAmelCase_  ,      lowerCAmelCase_     )     )     ).hexdigest()
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )

         __UpperCamelCase     ,       __UpperCamelCase						:						Union[str, Any]		     =	os.path.join(gettempdir()  ,      'ipykernel_12345'  ,      '321456789.py'     ), [0]
         __UpperCamelCase						:						Tuple		     =	mda(datasets.utils.py_utils.dumps(create_ipython_func(lowerCAmelCase_  ,      lowerCAmelCase_     )     )     ).hexdigest()
         __UpperCamelCase     ,       __UpperCamelCase						:						Tuple		     =	os.path.join(gettempdir()  ,      'ipykernel_12345'  ,      '321456789.py'     ), [1]
         __UpperCamelCase						:						int		     =	mda(datasets.utils.py_utils.dumps(create_ipython_func(lowerCAmelCase_  ,      lowerCAmelCase_     )     )     ).hexdigest()
         __UpperCamelCase     ,       __UpperCamelCase						:						Optional[int]		     =	os.path.join(gettempdir()  ,      'ipykernel_12345'  ,      '654123987.py'     ), [0]
         __UpperCamelCase						:						Any		     =	mda(datasets.utils.py_utils.dumps(create_ipython_func(lowerCAmelCase_  ,      lowerCAmelCase_     )     )     ).hexdigest()
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







     def lowerCAmelCase  (		self     ):
         __UpperCamelCase     ,       __UpperCamelCase						:						Dict		     =	[0], [1]

         def func():
             return foo, bar

         __UpperCamelCase						:						Optional[int]		     =	'__main__'

         def globalvars_mocka_side_effect(_lowerCamelCase  ,      *_lowerCamelCase  ,      **_lowerCamelCase     ):
             return {"foo": foo, "bar": bar}

         def globalvars_mocka_side_effect(_lowerCamelCase  ,      *_lowerCamelCase  ,      **_lowerCamelCase     ):
             return {"bar": bar, "foo": foo}

         with patch('dill.detect.globalvars'  ,      side_effect=lowerCAmelCase_     ) as globalvars_mocka:
             __UpperCamelCase						:						List[str]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
             self.assertGreater(globalvars_mocka.call_count  ,      0     )
         with patch('dill.detect.globalvars'  ,      side_effect=lowerCAmelCase_     ) as globalvars_mocka:
             __UpperCamelCase						:						List[Any]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
             self.assertGreater(globalvars_mocka.call_count  ,      0     )
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )






class       __lowercase		(    _lowerCamelCase    ):







     """simple docstring"""






     def lowerCAmelCase  (		self     ):
         __UpperCamelCase						:						Dict		     =	Hasher.hash('hello'     )
         __UpperCamelCase						:						Optional[int]		     =	Hasher.hash('hello'     )
         __UpperCamelCase						:						str		     =	Hasher.hash('there'     )
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







     def lowerCAmelCase  (		self     ):
         __UpperCamelCase						:						str		     =	Hasher.hash(Foo('hello'     )     )
         __UpperCamelCase						:						Optional[Any]		     =	Hasher.hash(Foo('hello'     )     )
         __UpperCamelCase						:						Optional[int]		     =	Hasher.hash(Foo('there'     )     )
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







     def lowerCAmelCase  (		self     ):
         __UpperCamelCase						:						Tuple		     =	Hasher()
         for x in ["hello", Foo('hello'     )]:
             hasher.update(lowerCAmelCase_     )
         __UpperCamelCase						:						Optional[int]		     =	hasher.hexdigest()
         __UpperCamelCase						:						Union[str, Any]		     =	Hasher()
         for x in ["hello", Foo('hello'     )]:
             hasher.update(lowerCAmelCase_     )
         __UpperCamelCase						:						Optional[Any]		     =	hasher.hexdigest()
         __UpperCamelCase						:						Dict		     =	Hasher()
         for x in ["there", Foo('there'     )]:
             hasher.update(lowerCAmelCase_     )
         __UpperCamelCase						:						Tuple		     =	hasher.hexdigest()
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







     def lowerCAmelCase  (		self     ):
         with self.assertRaises(pickle.PicklingError     ):
             Hasher.hash(UnpicklableCallable(Foo('hello'     )     )     )







     def lowerCAmelCase  (		self     ):
         __UpperCamelCase						:						Optional[Any]		     =	'abc'
         __UpperCamelCase						:						str		     =	[string, string]  # two strings have the same ids
         __UpperCamelCase						:						List[Any]		     =	[string, string]
         __UpperCamelCase						:						Any		     =	json.loads(f"""[\"{string}\", \"{string}\"]"""     )  # two strings have different ids
         self.assertIs(obja[0]  ,      lowerCAmelCase_     )
         self.assertIs(obja[0]  ,      obja[1]     )
         self.assertIs(obja[0]  ,      lowerCAmelCase_     )
         self.assertIs(obja[0]  ,      obja[1]     )
         self.assertIsNot(obja[0]  ,      lowerCAmelCase_     )
         self.assertIsNot(obja[0]  ,      obja[1]     )
         __UpperCamelCase						:						Any		     =	Hasher.hash(lowerCAmelCase_     )
         __UpperCamelCase						:						int		     =	Hasher.hash(lowerCAmelCase_     )
         __UpperCamelCase						:						Any		     =	Hasher.hash(lowerCAmelCase_     )
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







     @require_tiktoken
     def lowerCAmelCase  (		self     ):
         import tiktoken

         __UpperCamelCase						:						Union[str, Any]		     =	tiktoken.get_encoding('gpt2'     )
         __UpperCamelCase						:						str		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						Dict		     =	tiktoken.get_encoding('r50k_base'     )
         __UpperCamelCase						:						List[str]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						List[Any]		     =	tiktoken.get_encoding('gpt2'     )
         __UpperCamelCase						:						Union[str, Any]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







     @require_torch
     def lowerCAmelCase  (		self     ):
         import torch

         __UpperCamelCase						:						Tuple		     =	torch.tensor([1.0]     )
         __UpperCamelCase						:						int		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						List[Any]		     =	torch.tensor([2.0]     )
         __UpperCamelCase						:						List[str]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						Union[str, Any]		     =	torch.tensor([1.0]     )
         __UpperCamelCase						:						Optional[Any]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







     @require_spacy
     @require_spacy_model('en_core_web_sm'     )
     @require_spacy_model('fr_core_news_sm'     )
     @pytest.mark.integration
     def lowerCAmelCase  (		self     ):
         import spacy

         __UpperCamelCase						:						Any		     =	spacy.load('en_core_web_sm'     )
         __UpperCamelCase						:						List[Any]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						Tuple		     =	spacy.load('fr_core_news_sm'     )
         __UpperCamelCase						:						Optional[int]		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         __UpperCamelCase						:						str		     =	spacy.load('en_core_web_sm'     )
         __UpperCamelCase						:						int		     =	mda(datasets.utils.py_utils.dumps(lowerCAmelCase_     )     ).hexdigest()
         self.assertEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )
         self.assertNotEqual(lowerCAmelCase_  ,      lowerCAmelCase_     )







@pytest.mark.integration
def 			_UpperCamelCase			(  _a						:						Path      ):




    """simple docstring"""



    __UpperCamelCase						:						Optional[Any]		     =	tmp_path / 'dir1'
    __UpperCamelCase						:						Dict		     =	tmp_path / 'dir2'
    dira.mkdir()
    dira.mkdir()
    __UpperCamelCase						:						Dict		     =	'script.py'
    __UpperCamelCase						:						List[str]		     =	dedent(
        '\n    from datasets.fingerprint import Hasher\n    def foo():\n        pass\n    print(Hasher.hash(foo))\n    '      )
    __UpperCamelCase						:						str		     =	dira / script_filename
    __UpperCamelCase						:						List[str]		     =	dira / script_filename
    with script_patha.open('w'      ) as f:
        f.write(_a      )
    with script_patha.open('w'      ) as f:
        f.write(_a      )
    __UpperCamelCase						:						Optional[int]		     =	subprocess.check_output(['python', str(_a      )]      )
    __UpperCamelCase						:						Union[str, Any]		     =	subprocess.check_output(['python', str(_a      )]      )
    assert fingerprinta == fingerprinta







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    __UpperCamelCase						:						List[Any]		     =	{'a': [0, 1, 2]}
    __UpperCamelCase						:						List[str]		     =	DatasetChild(InMemoryTable.from_pydict(_a      )      )
    __UpperCamelCase						:						Tuple		     =	dataset.funca()._fingerprint
    assert expected_fingerprint == dataset.funca()._fingerprint
    assert expected_fingerprint != dataset.funca()._fingerprint

    with Pool(2      ) as p:
        assert expected_fingerprint == p.apply_async(dataset.funca      ).get()._fingerprint
        assert expected_fingerprint != p.apply_async(dataset.funca      ).get()._fingerprint







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    __UpperCamelCase						:						List[Any]		     =	{'a': [0, 1, 2]}






    class       __lowercase		(    datasets.Dataset    ):







         """simple docstring"""






         @fingerprint_transform(inplace=lowerCAmelCase_     )
         def lowerCAmelCase  (		self  ,      _lowerCamelCase     ):
             return DummyDatasetChild(self.data  ,      fingerprint=lowerCAmelCase_     )




    __UpperCamelCase						:						str		     =	DummyDatasetChild(InMemoryTable.from_pydict(_a      )      ).func()






    class       __lowercase		(    datasets.Dataset    ):







         """simple docstring"""






         @fingerprint_transform(inplace=lowerCAmelCase_  ,      version='1.0.0'     )
         def lowerCAmelCase  (		self  ,      _lowerCamelCase     ):
             return DummyDatasetChild(self.data  ,      fingerprint=lowerCAmelCase_     )




    __UpperCamelCase						:						Union[str, Any]		     =	DummyDatasetChild(InMemoryTable.from_pydict(_a      )      ).func()






    class       __lowercase		(    datasets.Dataset    ):







         """simple docstring"""






         @fingerprint_transform(inplace=lowerCAmelCase_  ,      version='2.0.0'     )
         def lowerCAmelCase  (		self  ,      _lowerCamelCase     ):
             return DummyDatasetChild(self.data  ,      fingerprint=lowerCAmelCase_     )




    __UpperCamelCase						:						Optional[Any]		     =	DummyDatasetChild(InMemoryTable.from_pydict(_a      )      ).func()

    assert len({fingeprint_no_version, fingeprint_a, fingeprint_a}      ) == 3







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    __UpperCamelCase						:						List[str]		     =	Hasher()
    hasher.update(lambda _a      : x      )





'''simple docstring'''





import argparse
import logging
import os
import sys
from collections import namedtuple

import torch
from modeling_bertabs import BertAbs, build_predictor
from torch.utils.data import DataLoader, SequentialSampler
from tqdm import tqdm

from transformers import BertTokenizer

from .utils_summarization import (
    CNNDMDataset,
    build_mask,
    compute_token_type_ids,
    encode_for_summarization,
    truncate_or_pad,
)


A							     = logging.getLogger(__name__)
logging.basicConfig(stream=sys.stdout, level=logging.INFO)


A							     = namedtuple("Batch", ["document_names", "batch_size", "src", "segs", "mask_src", "tgt_str"])
def 	snake_case_							(				_a  :				str      ):
      snake_case_    							=				BertTokenizer.from_pretrained("bert-base-uncased"			,       do_lower_case=_a      )
      snake_case_    							=				BertAbs.from_pretrained("remi/bertabs-finetuned-extractive-abstractive-summarization"      )
      model.to(args.device      )
      model.eval()

      snake_case_    							=				{
          "BOS": tokenizer.vocab["[unused0]"],
          "EOS": tokenizer.vocab["[unused1]"],
          "PAD": tokenizer.vocab["[PAD]"],
      }

      if args.compute_rouge:
            snake_case_    							=				[]
            snake_case_    							=				[]

            import nltk
            import rouge

            nltk.download("punkt"      )
            snake_case_    							=				rouge.Rouge(
                metrics=["rouge-n", "rouge-l"]			,       max_n=2			,       limit_length=_a			,       length_limit=args.beam_size			,       length_limit_type="words"			,       apply_avg=_a			,       apply_best=_a			,       alpha=0.5			,       weight_factor=1.2			,       stemming=_a			,       )

      # these (unused) arguments are defined to keep the compatibility
      # with the legacy code and will be deleted in a next iteration.
      snake_case_    							=				""
      snake_case_    							=				""

      snake_case_    							=				build_data_iterator(_a			,       _a      )
      snake_case_    							=				build_predictor(_a			,       _a			,       _a			,       _a      )

      logger.info("***** Running evaluation *****"      )
      logger.info("  Number examples = %d"			,       len(data_iterator.dataset      )      )
      logger.info("  Batch size = %d"			,       args.batch_size      )
      logger.info(""      )
      logger.info("***** Beam Search parameters *****"      )
      logger.info("  Beam size = %d"			,       args.beam_size      )
      logger.info("  Minimum length = %d"			,       args.min_length      )
      logger.info("  Maximum length = %d"			,       args.max_length      )
      logger.info("  Alpha (length penalty) = %.2f"			,       args.alpha      )
      logger.info("  Trigrams %s be blocked"			,       ("will" if args.block_trigram else "will NOT")      )

      for batch in tqdm(_a      ):
            snake_case_    							=				predictor.translate_batch(_a      )
            snake_case_    							=				predictor.from_batch(_a      )
            snake_case_    							=				[format_summary(_a      ) for t in translations]
            save_summaries(_a			,       args.summaries_output_dir			,       batch.document_names      )

            if args.compute_rouge:
                  reference_summaries += batch.tgt_str
                  generated_summaries += summaries

      if args.compute_rouge:
            snake_case_    							=				rouge_evaluator.get_scores(_a			,       _a      )
            snake_case_    							=				format_rouge_scores(_a      )
            save_rouge_scores(_a      )
            print(_a      )
def 	snake_case_							(				_a  :				List[Any]			,       _a  :				List[str]			,       _a  :				Any      ):
      for summary, document_name in zip(_a			,       _a      ):
            # Prepare the summary file's name
            if "." in document_name:
                  snake_case_    							=				".".join(document_name.split("."      )[:-1]      )
                  snake_case_    							=				document_name.split("."      )[-1]
                  snake_case_    							=				bare_document_name + "_summary." + extension
            else:
                  snake_case_    							=				document_name + "_summary"

            snake_case_    							=				os.path.join(_a			,       _a      )
            with open(_a			,       "w"      ) as output:
                  output.write(_a      )
def 	snake_case_							(				_a  :				Optional[int]      ):
      snake_case_							,						snake_case_							,						snake_case_    							=				translation
      snake_case_    							=				(
          raw_summary.replace("[unused0]"			,       ""      )
          .replace("[unused3]"			,       ""      )
          .replace("[PAD]"			,       ""      )
          .replace("[unused1]"			,       ""      )
          .replace(R" +"			,       " "      )
          .replace(" [unused2] "			,       ". "      )
          .replace("[unused2]"			,       ""      )
          .strip()
      )

      return summary
def 	snake_case_							(				_a  :				str      ):
      return """\n
****** ROUGE SCORES ******

** ROUGE 1
F1        >> {:.3f}
Precision >> {:.3f}
Recall    >> {:.3f}

** ROUGE 2
F1        >> {:.3f}
Precision >> {:.3f}
Recall    >> {:.3f}

** ROUGE L
F1        >> {:.3f}
Precision >> {:.3f}
Recall    >> {:.3f}""".format(
          scores["rouge-1"]["f"]			,       scores["rouge-1"]["p"]			,       scores["rouge-1"]["r"]			,       scores["rouge-2"]["f"]			,       scores["rouge-2"]["p"]			,       scores["rouge-2"]["r"]			,       scores["rouge-l"]["f"]			,       scores["rouge-l"]["p"]			,       scores["rouge-l"]["r"]			,       )
def 	snake_case_							(				_a  :				str      ):
      with open("rouge_scores.txt"			,       "w"      ) as output:
            output.write(_a      )
def 	snake_case_							(				_a  :				Tuple			,       _a  :				Union[str, Any]      ):
      snake_case_    							=				load_and_cache_examples(_a			,       _a      )
      snake_case_    							=				SequentialSampler(_a      )

      def collate_fn(_a  :				Tuple      ):
            return collate(_a			,       _a			,       block_size=512			,       device=args.device      )

      snake_case_    							=				DataLoader(
          _a			,       sampler=_a			,       batch_size=args.batch_size			,       collate_fn=_a			,       )

      return iterator
def 	snake_case_							(				_a  :				Optional[Any]			,       _a  :				Tuple      ):
      snake_case_    							=				CNNDMDataset(args.documents_dir      )
      return dataset
def 	snake_case_							(				_a  :				int			,       _a  :				int			,       _a  :				str			,       _a  :				List[Any]      ):
      snake_case_    							=				[x for x in data if not len(x[1]      ) == 0]  # remove empty_files
      snake_case_    							=				[name for name, _, _ in data]
      snake_case_    							=				[" ".join(_a      ) for _, _, summary_list in data]

      snake_case_    							=				[encode_for_summarization(_a			,       _a			,       _a      ) for _, story, summary in data]
      snake_case_    							=				torch.tensor(
          [truncate_or_pad(_a			,       _a			,       tokenizer.pad_token_id      ) for story, _ in encoded_text]      )
      snake_case_    							=				compute_token_type_ids(_a			,       tokenizer.cls_token_id      )
      snake_case_    							=				build_mask(_a			,       tokenizer.pad_token_id      )

      snake_case_    							=				Batch(
          document_names=_a			,       batch_size=len(_a      )			,       src=encoded_stories.to(_a      )			,       segs=encoder_token_type_ids.to(_a      )			,       mask_src=encoder_mask.to(_a      )			,       tgt_str=_a			,       )

      return batch
def 	snake_case_							(				_a  :				Tuple			,       _a  :				int      ):
      snake_case_    							=				summary_tokens.to("cpu"      ).numpy()
      snake_case_    							=				tokenizer.decode(_a      )
      snake_case_    							=				summary.split("."      )
      snake_case_    							=				[s + "." for s in sentences]
      return sentences
def 	snake_case_							(				):
      snake_case_    							=				argparse.ArgumentParser()
      parser.add_argument(
          "--documents_dir"			,       default=_a			,       type=_a			,       required=_a			,       help="The folder where the documents to summarize are located."			,       )
      parser.add_argument(
          "--summaries_output_dir"			,       default=_a			,       type=_a			,       required=_a			,       help="The folder in wich the summaries should be written. Defaults to the folder where the documents are"			,       )
      parser.add_argument(
          "--compute_rouge"			,       default=_a			,       type=_a			,       required=_a			,       help="Compute the ROUGE metrics during evaluation. Only available for the CNN/DailyMail dataset."			,       )
      # EVALUATION options
      parser.add_argument(
          "--no_cuda"			,       default=_a			,       type=_a			,       help="Whether to force the execution on CPU."			,       )
      parser.add_argument(
          "--batch_size"			,       default=4			,       type=_a			,       help="Batch size per GPU/CPU for training."			,       )
      # BEAM SEARCH arguments
      parser.add_argument(
          "--min_length"			,       default=50			,       type=_a			,       help="Minimum number of tokens for the summaries."			,       )
      parser.add_argument(
          "--max_length"			,       default=200			,       type=_a			,       help="Maixmum number of tokens for the summaries."			,       )
      parser.add_argument(
          "--beam_size"			,       default=5			,       type=_a			,       help="The number of beams to start with for each example."			,       )
      parser.add_argument(
          "--alpha"			,       default=0.9_5			,       type=_a			,       help="The value of alpha for the length penalty in the beam search."			,       )
      parser.add_argument(
          "--block_trigram"			,       default=_a			,       type=_a			,       help="Whether to block the existence of repeating trigrams in the text generated by beam search."			,       )
      snake_case_    							=				parser.parse_args()

      # Select device (distibuted not available)
      snake_case_    							=				torch.device("cuda" if torch.cuda.is_available() and not args.no_cuda else "cpu"      )

      # Check the existence of directories
      if not args.summaries_output_dir:
            snake_case_    							=				args.documents_dir

      if not documents_dir_is_valid(args.documents_dir      ):
            raise FileNotFoundError(
                "We could not find the directory you specified for the documents to summarize, or it was empty. Please"
                " specify a valid path."      )
      os.makedirs(args.summaries_output_dir			,       exist_ok=_a      )

      evaluate(_a      )
def 	snake_case_							(				_a  :				Optional[int]      ):
      if not os.path.exists(_a      ):
            return False

      snake_case_    							=				os.listdir(_a      )
      if len(_a      ) == 0:
            return False

      return True


if __name__ == "__main__":
     main()

