

import itertools
from dataclasses import dataclass
from typing import Optional

import pyarrow as pa

import datasets
from datasets.table import table_cast


lowercase_     =       datasets.utils.logging.get_logger(__name__)





@dataclass
class 			__lowerCAmelCase					( datasets.BuilderConfig							):
  _a						      =  None





class 			__lowerCAmelCase					( datasets.ArrowBasedBuilder							):
  _a						      =  ArrowConfig




  def   A__					(    self )  ->			Optional[Any]:
        '''simple docstring'''






        return datasets.DatasetInfo(features=self.config.features )




  def   A__					(    self							,       lowerCAmelCase )  ->			Any:
        '''simple docstring'''






        if not self.config.data_files:
              raise ValueError(F'''At least one data file must be specified, but got data_files={self.config.data_files}''' )
        _lowercase     	=dl_manager.download_and_extract(self.config.data_files )
        if isinstance(_snake_case							,       (str, list, tuple) ):
              _lowercase     	=data_files
              if isinstance(_snake_case							,       _snake_case ):
                    _lowercase     	=[files]
              # Use `dl_manager.iter_files` to skip hidden files in an extracted archive
              _lowercase     	=[dl_manager.iter_files(_snake_case ) for file in files]
              return [datasets.SplitGenerator(name=datasets.Split.TRAIN							,       gen_kwargs={'files': files} )]
        _lowercase     	=[]
        for split_name, files in data_files.items():
              if isinstance(_snake_case							,       _snake_case ):
                    _lowercase     	=[files]
              # Use `dl_manager.iter_files` to skip hidden files in an extracted archive
              _lowercase     	=[dl_manager.iter_files(_snake_case ) for file in files]
              # Infer features is they are stoed in the arrow schema
              if self.info.features is None:
                    for file in itertools.chain.from_iterable(_snake_case ):
                          with open(_snake_case							,       'rb' ) as f:
                                _lowercase     	=datasets.Features.from_arrow_schema(pa.ipc.open_stream(_snake_case ).schema )
                          break
              splits.append(datasets.SplitGenerator(name=_snake_case							,       gen_kwargs={'files': files} ) )
        return splits




  def   A__					(    self							,       lowerCAmelCase )  ->			pa.Table:
        '''simple docstring'''






        if self.info.features is not None:
              # more expensive cast to support nested features with keys in a different order
              # allows str <-> int/float or str to Audio for example
              _lowercase     	=table_cast(_snake_case							,       self.info.features.arrow_schema )
        return pa_table




  def   A__					(    self							,       lowerCAmelCase )  ->			Dict:
        '''simple docstring'''






        for file_idx, file in enumerate(itertools.chain.from_iterable(_snake_case ) ):
              with open(_snake_case							,       'rb' ) as f:
                    try:
                          for batch_idx, record_batch in enumerate(pa.ipc.open_stream(_snake_case ) ):
                                _lowercase     	=pa.Table.from_batches([record_batch] )
                                # Uncomment for debugging (will print the Arrow table size and elements)
                                # logger.warning(f"pa_table: {pa_table} num rows: {pa_table.num_rows}")
                                # logger.warning('\n'.join(str(pa_table.slice(i, 1).to_pydict()) for i in range(pa_table.num_rows)))
                                yield F'''{file_idx}_{batch_idx}''', self._cast_table(_snake_case )
                    except ValueError as e:
                          logger.error(F'''Failed to read file \'{file}\' with error {type(_snake_case )}: {e}''' )
                          raise





import unittest

from cer import CER


lowercase_     =       CER()





