






'''simple docstring'''




import contextlib
import csv
import json
import os
import sqlitea
import tarfile
import textwrap
import zipfile

import pyarrow as pa
import pyarrow.parquet as pq
import pytest

import datasets
import datasets.config







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  )							->     int:
		UpperCamelCase__				:					Tuple					=					10
		UpperCamelCase__				:					List[str]					=					datasets.Features(
		    {
		        'tokens': datasets.Sequence(datasets.Value('string')),
		        'labels': datasets.Sequence(datasets.ClassLabel(names=['negative', 'positive'])),
		        'answers': datasets.Sequence(
		            {
		                'text': datasets.Value('string'),
		                'answer_start': datasets.Value('int32'),
		            }),
		        'id': datasets.Value('int64'),
		    })
		UpperCamelCase__				:					Tuple					=					datasets.Dataset.from_dict(
		    {
		        'tokens': [['foo'] * 5] * n,
		        'labels': [[1] * 5] * n,
		        'answers': [{'answer_start': [97], 'text': ['1976']}] * 10,
		        'id': list(range(lowerCamelCase_)),
		    }     ,      features=lowerCamelCase_     ,      )
		return dataset







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_)							->     Dict:
		UpperCamelCase__				:					str					=					str(tmp_path_factory.mktemp('data') / 'file.arrow')
		dataset.map(cache_file_name=lowerCamelCase_)
		return filename


# FILE_CONTENT + files


