





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 				SCREAMING_SNAKE_CASE__							(	)   ->    Dict:
  _lowercase						    =			10
  _lowercase						    =			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'  ),
      }  )
  _lowercase						    =			datasets.Dataset.from_dict(
      {
          'tokens': [['foo'] * 5] * n,
          'labels': [[1] * 5] * n,
          'answers': [{'answer_start': [97], 'text': ['1976']}] * 10,
          'id': list(range(UpperCamelCase__  )  ),
      }      ,				features=UpperCamelCase__      ,				)
  return dataset

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:List[str]      ,				snake_case__	:Dict  )   ->    Dict:
  _lowercase						    =			str(tmp_path_factory.mktemp('data'  ) / 'file.arrow'  )
  dataset.map(cache_file_name=UpperCamelCase__  )
  return filename


# FILE_CONTENT + files


snake_case								=							"\\n    Text data.\n    Second line of data."

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:Dict  )   ->    Any:
  _lowercase						    =			tmp_path_factory.mktemp('data'  ) / 'file.txt'
  _lowercase						    =			FILE_CONTENT
  with open(UpperCamelCase__      ,				'w'  ) as f:
    f.write(UpperCamelCase__  )
  return filename

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

  _lowercase						    =			tmp_path_factory.mktemp('data'  ) / 'file.txt.bz2'
  _lowercase						    =			bytes(UpperCamelCase__      ,				'utf-8'  )
  with bza.open(UpperCamelCase__      ,				'wb'  ) as f:
    f.write(UpperCamelCase__  )
  return path

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:Tuple  )   ->    int:
  import gzip

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

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:Any  )   ->    Optional[int]:







  if datasets.config.LZ4_AVAILABLE:
    import lza.frame

    _lowercase						    =			tmp_path_factory.mktemp('data'  ) / 'file.txt.lz4'
    _lowercase						    =			bytes(UpperCamelCase__      ,				'utf-8'  )
    with lza.frame.open(UpperCamelCase__      ,				'wb'  ) as f:
      f.write(UpperCamelCase__  )
    return path

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:Optional[Any]      ,				snake_case__	:Union[str, Any]  )   ->    Any:







  if datasets.config.PY7ZR_AVAILABLE:
    import pyazr

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

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:Optional[Any]      ,				snake_case__	:Union[str, Any]  )   ->    Optional[int]:
  import tarfile

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

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:str  )   ->    List[Any]:
  import lzma

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

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:Optional[Any]      ,				snake_case__	:Optional[int]  )   ->    int:
  import zipfile

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

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:Any  )   ->    List[str]:







  if datasets.config.ZSTANDARD_AVAILABLE:
    import zstandard as zstd

    _lowercase						    =			tmp_path_factory.mktemp('data'  ) / 'file.txt.zst'
    _lowercase						    =			bytes(UpperCamelCase__      ,				'utf-8'  )
    with zstd.open(UpperCamelCase__      ,				'wb'  ) as f:
      f.write(UpperCamelCase__  )
    return path

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:Dict  )   ->    str:
  _lowercase						    =			tmp_path_factory.mktemp('data'  ) / 'file.xml'
  _lowercase						    =			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(UpperCamelCase__      ,				'w'  ) as f:
    f.write(UpperCamelCase__  )
  return filename


