





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    __lowercase   (      ) ->   str:


      '''simple docstring'''


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



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Optional[Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """file.arrow"""	)
      dataset.map(cache_file_name=_SCREAMING_SNAKE_CASE	)
      return filename


# FILE_CONTENT + files


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



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   str:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """file.txt"""
      SCREAMING_SNAKE_CASE		=						FILE_CONTENT
      with open(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.write(_SCREAMING_SNAKE_CASE	)
      return filename



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   Optional[Any]:


      '''simple docstring'''


      import bza

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



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   List[str]:


      '''simple docstring'''


      import gzip

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



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   Union[str, Any]:


      '''simple docstring'''


      if datasets.config.LZ4_AVAILABLE:
            import lza.frame

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



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Union[str, Any]:


      '''simple docstring'''


      if datasets.config.PY7ZR_AVAILABLE:
            import pyazr

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



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Any:


      '''simple docstring'''


      import tarfile

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



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   List[str]:


      '''simple docstring'''


      import lzma

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



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Tuple:


      '''simple docstring'''


      import zipfile

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



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   int:


      '''simple docstring'''


      if datasets.config.ZSTANDARD_AVAILABLE:
            import zstandard as zstd

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



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   List[Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """file.xml"""
      SCREAMING_SNAKE_CASE		=						textwrap.dedent(
          """\
    <?xml version=\"1.0\" encoding=\"UTF-8\" ?>
    <tmx version=\"1.4\">
      <header segtype=\"sentence\" srclang=\"ca\" />
      <body>
        <tu>
          <tuv xml:lang=\"ca\"><seg>Contingut 1</seg></tuv>
          <tuv xml:lang=\"en\"><seg>Content 1</seg></tuv>
        </tu>
        <tu>
          <tuv xml:lang=\"ca\"><seg>Contingut 2</seg></tuv>
          <tuv xml:lang=\"en\"><seg>Content 2</seg></tuv>
        </tu>
        <tu>
          <tuv xml:lang=\"ca\"><seg>Contingut 3</seg></tuv>
          <tuv xml:lang=\"en\"><seg>Content 3</seg></tuv>
        </tu>
        <tu>
          <tuv xml:lang=\"ca\"><seg>Contingut 4</seg></tuv>
          <tuv xml:lang=\"en\"><seg>Content 4</seg></tuv>
        </tu>
        <tu>
          <tuv xml:lang=\"ca\"><seg>Contingut 5</seg></tuv>
          <tuv xml:lang=\"en\"><seg>Content 5</seg></tuv>
        </tu>
      </body>
    </tmx>"""	)
      with open(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.write(_SCREAMING_SNAKE_CASE	)
      return filename


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

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

SCREAMING_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    __lowercase   (      ) ->   Optional[Any]:


      '''simple docstring'''


      return DATA_DICT_OF_LISTS



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   List[Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						datasets.Dataset.from_dict(_SCREAMING_SNAKE_CASE	)
      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset.arrow"""	)
      dataset.map(cache_file_name=_SCREAMING_SNAKE_CASE	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   str:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset.sqlite"""	)
      with contextlib.closing(sqlitea.connect(_SCREAMING_SNAKE_CASE	)	) as con:
            SCREAMING_SNAKE_CASE		=						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    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   Optional[int]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset.csv"""	)
      with open(_SCREAMING_SNAKE_CASE ,						"""w""" ,						newline=""""""	) as f:
            SCREAMING_SNAKE_CASE		=						csv.DictWriter(_SCREAMING_SNAKE_CASE ,						fieldnames=["""col_1""", """col_2""", """col_3"""]	)
            writer.writeheader()
            for item in DATA:
                  writer.writerow(_SCREAMING_SNAKE_CASE	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   List[Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset2.csv"""	)
      with open(_SCREAMING_SNAKE_CASE ,						"""w""" ,						newline=""""""	) as f:
            SCREAMING_SNAKE_CASE		=						csv.DictWriter(_SCREAMING_SNAKE_CASE ,						fieldnames=["""col_1""", """col_2""", """col_3"""]	)
            writer.writeheader()
            for item in DATA:
                  writer.writerow(_SCREAMING_SNAKE_CASE	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Tuple:


      '''simple docstring'''


      import bza

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



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Dict:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """dataset.csv.zip"""
      with zipfile.ZipFile(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename(_SCREAMING_SNAKE_CASE	)	)
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename(_SCREAMING_SNAKE_CASE	)	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Dict:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """dataset.csv.zip"""
      with zipfile.ZipFile(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename(csv_path.replace(""".csv""" ,						""".CSV"""	)	)	)
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename(csva_path.replace(""".csv""" ,						""".CSV"""	)	)	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Any:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """dataset_with_dir.csv.zip"""
      with zipfile.ZipFile(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.join("""main_dir""" ,						os.path.basename(_SCREAMING_SNAKE_CASE	)	)	)
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.join("""main_dir""" ,						os.path.basename(_SCREAMING_SNAKE_CASE	)	)	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   Union[str, Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset.parquet"""	)
      SCREAMING_SNAKE_CASE		=						pa.schema(
          {
              """col_1""": pa.string(),
              """col_2""": pa.intaa(),
              """col_3""": pa.floataa(),
          }	)
      with open(_SCREAMING_SNAKE_CASE ,						"""wb"""	) as f:
            SCREAMING_SNAKE_CASE		=						pq.ParquetWriter(_SCREAMING_SNAKE_CASE ,						schema=_SCREAMING_SNAKE_CASE	)
            SCREAMING_SNAKE_CASE		=						pa.Table.from_pydict({k: [DATA[i][k] for i in range(len(_SCREAMING_SNAKE_CASE	)	)] for k in DATA[0]} ,						schema=_SCREAMING_SNAKE_CASE	)
            writer.write_table(_SCREAMING_SNAKE_CASE	)
            writer.close()
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   int:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset.json"""	)
      SCREAMING_SNAKE_CASE		=						{"""data""": DATA}
      with open(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            json.dump(_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   List[str]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset.json"""	)
      SCREAMING_SNAKE_CASE		=						{"""data""": DATA_DICT_OF_LISTS}
      with open(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            json.dump(_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   List[Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset.jsonl"""	)
      with open(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            for item in DATA:
                  f.write(json.dumps(_SCREAMING_SNAKE_CASE	) + """\n"""	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   int:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset2.jsonl"""	)
      with open(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            for item in DATA:
                  f.write(json.dumps(_SCREAMING_SNAKE_CASE	) + """\n"""	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   str:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset_312.jsonl"""	)
      with open(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            for item in DATA_312:
                  f.write(json.dumps(_SCREAMING_SNAKE_CASE	) + """\n"""	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   Union[str, Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset-str.jsonl"""	)
      with open(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            for item in DATA_STR:
                  f.write(json.dumps(_SCREAMING_SNAKE_CASE	) + """\n"""	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   List[str]:


      '''simple docstring'''


      import gzip

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



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   str:


      '''simple docstring'''


      import gzip

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



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Union[str, Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """dataset.jsonl.zip"""
      with zipfile.ZipFile(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename(_SCREAMING_SNAKE_CASE	)	)
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename(_SCREAMING_SNAKE_CASE	)	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Union[str, Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """dataset_nested.jsonl.zip"""
      with zipfile.ZipFile(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.join("""nested""" ,						os.path.basename(_SCREAMING_SNAKE_CASE	)	)	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Union[str, Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """dataset_with_dir.jsonl.zip"""
      with zipfile.ZipFile(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.join("""main_dir""" ,						os.path.basename(_SCREAMING_SNAKE_CASE	)	)	)
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.join("""main_dir""" ,						os.path.basename(_SCREAMING_SNAKE_CASE	)	)	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   List[Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """dataset.jsonl.tar"""
      with tarfile.TarFile(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.add(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename(_SCREAMING_SNAKE_CASE	)	)
            f.add(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename(_SCREAMING_SNAKE_CASE	)	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   int:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """dataset_nested.jsonl.tar"""
      with tarfile.TarFile(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.add(_SCREAMING_SNAKE_CASE ,						arcname=os.path.join("""nested""" ,						os.path.basename(_SCREAMING_SNAKE_CASE	)	)	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   Tuple:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						["""0""", """1""", """2""", """3"""]
      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset.txt"""	)
      with open(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            for item in data:
                  f.write(item + """\n"""	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   str:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						["""0""", """1""", """2""", """3"""]
      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset2.txt"""	)
      with open(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            for item in data:
                  f.write(item + """\n"""	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   Union[str, Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						["""0""", """1""", """2""", """3"""]
      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """dataset.abc"""
      with open(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            for item in data:
                  f.write(item + """\n"""	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Any:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """dataset.text.zip"""
      with zipfile.ZipFile(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename(_SCREAMING_SNAKE_CASE	)	)
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename(_SCREAMING_SNAKE_CASE	)	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Optional[Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """dataset_with_dir.text.zip"""
      with zipfile.ZipFile(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.join("""main_dir""" ,						os.path.basename(_SCREAMING_SNAKE_CASE	)	)	)
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.join("""main_dir""" ,						os.path.basename(_SCREAMING_SNAKE_CASE	)	)	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   int:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """dataset.ext.zip"""
      with zipfile.ZipFile(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename("""unsupported.ext"""	)	)
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename("""unsupported_2.ext"""	)	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   str:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						"""\n""".join(["""First""", """Second\u2029with Unicode new line""", """Third"""]	)
      SCREAMING_SNAKE_CASE		=						str(tmp_path_factory.mktemp("""data"""	) / """dataset_with_unicode_new_lines.txt"""	)
      with open(_SCREAMING_SNAKE_CASE ,						"""w""" ,						encoding="""utf-8"""	) as f:
            f.write(_SCREAMING_SNAKE_CASE	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      ) ->   Tuple:


      '''simple docstring'''


      return os.path.join("""tests""" ,						"""features""" ,						"""data""" ,						"""test_image_rgb.jpg"""	)



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      ) ->   List[str]:


      '''simple docstring'''


      return os.path.join("""tests""" ,						"""features""" ,						"""data""" ,						"""test_audio_44100.wav"""	)



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE ,						_SCREAMING_SNAKE_CASE	) ->   Optional[Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						tmp_path_factory.mktemp("""data"""	) / """dataset.img.zip"""
      with zipfile.ZipFile(_SCREAMING_SNAKE_CASE ,						"""w"""	) as f:
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename(_SCREAMING_SNAKE_CASE	)	)
            f.write(_SCREAMING_SNAKE_CASE ,						arcname=os.path.basename(_SCREAMING_SNAKE_CASE	).replace(""".jpg""" ,						"""2.jpg"""	)	)
      return path



@pytest.fixture(scope="""session"""	)
def    __lowercase   (      _SCREAMING_SNAKE_CASE	) ->   Any:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						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 unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    TextToVideoSDPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, load_numpy, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()

@skip_mps
class 		UpperCamelCase__ (      lowerCAmelCase_							,      unittest.TestCase   ):



     '''simple docstring'''

     __snake_case    :	List[str]			=			TextToVideoSDPipeline
     __snake_case    :	int			=			TEXT_TO_IMAGE_PARAMS
     __snake_case    :	Tuple			=			TEXT_TO_IMAGE_BATCH_PARAMS
     # No `output_type`.
     __snake_case    :	Dict			=			frozenset(
         [
             "num_inference_steps",
             "generator",
             "latents",
             "return_dict",
             "callback",
             "callback_steps",
         ]   )






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




           '''simple docstring'''


           torch.manual_seed(0       )
           SCREAMING_SNAKE_CASE		=						UNetaDConditionModel(
               block_out_channels=(32, 64, 64, 64)						,layers_per_block=2						,sample_size=32						,in_channels=4						,out_channels=4						,down_block_types=("""CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """DownBlock3D""")						,up_block_types=("""UpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""")						,cross_attention_dim=32						,attention_head_dim=4						,)
           SCREAMING_SNAKE_CASE		=						DDIMScheduler(
               beta_start=0.00085						,beta_end=0.012						,beta_schedule="""scaled_linear"""						,clip_sample=lowerCamelCase__						,set_alpha_to_one=lowerCamelCase__						,)
           torch.manual_seed(0       )
           SCREAMING_SNAKE_CASE		=						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       )
           SCREAMING_SNAKE_CASE		=						CLIPTextConfig(
               bos_token_id=0						,eos_token_id=2						,hidden_size=32						,intermediate_size=37						,layer_norm_eps=1e-0_5						,num_attention_heads=4						,num_hidden_layers=5						,pad_token_id=1						,vocab_size=1000						,hidden_act="""gelu"""						,projection_dim=512						,)
           SCREAMING_SNAKE_CASE		=						CLIPTextModel(lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""       )

           SCREAMING_SNAKE_CASE		=						{
               """unet""": unet,
               """scheduler""": scheduler,
               """vae""": vae,
               """text_encoder""": text_encoder,
               """tokenizer""": tokenizer,
           }
           return components






     def 				SCREAMING_SNAKE_CASE__  (							self						:    int						,lowerCamelCase__						:    Optional[Any]						,lowerCamelCase__						:    int=0       )     ->		List[Any]:




           '''simple docstring'''


           if str(lowerCamelCase__       ).startswith("""mps"""       ):
                 SCREAMING_SNAKE_CASE		=						torch.manual_seed(lowerCamelCase__       )
           else:
                 SCREAMING_SNAKE_CASE		=						torch.Generator(device=lowerCamelCase__       ).manual_seed(lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						{
               """prompt""": """A painting of a squirrel eating a burger""",
               """generator""": generator,
               """num_inference_steps""": 2,
               """guidance_scale""": 6.0,
               """output_type""": """pt""",
           }
           return inputs






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[Any]       )     ->		List[str]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						"""cpu"""  # ensure determinism for the device-dependent torch.Generator
           SCREAMING_SNAKE_CASE		=						self.get_dummy_components()
           SCREAMING_SNAKE_CASE		=						TextToVideoSDPipeline(**lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						sd_pipe.to(lowerCamelCase__       )
           sd_pipe.set_progress_bar_config(disable=lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						self.get_dummy_inputs(lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						"""np"""
           SCREAMING_SNAKE_CASE		=						sd_pipe(**lowerCamelCase__       ).frames
           SCREAMING_SNAKE_CASE		=						frames[0][-3:, -3:, -1]

           assert frames[0].shape == (64, 64, 3)
           SCREAMING_SNAKE_CASE		=						np.array([158.0, 160.0, 153.0, 125.0, 100.0, 121.0, 111.0, 93.0, 113.0]       )

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






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Dict       )     ->		int:




           '''simple docstring'''


           self._test_attention_slicing_forward_pass(test_mean_pixel_difference=lowerCamelCase__						,expected_max_diff=3e-3       )






     @unittest.skipIf(
         torch_device != """cuda""" or not is_xformers_available()						,reason="""XFormers attention is only available with CUDA and `xformers` installed"""						,)
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Tuple       )     ->		Any:




           '''simple docstring'''


           self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=lowerCamelCase__						,expected_max_diff=1e-2       )






     @unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""       )
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[Any]       )     ->		List[Any]:




           '''simple docstring'''


           pass






     @unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""       )
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[Any]       )     ->		Any:




           '''simple docstring'''


           pass






     @unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline."""       )
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[int]       )     ->		Tuple:




           '''simple docstring'''


           pass






     def 				SCREAMING_SNAKE_CASE__  (							self						:    List[str]       )     ->		Optional[Any]:




           '''simple docstring'''


           return super().test_progress_bar()

@slow
@skip_mps
class 		UpperCamelCase__ (      unittest.TestCase   ):



     '''simple docstring'''

     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		Optional[int]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						load_numpy(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy"""       )

           SCREAMING_SNAKE_CASE		=						TextToVideoSDPipeline.from_pretrained("""damo-vilab/text-to-video-ms-1.7b"""       )
           SCREAMING_SNAKE_CASE		=						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config       )
           SCREAMING_SNAKE_CASE		=						pipe.to("""cuda"""       )

           SCREAMING_SNAKE_CASE		=						"""Spiderman is surfing"""
           SCREAMING_SNAKE_CASE		=						torch.Generator(device="""cpu"""       ).manual_seed(0       )

           SCREAMING_SNAKE_CASE		=						pipe(lowerCamelCase__						,generator=lowerCamelCase__						,num_inference_steps=25						,output_type="""pt"""       ).frames
           SCREAMING_SNAKE_CASE		=						video_frames.cpu().numpy()

           assert np.abs(expected_video - video       ).mean() < 5e-2






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		Any:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						load_numpy(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy"""       )

           SCREAMING_SNAKE_CASE		=						TextToVideoSDPipeline.from_pretrained("""damo-vilab/text-to-video-ms-1.7b"""       )
           SCREAMING_SNAKE_CASE		=						pipe.to("""cuda"""       )

           SCREAMING_SNAKE_CASE		=						"""Spiderman is surfing"""
           SCREAMING_SNAKE_CASE		=						torch.Generator(device="""cpu"""       ).manual_seed(0       )

           SCREAMING_SNAKE_CASE		=						pipe(lowerCamelCase__						,generator=lowerCamelCase__						,num_inference_steps=2						,output_type="""pt"""       ).frames
           SCREAMING_SNAKE_CASE		=						video_frames.cpu().numpy()

           assert np.abs(expected_video - video       ).mean() < 5e-2