class 			__lowerCAmelCase					( unittest.TestCase							):




  def   A__					(    self )  ->			Any:
        '''simple docstring'''






        _lowercase     	=['White House']
        _lowercase     	=['white house']
        # S = 2, D = 0, I = 0, N = 11, CER = 2 / 11
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.1818181818 ) < 1e-6 )




  def   A__					(    self )  ->			Dict:
        '''simple docstring'''






        _lowercase     	=['were wolf']
        _lowercase     	=['werewolf']
        # S = 0, D = 0, I = 1, N = 9, CER = 1 / 9
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.1111111 ) < 1e-6 )

        _lowercase     	=['werewolf']
        _lowercase     	=['weae     wolf']
        # S = 1, D = 1, I = 0, N = 8, CER = 0.25
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.25 ) < 1e-6 )

        # consecutive whitespaces case 1
        _lowercase     	=['were wolf']
        _lowercase     	=['were               wolf']
        # S = 0, D = 0, I = 0, N = 9, CER = 0
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.0 ) < 1e-6 )

        # consecutive whitespaces case 2
        _lowercase     	=['were   wolf']
        _lowercase     	=['were               wolf']
        # S = 0, D = 0, I = 0, N = 9, CER = 0
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.0 ) < 1e-6 )




  def   A__					(    self )  ->			Optional[Any]:
        '''simple docstring'''






        _lowercase     	=['werewolf']
        _lowercase     	=['weaewolf']
        # S = 1, D = 0, I = 0, N = 8, CER = 0.125
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.125 ) < 1e-6 )




  def   A__					(    self )  ->			Union[str, Any]:
        '''simple docstring'''






        _lowercase     	=['werewolf']
        _lowercase     	=['wereawolf']
        # S = 0, D = 1, I = 0, N = 8, CER = 0.125
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.125 ) < 1e-6 )




  def   A__					(    self )  ->			Any:
        '''simple docstring'''






        _lowercase     	=['werewolf']
        _lowercase     	=['wereolf']
        # S = 0, D = 0, I = 1, N = 8, CER = 0.125
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.125 ) < 1e-6 )




  def   A__					(    self )  ->			Any:
        '''simple docstring'''






        _lowercase     	=['werewolf']
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertEqual(lowerCAmelCase							,       0.0 )




  def   A__					(    self )  ->			List[Any]:
        '''simple docstring'''






        _lowercase     	=['werewolf', 'I am your father']
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertEqual(lowerCAmelCase							,       0.0 )

        _lowercase     	=['werewolf', 'I am your father', 'doge']
        _lowercase     	=['werxwolf', 'I       am your father', 'doge']
        # S = 1, D = 0, I = 0, N = 28, CER = 1 / 28
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.03571428 ) < 1e-6 )




  def   A__					(    self )  ->			str:
        '''simple docstring'''






        _lowercase     	=['My hovercraft', 'is full of eels']
        _lowercase     	=['My hovercraft is full', ' of eels']
        # S = 0, D = 0, I = 2, N = 28, CER = 2 / 28
        # whitespace at the front of " of eels" will be strip during preporcessing
        # so need to insert 2 whitespaces
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase							,       concatenate_texts=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.071428 ) < 1e-6 )




  def   A__					(    self )  ->			Union[str, Any]:
        '''simple docstring'''






        _lowercase     	=['我能吞下玻璃而不伤身体']
        _lowercase     	=[' 能吞虾玻璃而 不霜身体啦']
        # S = 3, D = 2, I = 0, N = 11, CER = 5 / 11
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.4545454545 ) < 1e-6 )

        _lowercase     	=['我能吞下玻璃', '而不伤身体']
        _lowercase     	=['我    能 吞 下 玻 璃', '而不伤身体']
        # S = 0, D = 5, I = 0, N = 11, CER = 5 / 11
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.454545454545 ) < 1e-6 )

        _lowercase     	=['我能吞下玻璃而不伤身体']
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertFalse(lowerCAmelCase							,       0.0 )




  def   A__					(    self )  ->			List[str]:
        '''simple docstring'''






        _lowercase     	=['']
        _lowercase     	=['Hypothesis']
        with self.assertRaises(lowerCAmelCase ):
              cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )


if __name__ == "__main__":
  unittest.main()


