
import multiprocessing
from typing import TYPE_CHECKING, Optional, Union

from .. import Dataset, Features, config
from ..formatting import query_table
from ..packaged_modules.sql.sql import Sql
from ..utils import logging
from .abc import AbstractDatasetInputStream


if TYPE_CHECKING:
    import sqlitea

    import sqlalchemy




class 					_lowerCamelCase			(   UpperCamelCase_			):





     """simple docstring"""







     def __init__(				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE = None    ,    __SCREAMING_SNAKE_CASE = None    ,    __SCREAMING_SNAKE_CASE = False    ,    **__SCREAMING_SNAKE_CASE    ,    )	-> Tuple:




           """simple docstring"""







           super().__init__(features=__SCREAMING_SNAKE_CASE    ,    cache_dir=__SCREAMING_SNAKE_CASE    ,    keep_in_memory=__SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Optional[Any]											=					Sql(
               cache_dir=__SCREAMING_SNAKE_CASE    ,    features=__SCREAMING_SNAKE_CASE    ,    sql=__SCREAMING_SNAKE_CASE    ,    con=__SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE    ,    )






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Any:




           """simple docstring"""







           UpperCamelCase__     : Dict											=					None
           UpperCamelCase__     : Optional[int]											=					None
           UpperCamelCase__     : Any											=					None
           UpperCamelCase__     : str											=					None

           self.builder.download_and_prepare(
               download_config=__SCREAMING_SNAKE_CASE    ,    download_mode=__SCREAMING_SNAKE_CASE    ,    verification_mode=__SCREAMING_SNAKE_CASE    ,    base_path=__SCREAMING_SNAKE_CASE    ,    )

           # Build dataset for splits
           UpperCamelCase__     : str											=					self.builder.as_dataset(
               split='''train'''    ,    verification_mode=__SCREAMING_SNAKE_CASE    ,    in_memory=self.keep_in_memory	)
           return dataset




class 					_lowerCamelCase			:





     """simple docstring"""







     def __init__(				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE = None    ,    __SCREAMING_SNAKE_CASE = None    ,    **__SCREAMING_SNAKE_CASE    ,    )	-> Dict:




           """simple docstring"""







           if num_proc is not None and num_proc <= 0:
                 raise ValueError(F'''num_proc {num_proc} must be an integer > 0.'''	)

           UpperCamelCase__     : Optional[int]											=					dataset
           UpperCamelCase__     : Any											=					name
           UpperCamelCase__     : Dict											=					con
           UpperCamelCase__     : List[str]											=					batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
           UpperCamelCase__     : str											=					num_proc
           UpperCamelCase__     : Optional[int]											=					to_sql_kwargs






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> int:




           """simple docstring"""







           UpperCamelCase__     : List[str]											=					self.to_sql_kwargs.pop('''sql'''    ,    __SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Union[str, Any]											=					self.to_sql_kwargs.pop('''con'''    ,    __SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Dict											=					self.to_sql_kwargs.pop('''index'''    ,    __SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : Optional[int]											=					self._write(index=__SCREAMING_SNAKE_CASE    ,    **self.to_sql_kwargs	)
           return written






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE	)	-> Any:




           """simple docstring"""







           UpperCamelCase__					,UpperCamelCase__					,UpperCamelCase__     : Optional[int]											=					args
           UpperCamelCase__     : List[str]											=					{**to_sql_kwargs, '''if_exists''': '''append'''} if offset > 0 else to_sql_kwargs
           UpperCamelCase__     : str											=					query_table(
               table=self.dataset.data    ,    key=slice(__SCREAMING_SNAKE_CASE    ,    offset + self.batch_size	)    ,    indices=self.dataset._indices    ,    )
           UpperCamelCase__     : Any											=					batch.to_pandas()
           UpperCamelCase__     : Optional[int]											=					df.to_sql(self.name    ,    self.con    ,    index=__SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE	)
           return num_rows or len(__SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE	)	-> int:




           """simple docstring"""







           UpperCamelCase__     : Dict											=					0

           if self.num_proc is None or self.num_proc == 1:
                 for offset in logging.tqdm(
                     range(0    ,    len(self.dataset	)    ,    self.batch_size	)    ,    unit='''ba'''    ,    disable=not logging.is_progress_bar_enabled()    ,    desc='''Creating SQL from Arrow format'''    ,    ):
                       written += self._batch_sql((offset, index, to_sql_kwargs)	)
           else:
                 UpperCamelCase__					,UpperCamelCase__     : Any											=					len(self.dataset	), self.batch_size
                 with multiprocessing.Pool(self.num_proc	) as pool:
                       for num_rows in logging.tqdm(
                           pool.imap(
                               self._batch_sql    ,    [(offset, index, to_sql_kwargs) for offset in range(0    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE	)]    ,    )    ,    total=(num_rows // batch_size) + 1 if num_rows % batch_size else num_rows // batch_size    ,    unit='''ba'''    ,    disable=not logging.is_progress_bar_enabled()    ,    desc='''Creating SQL from Arrow format'''    ,    ):
                             written += num_rows

           return written




import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import is_speech_available, is_vision_available
from transformers.testing_utils import require_torch


if is_vision_available():
    from transformers import TvltImageProcessor

if is_speech_available():
    from transformers import TvltFeatureExtractor

from transformers import TvltProcessor




@require_torch
class 					_lowerCamelCase			(   unittest.TestCase			):





     """simple docstring"""







     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Union[str, Any]:




           """simple docstring"""







           UpperCamelCase__     : int											=					'''ZinengTang/tvlt-base'''
           UpperCamelCase__     : Optional[Any]											=					tempfile.mkdtemp()






     def  __SCREAMING_SNAKE_CASE      (				self    ,    **__SCREAMING_SNAKE_CASE	)	-> Tuple:




           """simple docstring"""







           return TvltImageProcessor.from_pretrained(self.checkpoint    ,    **__SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self    ,    **__SCREAMING_SNAKE_CASE	)	-> str:




           """simple docstring"""







           return TvltFeatureExtractor.from_pretrained(self.checkpoint    ,    **__SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Optional[int]:




           """simple docstring"""







           shutil.rmtree(self.tmpdirname	)






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Any:




           """simple docstring"""







           UpperCamelCase__     : Optional[Any]											=					self.get_image_processor()
           UpperCamelCase__     : Tuple											=					self.get_feature_extractor()

           UpperCamelCase__     : Dict											=					TvltProcessor(image_processor=__SCREAMING_SNAKE_CASE    ,    feature_extractor=__SCREAMING_SNAKE_CASE	)
           processor.save_pretrained(self.tmpdirname	)
           UpperCamelCase__     : Optional[Any]											=					TvltProcessor.from_pretrained(self.tmpdirname	)

           self.assertIsInstance(processor.feature_extractor    ,    __SCREAMING_SNAKE_CASE	)
           self.assertIsInstance(processor.image_processor    ,    __SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> List[str]:




           """simple docstring"""







           UpperCamelCase__     : Union[str, Any]											=					self.get_image_processor()
           UpperCamelCase__     : int											=					self.get_feature_extractor()

           UpperCamelCase__     : Union[str, Any]											=					TvltProcessor(image_processor=__SCREAMING_SNAKE_CASE    ,    feature_extractor=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : str											=					np.ones([1_2_0_0_0]	)

           UpperCamelCase__     : Dict											=					feature_extractor(__SCREAMING_SNAKE_CASE    ,    return_tensors='''np'''	)
           UpperCamelCase__     : Dict											=					processor(audio=__SCREAMING_SNAKE_CASE    ,    return_tensors='''np'''	)

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






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Optional[int]:




           """simple docstring"""







           UpperCamelCase__     : Tuple											=					self.get_image_processor()
           UpperCamelCase__     : Tuple											=					self.get_feature_extractor()

           UpperCamelCase__     : str											=					TvltProcessor(image_processor=__SCREAMING_SNAKE_CASE    ,    feature_extractor=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : List[Any]											=					np.ones([3, 2_2_4, 2_2_4]	)

           UpperCamelCase__     : Optional[int]											=					image_processor(__SCREAMING_SNAKE_CASE    ,    return_tensors='''np'''	)
           UpperCamelCase__     : Tuple											=					processor(images=__SCREAMING_SNAKE_CASE    ,    return_tensors='''np'''	)

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






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> str:




           """simple docstring"""







           UpperCamelCase__     : List[str]											=					self.get_image_processor()
           UpperCamelCase__     : Optional[Any]											=					self.get_feature_extractor()

           UpperCamelCase__     : Union[str, Any]											=					TvltProcessor(image_processor=__SCREAMING_SNAKE_CASE    ,    feature_extractor=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : Dict											=					np.ones([1_2_0_0_0]	)
           UpperCamelCase__     : str											=					np.ones([3, 2_2_4, 2_2_4]	)

           UpperCamelCase__     : List[Any]											=					processor(audio=__SCREAMING_SNAKE_CASE    ,    images=__SCREAMING_SNAKE_CASE	)

           self.assertListEqual(list(inputs.keys()	)    ,    ['''audio_values''', '''audio_mask''', '''pixel_values''', '''pixel_mask''']	)

           # test if it raises when no input is passed
           with pytest.raises(__SCREAMING_SNAKE_CASE	):
                 processor()






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> List[Any]:




           """simple docstring"""







           UpperCamelCase__     : int											=					self.get_image_processor()
           UpperCamelCase__     : str											=					self.get_feature_extractor()

           UpperCamelCase__     : Dict											=					TvltProcessor(image_processor=__SCREAMING_SNAKE_CASE    ,    feature_extractor=__SCREAMING_SNAKE_CASE	)

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