lowerCAmelCase__          =						"""\
    Text data.
    Second line of data."""







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Union[str, Any]:
		UpperCamelCase__				:					Tuple					=					tmp_path_factory.mktemp('data') / 'file.txt'
		UpperCamelCase__				:					Tuple					=					FILE_CONTENT
		with open(lowerCamelCase_     ,      'w') as f:
				f.write(lowerCamelCase_)
		return filename







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Optional[int]:
		import bza

		UpperCamelCase__				:					Union[str, Any]					=					tmp_path_factory.mktemp('data') / 'file.txt.bz2'
		UpperCamelCase__				:					Optional[int]					=					bytes(lowerCamelCase_     ,      'utf-8')
		with bza.open(lowerCamelCase_     ,      'wb') as f:
				f.write(lowerCamelCase_)
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Union[str, Any]:
		import gzip

		UpperCamelCase__				:					List[str]					=					str(tmp_path_factory.mktemp('data') / 'file.txt.gz')
		UpperCamelCase__				:					Dict					=					bytes(lowerCamelCase_     ,      'utf-8')
		with gzip.open(lowerCamelCase_     ,      'wb') as f:
				f.write(lowerCamelCase_)
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Any:







		if datasets.config.LZ4_AVAILABLE:
				import lza.frame

				UpperCamelCase__				:					int					=					tmp_path_factory.mktemp('data') / 'file.txt.lz4'
				UpperCamelCase__				:					Union[str, Any]					=					bytes(lowerCamelCase_     ,      'utf-8')
				with lza.frame.open(lowerCamelCase_     ,      'wb') as f:
						f.write(lowerCamelCase_)
				return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_)							->     List[str]:







		if datasets.config.PY7ZR_AVAILABLE:
				import pyazr

				UpperCamelCase__				:					Optional[int]					=					tmp_path_factory.mktemp('data') / 'file.txt.7z'
				with pyazr.SevenZipFile(lowerCamelCase_     ,      'w') as archive:
						archive.write(lowerCamelCase_     ,      arcname=os.path.basename(lowerCamelCase_))
				return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_)							->     int:
		import tarfile

		UpperCamelCase__				:					List[Any]					=					tmp_path_factory.mktemp('data') / 'file.txt.tar'
		with tarfile.TarFile(lowerCamelCase_     ,      'w') as f:
				f.add(lowerCamelCase_     ,      arcname=os.path.basename(lowerCamelCase_))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     str:
		import lzma

		UpperCamelCase__				:					Optional[Any]					=					tmp_path_factory.mktemp('data') / 'file.txt.xz'
		UpperCamelCase__				:					Tuple					=					bytes(lowerCamelCase_     ,      'utf-8')
		with lzma.open(lowerCamelCase_     ,      'wb') as f:
				f.write(lowerCamelCase_)
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_)							->     List[str]:
		import zipfile

		UpperCamelCase__				:					Tuple					=					tmp_path_factory.mktemp('data') / 'file.txt.zip'
		with zipfile.ZipFile(lowerCamelCase_     ,      'w') as f:
				f.write(lowerCamelCase_     ,      arcname=os.path.basename(lowerCamelCase_))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Tuple:







		if datasets.config.ZSTANDARD_AVAILABLE:
				import zstandard as zstd

				UpperCamelCase__				:					Union[str, Any]					=					tmp_path_factory.mktemp('data') / 'file.txt.zst'
				UpperCamelCase__				:					Tuple					=					bytes(lowerCamelCase_     ,      'utf-8')
				with zstd.open(lowerCamelCase_     ,      'wb') as f:
						f.write(lowerCamelCase_)
				return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Optional[int]:
		UpperCamelCase__				:					Optional[Any]					=					tmp_path_factory.mktemp('data') / 'file.xml'
		UpperCamelCase__				:					Any					=					textwrap.dedent(
		    '\\n    <?xml version=\"1.0\" encoding=\"UTF-8\" ?>\n    <tmx version=\"1.4\">\n      <header segtype=\"sentence\" srclang=\"ca\" />\n      <body>\n        <tu>\n          <tuv xml:lang=\"ca\"><seg>Contingut 1</seg></tuv>\n          <tuv xml:lang=\"en\"><seg>Content 1</seg></tuv>\n        </tu>\n        <tu>\n          <tuv xml:lang=\"ca\"><seg>Contingut 2</seg></tuv>\n          <tuv xml:lang=\"en\"><seg>Content 2</seg></tuv>\n        </tu>\n        <tu>\n          <tuv xml:lang=\"ca\"><seg>Contingut 3</seg></tuv>\n          <tuv xml:lang=\"en\"><seg>Content 3</seg></tuv>\n        </tu>\n        <tu>\n          <tuv xml:lang=\"ca\"><seg>Contingut 4</seg></tuv>\n          <tuv xml:lang=\"en\"><seg>Content 4</seg></tuv>\n        </tu>\n        <tu>\n          <tuv xml:lang=\"ca\"><seg>Contingut 5</seg></tuv>\n          <tuv xml:lang=\"en\"><seg>Content 5</seg></tuv>\n        </tu>\n      </body>\n    </tmx>')
		with open(lowerCamelCase_     ,      'w') as f:
				f.write(lowerCamelCase_)
		return filename


lowerCAmelCase__          =						[
    {"""col_1""": """0""", """col_2""": 0, """col_3""": 0.0},
    {"""col_1""": """1""", """col_2""": 1, """col_3""": 1.0},
    {"""col_1""": """2""", """col_2""": 2, """col_3""": 2.0},
    {"""col_1""": """3""", """col_2""": 3, """col_3""": 3.0},
]
lowerCAmelCase__          =						[
    {"""col_1""": """4""", """col_2""": 4, """col_3""": 4.0},
    {"""col_1""": """5""", """col_2""": 5, """col_3""": 5.0},
]
lowerCAmelCase__          =						{
    """col_1""": ["""0""", """1""", """2""", """3"""],
    """col_2""": [0, 1, 2, 3],
    """col_3""": [0.0, 1.0, 2.0, 3.0],
}

lowerCAmelCase__          =						[
    {"""col_3""": 0.0, """col_1""": """0""", """col_2""": 0},
    {"""col_3""": 1.0, """col_1""": """1""", """col_2""": 1},
]