snake_case								=							[
    {"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},
]
snake_case								=							[
    {"col_1": "4", "col_2": 4, "col_3": 4.0},
    {"col_1": "5", "col_2": 5, "col_3": 5.0},
]
snake_case								=							{
    "col_1": ["0", "1", "2", "3"],
    "col_2": [0, 1, 2, 3],
    "col_3": [0.0, 1.0, 2.0, 3.0],
}

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

snake_case								=							[
    {"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 				SCREAMING_SNAKE_CASE__							(	)   ->    str:
  return DATA_DICT_OF_LISTS

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:Optional[Any]  )   ->    List[str]:
  _lowercase						    =			datasets.Dataset.from_dict(UpperCamelCase__  )
  _lowercase						    =			str(tmp_path_factory.mktemp('data'  ) / 'dataset.arrow'  )
  dataset.map(cache_file_name=UpperCamelCase__  )
  return path

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:Dict  )   ->    Any:
  _lowercase						    =			str(tmp_path_factory.mktemp('data'  ) / 'dataset.sqlite'  )
  with contextlib.closing(sqlitea.connect(UpperCamelCase__  )  ) as con:
    _lowercase						    =			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 				SCREAMING_SNAKE_CASE__							(	snake_case__	:List[Any]  )   ->    Optional[int]:
  _lowercase						    =			str(tmp_path_factory.mktemp('data'  ) / 'dataset.csv'  )
  with open(UpperCamelCase__      ,				'w'      ,				newline=''  ) as f:
    _lowercase						    =			csv.DictWriter(UpperCamelCase__      ,				fieldnames=['col_1', 'col_2', 'col_3']  )
    writer.writeheader()
    for item in DATA:
      writer.writerow(UpperCamelCase__  )
  return path

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

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:str      ,				snake_case__	:int  )   ->    str:
  import bza

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:Tuple      ,				snake_case__	:str      ,				snake_case__	:Dict      ,				snake_case__	:Any  )   ->    Optional[int]:
  _lowercase						    =			tmp_path_factory.mktemp('data'  ) / 'dataset_nested.jsonl.zip'
  with zipfile.ZipFile(UpperCamelCase__      ,				'w'  ) as f:
    f.write(UpperCamelCase__      ,				arcname=os.path.join('nested'      ,				os.path.basename(UpperCamelCase__  )  )  )
  return path

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

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

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

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

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

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

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

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

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:int      ,				snake_case__	:Any      ,				snake_case__	:Tuple  )   ->    int:
  _lowercase						    =			tmp_path_factory.mktemp('data'  ) / 'dataset.ext.zip'
  with zipfile.ZipFile(UpperCamelCase__      ,				'w'  ) as f:
    f.write(UpperCamelCase__      ,				arcname=os.path.basename('unsupported.ext'  )  )
    f.write(UpperCamelCase__      ,				arcname=os.path.basename('unsupported_2.ext'  )  )
  return path

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

@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	)   ->    Dict:
  return os.path.join('tests'      ,				'features'      ,				'data'      ,				'test_image_rgb.jpg'  )

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

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







@pytest.fixture(scope='session'  )
def 				SCREAMING_SNAKE_CASE__							(	snake_case__	:int  )   ->    Dict:
  _lowercase						    =			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



import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


_lowercase							:   Optional[int]						=						False