lowerCAmelCase__          =						[
    {"""col_1""": """s0""", """col_2""": 0, """col_3""": 0.0},
    {"""col_1""": """s1""", """col_2""": 1, """col_3""": 1.0},
    {"""col_1""": """s2""", """col_2""": 2, """col_3""": 2.0},
    {"""col_1""": """s3""", """col_2""": 3, """col_3""": 3.0},
]







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  )							->     Optional[int]:
		return DATA_DICT_OF_LISTS







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Dict:
		UpperCamelCase__				:					Optional[int]					=					datasets.Dataset.from_dict(lowerCamelCase_)
		UpperCamelCase__				:					Optional[int]					=					str(tmp_path_factory.mktemp('data') / 'dataset.arrow')
		dataset.map(cache_file_name=lowerCamelCase_)
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Tuple:
		UpperCamelCase__				:					List[Any]					=					str(tmp_path_factory.mktemp('data') / 'dataset.sqlite')
		with contextlib.closing(sqlitea.connect(lowerCamelCase_)) as con:
				UpperCamelCase__				:					List[Any]					=					con.cursor()
				cur.execute('CREATE TABLE dataset(col_1 text, col_2 int, col_3 real)')
				for item in DATA:
						cur.execute('INSERT INTO dataset(col_1, col_2, col_3) VALUES (?, ?, ?)'     ,      tuple(item.values()))
				con.commit()
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     str:
		UpperCamelCase__				:					List[Any]					=					str(tmp_path_factory.mktemp('data') / 'dataset.csv')
		with open(lowerCamelCase_     ,      'w'     ,      newline='') as f:
				UpperCamelCase__				:					Tuple					=					csv.DictWriter(lowerCamelCase_     ,      fieldnames=['col_1', 'col_2', 'col_3'])
				writer.writeheader()
				for item in DATA:
						writer.writerow(lowerCamelCase_)
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Optional[int]:
		UpperCamelCase__				:					List[Any]					=					str(tmp_path_factory.mktemp('data') / 'dataset2.csv')
		with open(lowerCamelCase_     ,      'w'     ,      newline='') as f:
				UpperCamelCase__				:					Optional[int]					=					csv.DictWriter(lowerCamelCase_     ,      fieldnames=['col_1', 'col_2', 'col_3'])
				writer.writeheader()
				for item in DATA:
						writer.writerow(lowerCamelCase_)
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_)							->     List[str]:
		import bza

		UpperCamelCase__				:					int					=					tmp_path_factory.mktemp('data') / 'dataset.csv.bz2'
		with open(lowerCamelCase_     ,      'rb') as f:
				UpperCamelCase__				:					Any					=					f.read()
		# data = bytes(FILE_CONTENT, "utf-8")
		with bza.open(lowerCamelCase_     ,      'wb') as f:
				f.write(lowerCamelCase_)
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_)							->     Any:
		UpperCamelCase__				:					Union[str, Any]					=					tmp_path_factory.mktemp('data') / 'dataset.csv.zip'
		with zipfile.ZipFile(lowerCamelCase_     ,      'w') as f:
				f.write(lowerCamelCase_     ,      arcname=os.path.basename(lowerCamelCase_))
				f.write(lowerCamelCase_     ,      arcname=os.path.basename(lowerCamelCase_))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_)							->     Any:
		UpperCamelCase__				:					Optional[Any]					=					tmp_path_factory.mktemp('data') / 'dataset.csv.zip'
		with zipfile.ZipFile(lowerCamelCase_     ,      'w') as f:
				f.write(lowerCamelCase_     ,      arcname=os.path.basename(csv_path.replace('.csv'     ,      '.CSV')))
				f.write(lowerCamelCase_     ,      arcname=os.path.basename(csva_path.replace('.csv'     ,      '.CSV')))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_)							->     int:
		UpperCamelCase__				:					Optional[int]					=					tmp_path_factory.mktemp('data') / 'dataset_with_dir.csv.zip'
		with zipfile.ZipFile(lowerCamelCase_     ,      'w') as f:
				f.write(lowerCamelCase_     ,      arcname=os.path.join('main_dir'     ,      os.path.basename(lowerCamelCase_)))
				f.write(lowerCamelCase_     ,      arcname=os.path.join('main_dir'     ,      os.path.basename(lowerCamelCase_)))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Tuple:
		UpperCamelCase__				:					List[Any]					=					str(tmp_path_factory.mktemp('data') / 'dataset.parquet')
		UpperCamelCase__				:					str					=					pa.schema(
		    {
		        'col_1': pa.string(),
		        'col_2': pa.intaa(),
		        'col_3': pa.floataa(),
		    })
		with open(lowerCamelCase_     ,      'wb') as f:
				UpperCamelCase__				:					Dict					=					pq.ParquetWriter(lowerCamelCase_     ,      schema=lowerCamelCase_)
				UpperCamelCase__				:					Any					=					pa.Table.from_pydict({k: [DATA[i][k] for i in range(len(lowerCamelCase_))] for k in DATA[0]}     ,      schema=lowerCamelCase_)
				writer.write_table(lowerCamelCase_)
				writer.close()
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     List[Any]:
		UpperCamelCase__				:					Union[str, Any]					=					str(tmp_path_factory.mktemp('data') / 'dataset.json')
		UpperCamelCase__				:					List[str]					=					{'data': DATA}
		with open(lowerCamelCase_     ,      'w') as f:
				json.dump(lowerCamelCase_     ,      lowerCamelCase_)
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Optional[int]:
		UpperCamelCase__				:					List[Any]					=					str(tmp_path_factory.mktemp('data') / 'dataset.json')
		UpperCamelCase__				:					Dict					=					{'data': DATA_DICT_OF_LISTS}
		with open(lowerCamelCase_     ,      'w') as f:
				json.dump(lowerCamelCase_     ,      lowerCamelCase_)
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Tuple:
		UpperCamelCase__				:					int					=					str(tmp_path_factory.mktemp('data') / 'dataset.jsonl')
		with open(lowerCamelCase_     ,      'w') as f:
				for item in DATA:
						f.write(json.dumps(lowerCamelCase_) + '\n')
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     List[str]:
		UpperCamelCase__				:					Union[str, Any]					=					str(tmp_path_factory.mktemp('data') / 'dataset2.jsonl')
		with open(lowerCamelCase_     ,      'w') as f:
				for item in DATA:
						f.write(json.dumps(lowerCamelCase_) + '\n')
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Dict:
		UpperCamelCase__				:					Tuple					=					str(tmp_path_factory.mktemp('data') / 'dataset_312.jsonl')
		with open(lowerCamelCase_     ,      'w') as f:
				for item in DATA_312:
						f.write(json.dumps(lowerCamelCase_) + '\n')
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     List[str]:
		UpperCamelCase__				:					str					=					str(tmp_path_factory.mktemp('data') / 'dataset-str.jsonl')
		with open(lowerCamelCase_     ,      'w') as f:
				for item in DATA_STR:
						f.write(json.dumps(lowerCamelCase_) + '\n')
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_)							->     Tuple:
		import gzip

		UpperCamelCase__				:					int					=					str(tmp_path_factory.mktemp('data') / 'dataset.txt.gz')
		with open(lowerCamelCase_     ,      'rb') as orig_file:
				with gzip.open(lowerCamelCase_     ,      'wb') as zipped_file:
						zipped_file.writelines(lowerCamelCase_)
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_)							->     int:
		import gzip

		UpperCamelCase__				:					Any					=					str(tmp_path_factory.mktemp('data') / 'dataset.jsonl.gz')
		with open(lowerCamelCase_     ,      'rb') as orig_file:
				with gzip.open(lowerCamelCase_     ,      'wb') as zipped_file:
						zipped_file.writelines(lowerCamelCase_)
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_)							->     Tuple:
		UpperCamelCase__				:					int					=					tmp_path_factory.mktemp('data') / 'dataset.jsonl.zip'
		with zipfile.ZipFile(lowerCamelCase_     ,      'w') as f:
				f.write(lowerCamelCase_     ,      arcname=os.path.basename(lowerCamelCase_))
				f.write(lowerCamelCase_     ,      arcname=os.path.basename(lowerCamelCase_))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_)							->     int:
		UpperCamelCase__				:					str					=					tmp_path_factory.mktemp('data') / 'dataset_nested.jsonl.zip'
		with zipfile.ZipFile(lowerCamelCase_     ,      'w') as f:
				f.write(lowerCamelCase_     ,      arcname=os.path.join('nested'     ,      os.path.basename(lowerCamelCase_)))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_)							->     str:
		UpperCamelCase__				:					Optional[Any]					=					tmp_path_factory.mktemp('data') / 'dataset_with_dir.jsonl.zip'
		with zipfile.ZipFile(lowerCamelCase_     ,      'w') as f:
				f.write(lowerCamelCase_     ,      arcname=os.path.join('main_dir'     ,      os.path.basename(lowerCamelCase_)))
				f.write(lowerCamelCase_     ,      arcname=os.path.join('main_dir'     ,      os.path.basename(lowerCamelCase_)))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_)							->     str:
		UpperCamelCase__				:					Union[str, Any]					=					tmp_path_factory.mktemp('data') / 'dataset.jsonl.tar'
		with tarfile.TarFile(lowerCamelCase_     ,      'w') as f:
				f.add(lowerCamelCase_     ,      arcname=os.path.basename(lowerCamelCase_))
				f.add(lowerCamelCase_     ,      arcname=os.path.basename(lowerCamelCase_))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_)							->     int:
		UpperCamelCase__				:					int					=					tmp_path_factory.mktemp('data') / 'dataset_nested.jsonl.tar'
		with tarfile.TarFile(lowerCamelCase_     ,      'w') as f:
				f.add(lowerCamelCase_     ,      arcname=os.path.join('nested'     ,      os.path.basename(lowerCamelCase_)))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Tuple:
		UpperCamelCase__				:					List[str]					=					['0', '1', '2', '3']
		UpperCamelCase__				:					Dict					=					str(tmp_path_factory.mktemp('data') / 'dataset.txt')
		with open(lowerCamelCase_     ,      'w') as f:
				for item in data:
						f.write(item + '\n')
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     List[str]:
		UpperCamelCase__				:					List[Any]					=					['0', '1', '2', '3']
		UpperCamelCase__				:					Any					=					str(tmp_path_factory.mktemp('data') / 'dataset2.txt')
		with open(lowerCamelCase_     ,      'w') as f:
				for item in data:
						f.write(item + '\n')
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Any:
		UpperCamelCase__				:					str					=					['0', '1', '2', '3']
		UpperCamelCase__				:					Tuple					=					tmp_path_factory.mktemp('data') / 'dataset.abc'
		with open(lowerCamelCase_     ,      'w') as f:
				for item in data:
						f.write(item + '\n')
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_)							->     Optional[int]:
		UpperCamelCase__				:					int					=					tmp_path_factory.mktemp('data') / 'dataset.text.zip'
		with zipfile.ZipFile(lowerCamelCase_     ,      'w') as f:
				f.write(lowerCamelCase_     ,      arcname=os.path.basename(lowerCamelCase_))
				f.write(lowerCamelCase_     ,      arcname=os.path.basename(lowerCamelCase_))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_)							->     List[Any]:
		UpperCamelCase__				:					Optional[Any]					=					tmp_path_factory.mktemp('data') / 'dataset_with_dir.text.zip'
		with zipfile.ZipFile(lowerCamelCase_     ,      'w') as f:
				f.write(lowerCamelCase_     ,      arcname=os.path.join('main_dir'     ,      os.path.basename(lowerCamelCase_)))
				f.write(lowerCamelCase_     ,      arcname=os.path.join('main_dir'     ,      os.path.basename(lowerCamelCase_)))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_)							->     List[str]:
		UpperCamelCase__				:					Union[str, Any]					=					tmp_path_factory.mktemp('data') / 'dataset.ext.zip'
		with zipfile.ZipFile(lowerCamelCase_     ,      'w') as f:
				f.write(lowerCamelCase_     ,      arcname=os.path.basename('unsupported.ext'))
				f.write(lowerCamelCase_     ,      arcname=os.path.basename('unsupported_2.ext'))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Dict:
		UpperCamelCase__				:					Optional[int]					=					'\n'.join(['First', 'Second\u2029with Unicode new line', 'Third'])
		UpperCamelCase__				:					str					=					str(tmp_path_factory.mktemp('data') / 'dataset_with_unicode_new_lines.txt')
		with open(lowerCamelCase_     ,      'w'     ,      encoding='utf-8') as f:
				f.write(lowerCamelCase_)
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  )							->     List[Any]:
		return os.path.join('tests'     ,      'features'     ,      'data'     ,      'test_image_rgb.jpg')







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  )							->     Tuple:
		return os.path.join('tests'     ,      'features'     ,      'data'     ,      'test_audio_44100.wav')







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_)							->     Any:
		UpperCamelCase__				:					Tuple					=					tmp_path_factory.mktemp('data') / 'dataset.img.zip'
		with zipfile.ZipFile(lowerCamelCase_     ,      'w') as f:
				f.write(lowerCamelCase_     ,      arcname=os.path.basename(lowerCamelCase_))
				f.write(lowerCamelCase_     ,      arcname=os.path.basename(lowerCamelCase_).replace('.jpg'     ,      '2.jpg'))
		return path







@pytest.fixture(scope='session')
def       __UpperCAmelCase						(  lowerCamelCase_)							->     Union[str, Any]:
		UpperCamelCase__				:					Dict					=					tmp_path_factory.mktemp('data_dir')

		(data_dir / "subdir").mkdir()
		with open(data_dir / 'subdir' / 'train.txt'     ,      'w') as f:
				f.write('foo\n' * 10)
		with open(data_dir / 'subdir' / 'test.txt'     ,      'w') as f:
				f.write('bar\n' * 10)
		# hidden file
		with open(data_dir / 'subdir' / '.test.txt'     ,      'w') as f:
				f.write('bar\n' * 10)

		# hidden directory
		(data_dir / ".subdir").mkdir()
		with open(data_dir / '.subdir' / 'train.txt'     ,      'w') as f:
				f.write('foo\n' * 10)
		with open(data_dir / '.subdir' / 'test.txt'     ,      'w') as f:
				f.write('bar\n' * 10)

		return data_dir












'''simple docstring'''




import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import TransformeraDModel, VQDiffusionPipeline, VQDiffusionScheduler, VQModel
from diffusers.pipelines.vq_diffusion.pipeline_vq_diffusion import LearnedClassifierFreeSamplingEmbeddings
from diffusers.utils import load_numpy, slow, torch_device
from diffusers.utils.testing_utils import require_torch_gpu