@skip_mps
class    _UpperCamelCase     (       __snake_case     ,		__snake_case     ,		__snake_case     ,		unittest.TestCase					):

				"""simple docstring"""







				lowerCAmelCase					=			StableDiffusionAttendAndExcitePipeline
				lowerCAmelCase					=			False
				lowerCAmelCase					=			TEXT_TO_IMAGE_PARAMS
				lowerCAmelCase					=			TEXT_TO_IMAGE_BATCH_PARAMS.union({'token_indices'}					)
				lowerCAmelCase					=			TEXT_TO_IMAGE_IMAGE_PARAMS
				lowerCAmelCase					=			TEXT_TO_IMAGE_IMAGE_PARAMS

				@classmethod
				def   _UpperCAmelCase       (    cls )						->					List[Any]:
					super().setUpClass()
					torch.use_deterministic_algorithms(a__ )

				@classmethod
				def   _UpperCAmelCase       (    cls )						->					Tuple:
					super().tearDownClass()
					torch.use_deterministic_algorithms(a__ )

				def   _UpperCAmelCase       (    self )						->					Dict:
					torch.manual_seed(0 )
					A             =					UNetaDConditionModel(
					    block_out_channels=(32, 64)					, layers_per_block=1					, sample_size=32					, in_channels=4					, out_channels=4					, down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")					, up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")					, cross_attention_dim=32					, attention_head_dim=(2, 4)					, use_linear_projection=a__					, )
					A             =					DDIMScheduler(
					    beta_start=0.0_00_85					, beta_end=0.0_12					, beta_schedule="""scaled_linear"""					, clip_sample=a__					, set_alpha_to_one=a__					, )
					torch.manual_seed(0 )
					A             =					AutoencoderKL(
					    block_out_channels=[32, 64]					, in_channels=3					, out_channels=3					, down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]					, up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]					, latent_channels=4					, sample_size=128					, )
					torch.manual_seed(0 )
					A             =					CLIPTextConfig(
					    bos_token_id=0					, eos_token_id=2					, hidden_size=32					, intermediate_size=37					, layer_norm_eps=1e-05					, num_attention_heads=4					, num_hidden_layers=5					, pad_token_id=1					, vocab_size=1000					, hidden_act="""gelu"""					, projection_dim=512					, )
					A             =					CLIPTextModel(a__ )
					A             =					CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""" )

					A             =					{
					    """unet""": unet,
					    """scheduler""": scheduler,
					    """vae""": vae,
					    """text_encoder""": text_encoder,
					    """tokenizer""": tokenizer,
					    """safety_checker""": None,
					    """feature_extractor""": None,
					}

					return components

				def   _UpperCAmelCase       (    self					, a__					, a__=0 )						->					Optional[Any]:
					if str(a__ ).startswith("""mps""" ):
						A             =					torch.manual_seed(a__ )
					else:
						A             =					torch.Generator(device=a__ ).manual_seed(a__ )
					A             =					A             =					{
					    """prompt""": """a cat and a frog""",
					    """token_indices""": [2, 5],
					    """generator""": generator,
					    """num_inference_steps""": 1,
					    """guidance_scale""": 6.0,
					    """output_type""": """numpy""",
					    """max_iter_to_alter""": 2,
					    """thresholds""": {0: 0.7},
					}
					return inputs

				def   _UpperCAmelCase       (    self )						->					Union[str, Any]:
					A             =					"""cpu"""

					A             =					self.get_dummy_components()
					A             =					self.pipeline_class(**a__ )
					pipe.to(a__ )
					pipe.set_progress_bar_config(disable=a__ )

					A             =					self.get_dummy_inputs(a__ )
					A             =					pipe(**a__ ).images
					A             =					image[0, -3:, -3:, -1]

					self.assertEqual(image.shape					, (1, 64, 64, 3) )
					A             =					np.array(
					    [0.63_90_53_64, 0.62_89_73_07, 0.48_59_90_17, 0.5_13_36_24, 0.5_55_00_48, 0.45_76_95_16, 0.50_32_69_73, 0.5_02_31_39, 0.45_38_44_96] )
					A             =					np.abs(image_slice.flatten() - expected_slice ).max()
					self.assertLessEqual(a__					, 1e-3 )

				def   _UpperCAmelCase       (    self )						->					List[Any]:
					super().test_cpu_offload_forward_pass(expected_max_diff=5e-4 )

				def   _UpperCAmelCase       (    self )						->					Dict:
					# NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
					self._test_inference_batch_consistent(batch_sizes=[1, 2] )

				def   _UpperCAmelCase       (    self )						->					Any:
					self._test_inference_batch_single_identical(batch_size=2					, expected_max_diff=7e-4 )

				def   _UpperCAmelCase       (    self )						->					Optional[Any]:
					super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3 )

				def   _UpperCAmelCase       (    self )						->					str:
					super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4 )

				def   _UpperCAmelCase       (    self )						->					int:
					super().test_save_load_local(expected_max_difference=5e-4 )

				def   _UpperCAmelCase       (    self )						->					Optional[Any]:
					super().test_save_load_optional_components(expected_max_difference=4e-4 )






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

				"""simple docstring"""







				@classmethod
				def   _UpperCAmelCase       (    cls )						->					Tuple:
					super().setUpClass()
					torch.use_deterministic_algorithms(a__ )

				@classmethod
				def   _UpperCAmelCase       (    cls )						->					Dict:
					super().tearDownClass()
					torch.use_deterministic_algorithms(a__ )

				def   _UpperCAmelCase       (    self )						->					Optional[Any]:
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()

				def   _UpperCAmelCase       (    self )						->					int:
					A             =					torch.manual_seed(51 )

					A             =					StableDiffusionAttendAndExcitePipeline.from_pretrained(
					    """CompVis/stable-diffusion-v1-4"""					, safety_checker=a__					, torch_dtype=torch.floataa )
					pipe.to("""cuda""" )

					A             =					"""a painting of an elephant with glasses"""
					A             =					[5, 7]

					A             =					pipe(
					    prompt=a__					, token_indices=a__					, guidance_scale=7.5					, generator=a__					, num_inference_steps=5					, max_iter_to_alter=5					, output_type="""numpy"""					, ).images[0]

					A             =					load_numpy(
					    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy""" )
					assert np.abs((expected_image - image).max() ) < 5e-1