lowerCAmelCase__          =						False







class       __lowercase					(unittest.TestCase    ):

	def      __UpperCamelCase							(      self   :						Optional[Any]):
			# clean up the VRAM after each test
			super().tearDown()
			gc.collect()
			torch.cuda.empty_cache()

	@property
	def      __UpperCamelCase							(      self   :						int):
			return 12

	@property
	def      __UpperCamelCase							(      self   :						Tuple):
			return 12

	@property
	def      __UpperCamelCase							(      self   :						Dict):
			return 32

	@property
	def      __UpperCamelCase							(      self   :						Optional[int]):
			torch.manual_seed(0)
			UpperCamelCase__				:					List[Any]					=					VQModel(
			    block_out_channels=[32, 64]  ,					in_channels=3  ,					out_channels=3  ,					down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']  ,					up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']  ,					latent_channels=3  ,					num_vq_embeddings=self.num_embed  ,					vq_embed_dim=3  ,					)
			return model

	@property
	def      __UpperCamelCase							(      self   :						Optional[Any]):
			UpperCamelCase__				:					Any					=					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')
			return tokenizer

	@property
	def      __UpperCamelCase							(      self   :						List[str]):
			torch.manual_seed(0)
			UpperCamelCase__				:					Optional[int]					=					CLIPTextConfig(
			    bos_token_id=0  ,					eos_token_id=2  ,					hidden_size=self.text_embedder_hidden_size  ,					intermediate_size=37  ,					layer_norm_eps=1e-05  ,					num_attention_heads=4  ,					num_hidden_layers=5  ,					pad_token_id=1  ,					vocab_size=1_000  ,					)
			return CLIPTextModel(UpperCAmelCase_)

	@property
	def      __UpperCamelCase							(      self   :						Optional[int]):
			torch.manual_seed(0)

			UpperCamelCase__				:					List[Any]					=					12
			UpperCamelCase__				:					Dict					=					12

			UpperCamelCase__				:					Union[str, Any]					=					{
			    'attention_bias': True,
			    'cross_attention_dim': 32,
			    'attention_head_dim': height * width,
			    'num_attention_heads': 1,
			    'num_vector_embeds': self.num_embed,
			    'num_embeds_ada_norm': self.num_embeds_ada_norm,
			    'norm_num_groups': 32,
			    'sample_size': width,
			    'activation_fn': 'geglu-approximate',
			}

			UpperCamelCase__				:					Tuple					=					TransformeraDModel(**UpperCAmelCase_)
			return model

	def      __UpperCamelCase							(      self   :						int):
			UpperCamelCase__				:					List[Any]					=					'cpu'

			UpperCamelCase__				:					List[str]					=					self.dummy_vqvae
			UpperCamelCase__				:					List[str]					=					self.dummy_text_encoder
			UpperCamelCase__				:					Optional[int]					=					self.dummy_tokenizer
			UpperCamelCase__				:					List[str]					=					self.dummy_transformer
			UpperCamelCase__				:					Dict					=					VQDiffusionScheduler(self.num_embed)
			UpperCamelCase__				:					List[Any]					=					LearnedClassifierFreeSamplingEmbeddings(learnable=UpperCAmelCase_)

			UpperCamelCase__				:					int					=					VQDiffusionPipeline(
			    vqvae=UpperCAmelCase_  ,					text_encoder=UpperCAmelCase_  ,					tokenizer=UpperCAmelCase_  ,					transformer=UpperCAmelCase_  ,					scheduler=UpperCAmelCase_  ,					learned_classifier_free_sampling_embeddings=UpperCAmelCase_  ,					)
			UpperCamelCase__				:					Optional[Any]					=					pipe.to(UpperCAmelCase_)
			pipe.set_progress_bar_config(disable=UpperCAmelCase_)

			UpperCamelCase__				:					Optional[Any]					=					'teddy bear playing in the pool'

			UpperCamelCase__				:					Dict					=					torch.Generator(device=UpperCAmelCase_).manual_seed(0)
			UpperCamelCase__				:					Any					=					pipe([prompt]  ,					generator=UpperCAmelCase_  ,					num_inference_steps=2  ,					output_type='np')
			UpperCamelCase__				:					Optional[Any]					=					output.images

			UpperCamelCase__				:					int					=					torch.Generator(device=UpperCAmelCase_).manual_seed(0)
			UpperCamelCase__				:					Any					=					pipe(
			    [prompt]  ,					generator=UpperCAmelCase_  ,					output_type='np'  ,					return_dict=UpperCAmelCase_  ,					num_inference_steps=2)[0]

			UpperCamelCase__				:					Optional[Any]					=					image[0, -3:, -3:, -1]
			UpperCamelCase__				:					Any					=					image_from_tuple[0, -3:, -3:, -1]

			assert image.shape == (1, 24, 24, 3)

			UpperCamelCase__				:					Any					=					np.array([0.65_51, 0.61_68, 0.50_08, 0.56_76, 0.56_59, 0.42_95, 0.60_73, 0.55_99, 0.49_92])

			assert np.abs(image_slice.flatten() - expected_slice).max() < 1e-2
			assert np.abs(image_from_tuple_slice.flatten() - expected_slice).max() < 1e-2





	def      __UpperCamelCase							(      self   :						Optional[int]):
			UpperCamelCase__				:					Optional[int]					=					'cpu'

			UpperCamelCase__				:					str					=					self.dummy_vqvae
			UpperCamelCase__				:					Any					=					self.dummy_text_encoder
			UpperCamelCase__				:					List[Any]					=					self.dummy_tokenizer
			UpperCamelCase__				:					Dict					=					self.dummy_transformer
			UpperCamelCase__				:					Optional[Any]					=					VQDiffusionScheduler(self.num_embed)
			UpperCamelCase__				:					Optional[Any]					=					LearnedClassifierFreeSamplingEmbeddings(
			    learnable=UpperCAmelCase_  ,					hidden_size=self.text_embedder_hidden_size  ,					length=tokenizer.model_max_length)

			UpperCamelCase__				:					str					=					VQDiffusionPipeline(
			    vqvae=UpperCAmelCase_  ,					text_encoder=UpperCAmelCase_  ,					tokenizer=UpperCAmelCase_  ,					transformer=UpperCAmelCase_  ,					scheduler=UpperCAmelCase_  ,					learned_classifier_free_sampling_embeddings=UpperCAmelCase_  ,					)
			UpperCamelCase__				:					str					=					pipe.to(UpperCAmelCase_)
			pipe.set_progress_bar_config(disable=UpperCAmelCase_)

			UpperCamelCase__				:					List[Any]					=					'teddy bear playing in the pool'

			UpperCamelCase__				:					Union[str, Any]					=					torch.Generator(device=UpperCAmelCase_).manual_seed(0)
			UpperCamelCase__				:					Any					=					pipe([prompt]  ,					generator=UpperCAmelCase_  ,					num_inference_steps=2  ,					output_type='np')
			UpperCamelCase__				:					int					=					output.images

			UpperCamelCase__				:					List[Any]					=					torch.Generator(device=UpperCAmelCase_).manual_seed(0)
			UpperCamelCase__				:					Optional[Any]					=					pipe(
			    [prompt]  ,					generator=UpperCAmelCase_  ,					output_type='np'  ,					return_dict=UpperCAmelCase_  ,					num_inference_steps=2)[0]

			UpperCamelCase__				:					Union[str, Any]					=					image[0, -3:, -3:, -1]
			UpperCamelCase__				:					Dict					=					image_from_tuple[0, -3:, -3:, -1]

			assert image.shape == (1, 24, 24, 3)

			UpperCamelCase__				:					str					=					np.array([0.66_93, 0.60_75, 0.49_59, 0.57_01, 0.55_83, 0.43_33, 0.61_71, 0.56_84, 0.49_88])

			assert np.abs(image_slice.flatten() - expected_slice).max() < 2.0
			assert np.abs(image_from_tuple_slice.flatten() - expected_slice).max() < 1e-2







@slow
@require_torch_gpu
class       __lowercase					(unittest.TestCase    ):

	def      __UpperCamelCase							(      self   :						Any):
			# clean up the VRAM after each test
			super().tearDown()
			gc.collect()
			torch.cuda.empty_cache()





	def      __UpperCamelCase							(      self   :						List[Any]):
			UpperCamelCase__				:					Optional[Any]					=					load_numpy(
			    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
			    '/vq_diffusion/teddy_bear_pool_classifier_free_sampling.npy')

			UpperCamelCase__				:					List[Any]					=					VQDiffusionPipeline.from_pretrained('microsoft/vq-diffusion-ithq')
			UpperCamelCase__				:					Any					=					pipeline.to(UpperCAmelCase_)
			pipeline.set_progress_bar_config(disable=UpperCAmelCase_)

			# requires GPU generator for gumbel softmax
			# don't use GPU generator in tests though
			UpperCamelCase__				:					Optional[int]					=					torch.Generator(device=UpperCAmelCase_).manual_seed(0)
			UpperCamelCase__				:					int					=					pipeline(
			    'teddy bear playing in the pool'  ,					num_images_per_prompt=1  ,					generator=UpperCAmelCase_  ,					output_type='np'  ,					)

			UpperCamelCase__				:					int					=					output.images[0]

			assert image.shape == (256, 256, 3)
			assert np.abs(expected_image - image).max() < 2.0




